5d60d33db1
Volledige Foodlinkk Command Center uitbreiding met social automatisering, reclamefolder filters, Proxmox monitoring en documentatie.
269 lines
9.5 KiB
Python
269 lines
9.5 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 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)
|