from __future__ import annotations import asyncio import json from datetime import date, datetime, timezone from typing import Any from app.config import settings from app.db import execute, fetch_all, fetch_one from app.services import ollama def _safe_count(table: str, where: str = "", params: tuple = ()) -> int: try: clause = f" WHERE {where}" if where else "" row = fetch_one(f"SELECT COUNT(*) AS c FROM {table}{clause}", params or None) return int(row["c"]) if row else 0 except Exception: return 0 def _safe_sum(table: str, column: str, where: str = "", params: tuple = ()) -> float: try: clause = f" WHERE {where}" if where else "" row = fetch_one(f"SELECT COALESCE(SUM({column}), 0) AS total FROM {table}{clause}", params or None) return float(row["total"]) if row else 0.0 except Exception: return 0.0 def serialize_stats(data: dict[str, Any]) -> dict[str, Any]: """JSON-safe copy of briefing stats (datetimes, decimals).""" def _default(o: Any) -> Any: if hasattr(o, "isoformat"): return o.isoformat() if hasattr(o, "__float__"): try: return float(o) except (TypeError, ValueError): pass return str(o) return json.loads(json.dumps(data, default=_default)) def collect_briefing_data() -> dict[str, Any]: data: dict[str, Any] = { "date": date.today().isoformat(), "generated_at": datetime.now(timezone.utc).isoformat(), } data["clients"] = _safe_count("clients") data["deals"] = _safe_count("deals") data["products"] = _safe_count("products") data["suppliers"] = _safe_count("suppliers") data["pipeline_eur"] = _safe_sum("deals", "value", "stage NOT IN ('won', 'lost')") data["pending_approvals"] = _safe_count("agent_events", "status = 'needs_approval'") try: data["deals_by_stage"] = fetch_all( "SELECT stage, COUNT(*) AS cnt, COALESCE(SUM(value), 0) AS total FROM deals GROUP BY stage ORDER BY cnt DESC" ) except Exception: data["deals_by_stage"] = [] try: data["recent_clients"] = fetch_all( "SELECT name, stage, email, created_at FROM clients ORDER BY created_at DESC LIMIT 5" ) except Exception: data["recent_clients"] = [] try: data["recent_events"] = fetch_all( """ SELECT agent_name, event_type, title, status, created_at FROM agent_events ORDER BY created_at DESC LIMIT 12 """ ) except Exception: data["recent_events"] = [] try: data["pending_items"] = fetch_all( """ SELECT agent_name, title, event_type, created_at FROM agent_events WHERE status = 'needs_approval' ORDER BY created_at DESC LIMIT 8 """ ) except Exception: data["pending_items"] = [] try: row = fetch_one( """ SELECT COUNT(*) AS docs, COALESCE(SUM(word_count), 0) AS words, COALESCE(AVG(sentiment_compound), 0) AS avg_sentiment FROM document_analytics """ ) data["nas_docs"] = int(row["docs"] or 0) if row else 0 data["nas_words"] = int(row["words"] or 0) if row else 0 data["nas_sentiment"] = round(float(row["avg_sentiment"] or 0), 3) if row else 0.0 except Exception: data["nas_docs"] = data["nas_words"] = 0 data["nas_sentiment"] = 0.0 try: data["nas_files"] = fetch_all( """ SELECT filename, doc_type, sentiment_label, word_count FROM document_analytics ORDER BY analyzed_at DESC LIMIT 8 """ ) except Exception: data["nas_files"] = [] try: data["top_words"] = fetch_all( """ SELECT lemma, SUM(count) AS total FROM document_word_counts WHERE NOT is_stopword GROUP BY lemma ORDER BY total DESC LIMIT 10 """ ) except Exception: data["top_words"] = [] try: data["calendar_events"] = fetch_all( """ SELECT ce.title, ce.starts_at, ce.ends_at, c.name AS client_name FROM calendar_events ce LEFT JOIN clients c ON c.id = ce.client_id WHERE ce.starts_at >= NOW() - INTERVAL '1 day' AND ce.starts_at <= NOW() + INTERVAL '7 days' ORDER BY ce.starts_at ASC LIMIT 10 """ ) except Exception: data["calendar_events"] = [] return data def build_template_report(data: dict[str, Any]) -> str: lines = [ f"# Foodlinkk Dagrapport — {data['date']}", "", f"*Gegenereerd: {data['generated_at'][:19]} UTC · Model: {settings.OLLAMA_MODEL}*", "", "## KPI's", f"- **Klanten:** {data['clients']}", f"- **Deals totaal:** {data['deals']}", f"- **Pipeline (actief):** €{data['pipeline_eur']:,.0f}", f"- **Producten:** {data['products']}", f"- **Leveranciers:** {data['suppliers']}", f"- **Openstaande goedkeuringen:** {data['pending_approvals']}", "", "## Pipeline per stage", ] if data.get("deals_by_stage"): for row in data["deals_by_stage"]: lines.append(f"- **{row.get('stage')}:** {row.get('cnt')} deals · €{float(row.get('total') or 0):,.0f}") else: lines.append("- Geen deals in database.") lines.extend(["", "## Recente klanten"]) for row in data.get("recent_clients") or []: lines.append(f"- {row.get('name')} ({row.get('stage')})") if not data.get("recent_clients"): lines.append("- Geen klanten.") lines.extend([ "", "## NAS share — documenten", f"- Ingelezen documenten: **{data['nas_docs']}**", f"- Totaal woorden geanalyseerd: **{data['nas_words']}**", f"- Gemiddeld sentiment: **{data['nas_sentiment']}**", "", ]) for row in data.get("nas_files") or []: lines.append(f"- {row.get('filename')} · {row.get('doc_type')} · sentiment: {row.get('sentiment_label')}") if data.get("top_words"): lines.extend(["", "## Top woorden (NAS corpus)"]) for row in data["top_words"]: lines.append(f"- {row.get('lemma')}: {row.get('total')}×") if data.get("calendar_events"): lines.extend(["", "## Agenda (7 dagen)"]) for row in data["calendar_events"]: ts = row.get("starts_at") ts_s = ts.isoformat()[:16] if hasattr(ts, "isoformat") else str(ts)[:16] lines.append(f"- [{ts_s}] {row.get('title')} ({row.get('client_name') or '-'})") lines.extend(["", "## Recente agent activiteit"]) for row in data.get("recent_events") or []: ts = row.get("created_at") ts_s = ts.isoformat()[:16] if hasattr(ts, "isoformat") else str(ts)[:16] lines.append(f"- [{ts_s}] **{row.get('agent_name')}** — {row.get('title') or row.get('event_type')}") if data.get("pending_items"): lines.extend(["", "## ⚠️ Wacht op jouw goedkeuring"]) for row in data["pending_items"]: lines.append(f"- {row.get('agent_name')}: {row.get('title')}") return "\n".join(lines) async def _ai_executive_summary(data: dict[str, Any], template: str) -> str: prompt = ( "Schrijf alleen deze twee secties in het Nederlands (markdown):\n" "## Samenvatting\n(4-6 zinnen voor CEO Aïssa)\n\n" "## Actiepunten vandaag\n(minimaal 5 concrete bullets)\n\n" f"Gebaseerd op:\n- Pipeline €{data['pipeline_eur']:,.0f}\n" f"- {data['clients']} klanten, {data['deals']} deals\n" f"- {data['pending_approvals']} goedkeuringen open\n" f"- {data['nas_docs']} NAS documenten\n" ) system = "Je bent Herman, co-CEO Foodlinkk. Kort, zakelijk, actionable." try: return await ollama.generate(prompt, system=system, timeout=120.0) except Exception: return "" def _save_briefing(content: str, data: dict[str, Any]) -> None: safe = serialize_stats(data) metadata = {"stats": safe, "model": settings.OLLAMA_MODEL, "type": "daily_ceo_report"} try: execute( "INSERT INTO daily_briefings (content, generated_by, metadata) VALUES (%s, %s, %s::jsonb)", (content, "herman", json.dumps(metadata)), ) except Exception: pass try: execute( """ INSERT INTO agent_events (agent_name, agent_type, event_type, title, body, status, channel, metadata) VALUES (%s, %s, %s, %s, %s, %s, %s, %s::jsonb) """, ( "herman", "herman_delegate", "briefing", f"CEO dagrapport {data['date']}", content[:2000], "completed", "dashboard", json.dumps({"stats": safe}), ), ) except Exception: pass async def generate_daily_briefing() -> tuple[str, dict[str, Any]]: data = collect_briefing_data() template = build_template_report(data) try: ai_part = await asyncio.wait_for(_ai_executive_summary(data, template), timeout=90.0) except (asyncio.TimeoutError, Exception): ai_part = "" if ai_part and len(ai_part.strip()) > 40: content = ai_part.strip() + "\n\n---\n\n" + template else: content = template + "\n\n---\n\n*AI-samenvatting niet beschikbaar (Ollama busy) — bovenstaande data is live uit je database.*" _save_briefing(content, data) return content, serialize_stats(data)