import json from datetime import date, datetime from typing import Any, Optional import httpx from fastapi import APIRouter, HTTPException from fastapi.responses import StreamingResponse from app.config import settings from app.db import execute, fetch_all, fetch_one from app.services.briefing import ( build_live_digest, collect_briefing_data, generate_daily_briefing, serialize_stats, stream_daily_briefing, ) router = APIRouter(prefix="/api", tags=["api"]) def _stats_payload() -> dict[str, Any]: stats: dict[str, Any] = { "deals": 0, "clients": 0, "pending_approvals": 0, "pipeline_value": 0, } try: row = fetch_one("SELECT COUNT(*) AS c FROM deals") stats["deals"] = int(row["c"]) if row else 0 except Exception: pass try: row = fetch_one("SELECT COUNT(*) AS c FROM clients") stats["clients"] = int(row["c"]) if row else 0 except Exception: pass try: row = fetch_one("SELECT COUNT(*) AS c FROM agent_events WHERE status = 'needs_approval'") stats["pending_approvals"] = int(row["c"]) if row else 0 except Exception: pass try: row = fetch_one( "SELECT COALESCE(SUM(value), 0) AS total FROM deals WHERE stage NOT IN ('won', 'lost')" ) stats["pipeline_value"] = float(row["total"]) if row else 0 except Exception: pass return stats @router.get("/server-time") async def server_time(): now = datetime.utcnow() return {"utc": now.isoformat() + "Z", "timezone": "Europe/Amsterdam"} @router.get("/herman/briefing/stats") async def herman_briefing_stats(): """Live stats from DB — always fresh for dashboard panels.""" stats = serialize_stats(collect_briefing_data()) bookmarks = [] bookmark_map = {} try: bookmarks = fetch_all( """SELECT b.rss_item_id, b.title, b.link, b.feed_name, b.created_at FROM rss_bookmarks b ORDER BY b.created_at DESC LIMIT 30""" ) for b in bookmarks: if b.get("created_at") and hasattr(b["created_at"], "isoformat"): b["created_at"] = b["created_at"].isoformat() bookmark_map[b["rss_item_id"]] = True except Exception: bookmarks = [] stats["rss_bookmarks"] = bookmarks stats["rss_bookmark_ids"] = list(bookmark_map.keys()) if not stats.get("rss_live") and not stats.get("trending_food"): try: stats["rss_live"] = fetch_all( """SELECT i.id, i.title, i.link, i.summary, f.name AS feed_name, f.category, i.published_at FROM rss_items i JOIN rss_feeds f ON f.id = i.feed_id AND f.is_active = TRUE ORDER BY i.published_at DESC NULLS LAST, i.fetched_at DESC LIMIT 25""" ) except Exception: stats["rss_live"] = [] return {"ok": True, "stats": stats, "bookmarks": bookmarks, "live_digest": build_live_digest(stats), "at": datetime.utcnow().isoformat()} @router.post("/herman/briefing") async def herman_briefing(): try: content, stats = await generate_daily_briefing() except Exception as exc: raise HTTPException(status_code=502, detail=str(exc)) from exc return {"ok": True, "content": content, "stats": stats, "generated_at": datetime.utcnow().isoformat()} @router.post("/herman/briefing/stream") async def herman_briefing_stream(): """SSE stream — live stappen tijdens dagrapport generatie.""" async def event_gen(): try: async for event in stream_daily_briefing(): yield f"data: {json.dumps(event, default=str)}\n\n" except Exception as exc: err = {"type": "error", "message": str(exc)} yield f"data: {json.dumps(err)}\n\n" return StreamingResponse( event_gen(), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, ) @router.get("/herman/briefing/latest") async def herman_briefing_latest(): try: row = fetch_one( "SELECT id, content, generated_by, metadata, created_at FROM daily_briefings ORDER BY created_at DESC LIMIT 1" ) except Exception as exc: raise HTTPException(status_code=500, detail=str(exc)) from exc live_stats = serialize_stats(collect_briefing_data()) if not row: return {"ok": True, "content": None, "stats": live_stats} if row.get("created_at") and hasattr(row["created_at"], "isoformat"): row["created_at"] = row["created_at"].isoformat() return {"ok": True, **row, "stats": live_stats} @router.get("/live/platform") async def live_platform(limit: int = 100, agent: Optional[str] = None): from app.services.platform_live import fetch_platform_events, platform_stats events = fetch_platform_events(limit=min(limit, 200), agent=agent) return {"ok": True, "stats": platform_stats(), "events": events} @router.get("/events") async def list_events(limit: int = 50): limit = max(1, min(limit, 200)) try: rows = fetch_all( """ SELECT id, agent_name, event_type, title, body, status, created_at FROM agent_events ORDER BY created_at DESC LIMIT %s """, (limit,), ) except Exception as exc: raise HTTPException(status_code=500, detail=str(exc)) from exc for row in rows: if row.get("created_at"): row["created_at"] = row["created_at"].isoformat() return {"events": rows}