# -----------------------------------------------------------------------------
# Food-Ka — Agrégateur de produits d'épicerie (province de Québec)
# Auteur : Simon-Pierre Boucher — contact@spboucher.ai
# statsdash.py : tableau de bord analytique — source unique de
#   GET /api/stats/dashboard (contrat commun ka-ui/stats/SPEC.md v2) et des
#   5 rapports PDF Groupe-KA (GET /api/stats/report, moteur foodka/kapdf.py).
#   Tout est calculé sur les données réelles : products (catalogue vivant),
#   price_log (historique des relevés de prix), sync_log (journal des syncs).
#   Rien d'inventé : une mesure indisponible = champ omis (section masquée).
#   Cache serveur : 5 minutes par période demandée.
# -----------------------------------------------------------------------------
from __future__ import annotations

import json
import statistics
import threading
import time
from datetime import date, datetime, timedelta
from pathlib import Path
from zoneinfo import ZoneInfo

from . import db

TZ = ZoneInfo("America/Toronto")
CACHE_TTL = 300  # secondes (SPEC : >= 5 min par période)

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

# période -> (libellé, nombre de jours) ; « annee » et « tout » sont calculés
_PERIOD_DAYS = {
    "auj": ("Aujourd'hui", 1),
    "7j": ("7 jours", 7),
    "30j": ("30 jours", 30),
    "3m": ("3 mois", 91),
    "6m": ("6 mois", 182),
    "12m": ("12 mois", 365),
}

_PRICE_SANE = "price IS NOT NULL AND price > 0 AND price <= 2000"

# Noms d'affichage des bannières (registre data/sources.json)
_SOURCES_PATH = Path(__file__).resolve().parent.parent / "data" / "sources.json"


def _source_registry() -> list[dict]:
    try:
        return json.loads(_SOURCES_PATH.read_text(encoding="utf-8"))["sources"]
    except Exception:
        return []


def _fmt_money(v: float | None) -> str:
    if v is None:
        return "—"
    return f"{v:,.2f}".replace(",", " ").replace(".", ",") + " $"


def _fmt_int(n: int) -> str:
    return f"{n:,}".replace(",", " ")


def _parse_date(s: str | None) -> date | None:
    if not s:
        return None
    try:
        return date.fromisoformat(s[:10])
    except ValueError:
        return None


def _day_start_ts(d: date) -> float:
    return datetime(d.year, d.month, d.day, tzinfo=TZ).timestamp()


def _resolve_period(con, period: str, from_s: str | None, to_s: str | None):
    """Retourne (from_date, to_date, label, period_id) — bornes inclusives."""
    today = datetime.now(TZ).date()
    f, t = _parse_date(from_s), _parse_date(to_s)
    if f and t:
        if t < f:
            f, t = t, f
        return f, min(t, today), f"du {f.isoformat()} au {t.isoformat()}", "perso"
    if period == "annee":
        return date(today.year, 1, 1), today, "Année en cours", period
    if period == "tout":
        row = con.execute("SELECT MIN(first_seen) m FROM products").fetchone()
        start = (datetime.fromtimestamp(row["m"], TZ).date()
                 if row and row["m"] else today)
        return start, today, "Toute la période", period
    label, days = _PERIOD_DAYS.get(period, _PERIOD_DAYS["30j"])
    if period not in _PERIOD_DAYS:
        label, period = _PERIOD_DAYS["30j"][0], "30j"
    return today - timedelta(days=days - 1), today, label, period


def _delta_pct(cur: float | None, prev: float | None) -> float | None:
    if cur is None or prev is None or prev == 0:
        return None
    return round(100 * (cur - prev) / prev, 1)


def _days_range(f: date, t: date) -> list[date]:
    n = (t - f).days + 1
    step = max(1, -(-n // 200))  # au plus ~200 points de série
    days = [f + timedelta(days=i) for i in range(0, n, step)]
    if days[-1] != t:
        days.append(t)
    return days


def _spark(points: list[dict], n: int = 30) -> list[dict]:
    """Sous-échantillonne une série pour la sparkline d'un KPI (≤ n points)."""
    if len(points) <= n:
        return points
    step = len(points) / (n - 1)
    out = [points[int(i * step)] for i in range(n - 1)]
    out.append(points[-1])
    return out


# ---------------------------------------------------------------------------
# Calcul principal
# ---------------------------------------------------------------------------

def _tracked_at(con, ts: float) -> int:
    """Produits suivis à l'instant ts (reconstruit via first_seen/last_seen)."""
    return con.execute(
        "SELECT COUNT(*) c FROM products"
        " WHERE first_seen IS NOT NULL AND first_seen <= ?"
        "   AND (active = 1 OR last_seen >= ?)", (ts, ts)).fetchone()["c"]


# variations de prix (LAG par produit sur price_log), bornées par ts — colonnes :
# uid, ts, price, prev, name, source, category
_MOVES_SQL = """WITH x AS (
       SELECT uid, ts, price,
              LAG(price) OVER (PARTITION BY uid ORDER BY ts) prev
       FROM price_log)
   SELECT x.uid, x.ts, x.price, x.prev, p.name, p.source, p.category
   FROM x JOIN products p ON p.uid = x.uid
   WHERE x.ts >= ? AND x.ts < ?
     AND x.price IS NOT NULL AND x.prev IS NOT NULL
     AND x.prev > 0 AND x.price > 0 AND x.price <> x.prev
     AND x.price <= 2000 AND x.prev <= 2000"""


def _compute(period: str, from_s: str | None, to_s: str | None) -> dict:
    con = db.connect()
    registry = _source_registry()
    names = {s["id"]: s.get("name") or s["id"] for s in registry}
    label_of = lambda src: names.get(src, src)  # noqa: E731

    f_date, t_date, label, period_id = _resolve_period(con, period, from_s, to_s)
    start = _day_start_ts(f_date)
    end = _day_start_ts(t_date + timedelta(days=1))
    now_ts = time.time()
    end_eff = min(end, now_ts)  # fin effective (la période inclut souvent « maintenant »)
    span = end - start
    prev_start, prev_end = start - span, start

    # ---- historique des relevés de prix (price_log) --------------------------
    first_log = con.execute("SELECT MIN(ts) m FROM price_log").fetchone()["m"]
    has_history = first_log is not None
    has_prev_history = bool(has_history and first_log < prev_end)

    releves_cur = con.execute(
        "SELECT COUNT(*) c FROM price_log WHERE ts >= ? AND ts < ?",
        (start, end)).fetchone()["c"]
    releves_prev = con.execute(
        "SELECT COUNT(*) c FROM price_log WHERE ts >= ? AND ts < ?",
        (prev_start, prev_end)).fetchone()["c"]

    # variations de prix détectées dans la période
    moves = con.execute(_MOVES_SQL, (start, end)).fetchall()
    drops = [m for m in moves if m["price"] < m["prev"]]
    hikes = [m for m in moves if m["price"] > m["prev"]]
    amp = [abs(m["price"] - m["prev"]) / m["prev"] for m in moves]
    amp_avg_pct = round(100 * statistics.mean(amp), 1) if amp else None

    drops_prev = hikes_prev = None
    if has_prev_history:
        mv_prev = con.execute(
            f"SELECT SUM(price < prev) d, SUM(price > prev) h FROM ({_MOVES_SQL})",
            (prev_start, prev_end)).fetchone()
        drops_prev, hikes_prev = mv_prev["d"] or 0, mv_prev["h"] or 0

    drops_by_day: dict[str, int] = {}
    hikes_by_day: dict[str, int] = {}
    for m in drops:
        d = datetime.fromtimestamp(m["ts"], TZ).date().isoformat()
        drops_by_day[d] = drops_by_day.get(d, 0) + 1
    for m in hikes:
        d = datetime.fromtimestamp(m["ts"], TZ).date().isoformat()
        hikes_by_day[d] = hikes_by_day.get(d, 0) + 1
    top_drops = sorted(
        ({"name": m["name"] or "", "source": m["source"],
          "old": m["prev"], "new": m["price"], "ts": m["ts"],
          "pct": round(100 * (m["prev"] - m["price"]) / m["prev"], 1)}
         for m in drops), key=lambda d: -d["pct"])
    top_hikes = sorted(
        ({"name": m["name"] or "", "source": m["source"],
          "old": m["prev"], "new": m["price"], "ts": m["ts"],
          "pct": round(100 * (m["price"] - m["prev"]) / m["prev"], 1)}
         for m in hikes), key=lambda d: -d["pct"])

    # ---- agrégats du catalogue --------------------------------------------------
    g = con.execute(
        f"""SELECT COUNT(*) total, SUM(on_sale) on_sale,
                   COUNT(DISTINCT source) sources,
                   COUNT(DISTINCT category) categories,
                   AVG(CASE WHEN {_PRICE_SANE} THEN price END) avg_price
            FROM products WHERE active=1""").fetchone()

    tracked_now = _tracked_at(con, end_eff)
    tracked_prev = _tracked_at(con, start) if start > (first_log or 0) - 1 else None

    src_cur = con.execute(
        "SELECT COUNT(DISTINCT source) c FROM sync_log WHERE ok=1 AND ts >= ? AND ts < ?",
        (start, end)).fetchone()["c"]
    src_prev = con.execute(
        "SELECT COUNT(DISTINCT source) c FROM sync_log WHERE ok=1 AND ts >= ? AND ts < ?",
        (prev_start, prev_end)).fetchone()["c"]

    avg_obs_cur = con.execute(
        "SELECT AVG(price) a FROM price_log WHERE ts >= ? AND ts < ?"
        " AND price > 0 AND price <= 2000", (start, end)).fetchone()["a"]
    avg_obs_prev = con.execute(
        "SELECT AVG(price) a FROM price_log WHERE ts >= ? AND ts < ?"
        " AND price > 0 AND price <= 2000", (prev_start, prev_end)).fetchone()["a"]

    new_cur = con.execute(
        "SELECT COUNT(*) c FROM products WHERE first_seen >= ? AND first_seen < ?",
        (start, end)).fetchone()["c"]
    new_prev = con.execute(
        "SELECT COUNT(*) c FROM products WHERE first_seen >= ? AND first_seen < ?",
        (prev_start, prev_end)).fetchone()["c"]

    # rabais moyen affiché (soldes actifs avec prix régulier connu)
    rabais_avg = con.execute(
        f"""SELECT AVG(100.0 * (regular_price - price) / regular_price) a
            FROM products WHERE active=1 AND on_sale=1 AND {_PRICE_SANE}
              AND regular_price IS NOT NULL AND regular_price > price""").fetchone()["a"]

    # ---- séries par jour --------------------------------------------------------
    data_start = (datetime.fromtimestamp(first_log, TZ).date()
                  if has_history else t_date)
    serie_from = max(f_date, data_start)
    days = _days_range(serie_from, t_date)
    iso = [d.isoformat() for d in days]

    pts_tracked = [{"t": d.isoformat(),
                    "v": _tracked_at(con, min(_day_start_ts(d + timedelta(days=1)), now_ts))}
                   for d in days]

    per_day = {r["d"]: r["c"] for r in con.execute(
        "SELECT date(ts,'unixepoch','localtime') d, COUNT(*) c FROM price_log"
        " WHERE ts >= ? AND ts < ? GROUP BY d", (start, end))}
    pts_releves = [{"t": d, "v": per_day.get(d, 0)} for d in iso]

    avg_day = {r["d"]: round(r["a"], 2) for r in con.execute(
        "SELECT date(ts,'unixepoch','localtime') d, AVG(price) a FROM price_log"
        " WHERE ts >= ? AND ts < ? AND price > 0 AND price <= 2000"
        " GROUP BY d", (start, end)) if r["a"] is not None}
    pts_avg = [{"t": d, "v": avg_day[d]} for d in iso if d in avg_day]

    # comparaison N-1 de l'indice de prix moyen (si historique antérieur)
    cmp_avg: list[dict] = []
    if has_prev_history:
        cmp_avg = [{"t": r["d"], "v": round(r["a"], 2)} for r in con.execute(
            "SELECT date(ts,'unixepoch','localtime') d, AVG(price) a FROM price_log"
            " WHERE ts >= ? AND ts < ? AND price > 0 AND price <= 2000"
            " GROUP BY d ORDER BY d", (prev_start, prev_end)) if r["a"] is not None]

    new_day = {r["d"]: r["c"] for r in con.execute(
        "SELECT date(first_seen,'unixepoch','localtime') d, COUNT(*) c FROM products"
        " WHERE first_seen >= ? AND first_seen < ? GROUP BY d", (start, end))}
    pts_new = [{"t": d, "v": new_day.get(d, 0)} for d in iso]

    pts_drops = [{"t": d, "v": drops_by_day.get(d, 0)} for d in iso]

    # ---- KPI (≥ 8, deltas honnêtes, sparklines quand une série existe) ----------
    def _kpi(id_, lbl, value, unit="", delta=None, spark=None):
        d = {"id": id_, "label": lbl, "value": value, "unit": unit}
        if delta is not None:
            d["delta_pct"] = delta
            d["direction"] = "up" if delta >= 0 else "down"
        else:
            d["delta_pct"] = None
        if spark and len(spark) >= 2:
            d["spark"] = _spark(spark)
        return d

    kpis = [
        _kpi("suivis", "Produits suivis (actifs)", g["total"], "",
             _delta_pct(tracked_now, tracked_prev), spark=pts_tracked),
        _kpi("releves", "Relevés de prix (période)", releves_cur, "",
             _delta_pct(releves_cur, releves_prev), spark=pts_releves),
        _kpi("baisses", "Baisses de prix détectées", len(drops), "",
             _delta_pct(len(drops), drops_prev), spark=pts_drops),
        _kpi("hausses", "Hausses de prix détectées", len(hikes), "",
             _delta_pct(len(hikes), hikes_prev)),
        _kpi("soldes", "Soldes actifs", g["on_sale"] or 0, ""),
        *([_kpi("rabais_moyen", "Rabais moyen affiché",
                round(rabais_avg, 1), "%")] if rabais_avg else []),
        _kpi("prix_moyen", "Prix moyen (produits actifs)",
             round(g["avg_price"], 2) if g["avg_price"] else 0, "$",
             _delta_pct(avg_obs_cur, avg_obs_prev), spark=pts_avg),
        _kpi("bannieres", "Bannières connectées", g["sources"], "",
             _delta_pct(src_cur, src_prev)),
        _kpi("nouveaux", "Nouveaux produits (période)", new_cur, "",
             _delta_pct(new_cur, new_prev) if new_prev else None, spark=pts_new),
        _kpi("categories", "Catégories", g["categories"], ""),
    ]

    # ---- jauges (taux & couvertures mesurés sur le catalogue réel) ---------------
    active_total = g["total"] or 0
    gauges = []
    if active_total:
        fresh7 = con.execute(
            "SELECT COUNT(*) c FROM products WHERE active=1 AND last_seen >= ?",
            (now_ts - 7 * 86400,)).fetchone()["c"]
        with_img = con.execute(
            "SELECT COUNT(*) c FROM products WHERE active=1"
            " AND images IS NOT NULL AND images <> '' AND images <> '[]'").fetchone()["c"]
        with_up = con.execute(
            "SELECT COUNT(*) c FROM products WHERE active=1"
            " AND unit_price IS NOT NULL").fetchone()["c"]
        gauges = [
            {"id": "fraicheur", "label": "Produits vus il y a moins de 7 jours",
             "value": round(100 * fresh7 / active_total, 1), "max": 100, "unit": "%"},
            {"id": "images", "label": "Produits avec image",
             "value": round(100 * with_img / active_total, 1), "max": 100, "unit": "%"},
            {"id": "prix_unitaire", "label": "Produits avec prix unitaire comparable",
             "value": round(100 * with_up / active_total, 1), "max": 100, "unit": "%"},
        ]
        if registry:
            gauges.append({"id": "couverture",
                           "label": "Bannières du registre avec produits actifs",
                           "value": g["sources"], "max": len(registry), "unit": ""})

    # ---- séries -------------------------------------------------------------------
    series = []
    if has_history and len(pts_releves) >= 2:
        series.append({"id": "releves", "title": "Relevés de prix par jour",
                       "unit": "relevés", "kind": "bar", "points": pts_releves})
    if len(pts_tracked) >= 2:
        series.append({"id": "suivis", "title": "Produits suivis par jour",
                       "unit": "produits", "kind": "line", "points": pts_tracked})
    if has_history and len(pts_avg) >= 2:
        s_avg = {"id": "prix_moyen", "title": "Indice de prix moyen relevé par jour",
                 "unit": "$", "kind": "area", "points": pts_avg}
        if len(cmp_avg) >= 2:
            s_avg["compare"] = cmp_avg
        series.append(s_avg)
    if has_history and len(pts_drops) >= 2 and drops:
        series.append({"id": "baisses", "title": "Baisses de prix détectées par jour",
                       "unit": "baisses", "kind": "line", "points": pts_drops})
    if len(pts_new) >= 2 and any(p["v"] for p in pts_new):
        series.append({"id": "nouveautes", "title": "Nouveaux produits par jour",
                       "unit": "produits", "kind": "bar", "points": pts_new})

    # ---- multi-courbes : prix moyen relevé par jour, top bannières (≤ 4) ----------
    multiseries = []
    if has_history:
        rows = con.execute(
            """SELECT p.source s, date(l.ts,'unixepoch','localtime') d,
                      AVG(l.price) a, COUNT(*) c
               FROM price_log l JOIN products p ON p.uid = l.uid
               WHERE l.ts >= ? AND l.ts < ? AND l.price > 0 AND l.price <= 2000
               GROUP BY s, d""", (start, end)).fetchall()
        by_src_day: dict[str, dict[str, float]] = {}
        src_obs: dict[str, int] = {}
        for r in rows:
            by_src_day.setdefault(r["s"], {})[r["d"]] = round(r["a"], 2)
            src_obs[r["s"]] = src_obs.get(r["s"], 0) + r["c"]
        # jusqu'à 4 bannières très actives partageant assez de jours communs
        chosen: list[str] = []
        common: set[str] = set()
        for s in sorted(src_obs, key=lambda s: -src_obs[s]):
            days_s = set(by_src_day[s])
            cand = (common & days_s) if chosen else days_s
            if len(cand) >= 3 and len(chosen) < 4:
                chosen.append(s)
                common = cand
        if len(chosen) >= 2 and len(common) >= 3:
            axis = sorted(common)
            multiseries.append({
                "id": "prix_bannieres",
                "title": "Prix moyen relevé par jour — bannières les plus actives",
                "unit": "$",
                "series": [{"label": label_of(s),
                            "points": [{"t": d, "v": by_src_day[s][d]} for d in axis]}
                           for s in chosen]})

    # ---- barres empilées ------------------------------------------------------------
    stacked = []
    if moves and len(iso) >= 2:
        stacked.append({
            "id": "variations",
            "title": "Variations de prix par jour — baisses vs hausses",
            "unit": "variations", "keys": ["Baisses", "Hausses"],
            "points": [{"t": d, "values": [drops_by_day.get(d, 0),
                                           hikes_by_day.get(d, 0)]} for d in iso]})
    adds = con.execute(
        """SELECT source s, date(first_seen,'unixepoch','localtime') d, COUNT(*) c
           FROM products WHERE first_seen >= ? AND first_seen < ?
           GROUP BY s, d""", (start, end)).fetchall()
    if adds and len(iso) >= 2:
        add_tot: dict[str, int] = {}
        add_day: dict[tuple[str, str], int] = {}
        for r in adds:
            add_tot[r["s"]] = add_tot.get(r["s"], 0) + r["c"]
            add_day[(r["s"], r["d"])] = r["c"]
        top_src = sorted(add_tot, key=lambda s: -add_tot[s])[:5]
        others = [s for s in add_tot if s not in top_src]
        keys = [label_of(s) for s in top_src] + (["Autres"] if others else [])
        pts_st = []
        for d in iso:
            vals = [add_day.get((s, d), 0) for s in top_src]
            if others:
                vals.append(sum(add_day.get((s, d), 0) for s in others))
            pts_st.append({"t": d, "values": vals})
        if any(sum(p["values"]) for p in pts_st):
            stacked.append({"id": "ajouts",
                            "title": "Nouveaux produits par jour et par bannière",
                            "unit": "produits", "keys": keys, "points": pts_st})

    # ---- répartitions ----------------------------------------------------------------
    by_src = con.execute(
        "SELECT source, COUNT(*) n, SUM(on_sale) sales FROM products WHERE active=1"
        " GROUP BY source ORDER BY n DESC").fetchall()
    donut_items = [{"label": label_of(r["source"]), "value": r["n"]}
                   for r in by_src[:8]]  # top 8 (limite du donut ka-ui/kapdf)
    by_cat = con.execute(
        "SELECT category, COUNT(*) n FROM products WHERE active=1 AND category<>''"
        " GROUP BY category ORDER BY n DESC").fetchall()
    breakdowns = [
        {"id": "bannieres", "title": "Produits actifs par bannière", "kind": "donut",
         "items": donut_items},
        {"id": "categories", "title": "Top catégories (produits actifs)", "kind": "bars",
         "items": [{"label": r["category"], "value": r["n"]} for r in by_cat[:14]]},
    ]
    sale_src = [{"label": label_of(r["source"]), "value": r["sales"] or 0}
                for r in sorted(by_src, key=lambda r: -(r["sales"] or 0))
                if (r["sales"] or 0) > 0][:14]
    if sale_src:
        breakdowns.append({"id": "soldes_bannieres",
                           "title": "Soldes actifs par bannière", "kind": "bars",
                           "items": sale_src})
    if drops:
        d_src_cur: dict[str, int] = {}
        for m in drops:
            d_src_cur[m["source"]] = d_src_cur.get(m["source"], 0) + 1
        d_src_prev: dict[str, int] = {}
        if has_prev_history:
            d_src_prev = {r["source"]: r["c"] for r in con.execute(
                f"""SELECT source, COUNT(*) c FROM ({_MOVES_SQL})
                    WHERE price < prev GROUP BY source""", (prev_start, prev_end))}
        breakdowns.append({
            "id": "baisses_bannieres",
            "title": "Baisses de prix détectées par bannière (période)", "kind": "bars",
            "items": [{"label": label_of(s), "value": n,
                       "delta_pct": _delta_pct(n, d_src_prev.get(s))}
                      for s, n in sorted(d_src_cur.items(), key=lambda kv: -kv[1])[:12]]})

    # ---- distributions (histogrammes) --------------------------------------------------
    distributions = []
    rabais_rows = con.execute(
        f"""SELECT 100.0 * (regular_price - price) / regular_price pct
            FROM products WHERE active=1 AND on_sale=1 AND {_PRICE_SANE}
              AND regular_price IS NOT NULL AND regular_price > price""").fetchall()
    if rabais_rows:
        edges = [(0, 10), (10, 20), (20, 30), (30, 40), (40, 50), (50, 101)]
        bins = [{"label": ("50 % +" if lo == 50 else f"{lo}-{hi} %"),
                 "value": sum(1 for r in rabais_rows if lo <= r["pct"] < hi)}
                for lo, hi in edges]
        distributions.append({"id": "rabais",
                              "title": "Distribution des rabais affichés (soldes actifs)",
                              "unit": "produits", "bins": bins})
    price_rows = con.execute(
        f"SELECT price FROM products WHERE active=1 AND {_PRICE_SANE}").fetchall()
    if price_rows:
        edges_p = [(0, 2, "0-2 $"), (2, 5, "2-5 $"), (5, 10, "5-10 $"),
                   (10, 20, "10-20 $"), (20, 50, "20-50 $"),
                   (50, 100, "50-100 $"), (100, 2001, "100 $ +")]
        bins_p = [{"label": lb,
                   "value": sum(1 for r in price_rows if lo <= r["price"] < hi)}
                  for lo, hi, lb in edges_p]
        distributions.append({"id": "prix",
                              "title": "Distribution des prix (produits actifs)",
                              "unit": "produits", "bins": bins_p})

    # ---- heatmap calendrier : relevés de prix par jour ---------------------------------
    heatmap = None
    if has_history:
        hm_start = max(start, now_ts - 183 * 86400)
        cells = [{"date": r["d"], "value": r["c"]} for r in con.execute(
            "SELECT date(ts,'unixepoch','localtime') d, COUNT(*) c FROM price_log"
            " WHERE ts >= ? AND ts < ? GROUP BY d ORDER BY d", (hm_start, end))]
        if cells:
            heatmap = {"title": "Relevés de prix par jour", "cells": cells}

    # ---- heatmap horaire 7×24 : relevés par jour de semaine × heure --------------------
    hourly = None
    if has_history:
        hcells = [{"dow": (r["w"] + 6) % 7, "hour": r["h"], "value": r["c"]}
                  for r in con.execute(
                      """SELECT CAST(strftime('%w', ts,'unixepoch','localtime') AS INT) w,
                                CAST(strftime('%H', ts,'unixepoch','localtime') AS INT) h,
                                COUNT(*) c
                         FROM price_log WHERE ts >= ? AND ts < ?
                         GROUP BY w, h""", (start, end))]
        if hcells:
            hourly = {"title": "Relevés de prix par jour de semaine et heure",
                      "cells": hcells}

    # ---- tableaux -----------------------------------------------------------------------
    def _cut(s: str, n: int = 26) -> str:
        s = (s or "").strip()
        return s if len(s) <= n else s[: n - 1] + "…"

    sale_rows = [
        [_cut(r["name"]), label_of(r["source"]), _fmt_money(r["price"]),
         _fmt_money(r["regular_price"]),
         f"−{round(100 * (r['regular_price'] - r['price']) / r['regular_price'])} %"]
        for r in con.execute(
            f"""SELECT name, source, price, regular_price FROM products
                WHERE active=1 AND on_sale=1 AND regular_price IS NOT NULL
                  AND {_PRICE_SANE} AND regular_price > price
                ORDER BY (regular_price - price) / regular_price DESC LIMIT 100""")]

    # prix moyens par catégorie + delta des prix relevés (période vs précédente)
    cat_prices: dict[str, list[float]] = {}
    cat_sales: dict[str, int] = {}
    for r in con.execute(
            f"""SELECT category, price, on_sale FROM products
                WHERE active=1 AND category<>'' AND {_PRICE_SANE}"""):
        cat_prices.setdefault(r["category"], []).append(r["price"])
        cat_sales[r["category"]] = cat_sales.get(r["category"], 0) + (r["on_sale"] or 0)
    cat_obs_cur = {r["cat"]: r["a"] for r in con.execute(
        """SELECT p.category cat, AVG(l.price) a
           FROM price_log l JOIN products p ON p.uid = l.uid
           WHERE l.ts >= ? AND l.ts < ? AND l.price > 0 AND l.price <= 2000
             AND p.category <> '' GROUP BY cat""", (start, end))}
    cat_obs_prev: dict[str, float] = {}
    if has_prev_history:
        cat_obs_prev = {r["cat"]: r["a"] for r in con.execute(
            """SELECT p.category cat, AVG(l.price) a
               FROM price_log l JOIN products p ON p.uid = l.uid
               WHERE l.ts >= ? AND l.ts < ? AND l.price > 0 AND l.price <= 2000
                 AND p.category <> '' GROUP BY cat""", (prev_start, prev_end))}

    def _fmt_delta(cat: str) -> str:
        dp = _delta_pct(cat_obs_cur.get(cat), cat_obs_prev.get(cat))
        if dp is None:
            return "—"
        return f"{'+' if dp >= 0 else '−'}{str(abs(dp)).replace('.', ',')} %"

    cat_rows = sorted(
        ([cat, len(v), _fmt_money(round(statistics.mean(v), 2)),
          _fmt_money(round(statistics.median(v), 2)), cat_sales.get(cat, 0),
          _fmt_delta(cat)]
         for cat, v in cat_prices.items()), key=lambda r: -r[1])

    # bannières : produits, soldes, relevés (période), fraîcheur de synchro
    rel_src = {r["s"]: r["c"] for r in con.execute(
        """SELECT p.source s, COUNT(*) c FROM price_log l
           JOIN products p ON p.uid = l.uid
           WHERE l.ts >= ? AND l.ts < ? GROUP BY s""", (start, end))}
    last_sync = {r["source"]: r["ts"] for r in con.execute(
        "SELECT source, MAX(ts) ts FROM sync_log WHERE ok=1 GROUP BY source")}
    src_price = {r["source"]: r["a"] for r in con.execute(
        f"""SELECT source, AVG(price) a FROM products
            WHERE active=1 AND {_PRICE_SANE} GROUP BY source""")}
    src_rows = [
        [label_of(r["source"]), r["n"], r["sales"] or 0,
         _fmt_money(round(src_price[r["source"]], 2))
         if src_price.get(r["source"]) else "—",
         _fmt_int(rel_src.get(r["source"], 0)),
         (datetime.fromtimestamp(last_sync[r["source"]], TZ).strftime("%Y-%m-%d %H:%M")
          if last_sync.get(r["source"]) else "—")]
        for r in by_src]

    tables = [
        {"id": "top_soldes", "title": "Top produits en solde (rabais les plus forts)",
         "columns": ["Produit", "Bannière", "Prix", "Prix rég.", "Rabais"],
         "rows": sale_rows},
        {"id": "prix_categories", "title": "Prix moyens par catégorie",
         "columns": ["Catégorie", "Produits", "Prix moyen", "Prix médian",
                     "En solde", "Δ prix relevés"],
         "rows": cat_rows},
        {"id": "bannieres", "title": "Bannières — produits, soldes & fraîcheur",
         "columns": ["Bannière", "Produits actifs", "En solde", "Prix moyen",
                     "Relevés (période)", "Dernière synchro"],
         "rows": src_rows},
    ]
    if top_drops:
        tables.insert(1, {
            "id": "baisses", "title": "Top baisses de prix détectées (période)",
            "columns": ["Produit", "Bannière", "Avant", "Après", "Baisse"],
            "rows": [[_cut(d["name"]), label_of(d["source"]), _fmt_money(d["old"]),
                      _fmt_money(d["new"]), f"−{d['pct']} %".replace(".", ",")]
                     for d in top_drops[:100]]})
    if top_hikes:
        tables.insert(2 if top_drops else 1, {
            "id": "hausses", "title": "Top hausses de prix détectées (période)",
            "columns": ["Produit", "Bannière", "Avant", "Après", "Hausse"],
            "rows": [[_cut(d["name"]), label_of(d["source"]), _fmt_money(d["old"]),
                      _fmt_money(d["new"]), f"+{d['pct']} %".replace(".", ",")]
                     for d in top_hikes[:100]]})

    # ---- records & faits marquants ------------------------------------------------------
    records = []
    best_sale = con.execute(
        f"""SELECT name, source, price, regular_price,
                   (regular_price - price) / regular_price pct
            FROM products WHERE active=1 AND on_sale=1 AND {_PRICE_SANE}
              AND regular_price IS NOT NULL AND regular_price > price
            ORDER BY pct DESC LIMIT 1""").fetchone()
    if best_sale:
        records.append({
            "label": "Record de promo — plus gros rabais affiché",
            "value": f"−{round(100 * best_sale['pct'])} % · {_cut(best_sale['name'], 34)} "
                     f"({label_of(best_sale['source'])})"})
    if top_drops:
        d0 = top_drops[0]
        records.append({
            "label": "Plus forte baisse détectée (période)",
            "value": f"−{str(d0['pct']).replace('.', ',')} % · {_cut(d0['name'], 34)} "
                     f"({_fmt_money(d0['old'])} → {_fmt_money(d0['new'])})",
            "date": datetime.fromtimestamp(d0["ts"], TZ).date().isoformat()})
    if top_hikes:
        h0 = top_hikes[0]
        records.append({
            "label": "Plus forte hausse détectée (période)",
            "value": f"+{str(h0['pct']).replace('.', ',')} % · {_cut(h0['name'], 34)} "
                     f"({_fmt_money(h0['old'])} → {_fmt_money(h0['new'])})",
            "date": datetime.fromtimestamp(h0["ts"], TZ).date().isoformat()})
    if has_history and per_day:
        rec_day = max(per_day.items(), key=lambda kv: kv[1])
        records.append({"label": "Jour record de relevés de prix",
                        "value": _fmt_int(rec_day[1]) + " relevés",
                        "date": rec_day[0]})
    if new_day:
        rec_new = max(new_day.items(), key=lambda kv: kv[1])
        records.append({"label": "Jour record de nouveaux produits",
                        "value": _fmt_int(rec_new[1]) + " produits",
                        "date": rec_new[0]})
    if moves:
        records.append({
            "label": "Variations de prix détectées (période)",
            "value": f"{_fmt_int(len(drops))} baisses · {_fmt_int(len(hikes))} hausses"})
    if amp_avg_pct is not None:
        records.append({"label": "Amplitude moyenne des variations de prix",
                        "value": f"±{amp_avg_pct} %".replace(".", ",")})
    if rel_src:
        top_rel = max(rel_src.items(), key=lambda kv: kv[1])
        records.append({"label": "Bannière la plus relevée (période)",
                        "value": f"{label_of(top_rel[0])} · "
                                 f"{_fmt_int(top_rel[1])} relevés"})
    if hourly and hourly["cells"]:
        dows = ["lundi", "mardi", "mercredi", "jeudi", "vendredi", "samedi", "dimanche"]
        peak = max(hourly["cells"], key=lambda c: c["value"])
        records.append({"label": "Heure de pointe des relevés",
                        "value": f"{dows[peak['dow']]} {peak['hour']} h · "
                                 f"{_fmt_int(peak['value'])} relevés"})
    cat_sale_share = [(cat, 100 * cat_sales.get(cat, 0) / len(v))
                      for cat, v in cat_prices.items()
                      if len(v) >= 50 and cat_sales.get(cat, 0) > 0]
    if cat_sale_share:
        top_cat = max(cat_sale_share, key=lambda kv: kv[1])
        records.append({"label": "Catégorie la plus en solde",
                        "value": (f"{top_cat[0]} · {round(top_cat[1], 1)} % des produits"
                                  ).replace(".", ",")})
    cat_deltas = [(cat, _delta_pct(cat_obs_cur.get(cat), cat_obs_prev.get(cat)))
                  for cat in cat_obs_cur if cat_obs_prev.get(cat)]
    cat_deltas = [(c, d) for c, d in cat_deltas if d is not None]
    if cat_deltas:
        up_cat = max(cat_deltas, key=lambda kv: kv[1])
        dn_cat = min(cat_deltas, key=lambda kv: kv[1])
        if up_cat[1] > 0:
            records.append({"label": "Catégorie en plus forte hausse (prix relevés)",
                            "value": f"{up_cat[0]} · +{str(up_cat[1]).replace('.', ',')} %"})
        if dn_cat[1] < 0:
            records.append({"label": "Catégorie en plus forte baisse (prix relevés)",
                            "value": f"{dn_cat[0]} · −{str(abs(dn_cat[1])).replace('.', ',')} %"})

    con.close()
    return {
        "updated": datetime.now(TZ).isoformat(timespec="seconds"),
        "period": {"from": f_date.isoformat(), "to": t_date.isoformat(),
                   "label": label, "id": period_id},
        "kpis": kpis,
        **({"gauges": gauges} if gauges else {}),
        "series": series,
        **({"multiseries": multiseries} if multiseries else {}),
        **({"stacked": stacked} if stacked else {}),
        "breakdowns": breakdowns,
        **({"distributions": distributions} if distributions else {}),
        **({"heatmap": heatmap} if heatmap else {}),
        **({"hourly": hourly} if hourly else {}),
        "tables": tables,
        "records": records,
    }


def dashboard(period: str = "30j", from_s: str | None = None,
              to_s: str | None = None) -> dict:
    """Tableau de bord (contrat SPEC.md v2) — mis en cache 5 minutes par période."""
    key = (period, from_s or "", to_s or "")
    now = time.time()
    with _cache_lock:
        hit = _cache.get(key)
        if hit and now - hit[0] < CACHE_TTL:
            return hit[1]
    data = _compute(period, from_s, to_s)
    with _cache_lock:
        _cache[key] = (now, data)
        if len(_cache) > 64:  # borne de sécurité (plages personnalisées)
            oldest = min(_cache, key=lambda k: _cache[k][0])
            _cache.pop(oldest, None)
    return data