# ==============================================================================
# Author: Simon-Pierre Boucher <contact@spboucher.ai>
# File:   restoka/stats.py
# Desc:   Tableau de bord analytique — contrat commun Groupe KA v2 (ka-ui/
#         stats/SPEC.md §2). Agrège la DB réelle (restaurants, menus,
#         item_price_log, sync_log, details RACJ/Yelp/MAPAQ) : KPI avec deltas
#         et sparklines, jauges de complétude, séries quotidiennes, multi-
#         courbes (prix moyen par ville), barres empilées (ajouts par source),
#         répartitions, distributions, géographie, heatmaps calendrier +
#         horaire, tableaux et records. AUCUNE stat inventée : une mesure
#         indisponible est simplement absente du JSON.
#         Cache mémoire 5 min par période.
# ==============================================================================
from __future__ import annotations

import json
import statistics
import threading
import time
from collections import Counter
from datetime import date, datetime, timedelta
from zoneinfo import ZoneInfo

from . import db

TZ = ZoneInfo("America/Toronto")
CACHE_TTL = 300  # ≥ 5 min (SPEC.md §2)

_cache: dict[tuple, tuple[float, dict]] = {}
_cache_lock = threading.Lock()

# libellés FR compacts pour cuisines/types (sous-ensemble de api.ts)
_CUISINE_LABELS = {
    "cafe-dessert": "Café & desserts", "autre": "Autre", "burgers": "Burgers",
    "pizza": "Pizza", "fast-food": "Restauration rapide", "poulet": "Poulet",
    "quebecois": "Québécois", "sushi-japonais": "Sushi & japonais",
    "italien": "Italien", "bbq-grillades": "BBQ & grillades",
    "chinois": "Chinois", "dejeuner-brunch": "Déjeuner & brunch",
    "mexicain": "Mexicain", "libanais-moyen-orient": "Libanais & M-O",
    "thai": "Thaï", "indien": "Indien", "grec": "Grec",
    "vietnamien": "Vietnamien", "coreen": "Coréen", "francais": "Français",
    "fruits-de-mer": "Fruits de mer", "vegetarien": "Végétarien",
}
_TYPE_LABELS = {
    "restaurant": "Restaurant", "fast-food": "Restauration rapide",
    "cafe": "Café", "bar": "Bar", "boulangerie-patisserie":
    "Boulangerie-pâtisserie", "casse-croute": "Casse-croûte",
    "traiteur": "Traiteur", "creme-glacee": "Crème glacée",
}
_CTX_LABELS = {"dine-in": "En salle", "takeout": "Pour emporter",
               "delivery": "Livraison"}
_PRICE_BINS = [
    ("< 5 $", 0, 5), ("5-10 $", 5, 10), ("10-15 $", 10, 15),
    ("15-20 $", 15, 20), ("20-25 $", 20, 25), ("25-30 $", 25, 30),
    ("30-40 $", 30, 40), ("40-50 $", 40, 50), ("50 $ +", 50, 1e9),
]
_CAP_BINS = [
    ("1-50", 1, 50), ("50-100", 50, 100), ("100-200", 100, 200),
    ("200-300", 200, 300), ("300-500", 300, 500), ("500 +", 500, 1e9),
]
_YELP_BINS = [
    ("< 3", 0, 3), ("3 à 3,5", 3, 3.5), ("3,5 à 4", 3.5, 4),
    ("4 à 4,5", 4, 4.5), ("4,5 à 5", 4.5, 5.01),
]

PERIOD_LABELS = {
    "auj": "Aujourd'hui", "7j": "7 jours", "30j": "30 jours",
    "3m": "3 mois", "6m": "6 mois", "12m": "12 mois",
    "annee": "Année en cours", "tout": "Toute la période",
}


def _day(ts: float) -> str:
    return datetime.fromtimestamp(ts, TZ).strftime("%Y-%m-%d")


def _epoch(d: date, end: bool = False) -> float:
    dt = datetime(d.year, d.month, d.day, tzinfo=TZ)
    if end:
        dt += timedelta(days=1)
    return dt.timestamp()


def _resolve_period(con, period: str, dfrom: str | None,
                    dto: str | None) -> tuple[date, date, str]:
    """(from, to, label) — bornes inclusives en dates locales."""
    today = datetime.now(TZ).date()
    if dfrom and dto:
        try:
            f = date.fromisoformat(dfrom)
            t = date.fromisoformat(dto)
            if f <= t:
                return f, t, f"du {f} au {t}"
        except ValueError:
            pass
    days = {"7j": 7, "30j": 30, "3m": 91, "6m": 182, "12m": 365}
    if period == "auj":
        return today, today, PERIOD_LABELS["auj"]
    if period in days:
        return today - timedelta(days=days[period] - 1), today, \
            PERIOD_LABELS[period]
    if period == "annee":
        return date(today.year, 1, 1), today, f"Année {today.year}"
    # tout : depuis la première fiche référencée
    row = con.execute("SELECT MIN(first_seen) m FROM restaurants").fetchone()
    start = date.fromtimestamp(row["m"]) if row and row["m"] else today
    return start, today, PERIOD_LABELS["tout"]


def _delta(cur: float, prev: float) -> float | None:
    if not prev:
        return None
    return round(100.0 * (cur - prev) / prev, 1)


def _kpi(id_, label, value, unit="", delta_pct=None, positive_is_up=True,
         spark=None):
    k = {"id": id_, "label": label, "value": value, "unit": unit,
         "delta_pct": delta_pct}
    if delta_pct is not None:
        up = delta_pct >= 0 if positive_is_up else delta_pct < 0
        k["direction"] = "up" if up else "down"
    if spark and len(spark) >= 2:
        k["spark"] = spark
    return k


def _spark(pts: list[dict], n: int = 30) -> list[dict]:
    """Sous-échantillonne une série quotidienne pour la sparkline (≤ n pts)."""
    if len(pts) <= n:
        return pts
    step = max(1, len(pts) // n)
    out = pts[::step]
    if out[-1] is not pts[-1]:
        out.append(pts[-1])
    return out


def _pct(part: int, total: int) -> float:
    return round(100.0 * part / total, 1) if total else 0.0


def _fr(n: float, dec: int = 2) -> str:
    return f"{n:.{dec}f}".replace(".", ",")


def _daily(rows: list, f: date, t: date, cumulative: bool = False,
           base: int = 0) -> list[dict]:
    """Série quotidienne bouchée à zéro sur [f, t] à partir de {jour: n}."""
    by_day = dict(rows)
    n_days = (t - f).days + 1
    pts, acc = [], base
    step = max(1, n_days // 366)  # plafonne le nombre de points
    d = f
    while d <= t:
        v = 0
        for k in range(step):
            v += by_day.get((d + timedelta(days=k)).isoformat(), 0)
        acc += v
        pts.append({"t": d.isoformat(), "v": acc if cumulative else v})
        d += timedelta(days=step)
    if pts and pts[-1]["t"] != t.isoformat():
        pts.append({"t": t.isoformat(), "v": acc if cumulative else 0})
    return pts


def _iter_menu_items(con):
    for r in con.execute(
            "SELECT m.sections, r.chain, r.name FROM menus m"
            " JOIN restaurants r ON r.uid=m.uid"
            " WHERE r.active=1 AND r.dup_of IS NULL"):
        try:
            sections = json.loads(r["sections"] or "[]")
        except ValueError:
            continue
        for sec in sections:
            for it in sec.get("items") or []:
                yield r, it


def _bin_counts(values: list[float], bins) -> list[dict]:
    counts = Counter()
    for v in values:
        for lbl, lo, hi in bins:
            if lo <= v < hi:
                counts[lbl] += 1
                break
    return [{"label": lbl, "value": counts[lbl]}
            for lbl, _, _ in bins if counts[lbl]]


def _build(period: str, dfrom: str | None, dto: str | None) -> dict:
    con = db.connect()
    try:
        f, t, label = _resolve_period(con, period, dfrom, dto)
        f_ts, t_ts = _epoch(f), _epoch(t, end=True)
        span = (t - f).days + 1
        pf, pt = f - timedelta(days=span), f - timedelta(days=1)
        pf_ts, pt_ts = _epoch(pf), _epoch(pt, end=True)

        A = "active=1 AND dup_of IS NULL"      # restos comptés partout

        # ------------------------------------------------------------ KPI ---
        total = con.execute(
            f"SELECT COUNT(*) n FROM restaurants WHERE {A}").fetchone()["n"]
        # proxy de stock par first_seen (croissance sur la période)
        stock_end = con.execute(
            f"SELECT COUNT(*) n FROM restaurants WHERE {A} AND first_seen<?",
            (t_ts,)).fetchone()["n"]
        stock_start = con.execute(
            f"SELECT COUNT(*) n FROM restaurants WHERE {A} AND first_seen<?",
            (f_ts,)).fetchone()["n"]

        with_menu = con.execute(
            f"SELECT COUNT(*) n FROM restaurants WHERE {A} AND EXISTS"
            " (SELECT 1 FROM menus m WHERE m.uid=restaurants.uid)"
        ).fetchone()["n"]

        items = con.execute(
            "SELECT COALESCE(SUM(m.item_count),0) n FROM menus m"
            " JOIN restaurants r ON r.uid=m.uid"
            " WHERE r.active=1 AND r.dup_of IS NULL").fetchone()["n"]

        chains = con.execute(
            f"SELECT COUNT(DISTINCT chain) n FROM restaurants WHERE {A}"
            " AND chain IS NOT NULL").fetchone()["n"]
        cities = con.execute(
            f"SELECT COUNT(DISTINCT city) n FROM restaurants WHERE {A}"
            " AND city<>''").fetchone()["n"]

        new_cur = con.execute(
            f"SELECT COUNT(*) n FROM restaurants WHERE {A}"
            " AND first_seen>=? AND first_seen<?", (f_ts, t_ts)).fetchone()["n"]
        new_prev = con.execute(
            f"SELECT COUNT(*) n FROM restaurants WHERE {A}"
            " AND first_seen>=? AND first_seen<?", (pf_ts, pt_ts)).fetchone()["n"]

        closed_cur = con.execute(
            "SELECT COUNT(*) n FROM restaurants WHERE dup_of IS NULL"
            " AND status IN ('temporarily_closed','closed')"
            " AND updated_at>=? AND updated_at<?", (f_ts, t_ts)).fetchone()["n"]
        closed_prev = con.execute(
            "SELECT COUNT(*) n FROM restaurants WHERE dup_of IS NULL"
            " AND status IN ('temporarily_closed','closed')"
            " AND updated_at>=? AND updated_at<?", (pf_ts, pt_ts)).fetchone()["n"]

        releves_cur = con.execute(
            "SELECT COUNT(*) n FROM item_price_log WHERE ts>=? AND ts<?",
            (f_ts, t_ts)).fetchone()["n"]
        releves_prev = con.execute(
            "SELECT COUNT(*) n FROM item_price_log WHERE ts>=? AND ts<?",
            (pf_ts, pt_ts)).fetchone()["n"]

        # -------------------------------------------- complétude des fiches ---
        comp = con.execute(
            f"""SELECT
                  SUM(lat IS NOT NULL AND lng IS NOT NULL) geo,
                  SUM(phone<>'') tel,
                  SUM(website<>'') web,
                  SUM(hours IS NOT NULL AND hours<>'' AND hours<>'{{}}') hrs,
                  SUM(price_range<>'') pr,
                  SUM(images IS NOT NULL AND images<>'' AND images<>'[]') img
                FROM restaurants WHERE {A}""").fetchone()

        # ------------------------- enrichissements (details : RACJ/Yelp/MAPAQ)
        permis_n = 0
        caps: list[float] = []
        max_cap: tuple[str | None, float] = (None, 0.0)
        yelp_ratings: list[float] = []
        best_yelp: tuple[str | None, float, int] = (None, 0.0, 0)
        mapaq_n = 0
        for r in con.execute(
                f"SELECT name, city, details FROM restaurants WHERE {A}"
                " AND details IS NOT NULL AND details<>''"):
            try:
                det = json.loads(r["details"])
            except ValueError:
                continue
            pa = det.get("permis_alcool")
            if pa:
                permis_n += 1
                cap = pa.get("capacite")
                if isinstance(cap, (int, float)) and cap > 0:
                    caps.append(float(cap))
                    if cap > max_cap[1]:
                        max_cap = (f"{r['name']} ({r['city']})", float(cap))
            y = det.get("yelp")
            if y and isinstance(y.get("rating"), (int, float)):
                yelp_ratings.append(float(y["rating"]))
                rc = int(y.get("review_count") or 0)
                if (y["rating"], rc) > (best_yelp[1], best_yelp[2]):
                    best_yelp = (f"{r['name']} ({r['city']})",
                                 float(y["rating"]), rc)
            if det.get("mapaq"):
                mapaq_n += 1

        # -------------------------------------------- séries quotidiennes ---
        new_by_day = [(r["d"], r["n"]) for r in con.execute(
            "SELECT date(first_seen,'unixepoch','localtime') d, COUNT(*) n"
            f" FROM restaurants WHERE {A} AND first_seen>=? AND first_seen<?"
            " GROUP BY d", (f_ts, t_ts))]
        # plats suivis : 1re apparition de chaque item dans l'historique
        items_first = [(r["d"], r["n"]) for r in con.execute(
            "SELECT date(m0,'unixepoch','localtime') d, COUNT(*) n FROM"
            " (SELECT MIN(ts) m0 FROM item_price_log"
            "  GROUP BY uid, price_context, item_key)"
            " WHERE m0>=? AND m0<? GROUP BY d", (f_ts, t_ts))]
        items_base = con.execute(
            "SELECT COUNT(*) n FROM (SELECT MIN(ts) m0 FROM item_price_log"
            " GROUP BY uid, price_context, item_key) WHERE m0<?",
            (f_ts,)).fetchone()["n"]
        releves_by_day = [(r["d"], r["n"]) for r in con.execute(
            "SELECT date(ts,'unixepoch','localtime') d, COUNT(*) n"
            " FROM item_price_log WHERE ts>=? AND ts<? GROUP BY d",
            (f_ts, t_ts))]
        updated_by_day = [(r["d"], r["n"]) for r in con.execute(
            "SELECT date(updated_at,'unixepoch','localtime') d, COUNT(*) n"
            f" FROM restaurants WHERE {A} AND updated_at>=? AND updated_at<?"
            " GROUP BY d", (f_ts, t_ts))]

        stock_pts = _daily(new_by_day, f, t, cumulative=True,
                           base=stock_start)
        new_pts = _daily(new_by_day, f, t)
        items_pts = _daily(items_first, f, t, cumulative=True,
                           base=items_base)
        releves_pts = _daily(releves_by_day, f, t)
        updated_pts = _daily(updated_by_day, f, t)

        kpis = [
            _kpi("total", "Restaurants référencés", total,
                 delta_pct=_delta(stock_end, stock_start),
                 spark=_spark(stock_pts)),
            _kpi("with_menu", "Restos avec menu & prix", with_menu),
            _kpi("items", "Plats & prix suivis", items,
                 spark=_spark(items_pts)),
            _kpi("releves", "Relevés de prix (période)", releves_cur,
                 delta_pct=_delta(releves_cur, releves_prev),
                 spark=_spark(releves_pts)),
            _kpi("chains", "Chaînes suivies", chains),
            _kpi("cities", "Villes couvertes", cities),
            _kpi("new", "Nouveaux référencés (période)", new_cur,
                 delta_pct=_delta(new_cur, new_prev),
                 spark=_spark(new_pts)),
            _kpi("closed", "Fermetures détectées (période)", closed_cur,
                 delta_pct=_delta(closed_cur, closed_prev),
                 positive_is_up=False),
            _kpi("permis", "Permis d'alcool reliés (RACJ)", permis_n),
            _kpi("yelp", "Notes Yelp reliées", len(yelp_ratings)),
        ]

        # ----------------------------------------------------------- jauges ---
        gauges = [
            {"id": "geo", "label": "Restos géolocalisés",
             "value": _pct(comp["geo"], total), "max": 100, "unit": "%"},
            {"id": "menu", "label": "Restos avec menu & prix",
             "value": _pct(with_menu, total), "max": 100, "unit": "%"},
            {"id": "tel", "label": "Fiches avec téléphone",
             "value": _pct(comp["tel"], total), "max": 100, "unit": "%"},
            {"id": "hours", "label": "Horaires structurés",
             "value": _pct(comp["hrs"], total), "max": 100, "unit": "%"},
            {"id": "web", "label": "Fiches avec site web",
             "value": _pct(comp["web"], total), "max": 100, "unit": "%"},
            {"id": "pr", "label": "Fourchette de prix estimée",
             "value": _pct(comp["pr"], total), "max": 100, "unit": "%"},
        ]

        # ----------------------------------------------------------- séries ---
        series = [
            {"id": "new_restos", "title": "Nouveaux restos référencés par jour",
             "unit": "restos", "kind": "line", "points": new_pts},
            {"id": "items_cum", "title": "Plats & prix suivis (cumul)",
             "unit": "plats", "kind": "area", "points": items_pts},
            {"id": "releves", "title": "Relevés de prix par jour",
             "unit": "relevés", "kind": "bar", "points": releves_pts},
            {"id": "updated", "title": "Fiches mises à jour par jour",
             "unit": "fiches", "kind": "bar", "points": updated_pts},
        ]

        # -------------------------------- multi-courbes : prix moyen / ville ---
        # jours de la période où des relevés existent (grille commune)
        grid_days = [r["d"] for r in con.execute(
            "SELECT DISTINCT date(ts,'unixepoch','localtime') d"
            " FROM item_price_log WHERE ts>=? AND ts<? AND price>0"
            " ORDER BY d", (f_ts, t_ts))]
        multiseries = []
        if len(grid_days) >= 2:
            city_day: dict[str, dict[str, float]] = {}
            city_tot: Counter = Counter()
            for r in con.execute(
                    "SELECT re.city c, date(l.ts,'unixepoch','localtime') d,"
                    " AVG(l.price) a, COUNT(*) n FROM item_price_log l"
                    " JOIN restaurants re ON re.uid=l.uid"
                    " WHERE l.ts>=? AND l.ts<? AND l.price>0 AND re.city<>''"
                    " GROUP BY c, d", (f_ts, t_ts)):
                city_day.setdefault(r["c"], {})[r["d"]] = round(r["a"], 2)
                city_tot[r["c"]] += r["n"]
            chosen = []
            for city, _n in city_tot.most_common():
                if all(d in city_day[city] for d in grid_days):
                    chosen.append(city)
                if len(chosen) == 4:
                    break
            if len(chosen) >= 2:
                multiseries.append({
                    "id": "prix_villes",
                    "title": "Prix moyen d'un plat relevé — grandes villes",
                    "unit": "$",
                    "series": [{"label": c, "points": [
                        {"t": d, "v": city_day[c][d]} for d in grid_days]}
                        for c in chosen]})

        # ------------------------------- empilé : éléments ajoutés / source ---
        src_added: dict[str, list] = {}
        for r in con.execute(
                "SELECT source s, date(ts,'unixepoch','localtime') d,"
                " SUM(added) n FROM sync_log WHERE ts>=? AND ts<?"
                " GROUP BY s, d", (f_ts, t_ts)):
            src_added.setdefault(r["s"], []).append((r["d"], r["n"] or 0))
        stacked = []
        if src_added:
            keys = sorted(src_added,
                          key=lambda s: -sum(n for _, n in src_added[s]))[:6]
            per_key = {k: _daily(src_added[k], f, t) for k in keys}
            grid = per_key[keys[0]]
            stacked.append({
                "id": "ajouts_sources",
                "title": "Éléments ajoutés par source (journal de sync)",
                "unit": "ajouts", "keys": keys,
                "points": [{"t": grid[i]["t"],
                            "values": [per_key[k][i]["v"] for k in keys]}
                           for i in range(len(grid))]})

        # ---------------------------------------------- répartitions (stock) ---
        cuisine_counts: Counter = Counter()
        for r in con.execute(
                f"SELECT cuisines FROM restaurants WHERE {A}"):
            for c in json.loads(r["cuisines"] or "[]"):
                cuisine_counts[c] += 1
        top_cuisines = [
            {"label": _CUISINE_LABELS.get(c, c.capitalize()), "value": n}
            for c, n in cuisine_counts.most_common(8)]

        max_item = (None, 0.0)           # (desc, prix) — record réel
        all_prices: list[float] = []
        for r, it in _iter_menu_items(con):
            p = it.get("price")
            if not isinstance(p, (int, float)) or p <= 0:
                continue
            all_prices.append(float(p))
            if p > max_item[1]:
                max_item = (f"{it.get('name')} — {r['chain'] or r['name']}", p)

        types = [{"label": _TYPE_LABELS.get(r["t"], r["t"] or "Autre"),
                  "value": r["n"]} for r in con.execute(
            f"SELECT establishment_type t, COUNT(*) n FROM restaurants"
            f" WHERE {A} AND establishment_type<>'' GROUP BY t"
            " ORDER BY n DESC")]

        ctx_items = [{"label": _CTX_LABELS.get(r["c"], r["c"]), "value": r["n"]}
                     for r in con.execute(
            "SELECT m.price_context c, COUNT(*) n FROM menus m"
            " JOIN restaurants r ON r.uid=m.uid"
            " WHERE r.active=1 AND r.dup_of IS NULL"
            " GROUP BY c ORDER BY n DESC")]

        gamme_items = [{"label": r["g"], "value": r["n"]} for r in con.execute(
            f"SELECT price_range g, COUNT(*) n FROM restaurants WHERE {A}"
            " AND price_range<>'' GROUP BY g ORDER BY LENGTH(g)")]

        breakdowns = [
            {"id": "cuisines", "title": "Restos par type de cuisine (top 8)",
             "kind": "donut", "items": top_cuisines},
            {"id": "contextes", "title": "Menus par contexte de prix",
             "kind": "donut", "items": ctx_items},
            {"id": "types", "title": "Par type d'établissement",
             "kind": "bar", "items": types},
        ]
        if gamme_items:
            breakdowns.append(
                {"id": "gammes",
                 "title": "Restos par fourchette de prix estimée",
                 "kind": "bar", "items": gamme_items})

        # ---------------------------------------------------- distributions ---
        distributions = []
        if all_prices:
            distributions.append(
                {"id": "prix_plats", "title": "Distribution des prix de plats",
                 "unit": "plats", "bins": _bin_counts(all_prices, _PRICE_BINS)})
        if caps:
            distributions.append(
                {"id": "capacites",
                 "title": "Capacité des salles (permis d'alcool RACJ)",
                 "unit": "restos", "bins": _bin_counts(caps, _CAP_BINS)})
        if yelp_ratings:
            distributions.append(
                {"id": "notes_yelp", "title": "Distribution des notes Yelp",
                 "unit": "restos",
                 "bins": _bin_counts(yelp_ratings, _YELP_BINS)})

        # -------------------------------------------------------------- géo ---
        geo = {"title": "Restos par région", "items": [
            {"label": r["region"], "value": r["n"]} for r in con.execute(
                f"SELECT region, COUNT(*) n FROM restaurants WHERE {A}"
                " AND region<>'' GROUP BY region ORDER BY n DESC")]}

        # ---------------------------------------------------------- heatmap ---
        heat = [{"date": r["d"], "value": r["n"]} for r in con.execute(
            "SELECT date(first_seen,'unixepoch','localtime') d, COUNT(*) n"
            f" FROM restaurants WHERE {A} GROUP BY d ORDER BY d")]
        heatmap = {"title": "Nouveaux restos référencés", "cells": heat}

        # ------------------------------------------- heatmap horaire (7×24) ---
        hourly_cells = [
            {"dow": r["w"], "hour": r["h"], "value": r["n"]}
            for r in con.execute(
                "SELECT (CAST(strftime('%w',ts,'unixepoch','localtime')"
                " AS INTEGER)+6)%7 w,"
                " CAST(strftime('%H',ts,'unixepoch','localtime') AS INTEGER) h,"
                " COUNT(*) n FROM item_price_log WHERE ts>=? AND ts<?"
                " GROUP BY w, h", (f_ts, t_ts))]
        hourly = ({"title": "Relevés de prix par heure", "cells": hourly_cells}
                  if hourly_cells else None)

        # --------------------------------------------------------- tableaux ---
        top_cities = [[r["city"], r["n"], r["wm"],
                       f"{100.0 * r['n'] / total:.1f} %".replace(".", ",")]
                      for r in con.execute(
            f"SELECT city, COUNT(*) n, SUM(EXISTS (SELECT 1 FROM menus m"
            f" WHERE m.uid=restaurants.uid)) wm FROM restaurants WHERE {A}"
            " AND city<>'' GROUP BY city ORDER BY n DESC LIMIT 25")]

        # top chaînes : succursales, plats du menu le plus complet, prix moyen
        chain_rows: dict[str, dict] = {}
        for r in con.execute(
                f"SELECT chain, COUNT(*) n FROM restaurants WHERE {A}"
                " AND chain IS NOT NULL GROUP BY chain"):
            chain_rows[r["chain"]] = {"locs": r["n"], "items": 0, "prices": []}
        for r in con.execute(
                "SELECT r.chain, m.item_count, m.sections FROM menus m"
                " JOIN restaurants r ON r.uid=m.uid"
                " WHERE r.active=1 AND r.dup_of IS NULL"
                " AND r.chain IS NOT NULL"):
            cr = chain_rows.get(r["chain"])
            if cr is None or (r["item_count"] or 0) <= cr["items"]:
                continue
            cr["items"] = r["item_count"] or 0
            try:
                secs = json.loads(r["sections"] or "[]")
            except ValueError:
                continue
            cr["prices"] = [it["price"] for s in secs
                            for it in s.get("items") or []
                            if isinstance(it.get("price"), (int, float))
                            and it["price"] > 0]
        top_chains = []
        for name, cr in sorted(chain_rows.items(),
                               key=lambda kv: -kv[1]["locs"])[:25]:
            avg = (f"{statistics.mean(cr['prices']):.2f} $".replace(".", ",")
                   if cr["prices"] else "—")
            top_chains.append([name, cr["locs"], cr["items"] or "—", avg])

        # top établissements par nombre de plats au menu
        seen_uids: set[str] = set()
        top_places = []
        for r in con.execute(
                "SELECT r.uid, r.name, r.city, m.price_context c,"
                " m.item_count ic, m.sections FROM menus m"
                " JOIN restaurants r ON r.uid=m.uid"
                " WHERE r.active=1 AND r.dup_of IS NULL"
                " ORDER BY m.item_count DESC LIMIT 60"):
            if r["uid"] in seen_uids:
                continue
            seen_uids.add(r["uid"])
            prices = []
            try:
                secs = json.loads(r["sections"] or "[]")
                prices = [it["price"] for s in secs
                          for it in s.get("items") or []
                          if isinstance(it.get("price"), (int, float))
                          and it["price"] > 0]
            except ValueError:
                pass
            avg = (f"{statistics.mean(prices):.2f} $".replace(".", ",")
                   if prices else "—")
            top_places.append([r["name"], r["city"] or "—",
                               _CTX_LABELS.get(r["c"], r["c"]),
                               r["ic"] or 0, avg])
            if len(top_places) == 25:
                break

        # sources & connecteurs : couverture + dernier sync
        src_restos = {r["source"]: (r["n"], r["wm"]) for r in con.execute(
            f"SELECT source, COUNT(*) n, SUM(EXISTS (SELECT 1 FROM menus m"
            f" WHERE m.uid=restaurants.uid)) wm FROM restaurants WHERE {A}"
            " GROUP BY source")}
        src_last: dict[str, dict] = {}
        for r in con.execute(
                "SELECT source, ts, added, ok, message FROM sync_log"
                " ORDER BY ts"):
            src_last[r["source"]] = dict(r)
        src_added_period = {r["source"]: r["n"] or 0 for r in con.execute(
            "SELECT source, SUM(added) n FROM sync_log"
            " WHERE ts>=? AND ts<? GROUP BY source", (f_ts, t_ts))}
        sources_rows = []
        for s in sorted(set(src_restos) | set(src_last),
                        key=lambda s: -(src_restos.get(s, (0, 0))[0])):
            n, wm = src_restos.get(s, (0, 0))
            last = src_last.get(s)
            when = (datetime.fromtimestamp(last["ts"], TZ)
                    .strftime("%Y-%m-%d %H:%M") if last else "—")
            state = ("—" if not last else
                     "ok" if last["ok"] and last["message"] == "ok"
                     else ("alerte" if last["ok"] else "échec"))
            sources_rows.append([s, n, wm or 0,
                                 src_added_period.get(s, 0), when, state])

        newest = [[r["name"], r["city"] or "—", r["region"] or "—",
                   _day(r["first_seen"])] for r in con.execute(
            f"SELECT name, city, region, first_seen FROM restaurants"
            f" WHERE {A} AND first_seen>=? AND first_seen<?"
            " ORDER BY first_seen DESC LIMIT 100", (f_ts, t_ts))]

        syncs = [[r["source"], datetime.fromtimestamp(r["ts"], TZ)
                  .strftime("%Y-%m-%d %H:%M"), r["found"], r["added"],
                  r["updated"], r["removed"],
                  "ok" if r["ok"] and r["message"] == "ok"
                  else ("alerte" if r["ok"] else "échec")]
                 for r in con.execute(
            "SELECT source, ts, found, added, updated, removed, ok, message"
            " FROM sync_log WHERE ts>=? AND ts<? ORDER BY ts DESC LIMIT 50",
            (f_ts, t_ts))]

        tables = [
            {"id": "villes", "title": "Top villes",
             "columns": ["Ville", "Restos", "Avec menu", "Part"],
             "rows": top_cities},
            {"id": "chaines", "title": "Top chaînes",
             "columns": ["Chaîne", "Succursales", "Plats au menu",
                         "Prix moyen"],
             "rows": top_chains},
        ]
        if top_places:
            tables.append(
                {"id": "etablissements",
                 "title": "Top établissements par plats au menu",
                 "columns": ["Restaurant", "Ville", "Contexte", "Plats",
                             "Prix moyen"],
                 "rows": top_places})
        if sources_rows:
            tables.append(
                {"id": "sources", "title": "Sources & connecteurs",
                 "columns": ["Source", "Restos actifs", "Avec menu",
                             "Ajoutés (période)", "Dernier sync", "État"],
                 "rows": sources_rows})
        if newest:
            tables.append(
                {"id": "nouveaux", "title": "Nouveaux restos de la période"
                 " (100 plus récents)",
                 "columns": ["Restaurant", "Ville", "Région", "Ajouté le"],
                 "rows": newest})
        if syncs:
            tables.append(
                {"id": "syncs", "title": "Journal des synchronisations",
                 "columns": ["Source", "Quand", "Trouvés", "Ajoutés",
                             "Mis à jour", "Retirés", "État"],
                 "rows": syncs})

        # ---------------------------------------------------------- records ---
        records = []
        rec_day = con.execute(
            "SELECT date(first_seen,'unixepoch','localtime') d, COUNT(*) n"
            f" FROM restaurants WHERE {A} GROUP BY d ORDER BY n DESC LIMIT 1"
        ).fetchone()
        if rec_day:
            records.append({"label": "Jour record de référencement",
                            "value": f"{rec_day['n']:,} restos".replace(",", " "),
                            "date": rec_day["d"]})
        big_menu = con.execute(
            "SELECT r.name, r.city, m.item_count ic FROM menus m"
            " JOIN restaurants r ON r.uid=m.uid"
            " WHERE r.active=1 AND r.dup_of IS NULL"
            " ORDER BY m.item_count DESC LIMIT 1").fetchone()
        if big_menu and big_menu["ic"]:
            records.append({"label": "Menu le plus étoffé",
                            "value": f"{big_menu['name']} ({big_menu['city']})"
                            f" — {big_menu['ic']} plats"})
        big_chain = max(chain_rows.items(), key=lambda kv: kv[1]["items"],
                        default=None)
        if big_chain and big_chain[1]["items"]:
            records.append({"label": "Chaîne au menu le plus étoffé",
                            "value": f"{big_chain[0]} — "
                            f"{big_chain[1]['items']} plats"})
        if top_cities:
            records.append({"label": "Ville la plus couverte",
                            "value": f"{top_cities[0][0]} — "
                            f"{top_cities[0][1]:,} restos".replace(",", " ")})
        if geo["items"]:
            g0 = geo["items"][0]
            records.append({"label": "Région la plus couverte",
                            "value": f"{g0['label']} — "
                            f"{g0['value']:,} restos".replace(",", " ")})
        if all_prices:
            med = statistics.median(all_prices)
            records.append({"label": "Prix médian d'un plat suivi",
                            "value": f"{med:.2f} $".replace(".", ",")})
        if max_item[0]:
            records.append({"label": "Plat le plus cher observé",
                            "value": f"{max_item[0][:44]} — "
                            f"{max_item[1]:.2f} $".replace(".", ",")})
        rec_price_day = con.execute(
            "SELECT date(ts,'unixepoch','localtime') d, COUNT(*) n"
            " FROM item_price_log GROUP BY d ORDER BY n DESC LIMIT 1"
        ).fetchone()
        if rec_price_day:
            records.append({"label": "Jour record de relevés de prix",
                            "value": f"{rec_price_day['n']:,} relevés"
                            .replace(",", " "),
                            "date": rec_price_day["d"]})
        if max_cap[0]:
            records.append({"label": "Plus grande salle (permis RACJ)",
                            "value": f"{max_cap[0][:44]} — "
                            f"{int(max_cap[1])} places"})
        if best_yelp[0]:
            records.append({"label": "Meilleure note Yelp reliée",
                            "value": f"{best_yelp[0][:40]} — "
                            f"{_fr(best_yelp[1], 1)}/5"
                            f" ({best_yelp[2]} avis)"})
        if src_added:
            bs_name, bs_rows = max(src_added.items(),
                                   key=lambda kv: sum(n for _, n in kv[1]))
            bs_tot = sum(n for _, n in bs_rows)
            if bs_tot:
                records.append(
                    {"label": "Source la plus productive (période)",
                     "value": f"{bs_name} — {bs_tot:,} ajouts"
                     .replace(",", " ")})
        if mapaq_n:
            records.append({"label": "Restos avec dossier MAPAQ relié",
                            "value": f"{mapaq_n} établissements"})

        out = {
            "updated": datetime.now(TZ).isoformat(timespec="seconds"),
            "period": {"from": f.isoformat(), "to": t.isoformat(),
                       "label": label},
            "kpis": kpis,
            "gauges": gauges,
            "series": series,
            "breakdowns": breakdowns,
            "geo": geo,
            "heatmap": heatmap,
            "tables": tables,
            "records": records,
        }
        if multiseries:
            out["multiseries"] = multiseries
        if stacked:
            out["stacked"] = stacked
        if distributions:
            out["distributions"] = distributions
        if hourly:
            out["hourly"] = hourly
        return out
    finally:
        con.close()


def dashboard(period: str = "30j", dfrom: str | None = None,
              dto: str | None = None) -> dict:
    key = (period, dfrom or "", dto or "")
    now = time.time()
    with _cache_lock:
        hit = _cache.get(key)
        if hit and now - hit[0] < CACHE_TTL:
            return hit[1]
    data = _build(period, dfrom, dto)
    with _cache_lock:
        _cache[key] = (time.time(), data)
    return data