from fastapi import FastAPI, HTTPException, BackgroundTasks from pydantic import BaseModel from typing import Optional, Any import aiohttp import os import requests from bring_api import Bring, BringItemOperation from bring_api.exceptions import BringAuthException, BringException, BringRequestException app = FastAPI(title="Bring! API", version="1.0.0") # --------------------------------------------------------------------------- # Configuration — override via environment variables # --------------------------------------------------------------------------- BRING_EMAIL = os.environ.get("BRING_EMAIL", "your@email.it") BRING_PASSWORD = os.environ.get("BRING_PASSWORD", "yourpassword") # --------------------------------------------------------------------------- # Session cache — one authenticated session reused across requests # --------------------------------------------------------------------------- _session_cache: dict = { "api": None, "session": None, } async def _get_api() -> Bring: """Return a cached, authenticated Bring instance, logging in if needed.""" if _session_cache["api"] is not None: return _session_cache["api"] session = aiohttp.ClientSession() api = Bring(session, BRING_EMAIL, BRING_PASSWORD) try: await api.login() except BringAuthException as e: await session.close() raise HTTPException(status_code=401, detail=f"Login failed: {e}") except BringException as e: await session.close() raise HTTPException(status_code=503, detail=f"Login failed: {e}") _session_cache.update({"api": api, "session": session}) return api async def _reset_session(): """Tear down the cached session so the next call triggers a fresh login.""" if _session_cache["session"]: await _session_cache["session"].close() _session_cache.update({"api": None, "session": None}) # --------------------------------------------------------------------------- # Webhook helper (same pattern as Cookidoo API) # --------------------------------------------------------------------------- def _send_webhook(webhook_url: str, payload: dict): try: requests.post(webhook_url, json=payload, timeout=10, headers={"Content-Type": "application/json"}) except Exception as e: print(f"Webhook error: {e}") # --------------------------------------------------------------------------- # Exception handler helper # --------------------------------------------------------------------------- async def _handle_bring_exception(e: BringException): msg = str(e) if isinstance(e, BringAuthException): await _reset_session() raise HTTPException(status_code=401, detail=f"Session expired — call /auth/refresh. Detail: {msg}") if isinstance(e, BringRequestException): if "authorization" in msg.lower() or "unauthorized" in msg.lower(): await _reset_session() raise HTTPException(status_code=401, detail=f"Session expired — call /auth/refresh. Detail: {msg}") raise HTTPException(status_code=502, detail=f"Bring request failed: {msg}") raise HTTPException(status_code=500, detail=msg) # --------------------------------------------------------------------------- # Pydantic models # --------------------------------------------------------------------------- class AddItemRequest(BaseModel): item_id: str specification: str = "" uuid: Optional[str] = None webhook_url: Optional[str] = None reference_id: Optional[str] = None class BatchAddRequest(BaseModel): items: list[AddItemRequest] webhook_url: Optional[str] = None reference_id: Optional[str] = None class ItemActionRequest(BaseModel): item_id: str uuid: Optional[str] = None webhook_url: Optional[str] = None reference_id: Optional[str] = None class UpdateItemRequest(BaseModel): item_id: str specification: str = "" webhook_url: Optional[str] = None reference_id: Optional[str] = None class NotifyRequest(BaseModel): notification_type: int = 1 webhook_url: Optional[str] = None reference_id: Optional[str] = None # --------------------------------------------------------------------------- # Routes # --------------------------------------------------------------------------- @app.get("/") def read_root(): return { "message": "Bring! API is running", "endpoints": { # Auth "POST /auth/refresh": "Forza un nuovo login e resetta la sessione", # Lists "GET /lists": "Tutte le liste della spesa", "GET /lists/{list_uuid}": "Articoli di una lista (attivi + recenti)", # Items "POST /lists/{list_uuid}/items": "Aggiungi un articolo", "POST /lists/{list_uuid}/items/batch": "Aggiungi più articoli", "POST /lists/{list_uuid}/items/{uuid}/update": "Aggiorna specification/quantità di un articolo", "POST /lists/{list_uuid}/items/{uuid}/complete": "Segna un articolo come acquistato", "POST /lists/{list_uuid}/items/{uuid}/remove": "Rimuovi un articolo dalla lista", # Notifications "POST /lists/{list_uuid}/notify": "Invia notifica push ai membri (tipi 1-4)", }, "notification_types": { 1: "Sto facendo la spesa", 2: "Ho finito la spesa", 3: "Ho aggiornato la lista", 4: "Articoli urgenti aggiunti", }, } # --- Auth ------------------------------------------------------------------- @app.post("/auth/refresh") async def auth_refresh(): """Forza un nuovo login scartando la sessione in cache.""" await _reset_session() await _get_api() return {"status": "ok", "message": "Session refreshed successfully"} # --- Lists ------------------------------------------------------------------ @app.get("/lists") async def get_lists(): api = await _get_api() try: data = await api.load_lists() lists = data.lists return { "count": len(lists), "lists": [ {"list_uuid": lst.listUuid, "name": lst.name, "theme": getattr(lst, "theme", None)} for lst in lists ], } except BringException as e: await _handle_bring_exception(e) @app.get("/lists/{list_uuid}") async def get_list(list_uuid: str, only_purchase: bool = False): api = await _get_api() try: raw = await api.get_list(list_uuid) all_lists = await api.load_lists() except BringException as e: await _handle_bring_exception(e) name = list_uuid for lst in all_lists.lists: if lst.listUuid == list_uuid: name = lst.name break def _items(src) -> list[dict]: return [ { "item_id": getattr(it, "itemId", getattr(it, "name", "")), "specification": getattr(it, "specification", getattr(it, "spec", "")), "uuid": getattr(it, "uuid", None), } for it in src ] items = getattr(raw, "items", raw) result = { "list_uuid": list_uuid, "name": name, "purchase": _items(getattr(items, "purchase", [])), } if not only_purchase: result["recently"] = _items(getattr(items, "recently", [])) return result # --- Items ------------------------------------------------------------------ @app.post("/lists/{list_uuid}/items") async def add_item(list_uuid: str, req: AddItemRequest, background_tasks: BackgroundTasks): api = await _get_api() try: payload: dict[str, Any] = {"itemId": req.item_id, "spec": req.specification} if req.uuid: payload["uuid"] = req.uuid await api.batch_update_list(list_uuid, payload, BringItemOperation.ADD) result = {"status": "ok", "item_id": req.item_id, "reference_id": req.reference_id} if req.webhook_url: background_tasks.add_task(_send_webhook, req.webhook_url, result) return result except BringException as e: await _handle_bring_exception(e) @app.post("/lists/{list_uuid}/items/batch") async def batch_add_items(list_uuid: str, req: BatchAddRequest, background_tasks: BackgroundTasks): api = await _get_api() try: payload = [ {**({"itemId": it.item_id, "spec": it.specification}), **({"uuid": it.uuid} if it.uuid else {})} for it in req.items ] await api.batch_update_list(list_uuid, payload, BringItemOperation.ADD) result = {"status": "ok", "items_added": len(payload), "reference_id": req.reference_id} if req.webhook_url: background_tasks.add_task(_send_webhook, req.webhook_url, result) return result except BringException as e: await _handle_bring_exception(e) @app.post("/lists/{list_uuid}/items/{uuid}/update") async def update_item(list_uuid: str, uuid: str, req: UpdateItemRequest, background_tasks: BackgroundTasks): """Aggiorna la specification (quantità) di un articolo identificato dal suo uuid.""" api = await _get_api() try: await api.batch_update_list(list_uuid, {"itemId": req.item_id, "uuid": uuid}, BringItemOperation.REMOVE) await api.batch_update_list(list_uuid, {"itemId": req.item_id, "uuid": uuid, "spec": req.specification}, BringItemOperation.ADD) result = {"status": "ok", "uuid": uuid, "item_id": req.item_id, "specification": req.specification, "reference_id": req.reference_id} if req.webhook_url: background_tasks.add_task(_send_webhook, req.webhook_url, result) return result except BringException as e: await _handle_bring_exception(e) @app.post("/lists/{list_uuid}/items/{uuid}/complete") async def complete_item(list_uuid: str, uuid: str, req: ItemActionRequest, background_tasks: BackgroundTasks): """Segna come acquistato l'articolo identificato dal suo uuid.""" api = await _get_api() try: await api.batch_update_list(list_uuid, {"itemId": req.item_id, "uuid": uuid}, BringItemOperation.COMPLETE) result = {"status": "ok", "uuid": uuid, "item_id": req.item_id, "reference_id": req.reference_id} if req.webhook_url: background_tasks.add_task(_send_webhook, req.webhook_url, result) return result except BringException as e: await _handle_bring_exception(e) @app.post("/lists/{list_uuid}/items/{uuid}/remove") async def remove_item(list_uuid: str, uuid: str, req: ItemActionRequest, background_tasks: BackgroundTasks): """Rimuove l'articolo identificato dal suo uuid.""" api = await _get_api() try: await api.batch_update_list(list_uuid, {"itemId": req.item_id, "uuid": uuid}, BringItemOperation.REMOVE) result = {"status": "ok", "uuid": uuid, "item_id": req.item_id, "reference_id": req.reference_id} if req.webhook_url: background_tasks.add_task(_send_webhook, req.webhook_url, result) return result except BringException as e: await _handle_bring_exception(e) # --- Notifications ---------------------------------------------------------- @app.post("/lists/{list_uuid}/notify") async def notify_list(list_uuid: str, req: NotifyRequest, background_tasks: BackgroundTasks): """ Invia una notifica push ai membri della lista condivisa. Tipi disponibili: - 1: Sto facendo la spesa per [lista] - 2: Ho finito la spesa per [lista] - 3: Ho aggiornato la lista [lista] - 4: Articoli urgenti aggiunti """ api = await _get_api() try: await api.notify(list_uuid, req.notification_type) result = {"status": "ok", "notification_type": req.notification_type, "reference_id": req.reference_id} if req.webhook_url: background_tasks.add_task(_send_webhook, req.webhook_url, result) return result except BringException as e: await _handle_bring_exception(e) # --- Debug ------------------------------------------------------------------ @app.get("/debug/list/{list_uuid}") async def debug_list(list_uuid: str): api = await _get_api() raw = await api.get_list(list_uuid) return { "type": str(type(raw)), "dir": [a for a in dir(raw) if not a.startswith("_")], "repr": repr(raw), } # --- Health ----------------------------------------------------------------- @app.get("/health") def health_check(): return { "status": "healthy", "session_active": _session_cache["api"] is not None, }