from __future__ import annotations import asyncio import json import re from datetime import date, datetime, timezone from typing import Any import httpx from app.config import settings from app.db import execute, fetch_all, fetch_one from app.services import market_stocks, llm_router 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]: 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 _briefing_step( phase: str, source: str, agent: str, status: str, message: str, detail: str | None = None, ) -> dict[str, Any]: return { "type": "step", "phase": phase, "source": source, "agent": agent, "status": status, "message": message, "detail": detail, "at": datetime.now(timezone.utc).isoformat(), } def _collect_crm_core(data: dict[str, Any]) -> None: 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'") def _collect_agent_queue(data: dict[str, Any]) -> None: try: data["pending_approval_requests"] = fetch_all( """SELECT id, agent_key, action_type, title, query_payload, created_at FROM agent_action_requests WHERE status = 'pending' ORDER BY created_at ASC LIMIT 15""" ) data["pending_approvals"] = len(data["pending_approval_requests"]) except Exception: data["pending_approval_requests"] = [] try: data["recent_executed_actions"] = fetch_all( """SELECT id, agent_key, action_type, title, result, executed_at, approved_by FROM agent_action_requests WHERE status = 'executed' AND executed_at >= NOW() - INTERVAL '24 hours' ORDER BY executed_at DESC LIMIT 12""" ) except Exception: data["recent_executed_actions"] = [] def _collect_projects_ops(data: dict[str, Any]) -> None: try: data["project_assets_recent"] = fetch_all( """SELECT pa.title, pa.asset_type, pa.source_agent, pa.created_at, cp.name AS project_name FROM project_assets pa JOIN cockpit_projects cp ON cp.id = pa.project_id WHERE pa.created_at >= NOW() - INTERVAL '24 hours' ORDER BY pa.created_at DESC LIMIT 15""" ) except Exception: data["project_assets_recent"] = [] try: data["ops_maintenance_open"] = fetch_all( """SELECT severity, title, body, created_at FROM ops_maintenance_notes WHERE resolved = false ORDER BY created_at DESC LIMIT 8""" ) except Exception: data["ops_maintenance_open"] = [] try: data["config_backups_recent"] = fetch_all( """SELECT status, message, commit_ref, created_at FROM config_backups ORDER BY created_at DESC LIMIT 5""" ) except Exception: data["config_backups_recent"] = [] try: data["sysops_activity_24h"] = fetch_all( """SELECT action_type, title, body, commit_ref, files_changed, status, created_at FROM sysops_activity WHERE created_at >= NOW() - INTERVAL '24 hours' ORDER BY created_at DESC LIMIT 20""" ) except Exception: data["sysops_activity_24h"] = [] try: data["sysops_events_24h"] = fetch_all( """SELECT event_type, title, body, status, created_at, metadata FROM agent_events WHERE LOWER(agent_name) = 'sysops' AND created_at >= NOW() - INTERVAL '24 hours' ORDER BY created_at DESC LIMIT 15""" ) except Exception: data["sysops_events_24h"] = [] def _collect_pipeline_events(data: dict[str, Any]) -> None: 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["recent_handoffs"] = fetch_all( """SELECT from_agent, to_agent, handoff_type, status, created_at, correlation_id::text FROM agent_handoffs WHERE created_at >= NOW() - INTERVAL '24 hours' ORDER BY created_at DESC LIMIT 20""" ) except Exception: data["recent_handoffs"] = [] 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"] = [] def _collect_nas_analytics(data: dict[str, Any]) -> None: 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"] = [] def _collect_retail_intel(data: dict[str, Any]) -> None: data["supermarkets"] = _safe_count("supermarkets") data["clients_active"] = _safe_count("clients", "stage = 'active'") data["clients_total"] = _safe_count("clients") data["crm_partnerships"] = _safe_count("supermarkets", "partnership_status = 'active'") data["wholesalers"] = _safe_count("wholesalers") data["rss_bookmarks"] = _safe_count("rss_bookmarks") data["promo_campaigns"] = _safe_count("promo_campaigns", "status = 'active'") try: data["top_opportunities"] = fetch_all( """SELECT s.name, s.chain, s.city, s.address, s.postcode, ros.halal_opportunity_score FROM retail_opportunity_scores ros JOIN supermarkets s ON s.id = ros.supermarket_id WHERE s.name NOT ILIKE '%masala%' AND s.name NOT ILIKE 'TEST%' AND COALESCE(s.data_source,'') NOT ILIKE '%test_exclude%' ORDER BY ros.halal_opportunity_score DESC LIMIT 8""" ) except Exception: data["top_opportunities"] = [] try: data["milestones_pending"] = fetch_all( """SELECT sm.title, sm.milestone_type, sm.status, sm.target_date, sm.value_eur, s.name AS store_name, s.chain, c.name AS client_name FROM sales_milestones sm LEFT JOIN supermarkets s ON s.id = sm.supermarket_id LEFT JOIN clients c ON c.id = sm.client_id WHERE sm.status IN ('pending', 'in_progress') ORDER BY sm.target_date ASC NULLS LAST, sm.created_at DESC LIMIT 8""" ) except Exception: data["milestones_pending"] = [] try: data["milestones_recent"] = fetch_all( """SELECT sm.title, sm.milestone_type, sm.status, sm.completed_at, sm.value_eur, s.name AS store_name, s.chain FROM sales_milestones sm LEFT JOIN supermarkets s ON s.id = sm.supermarket_id ORDER BY sm.created_at DESC LIMIT 5""" ) except Exception: data["milestones_recent"] = [] def _collect_rss_market(data: dict[str, Any]) -> None: try: data["rss_highlights"] = fetch_all( """SELECT i.id, i.title, i.link, i.summary, f.name AS feed_name, f.url AS feed_url FROM rss_items i JOIN rss_feeds f ON f.id = i.feed_id AND f.is_active = TRUE WHERE i.title ILIKE ANY (ARRAY['%kant%','%maaltijd%','%supermarkt%','%retail%','%halal%','%jumbo%','%meal%']) ORDER BY i.published_at DESC NULLS LAST LIMIT 8""" ) except Exception: data["rss_highlights"] = [] try: data["market_trends"] = fetch_all( "SELECT trend_name, description, opportunity_score FROM market_trends ORDER BY updated_at DESC LIMIT 4" ) except Exception: data["market_trends"] = [] try: quotes = market_stocks.fetch_retail_quotes() data["market_stocks"] = quotes data["market_summary"] = market_stocks.market_summary(quotes) except Exception: data["market_stocks"] = [] data["market_summary"] = {} try: data["regulation_highlights"] = fetch_all( """SELECT i.title, i.link, i.summary, f.name AS feed_name, f.url AS feed_url, f.category FROM rss_items i JOIN rss_feeds f ON f.id = i.feed_id AND f.is_active = TRUE WHERE f.category IN ('regelgeving', 'cbs') ORDER BY i.published_at DESC NULLS LAST LIMIT 8""" ) except Exception: data["regulation_highlights"] = [] try: data["trending_food"] = fetch_all( """SELECT i.id, i.title, i.link, i.summary, f.name AS feed_name, f.url AS feed_url, f.category, i.published_at FROM rss_items i JOIN rss_feeds f ON f.id = i.feed_id AND f.is_active = TRUE WHERE f.category IN ('food', 'markt', 'supermarkt', 'retail', 'kant-en-klaar') OR f.name ILIKE '%retaildetail%' ORDER BY i.published_at DESC NULLS LAST, i.fetched_at DESC LIMIT 10""" ) except Exception: data["trending_food"] = [] if not data.get("trending_food"): try: data["trending_food"] = fetch_all( """SELECT i.id, i.title, i.link, i.summary, f.name AS feed_name, f.url AS feed_url, 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 12""" ) except Exception: data["trending_food"] = [] try: data["food_market_highlights"] = fetch_all( """SELECT i.title, i.link, i.summary, f.name AS feed_name, f.url AS feed_url, f.category FROM rss_items i JOIN rss_feeds f ON f.id = i.feed_id AND f.is_active = TRUE WHERE f.category IN ('food', 'markt', 'supermarkt', 'kant-en-klaar', 'retail') OR f.name ILIKE '%retaildetail%' ORDER BY i.published_at DESC NULLS LAST LIMIT 10""" ) except Exception: data["food_market_highlights"] = [] if not data.get("food_market_highlights"): data["food_market_highlights"] = list(data.get("trending_food") or [])[:10] try: data["rss_live"] = fetch_all( """SELECT i.id, i.title, i.link, i.summary, f.name AS feed_name, f.url AS feed_url, 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: data["rss_live"] = list(data.get("trending_food") or []) data["rss_items"] = _safe_count("rss_items") def _classify_revenue_brand(name: str) -> str: n = (name or "").lower() if any(k in n for k in ("foodlinkk", "linknbit", "linknbi", "wereldexport")): return "foodlinkk" if "cucina" in n and "foodlinkk" not in n: return "cucina" chains = ("ah", "jumbo", "picnic", "plus", "vomar", "dirk", "hoogvliet", "deka", "spar", "doner") if any(n == c or n.startswith(c + " ") for c in chains): return "cucina" if any(k in n for k in ("foodservice", "snack", "diepvries", "halal", "gas stunning", "marketing budget adds cucina")): return "cucina" if any(k in n for k in ("fresh supplier", "boerschappen", "subsidies cucina")): return "foodlinkk" if any(k in n for k in ("total earnings", "loonkosten", "total expenses", "verkopen cucina")): return "shared" return "other" def _collect_revenue_cockpit(data: dict[str, Any]) -> None: try: from app.services import revenue_cockpit as rc projects = rc.list_projects() stats = rc.dashboard_stats() by_brand: dict[str, list] = {"cucina": [], "foodlinkk": [], "shared": [], "other": []} for p in projects: if (p.get("status") or "") != "active": continue brand = _classify_revenue_brand(p.get("name") or "") by_brand[brand].append(p) data["revenue"] = { "stats": stats, "by_brand": { k: sorted(v, key=lambda x: float(x.get("margin_month") or 0), reverse=True)[:6] for k, v in by_brand.items() }, } except Exception: data["revenue"] = {} def _collect_export_intel(data: dict[str, Any]) -> None: try: url = f"{settings.TOOLS_API_URL.rstrip('/')}/export-intel/stats" r = httpx.get(url, timeout=4.0) data["export_intel"] = r.json() if r.status_code == 200 else {} except Exception: data["export_intel"] = {} def collect_briefing_data(*, skip_halal: bool = False) -> dict[str, Any]: data: dict[str, Any] = { "date": date.today().isoformat(), "generated_at": datetime.now(timezone.utc).isoformat(), } _collect_crm_core(data) _collect_agent_queue(data) _collect_projects_ops(data) _collect_pipeline_events(data) _collect_nas_analytics(data) _collect_retail_intel(data) _collect_rss_market(data) _collect_revenue_cockpit(data) _collect_export_intel(data) if not skip_halal: _collect_halal_recommendations(data) data["activity_log"] = _build_activity_log(data) return data def _collect_halal_recommendations(data: dict[str, Any]) -> None: try: url = f"{settings.TOOLS_API_URL.rstrip('/')}/recommendations/halal/live" r = httpx.get(url, params={"limit": 5, "brand": "cucina"}, timeout=20.0) data["halal_recommendations"] = r.json() if r.status_code == 200 else {"items": []} except Exception: data["halal_recommendations"] = {"items": []} # Legacy veld voor backwards compat — eerste 3 engine results items = (data.get("halal_recommendations") or {}).get("items") or [] data["top_opportunities"] = [ { "id": i.get("store_id"), "name": (i.get("title") or "").split(" · ", 1)[-1], "chain": i.get("chain"), "city": i.get("city"), "halal_opportunity_score": i.get("halal_opportunity_score") or i.get("score"), "composite_score": i.get("score"), "reasons": i.get("reasons"), } for i in items[:5] ] async def stream_daily_briefing(): """Yield SSE step events while building the CEO daily report.""" data: dict[str, Any] = { "date": date.today().isoformat(), "generated_at": datetime.now(timezone.utc).isoformat(), } yield _briefing_step("init", "dashboard", "herman", "running", "Herman start CEO dagrapport") yield _briefing_step("crm", "CRM", "crm", "running", "Pipeline & klanten ophalen…") _collect_crm_core(data) yield _briefing_step( "crm", "CRM", "crm", "ok", f"{data['clients']} klanten · {data['deals']} deals · pipeline €{data['pipeline_eur']:,.0f}", ) yield _briefing_step("agents", "Herman", "herman", "running", "Open beslissingen filteren…") _collect_agent_queue(data) data["pending_approval_requests"] = _fresh_approvals(data.get("pending_approval_requests") or []) data["pending_approvals"] = len(data["pending_approval_requests"]) pending = data.get("pending_approval_requests") or [] yield _briefing_step( "agents", "Herman", "herman", "ok", f"{len(pending)} relevante beslissingen", ) yield _briefing_step("projects", "Operations", "product", "running", "Milestones & projectstatus…") _collect_projects_ops(data) yield _briefing_step( "projects", "Operations", "product", "ok", f"{len(data.get('milestones_pending') or [])} open milestones", ) yield _briefing_step("pipeline", "Finance", "finance", "running", "Pipeline stages laden…") _collect_pipeline_events(data) stages = len(data.get("deals_by_stage") or []) yield _briefing_step( "pipeline", "Finance", "finance", "ok", f"{stages} pipeline stages", ) # NAS collected for internal stats but not streamed as CEO noise _collect_nas_analytics(data) yield _briefing_step("retail", "Retail", "retail", "running", "Supermarkten & partnerships…") _collect_retail_intel(data) yield _briefing_step( "retail", "Retail", "retail", "ok", f"{data.get('supermarkets', 0)} filialen · {data.get('crm_partnerships', 0)} partnerships · " f"{len(data.get('milestones_pending') or [])} milestones", ) yield _briefing_step("rss", "Markt", "marketing", "running", "Food & retail nieuws selecteren…") _collect_rss_market(data) trend_n = len(data.get("trending_food") or []) yield _briefing_step( "rss", "Markt", "marketing", "ok", f"{trend_n} relevante food-headlines", ) data["activity_log"] = _build_activity_log(data) model_label = getattr(settings, "HERMES_API_MODEL", None) or "hermes-agent" yield _briefing_step( "ai", f"Herman ({model_label})", "herman", "running", "Herman schrijft CEO-briefing…", ) try: ai_part = await asyncio.wait_for(_ai_executive_summary(data), timeout=120.0) except (asyncio.TimeoutError, Exception) as exc: ai_part = "" yield _briefing_step( "ai", "Herman", "herman", "warn", "AI timeout — template samenvatting", str(exc)[:120], ) else: yield _briefing_step( "ai", f"Herman ({model_label})", "herman", "ok", f"Briefing klaar ({len(ai_part or '')} tekens)", ) yield _briefing_step("compose", "Herman", "herman", "running", "Rapport samenstellen & opslaan…") template = build_template_report(data) if ai_part and len(ai_part.strip()) > 80: content = ai_part.strip() + "\n\n---\n\n" + template else: content = _fallback_summary(data) + "\n\n---\n\n" + template _save_briefing(content, data) stats = serialize_stats(data) yield _briefing_step("compose", "PostgreSQL", "herman", "ok", "Dagrapport opgeslagen in daily_briefings") yield { "type": "done", "content": content, "stats": stats, "generated_at": datetime.now(timezone.utc).isoformat(), } def _fresh_approvals(rows: list | None, *, max_age_days: int = 14) -> list: """Drop stale agent approvals so SysOps noise does not dominate the CEO digest.""" from datetime import datetime, timezone, timedelta out = [] cutoff = datetime.now(timezone.utc) - timedelta(days=max_age_days) for row in rows or []: ts = row.get("created_at") or row.get("requested_at") or row.get("updated_at") if ts is None: continue if hasattr(ts, "tzinfo") and ts.tzinfo is None: ts = ts.replace(tzinfo=timezone.utc) try: if ts < cutoff: continue except TypeError: continue agent = (row.get("agent_key") or "").lower() action = (row.get("action_type") or "").lower() # Keep business agents; skip routine SysOps unless critical keyword if agent in ("sysops", "ops", "it") or action in ("maintenance_scan", "config_backup", "gitea_sync"): title = (row.get("title") or "").lower() if not any(k in title for k in ("critical", "kritiek", "down", "security", "breach")): continue out.append(row) return out def _build_activity_log(data: dict[str, Any]) -> list[str]: lines: list[str] = [] for row in data.get("pending_approval_requests") or []: agent = row.get("agent_key") or "agent" action = row.get("action_type") or "actie" title = row.get("title") or "" if action == "maintenance_scan": lines.append(f"⏳ SysOps vraagt toestemming voor update-scan: {title}") elif action == "config_backup": lines.append(f"⏳ SysOps vraagt goedkeuring backup: {title}") else: lines.append(f"⏳ {agent} wacht op goedkeuring ({action}): {title}") for row in data.get("recent_executed_actions") or []: lines.append(f"✓ Uitgevoerd door {row.get('agent_key')}: {row.get('title')}") for row in data.get("project_assets_recent") or []: ts = row.get("created_at") ts_s = ts.isoformat()[:16] if hasattr(ts, "isoformat") else str(ts or "")[:16] lines.append( f"📁 Project asset ({row.get('project_name')}): {row.get('title')} " f"[{row.get('asset_type')} · {row.get('source_agent')}] {ts_s}" ) for row in data.get("ops_maintenance_open") or []: lines.append(f"🔧 IT Ops [{row.get('severity')}]: {row.get('title')}") for row in data.get("config_backups_recent") or []: lines.append(f"💾 Backup {row.get('status')}: {row.get('message') or row.get('commit_ref')}") for row in data.get("sysops_activity_24h") or []: ts = row.get("created_at") ts_s = ts.isoformat()[11:16] if hasattr(ts, "isoformat") else "" cref = f" [{row.get('commit_ref')}]" if row.get("commit_ref") else "" lines.append(f"🖥️ SysOps {row.get('action_type')}: {row.get('title')}{cref} ({ts_s})") for row in data.get("sysops_events_24h") or []: if (row.get("event_type") or "") == "gitea_sync": continue ts = row.get("created_at") ts_s = ts.isoformat()[11:16] if hasattr(ts, "isoformat") else "" lines.append(f"🔧 SysOps: {row.get('title')} ({ts_s})") for row in (data.get("recent_events") or [])[:8]: ts = row.get("created_at") ts_s = ts.isoformat()[11:16] if hasattr(ts, "isoformat") else "" lines.append(f"⚡ {row.get('agent_name')}: {row.get('title')} ({ts_s})") for row in (data.get("recent_handoffs") or [])[:8]: ts = row.get("created_at") ts_s = ts.isoformat()[11:16] if hasattr(ts, "isoformat") else "" lines.append( f"🔗 {row.get('from_agent')} → {row.get('to_agent')}: {row.get('handoff_type')} ({ts_s})" ) return lines[:25] def _strip_activity_prefix(line: str) -> str: for prefix in ("⏳ ", "✓ ", "📁 ", "🔧 ", "💾 ", "🖥️ ", "⚡ ", "🔗 "): if line.startswith(prefix): return line[len(prefix) :].strip() return line.strip() def _short_activity(line: str, max_len: int = 88) -> str: s = _strip_activity_prefix(line) s = re.sub(r" — \d{4}-\d{2}-\d{2}$", "", s) s = re.sub(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}(:\d{2})?", "", s).strip() s = re.sub(r"\s+", " ", s) if len(s) > max_len: return s[: max_len - 1].rstrip() + "…" return s def _fmt_eur(val: float | int | None) -> str: try: return f"€{float(val or 0):,.0f}" except (TypeError, ValueError): return "€—" def _revenue_line(p: dict[str, Any]) -> str: mm = p.get("margin_month") margin = f" · marge/maand {_fmt_eur(mm)}" if mm else "" obj = p.get("open_objectives") or 0 extra = f" · {obj} open stappen" if obj else "" return f"{p.get('name')}{margin}{extra}" def build_live_digest(data: dict[str, Any]) -> dict[str, Any]: """CEO digest: B2B + B2C + focus + trends + nieuws — geen SysOps.""" revenue = data.get("revenue") or {} rev_stats = revenue.get("stats") or {} rev_goals = rev_stats.get("goals") or {} export = data.get("export_intel") or {} market = data.get("market_summary") or {} trends = data.get("market_trends") or [] halal_items = (data.get("halal_recommendations") or {}).get("items") or [] # —— Foodlinkk B2B (compact) —— fl_lines: list[str] = [ f"Pipeline {_fmt_eur(data.get('pipeline_eur'))} · {data.get('deals', 0)} deals · {data.get('clients', 0)} klanten", ] if export.get("tenders_open") or export.get("distributors"): fl_lines.append( f"Export: {export.get('distributors', 0):,} distributeurs · {export.get('tenders_open', 0)} open tenders" ) fl_projects = (revenue.get("by_brand") or {}).get("foodlinkk") or [] fl_projects = sorted( [p for p in fl_projects if p.get("margin_month")], key=lambda p: -float(p.get("margin_month") or 0), ) n_rev = 0 for p in fl_projects: name_l = (p.get("name") or "").lower() if any(x in name_l for x in ("verkopen", "2030", "total earnings", "total expense", "loonkosten", "boekhouder", "linknbit")): continue fl_lines.append(_revenue_line(p)) n_rev += 1 if n_rev >= 2: break if export.get("crm_linked"): fl_lines.append(f"Export↔CRM: {export.get('crm_linked')} gekoppeld") # —— Cucina B2C (compact) —— cu_lines: list[str] = [] if rev_stats: cu_lines.append( f"Marge/maand {_fmt_eur(rev_stats.get('total_margin_month'))} · " f"{rev_stats.get('open_objectives', 0)} open stappen" ) cu_lines.append( f"Retail: {data.get('supermarkets', 0):,} filialen · {data.get('crm_partnerships', 0)} partnerships" ) cu_projects = (revenue.get("by_brand") or {}).get("cucina") or [] deal_bits = [] for p in cu_projects[:4]: name = (p.get("name") or "").strip() if not name or "marketing" in name.lower() or "boekhouder" in name.lower(): continue mm = p.get("margin_month") bit = name if mm: bit += f" {_fmt_eur(mm)}/m" if p.get("open_objectives"): bit += f" · {p.get('open_objectives')} stappen" deal_bits.append(bit) if len(deal_bits) >= 3: break if deal_bits: cu_lines.append("Deals: " + " · ".join(deal_bits)) if data.get("promo_campaigns"): cu_lines.append(f"Marketing: {data.get('promo_campaigns')} actieve promo's") if halal_items: rec = halal_items[0] cu_lines.append( f"Halal-kans: {rec.get('chain')} · {rec.get('city')} (score {int(float(rec.get('halal_opportunity_score') or rec.get('score') or 0))})" ) sections = [ {"brand": "foodlinkk", "title": "Foodlinkk B2B", "icon": "🌍", "lines": fl_lines}, {"brand": "cucina", "title": "Cucina B2C", "icon": "🍽️", "lines": cu_lines}, ] # —— Focus vandaag (business only, no sysops) —— focus_lines: list[str] = [] for row in (data.get("milestones_pending") or [])[:2]: title = (row.get("title") or "").strip() if title: focus_lines.append(title[:90]) for p in cu_projects: if p.get("open_objectives") and (p.get("name") or "").lower() in ("ah", "jumbo", "picnic", "plus"): focus_lines.append(f"Cucina follow-up: {p.get('name')} ({p.get('open_objectives')} open stappen)") break for rec in halal_items[:2]: reason = ((rec.get("reasons") or [""])[0] or "")[:60] focus_lines.append(f"Outreach {rec.get('chain')} · {rec.get('city')}" + (f" — {reason}" if reason else "")) if export.get("tenders_open"): focus_lines.append(f"Bekijk {export.get('tenders_open')} open export-tenders") # Skip sysops / infra approvals entirely if not focus_lines: focus_lines.append("Pipeline en Cucina retail-deals nalopen") sections.append({"brand": "platform", "title": "Focus vandaag", "icon": "🎯", "lines": focus_lines[:4]}) # —— Trends —— trend_lines: list[str] = [] for tr in trends[:4]: name = (tr.get("trend_name") or tr.get("name") or "").strip() score = tr.get("opportunity_score") if not name: continue if score is not None: pct = int(float(score) * 100) if float(score) <= 1 else int(float(score)) trend_lines.append(f"{name} ↑{pct}%") else: trend_lines.append(name) if market.get("best_performer"): bp = market["best_performer"] trend_lines.append(f"Aandeel {bp.get('name')}: {bp.get('change_pct', 0):+.1f}%") if not trend_lines: trend_lines.append("Geen trenddata — check Retail 360") # —— Nieuws (max 5, food only) —— news: list[dict[str, Any]] = [] seen: set[str] = set() skip = ("mediamarkt", "ceconomy", "elektro", "airco", "gaming", "smartphone", "sysops", "gitea", "backup") for row in list(data.get("trending_food") or []) + list(data.get("rss_highlights") or []) + list(data.get("rss_live") or []): title = str(row.get("title") or "").strip() low = title.lower() if not title or title in seen: continue if any(x in low for x in skip): continue seen.add(title) news.append({ "title": title[:110], "link": row.get("link") or "#", "source": row.get("feed_name") or "RSS", }) if len(news) >= 5: break actions = focus_lines[:4] long_term = [ "Cucina: schaal AH/Jumbo/PLUS naar actieve partnerships", "Foodlinkk: export-distributeurs koppelen aan CRM-deals", "Revenue: open stappen wekelijks reviewen", ] if rev_goals.get("horizon_text"): long_term.insert(0, str(rev_goals["horizon_text"]).strip().split("\n")[0][:100]) summary = f"{data.get('date')} · Foodlinkk B2B + Cucina B2C" return { "summary": summary, "sections": sections, "b2b": fl_lines, "b2c": cu_lines, "focus": focus_lines[:4], "trends": trend_lines[:5], "news": news, "actions": actions, "long_term": long_term[:4], "updated_at": data.get("generated_at"), "activity_count": len(data.get("activity_log") or []), "source": "live", } def format_live_digest_text(digest: dict[str, Any]) -> str: """Platte tekst voor Telegram / Herman chat — halal zit in Cucina-sectie.""" lines = ["📋 Herman briefing · live", "", digest.get("summary") or ""] for section in digest.get("sections") or []: icon = section.get("icon") or "" title = section.get("title") or "" lines.append("") lines.append(f"{icon} {title}".strip()) for line in section.get("lines") or []: lines.append(f" · {line}") actions = digest.get("actions") or [] if actions: lines.extend(["", "⚡ Korte termijn"]) for action in actions[:6]: lines.append(f" · {action}") long_term = digest.get("long_term") or [] if long_term: lines.extend(["", "🎯 Lange termijn"]) for item in long_term[:5]: lines.append(f" · {item}") updated = digest.get("updated_at") if updated: ts = str(updated)[:19].replace("T", " ") n = digest.get("activity_count") extra = f" · {n} activiteiten" if n is not None else "" lines.extend(["", f"Bijgewerkt {ts} UTC{extra}"]) return "\n".join(lines) def build_live_digest_text(data: dict[str, Any] | None = None) -> str: payload = data if data is not None else collect_briefing_data() return format_live_digest_text(build_live_digest(payload)) def build_template_report(data: dict[str, Any]) -> str: lines = [ f"# Foodlinkk Dagrapport — {data['date']}", "", f"*Gegenereerd: {data['generated_at'][:19]} UTC · Herman via OpenCode*", "", "## KPI's", f"- **Klanten:** {data['clients']} · **Deals:** {data['deals']} · **Pipeline:** €{data['pipeline_eur']:,.0f}", f"- **Supermarkten in DB:** {data.get('supermarkets', 0)} · **CRM partnerships:** {data.get('crm_partnerships', 0)}", f"- **Groothandels:** {data.get('wholesalers', 0)} · **Goedkeuringen open:** {data['pending_approvals']}", "", ] if data.get("top_opportunities"): lines.extend(["## Top halal-markt kansen (Retail 360)"]) for row in data["top_opportunities"]: score = round(float(row.get("halal_opportunity_score") or 0)) lines.append(f"- **{row.get('chain')} · {row.get('name')}** ({row.get('city')}) — score {score}/100") lines.append("") if data.get("milestones_pending"): lines.extend(["## Sales milestones — open"]) for row in data["milestones_pending"]: td = row.get("target_date") td_s = td.isoformat()[:10] if hasattr(td, "isoformat") else str(td or "—")[:10] lines.append(f"- [{td_s}] **{row.get('title')}** · {row.get('chain') or ''} {row.get('store_name') or ''} · €{row.get('value_eur') or '—'}") lines.append("") if data.get("rss_highlights"): lines.extend(["## Kant-en-klaar & supermarkt nieuws"]) for row in data["rss_highlights"]: lines.append(f"- [{row.get('feed_name')}] {row.get('title')}") lines.append("") if data.get("market_trends"): lines.extend(["## Markt trends"]) for row in data["market_trends"]: pct = round(float(row.get("opportunity_score") or 0) * 100) lines.append(f"- **{row.get('trend_name')}** ({pct}% kans) — {row.get('description') or ''}") lines.append("") lines.extend(["## Pipeline per stage"]) for row in data.get("deals_by_stage") or []: lines.append(f"- **{row.get('stage')}:** {row.get('cnt')} deals · €{float(row.get('total') or 0):,.0f}") if not data.get("deals_by_stage"): lines.append("- Geen deals in database.") 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 '-'})") fresh_appr = _fresh_approvals(data.get("pending_approval_requests") or []) if fresh_appr: lines.extend(["", "## Open beslissingen"]) for row in fresh_appr[:8]: lines.append(f"- **{row.get('agent_key')}**: {row.get('title')}") return "\n".join(lines) async def _ai_executive_summary(data: dict[str, Any]) -> str: opp_lines = "" for row in data.get("top_opportunities") or []: opp_lines += f"- {row.get('chain')} {row.get('name')} ({row.get('city')}): score {round(float(row.get('halal_opportunity_score') or 0))}\n" ms_lines = "" for row in data.get("milestones_pending") or []: ms_lines += f"- {row.get('title')} ({row.get('chain') or 'CRM'}) deadline {row.get('target_date') or '?'}\n" news_lines = "" for row in (data.get("trending_food") or [])[:5]: title = str(row.get("title") or "").strip() if title: news_lines += f"- {title}\n" fresh = _fresh_approvals(data.get("pending_approval_requests") or []) decision_lines = "" for row in fresh[:5]: decision_lines += f"- @{row.get('agent_key')}: {row.get('title') or row.get('action_type')}\n" prompt = ( "Schrijf een strak CEO-memo in het Nederlands (markdown) voor Aïssa (Foodlinkk / Cucina, " "halal kant-en-klaar). Geen IT, geen SysOps, geen NAS, geen agent-mesh.\n\n" "## Stand van zaken\n(4-6 zinnen: business context vandaag)\n\n" "## Pipeline & revenue\n(bullets met cijfers en wat dat betekent)\n\n" "## Retail & partnerships\n(supermarkten, milestones, halal-kansen)\n\n" "## Markt\n(max 3 relevante food/retail signalen)\n\n" "## Beslissingen vandaag\n(3-5 concrete CEO-acties)\n\n" f"Data ({data['date']}):\n" f"- Pipeline €{data['pipeline_eur']:,.0f}, {data['clients']} klanten, {data['deals']} deals\n" f"- {data.get('supermarkets',0)} filialen, {data.get('crm_partnerships',0)} partnerships\n" f"Top kansen:\n{opp_lines or '- geen data'}\n" f"Milestones:\n{ms_lines or '- geen'}\n" f"Markt:\n{news_lines or '- geen'}\n" f"Open beslissingen:\n{decision_lines or '- geen'}\n" ) system = ( "Je bent Herman, AI co-CEO van Foodlinkk. Schrijf als een executive briefing: " "helder, zakelijk, actionable. Focus op omzet, retail, partnerships en keuzes. " "Noem geen servers, backups, Gitea, NAS of SysOps. " "Antwoord UITSLUITEND met de markdown-memo. Geen voorwoord, geen bestandsnamen, " "geen vragen aan de gebruiker, geen tools of Telegram-aanbod." ) try: raw = await llm_router.generate(prompt, system=system, timeout=120.0) except Exception: return "" text = (raw or "").strip() for marker in ("## Stand van zaken", "## Samenvatting", "# "): idx = text.find(marker) if idx > 0: text = text[idx:] break cut_at = len(text) for m in ["Wil je dat ik", "Wil je deze", "Shall I", "Memo geschreven"]: idx = text.find(m) if 0 < idx < cut_at: cut_at = idx text = text[:cut_at].rstrip() return text.strip() return text.strip() def _fallback_summary(data: dict[str, Any]) -> str: digest = build_live_digest(data) lines = ["## Samenvatting", digest["summary"]] lines.extend(["", "## Actiepunten vandaag — korte termijn"]) for a in digest["actions"]: lines.append(f"- {a}") lines.extend(["", "## Lange termijn focus"]) for a in digest["long_term"]: lines.append(f"- {a}") return "\n".join(lines) def _save_briefing(content: str, data: dict[str, Any]) -> None: safe = serialize_stats(data) metadata = {"stats": safe, "model": getattr(settings, "HERMES_API_MODEL", None) or "hermes-agent", "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 llm_memory (category, subject, content, source, metadata, updated_at) VALUES ('herman_daily', %s, %s, 'herman', %s::jsonb, NOW())""", ( f"Briefing {data['date']}", content[:8000], json.dumps({"date": data["date"], "activity_count": len(data.get("activity_log") or [])}), ), ) 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, "activity_log": data.get("activity_log", [])}), ), ) 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), timeout=120.0) except (asyncio.TimeoutError, Exception): ai_part = "" if ai_part and len(ai_part.strip()) > 80: content = ai_part.strip() + "\n\n---\n\n" + template else: content = _fallback_summary(data) + "\n\n---\n\n" + template _save_briefing(content, data) stats = serialize_stats(data) stats["live_digest"] = build_live_digest(data) return content, stats