271 lines
8.2 KiB
Python
271 lines
8.2 KiB
Python
"""API de autoayuno: catálogo compartido de alimentos (alimentado de Open Food
|
|
Facts bajo demanda) y recetas publicables. Sin cuentas: publicar es anónimo con
|
|
un deviceToken."""
|
|
import uuid
|
|
|
|
from fastapi import Depends, FastAPI, HTTPException
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from pydantic import BaseModel
|
|
from sqlalchemy import func, text
|
|
from sqlalchemy.orm import Session
|
|
|
|
|
|
def like_pattern(q: str) -> str:
|
|
"""Escapa comodines de LIKE (%, _, \\) para tratar la query como texto."""
|
|
q = q.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
|
|
return f"%{q.lower()}%"
|
|
|
|
|
|
def clamp_limit(limit: int) -> int:
|
|
return max(1, min(limit, 50))
|
|
|
|
import off
|
|
from db import Base, engine, get_db
|
|
from models import Food, Recipe
|
|
|
|
|
|
# Columnas añadidas a `recipes` después de la v1 (id/name/servings/items). En
|
|
# Postgres la tabla ya existe y `create_all` NO la altera, así que las añadimos a
|
|
# mano (idempotente). En SQLite fresco no hace falta: `create_all` ya las trae.
|
|
_RECIPE_NEW_COLUMNS = {
|
|
"category": "VARCHAR DEFAULT 'otros'",
|
|
"prep_minutes": "INTEGER",
|
|
"steps": "JSON",
|
|
"notes": "VARCHAR",
|
|
"author_id": "VARCHAR",
|
|
"author_name": "VARCHAR",
|
|
"parent_id": "VARCHAR",
|
|
"parent_author_name": "VARCHAR",
|
|
"root_id": "VARCHAR",
|
|
"root_author_name": "VARCHAR",
|
|
"visibility": "VARCHAR DEFAULT 'public'",
|
|
"updated_at": "TIMESTAMP",
|
|
}
|
|
|
|
|
|
def _migrate_recipes():
|
|
"""Añade en Postgres las columnas nuevas de `recipes` si faltan (best-effort)."""
|
|
if not engine.url.get_backend_name().startswith("postgres"):
|
|
return
|
|
try:
|
|
with engine.begin() as conn:
|
|
for name, ddl in _RECIPE_NEW_COLUMNS.items():
|
|
try:
|
|
conn.execute(text(
|
|
f"ALTER TABLE recipes ADD COLUMN IF NOT EXISTS {name} {ddl}"))
|
|
except Exception:
|
|
pass
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def init_db(retries: int = 10):
|
|
"""Crea las tablas, reintentando por si la BD tarda en levantar. No tumba
|
|
el arranque si falla: /health seguirá respondiendo."""
|
|
import time
|
|
for _ in range(retries):
|
|
try:
|
|
Base.metadata.create_all(bind=engine)
|
|
_migrate_recipes()
|
|
return True
|
|
except Exception:
|
|
time.sleep(2)
|
|
return False
|
|
|
|
|
|
# En import: un intento best-effort (crea tablas ya para tests/SQLite).
|
|
init_db(retries=1)
|
|
|
|
app = FastAPI(title="autoayuno-api")
|
|
|
|
|
|
@app.on_event("startup")
|
|
def _startup():
|
|
init_db()
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
|
|
@app.get("/health")
|
|
def health():
|
|
return {"status": "ok"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Alimentos
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class PortionIn(BaseModel):
|
|
name: str
|
|
grams: float
|
|
|
|
|
|
class FoodIn(BaseModel):
|
|
id: str | None = None
|
|
name: str
|
|
kcalPer100g: float
|
|
zone: str = "otros"
|
|
portions: list[PortionIn] = [] # validado: cada ración necesita name+grams
|
|
barcode: str | None = None
|
|
deviceToken: str | None = None
|
|
|
|
|
|
@app.get("/foods")
|
|
def search_foods(q: str = "", limit: int = 20, db: Session = Depends(get_db)):
|
|
q = (q or "").strip()
|
|
limit = clamp_limit(limit)
|
|
if len(q) < 2:
|
|
return []
|
|
local = (
|
|
db.query(Food)
|
|
.filter(func.lower(Food.name).like(like_pattern(q), escape="\\"))
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
results = [f.to_app() for f in local]
|
|
|
|
# Si hay pocos, enriquecemos con Open Food Facts y guardamos.
|
|
if len(local) < 5:
|
|
items = off.search(q)
|
|
# Dedupe en una sola query (evita N+1 contra Postgres).
|
|
off_ids = [i["off_id"] for i in items if i.get("off_id")]
|
|
existing = set()
|
|
if off_ids:
|
|
existing = {
|
|
r[0]
|
|
for r in db.query(Food.off_id).filter(Food.off_id.in_(off_ids)).all()
|
|
}
|
|
seen_names = {f.name.lower() for f in local}
|
|
for item in items:
|
|
if item.get("off_id") and item["off_id"] in existing:
|
|
continue
|
|
if item["name"].lower() in seen_names:
|
|
continue
|
|
food = Food(
|
|
id=str(uuid.uuid4()),
|
|
name=item["name"],
|
|
kcal_100g=item["kcal_100g"],
|
|
zone="otros",
|
|
portions=[],
|
|
barcode=item.get("barcode"),
|
|
source="off",
|
|
off_id=item.get("off_id"),
|
|
)
|
|
db.add(food)
|
|
seen_names.add(item["name"].lower())
|
|
if item.get("off_id"):
|
|
existing.add(item["off_id"])
|
|
results.append(food.to_app())
|
|
db.commit()
|
|
return results[:limit]
|
|
|
|
|
|
@app.post("/foods")
|
|
def add_food(body: FoodIn, db: Session = Depends(get_db)):
|
|
portions = [p.model_dump() for p in body.portions]
|
|
# Upsert por id del cliente: republicar actualiza en vez de duplicar.
|
|
food = db.get(Food, body.id) if body.id else None
|
|
if food is None:
|
|
food = Food(id=body.id or str(uuid.uuid4()), source="user")
|
|
db.add(food)
|
|
food.name = body.name.strip()[:120]
|
|
food.kcal_100g = body.kcalPer100g
|
|
food.zone = body.zone
|
|
food.portions = portions
|
|
food.barcode = body.barcode
|
|
food.author_token = body.deviceToken
|
|
db.commit()
|
|
return food.to_app()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Recetas
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class FoodSnapshotIn(BaseModel):
|
|
name: str
|
|
kcalPer100g: float
|
|
zone: str = "otros"
|
|
portions: list[PortionIn] = []
|
|
|
|
|
|
class RecipeItemIn(BaseModel):
|
|
food: FoodSnapshotIn
|
|
grams: float
|
|
portionLabel: str | None = None
|
|
|
|
|
|
class RecipeIn(BaseModel):
|
|
id: str | None = None
|
|
name: str
|
|
servings: int = 1
|
|
items: list[RecipeItemIn] = [] # validado: cada item necesita food+grams
|
|
category: str = "otros"
|
|
prepMinutes: int | None = None
|
|
steps: list[str] = []
|
|
notes: str | None = None
|
|
authorId: str | None = None
|
|
authorName: str | None = None
|
|
parentId: str | None = None
|
|
parentAuthorName: str | None = None
|
|
rootId: str | None = None
|
|
rootAuthorName: str | None = None
|
|
visibility: str = "public"
|
|
deviceToken: str | None = None
|
|
|
|
|
|
@app.get("/recipes")
|
|
def search_recipes(q: str = "", limit: int = 20, db: Session = Depends(get_db)):
|
|
# Buscar solo devuelve recetas PÚBLICAS (las privadas/no listadas no salen).
|
|
query = db.query(Recipe).filter(Recipe.visibility == "public")
|
|
q = (q or "").strip()
|
|
if q:
|
|
query = query.filter(func.lower(Recipe.name).like(like_pattern(q), escape="\\"))
|
|
return [r.to_app() for r in query.limit(clamp_limit(limit)).all()]
|
|
|
|
|
|
@app.get("/recipes/{recipe_id}")
|
|
def get_recipe(recipe_id: str, db: Session = Depends(get_db)):
|
|
r = db.query(Recipe).filter(Recipe.id == recipe_id).first()
|
|
if r is None:
|
|
raise HTTPException(status_code=404, detail="receta no encontrada")
|
|
return r.to_app()
|
|
|
|
|
|
@app.get("/cookbook/{author_id}")
|
|
def cookbook(author_id: str, db: Session = Depends(get_db)):
|
|
"""Todas las recetas de un autor (para respaldo/restaurar; usado en B3)."""
|
|
rows = db.query(Recipe).filter(Recipe.author_id == author_id).all()
|
|
return [r.to_app() for r in rows]
|
|
|
|
|
|
@app.post("/recipes")
|
|
def add_recipe(body: RecipeIn, db: Session = Depends(get_db)):
|
|
items = [i.model_dump() for i in body.items]
|
|
r = db.get(Recipe, body.id) if body.id else None
|
|
if r is None:
|
|
r = Recipe(id=body.id or str(uuid.uuid4()))
|
|
db.add(r)
|
|
r.name = body.name.strip()[:140]
|
|
r.servings = body.servings
|
|
r.items = items
|
|
r.category = (body.category or "otros")[:20]
|
|
r.prep_minutes = body.prepMinutes
|
|
r.steps = [str(s)[:500] for s in (body.steps or [])][:60]
|
|
r.notes = (body.notes or None)
|
|
r.author_id = body.authorId
|
|
r.author_name = (body.authorName or None)
|
|
r.parent_id = body.parentId
|
|
r.parent_author_name = body.parentAuthorName
|
|
r.root_id = body.rootId
|
|
r.root_author_name = body.rootAuthorName
|
|
r.visibility = body.visibility if body.visibility in (
|
|
"private", "unlisted", "public") else "public"
|
|
r.author_token = body.deviceToken
|
|
db.commit()
|
|
return r.to_app()
|