Aissa
4dd1b8265f
Ship export intel with contact filters and CRM push, Herman live digest with Telegram briefing, halal recommendation engine, revenue cockpit, and dashboard polling/WebSocket fixes.
873 lines
35 KiB
Python
873 lines
35 KiB
Python
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 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, ros.halal_opportunity_score
|
|
FROM retail_opportunity_scores ros
|
|
JOIN supermarkets s ON s.id = ros.supermarket_id
|
|
ORDER BY ros.halal_opportunity_score DESC LIMIT 5"""
|
|
)
|
|
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 collect_briefing_data() -> 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)
|
|
data["activity_log"] = _build_activity_log(data)
|
|
return data
|
|
|
|
|
|
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", "PostgreSQL", "crm", "running", "Ophalen klanten, deals & pipeline uit CRM database…")
|
|
_collect_crm_core(data)
|
|
yield _briefing_step(
|
|
"crm", "PostgreSQL", "crm", "ok",
|
|
f"{data['clients']} klanten · {data['deals']} deals · pipeline €{data['pipeline_eur']:,.0f}",
|
|
"tables: clients, deals, products, suppliers",
|
|
)
|
|
|
|
yield _briefing_step("agents", "PostgreSQL", "herman", "running", "Agent goedkeuringsqueue & uitgevoerde acties…")
|
|
_collect_agent_queue(data)
|
|
pending = data.get("pending_approval_requests") or []
|
|
yield _briefing_step(
|
|
"agents", "PostgreSQL", "herman", "ok",
|
|
f"{len(pending)} open goedkeuringen · {len(data.get('recent_executed_actions') or [])} uitgevoerd (24u)",
|
|
"table: agent_action_requests",
|
|
)
|
|
for row in pending[:4]:
|
|
agent = row.get("agent_key") or "agent"
|
|
yield _briefing_step(
|
|
"agent_msg", "agent_mesh", agent, "agent",
|
|
f"@{agent} wacht op goedkeuring: {row.get('title') or row.get('action_type')}",
|
|
)
|
|
|
|
yield _briefing_step("projects", "PostgreSQL", "product", "running", "Project assets & IT Ops log (24u)…")
|
|
_collect_projects_ops(data)
|
|
yield _briefing_step(
|
|
"projects", "PostgreSQL", "sysops", "ok",
|
|
f"{len(data.get('project_assets_recent') or [])} project assets · "
|
|
f"{len(data.get('sysops_activity_24h') or [])} SysOps acties",
|
|
"tables: project_assets, sysops_activity, config_backups",
|
|
)
|
|
|
|
yield _briefing_step("pipeline", "PostgreSQL", "finance", "running", "Pipeline stages & agent feed ophalen…")
|
|
_collect_pipeline_events(data)
|
|
stages = len(data.get("deals_by_stage") or [])
|
|
events = data.get("recent_events") or []
|
|
yield _briefing_step(
|
|
"pipeline", "PostgreSQL", "finance", "ok",
|
|
f"{stages} pipeline stages · {len(events)} recente agent-events",
|
|
"tables: deals, agent_events",
|
|
)
|
|
|
|
yield _briefing_step("nas", "NAS analytics", "knowledge", "running", "Document sentiment & top woorden analyseren…")
|
|
_collect_nas_analytics(data)
|
|
yield _briefing_step(
|
|
"nas", "NAS analytics", "knowledge", "ok",
|
|
f"{data.get('nas_docs', 0)} documenten · sentiment {data.get('nas_sentiment', 0):.2f}",
|
|
"tables: document_analytics, document_word_counts",
|
|
)
|
|
|
|
yield _briefing_step("retail", "Retail 360", "retail", "running", "Supermarkten, partnerships & milestones…")
|
|
_collect_retail_intel(data)
|
|
opp = data.get("top_opportunities") or []
|
|
yield _briefing_step(
|
|
"retail", "Retail 360", "retail", "ok",
|
|
f"{data.get('supermarkets', 0)} supermarkten · {data.get('crm_partnerships', 0)} partnerships · "
|
|
f"{len(data.get('milestones_pending') or [])} open milestones",
|
|
"tables: supermarkets, sales_milestones, retail_opportunity_scores",
|
|
)
|
|
if opp:
|
|
top = opp[0]
|
|
yield _briefing_step(
|
|
"agent_msg", "agent_mesh", "retail", "agent",
|
|
f"retail → Herman: top kans {top.get('chain')} {top.get('name')} ({top.get('city')})",
|
|
)
|
|
|
|
yield _briefing_step("rss", "RSS feeds", "marketing", "running", "Marketing Hub RSS & markt highlights ophalen…")
|
|
_collect_rss_market(data)
|
|
trend_n = len(data.get("trending_food") or [])
|
|
yield _briefing_step(
|
|
"rss", "RSS feeds", "marketing", "ok",
|
|
f"{data.get('rss_items', 0)} RSS items · {trend_n} food trends · {data.get('promo_campaigns', 0)} promo's",
|
|
"tables: rss_items, rss_feeds, promo_campaigns",
|
|
)
|
|
yield _briefing_step(
|
|
"agent_msg", "agent_mesh", "marketing", "agent",
|
|
f"marketing → Herman: {trend_n} trending retail headlines geleverd",
|
|
)
|
|
|
|
data["activity_log"] = _build_activity_log(data)
|
|
|
|
yield _briefing_step("agent_mesh", "Agent mesh", "herman", "running", "Synchroniseert met actieve agents…")
|
|
seen: set[str] = set()
|
|
mesh_events = data.get("recent_events") or []
|
|
for ev in mesh_events[:10]:
|
|
agent = (ev.get("agent_name") or "agent").lower()
|
|
if agent in seen:
|
|
continue
|
|
seen.add(agent)
|
|
yield _briefing_step(
|
|
"agent_msg", "agent_mesh", agent, "agent",
|
|
f"@{agent}: {ev.get('title') or ev.get('event_type')}",
|
|
)
|
|
for agent_key in ("bizdev", "finance", "sourcing", "halal", "packaging", "hr"):
|
|
yield _briefing_step(
|
|
"agent_msg", "agent_mesh", "herman", "agent",
|
|
f"Herman → {agent_key}: briefing context gedeeld",
|
|
)
|
|
yield _briefing_step(
|
|
"agent_mesh", "Agent mesh", "herman", "ok",
|
|
f"{len(seen)} agents met live activiteit · activity log {len(data.get('activity_log') or [])} regels",
|
|
)
|
|
|
|
yield _briefing_step(
|
|
"ai", f"Ollama ({settings.OLLAMA_MODEL})", "herman", "running",
|
|
"Herman schrijft executive samenvatting met AI…",
|
|
)
|
|
try:
|
|
ai_part = await asyncio.wait_for(_ai_executive_summary(data), timeout=25.0)
|
|
except (asyncio.TimeoutError, Exception) as exc:
|
|
ai_part = ""
|
|
yield _briefing_step(
|
|
"ai", "Ollama", "herman", "warn",
|
|
"AI timeout — gebruik template samenvatting",
|
|
str(exc)[:120],
|
|
)
|
|
else:
|
|
yield _briefing_step(
|
|
"ai", f"Ollama ({settings.OLLAMA_MODEL})", "herman", "ok",
|
|
f"Samenvatting 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 _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 build_live_digest(data: dict[str, Any]) -> dict[str, Any]:
|
|
"""Live samenvatting zonder LLM — volgt cockpit-activiteit (polling / websocket)."""
|
|
activity = data.get("activity_log") or []
|
|
opp = data.get("top_opportunities") or []
|
|
ms = data.get("milestones_pending") or []
|
|
pending_n = int(data.get("pending_approvals") or 0)
|
|
executed = data.get("recent_executed_actions") or []
|
|
sysops = data.get("sysops_activity_24h") or []
|
|
|
|
summary_parts = [
|
|
(
|
|
f"Vandaag ({data['date']}) staat er €{data['pipeline_eur']:,.0f} in je pipeline, "
|
|
f"{data.get('crm_partnerships', 0)} actieve supermarkt-partnerships en "
|
|
f"{pending_n} open goedkeuring{'en' if pending_n != 1 else ''}."
|
|
)
|
|
]
|
|
|
|
if activity:
|
|
recent = [_strip_activity_prefix(a) for a in activity[:4]]
|
|
summary_parts.append("Recent in het Cockpit: " + "; ".join(recent[:3]) + ".")
|
|
elif executed:
|
|
row = executed[0]
|
|
summary_parts.append(
|
|
f"Laatste uitgevoerde actie: {row.get('agent_key')} — {row.get('title')}."
|
|
)
|
|
elif sysops:
|
|
row = sysops[0]
|
|
summary_parts.append(f"Laatste SysOps: {row.get('title')}.")
|
|
elif data.get("recent_events"):
|
|
ev = data["recent_events"][0]
|
|
summary_parts.append(
|
|
f"Laatste agent-event: {ev.get('agent_name')} — {ev.get('title')}."
|
|
)
|
|
else:
|
|
summary_parts.append(
|
|
"Nog geen nieuwe activiteit vandaag — gebruik Voice, Agents of Export Intel om Herman te voeden."
|
|
)
|
|
|
|
if opp:
|
|
top = opp[0]
|
|
summary_parts.append(
|
|
f"Top retail-kans: {top.get('chain')} · {top.get('name')} in {top.get('city')} "
|
|
f"(score {round(float(top.get('halal_opportunity_score') or 0))}/100)."
|
|
)
|
|
|
|
actions: list[str] = []
|
|
for row in data.get("pending_approval_requests") or []:
|
|
label = row.get("title") or row.get("action_type") or "goedkeuring"
|
|
actions.append(f"Keur goed: @{row.get('agent_key')} — {label}")
|
|
for row in executed[:2]:
|
|
actions.append(f"Follow-up na {row.get('agent_key')}: {row.get('title')}")
|
|
for row in ms[:2]:
|
|
actions.append(f"Milestone: {row.get('title')} ({row.get('chain') or 'CRM'})")
|
|
if pending_n and not actions:
|
|
actions.append(f"Behandel {pending_n} open agent-goedkeuring(en) op het dashboard")
|
|
if not actions:
|
|
actions.extend([
|
|
"Open Retail 360 voor top halal-gap filialen",
|
|
"Check Marketing Live Feed voor kant-en-klaar trends",
|
|
])
|
|
|
|
long_term = [
|
|
"Schaal CRM partnerships van proposal naar actief in top-10 kans-filialen",
|
|
"Halal kant-en-klaar listing bij regio's met hoogste demografische vraag",
|
|
"Wekelijks milestones review in Retail 360",
|
|
]
|
|
if sysops:
|
|
long_term.insert(0, f"IT Ops: {len(sysops)} SysOps-acties in de laatste 24 uur vastleggen en reviewen")
|
|
|
|
return {
|
|
"summary": " ".join(summary_parts),
|
|
"actions": actions[:6],
|
|
"long_term": long_term[:5],
|
|
"updated_at": data.get("generated_at"),
|
|
"activity_count": len(activity),
|
|
"source": "live",
|
|
}
|
|
|
|
|
|
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']} · **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 '-'})")
|
|
|
|
if data.get("sysops_activity_24h"):
|
|
lines.extend(["", "## SysOps IT — laatste 24 uur"])
|
|
for row in data["sysops_activity_24h"][:12]:
|
|
ts = row.get("created_at")
|
|
ts_s = ts.isoformat()[11:16] if hasattr(ts, "isoformat") else ""
|
|
cref = f" · commit `{row.get('commit_ref')}`" if row.get("commit_ref") else ""
|
|
lines.append(f"- [{ts_s}] **{row.get('title')}**{cref}")
|
|
if row.get("body"):
|
|
lines.append(f" {str(row.get('body'))[:200]}")
|
|
|
|
if data.get("pending_approval_requests"):
|
|
lines.extend(["", "## ⏳ Wacht op jouw goedkeuring (agents)"])
|
|
for row in data["pending_approval_requests"]:
|
|
action = row.get("action_type") or ""
|
|
label = "Update-scan VM106" if action == "maintenance_scan" else action
|
|
lines.append(f"- **{row.get('agent_key')}** · {label}: {row.get('title')}")
|
|
|
|
if data.get("activity_log"):
|
|
lines.extend(["", "## Herman activiteitenlog (24u)"])
|
|
for entry in data["activity_log"][:20]:
|
|
lines.append(f"- {entry}")
|
|
|
|
if data.get("recent_handoffs"):
|
|
lines.extend(["", "## Agent samenwerking (handoffs 24u)"])
|
|
for row in data["recent_handoffs"][:15]:
|
|
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}"
|
|
)
|
|
|
|
if data.get("pending_items"):
|
|
lines.extend(["", "## Legacy goedkeuringen"])
|
|
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]) -> 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"
|
|
|
|
activity = "\n".join((data.get("activity_log") or [])[:15]) or "- Geen recente agent-acties"
|
|
|
|
prompt = (
|
|
"Schrijf in het Nederlands (markdown) voor CEO Aïssa van Foodlinkk (halal kant-en-klaar maaltijden):\n\n"
|
|
"## Samenvatting\n(5-7 zinnen: pipeline, retail, IT ops, agent activiteit vandaag)\n\n"
|
|
"## Actiepunten vandaag — korte termijn\n(minimaal 5 bullets — incl. open goedkeuringen SysOps scan/backup)\n\n"
|
|
"## Lange termijn focus\n(3-5 bullets)\n\n"
|
|
"## Herman documentatie — wat er gebeurde\n(korte chronologische samenvatting van agent-acties, project assets, backups)\n\n"
|
|
f"Data vandaag ({data['date']}):\n"
|
|
f"- Pipeline €{data['pipeline_eur']:,.0f}, {data['clients']} klanten, {data['deals']} deals\n"
|
|
f"- {data.get('supermarkets',0)} supermarkten, {data.get('crm_partnerships',0)} actieve CRM partnerships\n"
|
|
f"- {data['pending_approvals']} goedkeuringen open in approval queue\n"
|
|
f"Activiteitenlog:\n{activity}\n"
|
|
f"Top kansen:\n{opp_lines or '- geen data'}\n"
|
|
f"Milestones open:\n{ms_lines or '- geen milestones'}\n"
|
|
)
|
|
system = (
|
|
"Je bent Herman, AI co-CEO van Foodlinkk. Documenteer en vat samen wat agents en IT hebben gedaan. "
|
|
"Noem expliciet openstaande SysOps scan/backup verzoeken als die in de log staan. "
|
|
"Schrijf warm, professioneel, actionable."
|
|
)
|
|
try:
|
|
return await llm_router.generate(prompt, system=system, timeout=120.0)
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
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}")
|
|
activity = data.get("activity_log") or []
|
|
if activity:
|
|
lines.extend(["", "## Herman documentatie — wat er gebeurde"])
|
|
for entry in activity[:8]:
|
|
lines.append(f"- {_strip_activity_prefix(entry)}")
|
|
return "\n".join(lines)
|
|
|
|
|
|
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 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=25.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
|