Add live CEO briefing, Wereldexport CRM, halal engine, and realtime cockpit.

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.
2026-07-19 18:25:01 +00:00
parent 544d9e611d
commit 4dd1b8265f
25 changed files with 2757 additions and 108 deletions
@@ -285,37 +285,41 @@ def get_entity(entity_id: int) -> dict[str, Any]:
@router.get("/contacts")
 def list_contacts(
    country: Optional[str] = Query(None),
+    region: Optional[str] = Query(None),
    entity_type: Optional[str] = Query(None),
    has_email: Optional[bool] = Query(None),
+    crm_linked: Optional[bool] = Query(None),
    q: Optional[str] = Query(None),
    limit: int = Query(200, ge=1, le=500),
    offset: int = Query(0, ge=0),
 ) -> dict[str, Any]:
-    clauses = ["1=1"]
-    params: list[Any] = []
-    if country:
-        clauses.append("e.country_iso2 = %s")
-        params.append(country.upper())
-    if entity_type:
-        clauses.append("e.entity_type = %s")
-        params.append(entity_type)
-    if has_email:
-        clauses.append("c.email IS NOT NULL AND c.email <> ''")
+    join = ""
+    if region:
+        join = "JOIN export_territories t ON e.territory_code = t.code"
+    where_sql, params = _entity_filters(country, region, entity_type, None, False, crm_linked)
+    if has_email is True:
+        where_sql += " AND c.email IS NOT NULL AND c.email <> ''"
+    elif has_email is False:
+        where_sql += " AND (c.email IS NULL OR c.email = '')"
    if q:
-        clauses.append(
-            "(e.name ILIKE %s OR c.email ILIKE %s OR c.name ILIKE %s OR e.city ILIKE %s)"
-        )
        like = f"%{q}%"
-        params.extend([like, like, like, like])
+        where_sql += (
+            " AND (e.name ILIKE %s OR e.city ILIKE %s OR c.email ILIKE %s"
+            " OR c.name ILIKE %s OR c.phone ILIKE %s OR c.mobile ILIKE %s)"
+        )
+        params.extend([like, like, like, like, like, like])

-    where = " AND ".join(clauses)
    rows = fetch_all(
        f"""
-        SELECT c.*, e.name AS entity_name, e.entity_type, e.country_iso2, e.city AS entity_city
+        SELECT c.*, e.name AS entity_name, e.entity_type, e.country_iso2,
+               e.city AS entity_city, e.client_id, e.deal_id,
+               cl.name AS crm_client_name
        FROM export_entity_contacts c
        JOIN export_market_entities e ON e.id = c.entity_id
-        WHERE {where}
-        ORDER BY e.name, c.is_primary DESC
+        LEFT JOIN clients cl ON cl.id = e.client_id
+        {join}
+        WHERE {where_sql}
+        ORDER BY e.name, e.city NULLS LAST, c.is_primary DESC, c.email NULLS LAST
        LIMIT %s OFFSET %s
        """,
        tuple(params) + (limit, offset),
@@ -324,19 +328,33 @@ def list_contacts(
        f"""
        SELECT COUNT(*) AS n FROM export_entity_contacts c
        JOIN export_market_entities e ON e.id = c.entity_id
-        WHERE {where}
+        {join}
+        WHERE {where_sql}
        """,
        tuple(params),
    )
-    return {"items": rows, "total": int(total["n"] or 0)}
+    return {"items": rows, "total": int(total["n"] or 0), "limit": limit, "offset": offset}


@router.get("/contacts/export.csv")
 def export_contacts_csv(
    country: Optional[str] = Query(None),
+    region: Optional[str] = Query(None),
    entity_type: Optional[str] = Query(None),
+    has_email: Optional[bool] = Query(None),
+    crm_linked: Optional[bool] = Query(None),
+    q: Optional[str] = Query(None),
 ) -> StreamingResponse:
-    data = list_contacts(country=country, entity_type=entity_type, limit=5000, offset=0)
+    data = list_contacts(
+        country=country,
+        region=region,
+        entity_type=entity_type,
+        has_email=has_email,
+        crm_linked=crm_linked,
+        q=q,
+        limit=5000,
+        offset=0,
+    )
    buf = io.StringIO()
    writer = csv.writer(buf)
    writer.writerow(
@@ -0,0 +1,276 @@
+"""Live halal retail recommendation engine — diverse, actionable, data-driven."""
+from __future__ import annotations
+
+import json
+from datetime import date, datetime, timedelta, timezone
+from typing import Any
+
+from app import retail_opportunities
+from app.db import execute, execute_returning, fetch_all, fetch_one, json_param
+from app.middleware import log_agent_event
+
+MIN_SCORE = 35.0
+
+CANDIDATE_SQL = """
+    SELECT s.id, s.name, s.chain, s.city, s.province, s.postcode,
+           s.partnership_status, s.halal_certified, s.has_halal_section,
+           s.phone, s.email,
+           ros.halal_opportunity_score, ros.market_potential_score, ros.factors, ros.computed_at,
+           (a.religious_composition->>'muslim_proxy_pct')::float AS muslim_proxy_pct,
+           (a.ethnic_composition->>'niet_westers_pct')::float AS niet_westers_pct,
+           a.population, a.avg_income,
+           sp.manager_email, sp.manager_phone,
+           EXISTS(SELECT 1 FROM client_supermarket_links csl WHERE csl.supermarket_id = s.id) AS crm_linked
+    FROM supermarkets s
+    JOIN retail_opportunity_scores ros ON ros.supermarket_id = s.id
+    LEFT JOIN area_analysis a ON a.postcode = s.postcode
+    LEFT JOIN supermarket_profiles sp ON sp.supermarket_id = s.id
+    WHERE COALESCE(s.partnership_status, 'none') NOT IN ('active', 'contract', 'won')
+      AND COALESCE(s.has_halal_section, false) IS NOT TRUE
+      AND ros.halal_opportunity_score >= %s
+      AND s.postcode <> '0000AA'
+    LIMIT 500
+"""
+
+
+def _parse_factors(row: dict[str, Any]) -> dict[str, Any]:
+    factors = row.get("factors") or {}
+    if isinstance(factors, str):
+        try:
+            factors = json.loads(factors)
+        except json.JSONDecodeError:
+            factors = {}
+    return factors if isinstance(factors, dict) else {}
+
+
+def _composite_score(row: dict[str, Any], day_seed: int) -> float:
+    halal = float(row.get("halal_opportunity_score") or 0)
+    market = float(row.get("market_potential_score") or 0)
+    muslim = float(row.get("muslim_proxy_pct") or 0)
+    factors = _parse_factors(row)
+    halal_gap = float(factors.get("halal_gap") or 0)
+
+    score = halal * 0.32 + halal_gap * 0.28 + muslim * 0.22 + market * 0.12
+    if row.get("phone") or row.get("email") or row.get("manager_email") or row.get("manager_phone"):
+        score += 8
+    if row.get("crm_linked"):
+        score -= 22
+    if row.get("halal_certified"):
+        score -= 18
+    pop = int(row.get("population") or 0)
+    if pop > 80000:
+        score += 5
+    jitter = ((int(row["id"]) * 17 + day_seed) % 11) - 5
+    return round(score + jitter, 2)
+
+
+def _reasons(row: dict[str, Any]) -> list[str]:
+    factors = _parse_factors(row)
+    muslim = float(row.get("muslim_proxy_pct") or factors.get("muslim_proxy_pct") or 0)
+    reasons: list[str] = []
+    if not row.get("has_halal_section"):
+        reasons.append(f"Geen halal schap — {muslim:.0f}% moslim-demografie")
+    gap = float(factors.get("halal_gap") or 0)
+    if gap >= 25:
+        reasons.append(f"Halal-gap {gap:.0f}/100")
+    pop = int(row.get("population") or 0)
+    if pop > 40000:
+        reasons.append(f"Catchment {pop:,} inwoners")
+    if row.get("phone") or row.get("manager_phone"):
+        reasons.append("Telefoon beschikbaar")
+    chain = row.get("chain") or ""
+    if chain in ("Jumbo", "Albert Heijn", "PLUS", "Dirk"):
+        reasons.append(f"Strategische keten: {chain}")
+    status = row.get("partnership_status") or "none"
+    if status in ("none", "prospect", "lead"):
+        reasons.append("Nog geen actief partnership — eerste mover")
+    return reasons[:4] or ["Halal kant-en-klaar listing kans"]
+
+
+def _diverse_pick(candidates: list[dict[str, Any]], limit: int) -> list[dict[str, Any]]:
+    picked: list[dict[str, Any]] = []
+    chains_seen: set[str] = set()
+    cities_seen: set[str] = set()
+    ids_seen: set[int] = set()
+    for row in candidates:
+        sid = int(row["id"])
+        if sid in ids_seen:
+            continue
+        chain = (row.get("chain") or "").lower()
+        city = (row.get("city") or "").lower()
+        name = (row.get("name") or "").lower()
+        chain_city = f"{chain}|{city}|{name[:20]}"
+        if chain_city in chains_seen and len(picked) < max(1, limit - 1):
+            continue
+        if chain in chains_seen and len(picked) < max(1, limit - 1):
+            continue
+        if city in cities_seen and len(cities_seen) >= 2 and len(picked) < max(1, limit - 1):
+            continue
+        picked.append(row)
+        ids_seen.add(sid)
+        chains_seen.add(chain)
+        chains_seen.add(chain_city)
+        cities_seen.add(city)
+        if len(picked) >= limit:
+            return picked
+    for row in candidates:
+        if row in picked:
+            continue
+        picked.append(row)
+        if len(picked) >= limit:
+            break
+    return picked
+
+
+def get_dismissed_store_ids() -> set[int]:
+    rows = fetch_all(
+        """SELECT related_entity_id FROM ai_recommendations
+           WHERE recommendation_type LIKE 'halal_%%' AND status = 'dismissed'
+             AND related_entity_id IS NOT NULL
+             AND updated_at >= NOW() - INTERVAL '30 days'"""
+    )
+    return {int(r["related_entity_id"]) for r in rows if r.get("related_entity_id")}
+
+
+def _serialize_row(row: dict[str, Any]) -> dict[str, Any]:
+    out: dict[str, Any] = {}
+    for k, v in row.items():
+        if k.startswith("_"):
+            continue
+        if hasattr(v, "isoformat"):
+            out[k] = v.isoformat()
+        elif type(v).__name__ == "Decimal":
+            out[k] = float(v)
+        else:
+            out[k] = v
+    return out
+
+
+def build_recommendation(row: dict[str, Any], rank: int) -> dict[str, Any]:
+    score = row.get("_composite_score", row.get("halal_opportunity_score"))
+    reasons = _reasons(row)
+    title = f"{row.get('chain')} · {row.get('name')} ({row.get('city')})"
+    priority = "high" if float(score) >= 62 else "medium" if float(score) >= 45 else "normal"
+    return {
+        "rank": rank,
+        "store_id": int(row["id"]),
+        "recommendation_type": "halal_retail_cucina",
+        "brand": "cucina",
+        "title": title,
+        "description": " · ".join(reasons),
+        "score": round(float(score), 1),
+        "halal_opportunity_score": float(row.get("halal_opportunity_score") or 0),
+        "market_potential_score": float(row.get("market_potential_score") or 0),
+        "muslim_proxy_pct": float(row.get("muslim_proxy_pct") or 0),
+        "priority": priority,
+        "reasons": reasons,
+        "chain": row.get("chain"),
+        "city": row.get("city"),
+        "province": row.get("province"),
+        "partnership_status": row.get("partnership_status"),
+        "actions": [
+            {"label": "Retail 360", "href": f"/retail?open={row['id']}"},
+            {"label": "Naar CRM", "href": f"/retail?open={row['id']}&crm=1"},
+        ],
+    }
+
+
+def _upsert_recommendation(item: dict[str, Any]) -> None:
+    store_id = item["store_id"]
+    title = item["title"][:255]
+    existing = fetch_one(
+        """SELECT id FROM ai_recommendations
+           WHERE recommendation_type = %s AND related_entity_id = %s AND status = 'pending'""",
+        (item["recommendation_type"], store_id),
+    )
+    payload = json_param({"score": item["score"], "reasons": item["reasons"]})
+    if existing:
+        execute(
+            """UPDATE ai_recommendations SET description = %s, impact_score = %s,
+               priority = %s, data_sources = %s::jsonb, updated_at = NOW() WHERE id = %s""",
+            (
+                item["description"][:2000],
+                min(0.99, float(item["score"]) / 100),
+                item["priority"],
+                payload,
+                existing["id"],
+            ),
+        )
+        return
+    execute_returning(
+        """INSERT INTO ai_recommendations
+           (recommendation_type, title, description, priority, impact_score, confidence_score,
+            data_sources, action_items, generated_by, related_entity_type, related_entity_id, status, expires_at)
+           VALUES (%s,%s,%s,%s,%s,%s,%s,%s,'halal_engine','supermarket',%s,'pending',%s)
+           RETURNING id""",
+        (
+            item["recommendation_type"],
+            title,
+            item["description"][:2000],
+            item["priority"],
+            min(0.99, float(item["score"]) / 100),
+            min(0.95, 0.55 + float(item.get("muslim_proxy_pct") or 0) / 200),
+            payload,
+            [a["label"] for a in item.get("actions", [])],
+            store_id,
+            datetime.now(timezone.utc) + timedelta(days=14),
+        ),
+    )
+
+
+def _maybe_refresh_scores(force: bool = False) -> bool:
+    """Full recompute is expensive (~60s) — only on explicit refresh or empty DB."""
+    if force:
+        retail_opportunities.compute_all_scores(5000)
+        return True
+    row = fetch_one("SELECT COUNT(*) AS n FROM retail_opportunity_scores")
+    if not row or int(row.get("n") or 0) == 0:
+        retail_opportunities.compute_all_scores(5000)
+        return True
+    return False
+
+
+def live_halal_recommendations(
+    limit: int = 5,
+    refresh_scores: bool = False,
+    brand: str = "cucina",
+) -> dict[str, Any]:
+    refreshed = _maybe_refresh_scores(refresh_scores)
+    day_seed = int(date.today().strftime("%Y%m%d"))
+    dismissed = get_dismissed_store_ids()
+
+    rows = fetch_all(CANDIDATE_SQL, (MIN_SCORE,))
+    scored: list[dict[str, Any]] = []
+    for r in rows:
+        sid = int(r["id"])
+        if sid in dismissed:
+            continue
+        r["_composite_score"] = _composite_score(r, day_seed)
+        scored.append(r)
+    scored.sort(key=lambda x: x["_composite_score"], reverse=True)
+    picked = _diverse_pick(scored, limit)
+    items = [build_recommendation(r, i + 1) for i, r in enumerate(picked)]
+
+    for item in items:
+        _upsert_recommendation(item)
+
+    if items:
+        log_agent_event(
+            agent_name="halal_engine",
+            event_type="recommendations",
+            title=f"Halal kansen: {len(items)} live recommendations",
+            metadata={"store_ids": [i["store_id"] for i in items], "top_score": items[0]["score"]},
+        )
+
+    return {
+        "ok": True,
+        "brand": brand,
+        "generated_at": datetime.now(timezone.utc).isoformat(),
+        "items": items,
+        "meta": {
+            "candidates_evaluated": len(rows),
+            "dismissed_skipped": len(dismissed),
+            "scores_refreshed": refreshed or refresh_scores,
+            "engine": "halal_reco_v1",
+        },
+    }
@@ -10,6 +10,7 @@ from typing import Any, Optional
 from fastapi import APIRouter, HTTPException, Query

 from app.db import execute, execute_returning, fetch_all, fetch_one, json_param
+from app.halal_reco_engine import live_halal_recommendations
 from app.middleware import log_agent_event

 router = APIRouter(prefix="/recommendations", tags=["recommendations"])
@@ -154,6 +155,21 @@ def approve_recommendation(rec_id: int) -> dict[str, Any]:
    return _serialize(row)


+@router.get("/halal/live")
+def halal_recommendations_live(
+    limit: int = Query(5, ge=1, le=20),
+    refresh: bool = False,
+    brand: str = Query("cucina"),
+) -> dict[str, Any]:
+    """Live halal retail kansen — divers, geen actieve partnerships, composite score."""
+    return live_halal_recommendations(limit=limit, refresh_scores=refresh, brand=brand)
+
+
+@router.post("/halal/refresh")
+def halal_recommendations_refresh(limit: int = Query(5, ge=1, le=20)) -> dict[str, Any]:
+    return live_halal_recommendations(limit=limit, refresh_scores=True)
+
+
@router.post("/{rec_id}/dismiss")
 def dismiss_recommendation(rec_id: int) -> dict[str, Any]:
    row = execute_returning(