added fitness
This commit is contained in:
@@ -0,0 +1,62 @@
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
|
||||
from app.config import get_settings
|
||||
|
||||
|
||||
class OpenFoodFactsClient:
|
||||
def __init__(self) -> None:
|
||||
settings = get_settings()
|
||||
self.base_url = settings.openfoodfacts_base_url.rstrip("/")
|
||||
|
||||
def search(self, query: str, limit: int = 5) -> list[dict[str, Any]]:
|
||||
with httpx.Client(timeout=20.0) as client:
|
||||
response = client.get(
|
||||
f"{self.base_url}/cgi/search.pl",
|
||||
params={
|
||||
"search_terms": query,
|
||||
"search_simple": 1,
|
||||
"action": "process",
|
||||
"json": 1,
|
||||
"page_size": limit,
|
||||
"lc": "ru",
|
||||
},
|
||||
)
|
||||
response.raise_for_status()
|
||||
products = response.json().get("products") or []
|
||||
|
||||
out: list[dict[str, Any]] = []
|
||||
for p in products[:limit]:
|
||||
nutriments = p.get("nutriments") or {}
|
||||
out.append(
|
||||
{
|
||||
"name": p.get("product_name") or p.get("product_name_ru") or query,
|
||||
"brand": p.get("brands", ""),
|
||||
"barcode": p.get("code"),
|
||||
"calories_per_100g": nutriments.get("energy-kcal_100g"),
|
||||
"protein_g_per_100g": nutriments.get("proteins_100g"),
|
||||
"fat_g_per_100g": nutriments.get("fat_100g"),
|
||||
"carbs_g_per_100g": nutriments.get("carbohydrates_100g"),
|
||||
}
|
||||
)
|
||||
return out
|
||||
|
||||
def get_by_barcode(self, barcode: str) -> dict[str, Any] | None:
|
||||
with httpx.Client(timeout=20.0) as client:
|
||||
response = client.get(f"{self.base_url}/api/v2/product/{barcode}.json")
|
||||
if response.status_code == 404:
|
||||
return None
|
||||
response.raise_for_status()
|
||||
product = response.json().get("product")
|
||||
if not product:
|
||||
return None
|
||||
nutriments = product.get("nutriments") or {}
|
||||
return {
|
||||
"name": product.get("product_name") or product.get("product_name_ru"),
|
||||
"barcode": barcode,
|
||||
"calories_per_100g": nutriments.get("energy-kcal_100g"),
|
||||
"protein_g_per_100g": nutriments.get("proteins_100g"),
|
||||
"fat_g_per_100g": nutriments.get("fat_100g"),
|
||||
"carbs_g_per_100g": nutriments.get("carbohydrates_100g"),
|
||||
}
|
||||
@@ -0,0 +1,48 @@
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
|
||||
from app.config import get_settings
|
||||
|
||||
|
||||
class WgerClient:
|
||||
def __init__(self) -> None:
|
||||
settings = get_settings()
|
||||
self.base_url = settings.wger_base_url.rstrip("/")
|
||||
|
||||
def search_exercises(self, query: str, limit: int = 8) -> list[dict[str, Any]]:
|
||||
with httpx.Client(timeout=20.0) as client:
|
||||
response = client.get(
|
||||
f"{self.base_url}/exercise/search/",
|
||||
params={"term": query, "language": "ru"},
|
||||
)
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
sug = data.get("suggestions", data) if isinstance(data, dict) else []
|
||||
if isinstance(sug, dict):
|
||||
results = sug.get("results", [])
|
||||
elif isinstance(sug, list):
|
||||
results = sug
|
||||
else:
|
||||
results = []
|
||||
|
||||
out: list[dict[str, Any]] = []
|
||||
for item in results[:limit]:
|
||||
if isinstance(item, dict):
|
||||
name = item.get("value") or item.get("name") or str(item)
|
||||
out.append({"name": name, "data": item})
|
||||
elif isinstance(item, str):
|
||||
out.append({"name": item})
|
||||
if out:
|
||||
return out
|
||||
|
||||
response2 = client.get(
|
||||
f"{self.base_url}/exerciseinfo/",
|
||||
params={"language": 2, "limit": limit},
|
||||
)
|
||||
response2.raise_for_status()
|
||||
for item in (response2.json().get("results") or [])[:limit]:
|
||||
name = item.get("name") or f"#{item.get('id')}"
|
||||
if query.lower() in name.lower():
|
||||
out.append({"id": item.get("id"), "name": name, "category": item.get("category")})
|
||||
return out[:limit]
|
||||
Reference in New Issue
Block a user