# -----------------------------------------------------------------------------
# Auto-Ka — Agrégateur de voitures usagées à vendre (province de Québec)
# Auteur : Simon-Pierre Boucher — contact@spboucher.ai
# statsdash.py : tableau de bord analytique — construit le JSON du contrat
#                commun Groupe KA v2 (/api/stats/dashboard, voir ka/stats/SPEC.md)
#                à partir des données réelles (vehicles, price_log, sync_log).
#                Sert aussi de source unique aux 5 rapports PDF (autoka/kapdf.py).
#
# Principes :
#   - AUCUNE statistique inventée : tout vient de la base SQLite. Un champ sans
#     donnée mesurable est simplement omis (le front affiche « Pas encore
#     mesuré »).
#   - L'« inventaire à la date t » est reconstruit depuis le cycle de vie réel
#     des annonces : first_seen (arrivée) et updated_at des annonces
#     désactivées (retrait). Les prix/km historiques par jour utilisent le
#     dernier prix/km connu de chaque véhicule (approximation documentée).
#   - Les séries sont bornées au début réel des données (12 août 2026) :
#     avant, rien n'était mesuré — on ne trace pas de faux zéros. De même,
#     les deltas vs période précédente ne sont émis que si la référence
#     existait déjà (start > début des données).
#   - Cache serveur de 5 minutes par période.
# -----------------------------------------------------------------------------
from __future__ import annotations

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

from . import db

TZ = ZoneInfo("America/Toronto")
CACHE_TTL = 300  # 5 minutes

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

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",
}
PERIOD_DAYS = {"auj": 1, "7j": 7, "30j": 30, "3m": 90, "6m": 180, "12m": 365}

# Filtre « type de vendeur » : les annonces de particuliers (Kijiji) portent
# un dealer_name « Particulier (…) » — tout le reste vient de commerces.
PRIV = "dealer_name LIKE 'Particulier%'"


# --- utilitaires ---------------------------------------------------------------

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


def _to_date(ts: float) -> date:
    return datetime.fromtimestamp(ts, TZ).date()


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


def _fr_money(p) -> str:
    if p is None:
        return "—"
    return _fr_int(p) + " $"


def _fr_km(k) -> str:
    if k is None:
        return "—"
    return _fr_int(k) + " km"


def _fr_pct(p) -> str:
    if p is None:
        return "—"
    return ("+" if p >= 0 else "−") + f"{abs(p):.1f}".replace(".", ",") + " %"


def _delta(cur, prev) -> tuple[float | None, str | None]:
    """Variation en % vs période précédente ; None si non mesurable."""
    if cur is None or prev is None or not prev:
        return None, None
    pct = round(100.0 * (cur - prev) / prev, 1)
    return pct, ("up" if pct >= 0 else "down")


def _resolve_period(period: str, from_: str | None, to: str | None):
    """(start_ts, end_ts, prev_start_ts, prev_end_ts, label, from_iso, to_iso)."""
    now = datetime.now(TZ)
    today = now.date()
    if from_ and to:
        d0 = date.fromisoformat(from_)
        d1 = date.fromisoformat(to)
        if d1 < d0:
            d0, d1 = d1, d0
        start = _day_start(d0)
        end = min(now.timestamp(), _day_start(d1 + timedelta(days=1)))
        label = f"du {d0.isoformat()} au {d1.isoformat()}"
    elif period == "tout":
        con = db.connect()
        row = con.execute("SELECT MIN(first_seen) m FROM vehicles").fetchone()
        con.close()
        start = row["m"] or now.timestamp()
        end = now.timestamp()
        label = PERIOD_LABELS["tout"]
    elif period == "annee":
        start = _day_start(date(today.year, 1, 1))
        end = now.timestamp()
        label = f"Année {today.year}"
    else:
        days = PERIOD_DAYS.get(period, 30)
        start = _day_start(today - timedelta(days=days - 1))
        end = now.timestamp()
        label = PERIOD_LABELS.get(period, PERIOD_LABELS["30j"])
    span = max(end - start, 1.0)
    return start, end, start - span, start, label, \
        _to_date(start).isoformat(), _to_date(end - 1).isoformat()


# --- reconstruction de l'inventaire par jour ------------------------------------

def _lifecycle_rows(con):
    """(arrivée, retrait|None, prix, km, marque) pour chaque annonce auto."""
    return [
        (r["first_seen"], r["updated_at"] if not r["active"] else None,
         r["price"], r["mileage_km"], r["make"])
        for r in con.execute(
            "SELECT first_seen, active, updated_at, price, mileage_km, make"
            " FROM vehicles WHERE kind='auto' AND first_seen IS NOT NULL")
    ]


def _daily_series(rows, start_ts: float, end_ts: float):
    """Par jour : inventaire actif, prix/km moyens de l'inventaire, arrivées,
    retraits.

    Balayage d'événements (arrivées/retraits triés) — l'inventaire au soir du
    jour J = annonces arrivées avant la fin de J et pas encore retirées.
    """
    if not rows:
        return []
    data_start = min(r[0] for r in rows)
    d = max(_to_date(start_ts), _to_date(data_start))
    d_end = _to_date(end_ts - 1)
    if d > d_end:
        return []
    adds = sorted(rows, key=lambda r: r[0])
    rems = sorted((r for r in rows if r[1] is not None), key=lambda r: r[1])
    new_by_day = Counter(_to_date(r[0]) for r in rows)
    gone_by_day = Counter(_to_date(r[1]) for r in rows if r[1] is not None)
    ai = ri = count = n_price = n_km = 0
    sum_price = sum_km = 0.0
    out = []
    while d <= d_end:
        cutoff = _day_start(d + timedelta(days=1))
        while ai < len(adds) and adds[ai][0] < cutoff:
            count += 1
            if adds[ai][2] is not None:
                sum_price += adds[ai][2]
                n_price += 1
            if adds[ai][3] is not None:
                sum_km += adds[ai][3]
                n_km += 1
            ai += 1
        while ri < len(rems) and rems[ri][1] < cutoff:
            count -= 1
            if rems[ri][2] is not None:
                sum_price -= rems[ri][2]
                n_price -= 1
            if rems[ri][3] is not None:
                sum_km -= rems[ri][3]
                n_km -= 1
            ri += 1
        out.append({
            "t": d.isoformat(),
            "inv": count,
            "avg_price": round(sum_price / n_price) if n_price else None,
            "avg_km": round(sum_km / n_km) if n_km else None,
            "new": new_by_day.get(d, 0),
            "gone": gone_by_day.get(d, 0),
        })
        d += timedelta(days=1)
    return out


def _snapshot(con, t: float):
    """Indicateurs de l'inventaire actif reconstitué à l'instant t."""
    return dict(con.execute(
        """SELECT COUNT(*) n, AVG(price) avg_price, AVG(mileage_km) avg_km,
                  AVG(year) avg_year,
                  COUNT(DISTINCT CASE WHEN NOT {priv} THEN dealer_name END) dealers,
                  SUM(CASE WHEN {priv} THEN 1 ELSE 0 END) private,
                  COUNT(DISTINCT source) sources
           FROM vehicles WHERE kind='auto' AND first_seen<=?
             AND (active=1 OR updated_at>?)""".format(priv=PRIV),
        (t, t)).fetchone())


def _snapshot_counts(con, t: float, expr: str) -> dict[str, int]:
    """Effectifs par catégorie de l'inventaire reconstitué à l'instant t."""
    return {r["lab"]: r["n"] for r in con.execute(
        f"""SELECT {expr} lab, COUNT(*) n FROM vehicles
            WHERE kind='auto' AND first_seen<=? AND (active=1 OR updated_at>?)
            GROUP BY lab""", (t, t))}


def _price_drops(con, t0: float, t1: float) -> int:
    """Baisses de prix observées dans price_log entre t0 et t1."""
    return con.execute(
        """WITH pl AS (
             SELECT uid, ts, price,
                    LAG(price) OVER (PARTITION BY uid ORDER BY ts) prev_price
             FROM price_log WHERE price IS NOT NULL)
           SELECT COUNT(*) n FROM pl
           WHERE ts>=? AND ts<? AND prev_price IS NOT NULL
             AND prev_price > price""", (t0, t1)).fetchone()["n"]


# --- construction du dashboard ---------------------------------------------------

def _build(period: str, from_: str | None, to: str | None) -> dict:
    start, end, pstart, pend, label, f_iso, t_iso = _resolve_period(period, from_, to)
    con = db.connect()
    now_ts = time.time()

    # --- KPI : maintenant vs début de période / période précédente -------------
    cur = _snapshot(con, min(end, now_ts))
    # Pas de delta si la période commence avant le début réel des données :
    # l'inventaire de référence n'existait pas encore (rien d'inventé).
    data_start = con.execute(
        "SELECT MIN(first_seen) m FROM vehicles WHERE kind='auto'").fetchone()["m"]
    has_ref = data_start is not None and start > data_start
    if has_ref:
        ref = _snapshot(con, start)
    else:
        ref = {"n": None, "avg_price": None, "avg_km": None, "avg_year": None,
               "dealers": None, "private": None, "sources": None}

    def _count(sql, args):
        return con.execute(sql, args).fetchone()["n"]

    new_cur = _count("SELECT COUNT(*) n FROM vehicles WHERE kind='auto'"
                     " AND first_seen>=? AND first_seen<?", (start, end))
    new_prev = _count("SELECT COUNT(*) n FROM vehicles WHERE kind='auto'"
                      " AND first_seen>=? AND first_seen<?", (pstart, pend))
    gone_cur = _count("SELECT COUNT(*) n FROM vehicles WHERE kind='auto'"
                      " AND active=0 AND updated_at>=? AND updated_at<?", (start, end))
    gone_prev = _count("SELECT COUNT(*) n FROM vehicles WHERE kind='auto'"
                       " AND active=0 AND updated_at>=? AND updated_at<?", (pstart, pend))
    drops_cur = _price_drops(con, start, end)
    drops_prev = _price_drops(con, pstart, pend) if has_ref else None

    # --- séries quotidiennes (avant les KPI : sparklines) -----------------------
    rows = _lifecycle_rows(con)
    daily = _daily_series(rows, start, end)
    prev_daily = _daily_series(rows, pstart, pend)
    full_prev = len(prev_daily) == len(daily) and len(daily) > 1

    def _spark(key):
        pts = [{"t": p["t"], "v": p[key]} for p in daily if p[key] is not None]
        return pts if len(pts) >= 2 else None

    def _kpi(id_, label_, value, unit="", prev=None, spark=None):
        pct, direction = _delta(value, prev)
        k = {"id": id_, "label": label_, "value": value, "unit": unit,
             "delta_pct": pct, "direction": direction}
        if spark:
            k["spark"] = spark
        return k

    kpis = [
        _kpi("actifs", "Véhicules actifs", cur["n"], "", ref["n"], _spark("inv")),
        _kpi("nouveaux", "Nouveaux véhicules (période)", new_cur, "",
             new_prev if has_ref else None, _spark("new")),
        _kpi("retires", "Vendus / retirés (période)", gone_cur, "",
             gone_prev if has_ref else None, _spark("gone")),
        _kpi("prix", "Prix moyen (inventaire actif)",
             round(cur["avg_price"]) if cur["avg_price"] else None, "$",
             round(ref["avg_price"]) if ref["avg_price"] else None,
             _spark("avg_price")),
        _kpi("km", "Km moyen (inventaire actif)",
             round(cur["avg_km"]) if cur["avg_km"] else None, "km",
             round(ref["avg_km"]) if ref["avg_km"] else None, _spark("avg_km")),
        # année moyenne : valeur pré-formatée (« 2021,1 » — pas de séparateur
        # de milliers) ; un delta en % n'aurait aucun sens sur un millésime.
        _kpi("annee", "Année-modèle moyenne",
             f"{cur['avg_year']:.1f}".replace(".", ",")
             if cur["avg_year"] else None, ""),
        _kpi("dealers", "Concessionnaires actifs", cur["dealers"], "", ref["dealers"]),
        _kpi("particuliers", "Annonces de particuliers", cur["private"], "",
             ref["private"]),
        _kpi("baisses", "Baisses de prix (période)", drops_cur, "", drops_prev),
        _kpi("sources", "Sources actives", cur["sources"], "", ref["sources"]),
    ]
    kpis = [k for k in kpis if k["value"] is not None]

    # --- jauges : complétude des fiches de l'inventaire actif -------------------
    cov = con.execute(
        """SELECT COUNT(*) n,
                  SUM(CASE WHEN images IS NOT NULL AND images<>'[]'
                           AND images<>'' THEN 1 ELSE 0 END) img,
                  SUM(CASE WHEN mileage_km IS NOT NULL THEN 1 ELSE 0 END) km,
                  SUM(CASE WHEN price IS NOT NULL THEN 1 ELSE 0 END) prix,
                  SUM(CASE WHEN vin<>'' THEN 1 ELSE 0 END) vin,
                  SUM(CASE WHEN carfax_url<>'' THEN 1 ELSE 0 END) carfax,
                  SUM(CASE WHEN lat IS NOT NULL THEN 1 ELSE 0 END) geo
           FROM vehicles WHERE active=1 AND kind='auto'""").fetchone()
    gauges = []
    if cov["n"]:
        def _gauge(id_, label_, num, help_=None):
            g = {"id": id_, "label": label_,
                 "value": round(100.0 * num / cov["n"], 1), "max": 100, "unit": "%"}
            if help_:
                g["help"] = help_
            return g
        gauges = [
            _gauge("photos", "Fiches avec photos", cov["img"]),
            _gauge("km", "Kilométrage renseigné", cov["km"]),
            _gauge("prix", "Prix affiché", cov["prix"]),
            _gauge("vin", "NIV (VIN) connu", cov["vin"]),
            _gauge("geo", "Fiches géolocalisées", cov["geo"]),
            _gauge("carfax", "Rapport Carfax lié", cov["carfax"],
                   "Part des annonces actives dont la source publie un lien Carfax"),
        ]

    # --- séries d'évolution ------------------------------------------------------
    series = []
    if daily:
        series.append({
            "id": "inv", "title": "Inventaire actif par jour", "unit": "véhicules",
            "kind": "line", "points": [{"t": p["t"], "v": p["inv"]} for p in daily],
            **({"compare": [{"t": p["t"], "v": p["inv"]} for p in prev_daily]}
               if full_prev else {}),
        })
        series.append({
            "id": "new", "title": "Nouveaux véhicules par jour", "unit": "véhicules",
            "kind": "bar", "points": [{"t": p["t"], "v": p["new"]} for p in daily],
        })
        series.append({
            "id": "gone", "title": "Véhicules vendus / retirés par jour",
            "unit": "véhicules", "kind": "bar",
            "points": [{"t": p["t"], "v": p["gone"]} for p in daily],
        })
        price_pts = [{"t": p["t"], "v": p["avg_price"]} for p in daily
                     if p["avg_price"] is not None]
        if price_pts:
            series.append({
                "id": "avg_price", "title": "Prix moyen de l'inventaire par jour",
                "unit": "$", "kind": "area", "points": price_pts,
            })

    # --- multi-courbes : prix moyen par grande marque ----------------------------
    multiseries = []
    top_makes = [r["make"] for r in con.execute(
        "SELECT make FROM vehicles WHERE active=1 AND kind='auto' AND make<>''"
        " GROUP BY make ORDER BY COUNT(*) DESC LIMIT 4")]
    if top_makes and daily:
        by_make_rows = defaultdict(list)
        for r in rows:
            if r[4] in top_makes:
                by_make_rows[r[4]].append(r)
        make_daily = {m: _daily_series(by_make_rows[m], start, end)
                      for m in top_makes}
        # domaine commun : jours où chaque marque a un prix moyen mesuré
        commons = None
        for m in top_makes:
            days_m = {p["t"] for p in make_daily[m] if p["avg_price"] is not None}
            commons = days_m if commons is None else commons & days_m
        commons = commons or set()
        if len(commons) >= 2:
            ms_series = [{
                "label": m,
                "points": [{"t": p["t"], "v": p["avg_price"]}
                           for p in make_daily[m] if p["t"] in commons],
            } for m in top_makes]
            multiseries.append({
                "id": "prix_marques",
                "title": "Prix moyen de l'inventaire — top 4 des marques",
                "unit": "$", "series": ms_series,
            })

    # --- barres empilées : arrivées par source ------------------------------------
    stacked = []
    src_day = defaultdict(Counter)   # jour -> source -> n
    src_tot = Counter()
    for r in con.execute(
            "SELECT first_seen, source FROM vehicles WHERE kind='auto'"
            " AND first_seen>=? AND first_seen<?", (start, end)):
        d_ = _to_date(r["first_seen"]).isoformat()
        src_day[d_][r["source"]] += 1
        src_tot[r["source"]] += 1
    if src_tot:
        SRC_LABELS = {"otogo": "Otogo", "kijiji": "Kijiji",
                      "autotrader": "AutoTrader", "cargurus": "CarGurus",
                      "automobileendirect": "AutomobileEnDirect",
                      "hgregoire": "HGrégoire"}
        tops = [s for s, _ in src_tot.most_common(4)]
        keys = [SRC_LABELS.get(s, s) for s in tops]
        others = len(src_tot) > len(tops)
        if others:
            keys.append("Autres")
        pts = []
        for d_ in sorted(src_day):
            vals = [src_day[d_].get(s, 0) for s in tops]
            if others:
                vals.append(sum(src_day[d_].values()) - sum(vals))
            pts.append({"t": d_, "values": vals})
        if len(pts) >= 2:
            stacked.append({"id": "src", "title": "Nouveaux véhicules par source",
                            "unit": "véhicules", "keys": keys, "points": pts})

    # --- répartitions (inventaire actif, deltas vs début de période) --------------
    FUEL_EXPR = ("CASE WHEN fuel IN ('', 'N.D.') THEN 'Non précisé'"
                 " ELSE fuel END")
    TRANS_EXPR = ("CASE WHEN transmission IN ('', 'NA', 'N.D.')"
                  " THEN 'Non précisé' ELSE transmission END")
    BODY_EXPR = "CASE WHEN body_type='' THEN 'Non précisé' ELSE body_type END"
    SELLER_EXPR = (f"CASE WHEN {PRIV} THEN 'Particuliers'"
                   " ELSE 'Concessionnaires' END")

    def _items(expr, limit=None, where=""):
        cur_rows = con.execute(
            f"SELECT {expr} lab, COUNT(*) n FROM vehicles"
            f" WHERE active=1 AND kind='auto'{where}"
            f" GROUP BY lab ORDER BY n DESC" + (f" LIMIT {limit}" if limit else "")
        ).fetchall()
        prev_counts = _snapshot_counts(con, start, expr) if has_ref else {}
        out = []
        for r in cur_rows:
            pct, _ = _delta(r["n"], prev_counts.get(r["lab"]))
            out.append({"label": r["lab"] or "Non précisé", "value": r["n"],
                        **({"delta_pct": pct} if pct is not None else {})})
        return out

    by_fuel = _items(FUEL_EXPR)
    by_trans = _items(TRANS_EXPR, limit=8)
    by_body = _items(BODY_EXPR, limit=10)
    by_seller = _items(SELLER_EXPR)
    by_make = _items("make", limit=12, where=" AND make<>''")

    year_rows = con.execute(
        "SELECT year, COUNT(*) n FROM vehicles WHERE active=1 AND kind='auto'"
        " AND year IS NOT NULL GROUP BY year ORDER BY year").fetchall()
    # 11 années récentes + un groupe « antérieures » = 12 barres max, aucune
    # année récente escamotée par la limite d'affichage des graphiques.
    by_year, older = [], 0
    cutoff = max((r["year"] for r in year_rows), default=0) - 10
    for r in year_rows:
        if r["year"] < cutoff:
            older += r["n"]
        else:
            by_year.append({"label": str(r["year"]), "value": r["n"]})
    if older:
        by_year.insert(0, {"label": f"≤ {cutoff - 1}", "value": older})

    breakdowns = [
        {"id": "make", "title": "Top 12 des marques", "kind": "donut", "items": by_make},
        {"id": "fuel", "title": "Par carburant", "kind": "donut", "items": by_fuel},
        {"id": "seller", "title": "Concessionnaires vs particuliers",
         "kind": "donut", "items": by_seller},
        {"id": "trans", "title": "Par boîte de vitesses", "kind": "bar", "items": by_trans},
        {"id": "body", "title": "Par carrosserie", "kind": "bar", "items": by_body},
        {"id": "year", "title": "Par année-modèle", "kind": "bar", "items": by_year},
    ]
    breakdowns = [b for b in breakdowns if len(b["items"]) > 1]

    # --- distributions : prix et kilométrage --------------------------------------
    def _histo(col, width, cap, fmt):
        rows_ = con.execute(
            f"SELECT CAST({col}/{width} AS INTEGER) b, COUNT(*) n FROM vehicles"
            f" WHERE active=1 AND kind='auto' AND {col} IS NOT NULL AND {col}>=0"
            " GROUP BY b ORDER BY b").fetchall()
        if not rows_:
            return None
        n_bins = cap // width
        bins = [{"label": fmt(i), "value": 0} for i in range(n_bins)]
        over = {"label": fmt(n_bins), "value": 0}
        for r in rows_:
            if r["b"] < n_bins:
                bins[r["b"]]["value"] += r["n"]
            else:
                over["value"] += r["n"]
        if over["value"]:
            bins.append(over)
        while bins and bins[0]["value"] == 0:
            bins.pop(0)
        return bins

    def _fmt_price_bin(i):
        lo, hi = i * 5, i * 5 + 5
        return f"{lo}–{hi} k$" if hi <= 100 else "100 k$ +"

    def _fmt_km_bin(i):
        lo, hi = i * 25, i * 25 + 25
        return f"{lo}–{hi} k km" if hi <= 300 else "300 k km +"

    distributions = []
    price_bins = _histo("price", 5000, 100000, _fmt_price_bin)
    if price_bins:
        distributions.append({
            "id": "prix", "title": "Distribution des prix (tranches de 5 000 $)",
            "unit": "véhicules", "bins": price_bins})
    km_bins = _histo("mileage_km", 25000, 300000, _fmt_km_bin)
    if km_bins:
        distributions.append({
            "id": "km", "title": "Distribution du kilométrage (tranches de 25 000 km)",
            "unit": "véhicules", "bins": km_bins})

    # --- répartition géographique ---------------------------------------------------
    geo = {"title": "Par région", "items": _items("region", where=" AND region<>''")}

    # --- calendrier + activité horaire ----------------------------------------------
    heatmap = {"title": "Nouveaux véhicules par jour",
               "cells": [{"date": p["t"], "value": p["new"]} for p in daily]}

    hour_counter = Counter()
    for r in con.execute(
            "SELECT first_seen FROM vehicles WHERE kind='auto'"
            " AND first_seen>=? AND first_seen<?", (start, end)):
        dt = datetime.fromtimestamp(r["first_seen"], TZ)
        hour_counter[(dt.weekday(), dt.hour)] += 1
    hourly = None
    if hour_counter:
        hourly = {"title": "Nouveaux véhicules détectés par heure (synchros)",
                  "cells": [{"dow": k[0], "hour": k[1], "value": v}
                            for k, v in sorted(hour_counter.items())]}

    # --- tableaux détaillés (inventaire actif) ------------------------------------
    def _table_rows(sql, args=()):
        return [list(r) for r in con.execute(sql, args).fetchall()]

    make_prev = _snapshot_counts(con, start, "make") if has_ref else {}

    def _mk_delta(make_, n):
        pct, _ = _delta(n, make_prev.get(make_))
        return _fr_pct(pct)

    tables = [
        {"id": "makes", "title": "Top marques — volume, prix et km moyens",
         "columns": ["Marque", "Véhicules", "Prix moyen", "Km moyen", "Δ période"],
         "rows": [[m, n, _fr_money(p), _fr_km(k), _mk_delta(m, n)]
                  for m, n, p, k in _table_rows(
             "SELECT make, COUNT(*), AVG(price), AVG(mileage_km) FROM vehicles"
             " WHERE active=1 AND kind='auto' AND make<>''"
             " GROUP BY make ORDER BY COUNT(*) DESC LIMIT 25")]},
        {"id": "models", "title": "Top modèles — volume, prix et km moyens",
         "columns": ["Modèle", "Véhicules", "Prix moyen", "Km moyen"],
         "rows": [[m, n, _fr_money(p), _fr_km(k)] for m, n, p, k in _table_rows(
             "SELECT make || ' ' || model, COUNT(*), AVG(price), AVG(mileage_km)"
             " FROM vehicles WHERE active=1 AND kind='auto' AND make<>''"
             " AND model<>'' GROUP BY make, model ORDER BY COUNT(*) DESC LIMIT 25")]},
        {"id": "cities", "title": "Top villes — inventaire, prix et km moyens",
         "columns": ["Ville", "Région", "Véhicules", "Prix moyen", "Km moyen"],
         "rows": [[c, rg or "—", n, _fr_money(p), _fr_km(k)]
                  for c, rg, n, p, k in _table_rows(
             "SELECT city, MAX(region), COUNT(*), AVG(price), AVG(mileage_km)"
             " FROM vehicles WHERE active=1 AND kind='auto' AND city<>''"
             " GROUP BY city ORDER BY COUNT(*) DESC LIMIT 25")]},
        {"id": "dealers", "title": "Top concessionnaires — inventaire et prix moyen",
         "columns": ["Concessionnaire", "Région", "Véhicules", "Prix moyen"],
         "rows": [[d, rg or "—", n, _fr_money(p)] for d, rg, n, p in _table_rows(
             "SELECT dealer_name, MAX(region), COUNT(*), AVG(price) FROM vehicles"
             " WHERE active=1 AND kind='auto' AND dealer_name<>''"
             f" AND NOT {PRIV}"
             " GROUP BY dealer_name ORDER BY COUNT(*) DESC LIMIT 25")]},
    ]

    # sources & fraîcheur : inventaire actif + arrivées de la période + dernière
    # synchro réussie (sync_log = journal réel du pipeline d'ingestion)
    last_sync = {r["source"]: r for r in con.execute(
        """SELECT source, MAX(ts) ts, ok FROM sync_log GROUP BY source""")}
    src_rows = _table_rows(
        """SELECT source, COUNT(*),
                  SUM(CASE WHEN first_seen>=? AND first_seen<? THEN 1 ELSE 0 END)
           FROM vehicles WHERE active=1 AND kind='auto'
           GROUP BY source ORDER BY COUNT(*) DESC LIMIT 25""", (start, end))
    if src_rows:
        rows_out = []
        for s, n, added in src_rows:
            ls = last_sync.get(s)
            when = (datetime.fromtimestamp(ls["ts"], TZ)
                    .strftime("%Y-%m-%d %H:%M") if ls else "—")
            ok = ("OK" if ls and ls["ok"] else ("Erreur" if ls else "—"))
            rows_out.append([s, n, added, when, ok])
        tables.append({
            "id": "sources", "title": "Sources — inventaire, ajouts et fraîcheur",
            "columns": ["Source", "Véhicules actifs", "Ajouts (période)",
                        "Dernière synchro", "Statut"],
            "rows": rows_out})

    # --- records & faits marquants -------------------------------------------------
    records = []
    if daily:
        best = max(daily, key=lambda p: p["new"])
        if best["new"]:
            records.append({"label": "Jour record d'arrivées",
                            "value": _fr_int(best["new"]) + " véhicules",
                            "date": best["t"]})
    fastest = con.execute(
        """SELECT title, year, updated_at - first_seen dur, updated_at
           FROM vehicles WHERE kind='auto' AND active=0
             AND updated_at>=? AND updated_at<? AND updated_at>first_seen
           ORDER BY dur ASC LIMIT 1""", (start, end)).fetchone()
    if fastest:
        days_ = fastest["dur"] / 86400
        dur_txt = (f"{fastest['dur'] / 3600:.0f} h" if days_ < 1
                   else f"{days_:.1f} jours".replace(".", ","))
        records.append({"label": f"Vente la plus rapide — {fastest['title']}",
                        "value": dur_txt,
                        "date": _to_date(fastest["updated_at"]).isoformat()})
    drop = con.execute(
        """WITH pl AS (
             SELECT uid, ts, price,
                    LAG(price) OVER (PARTITION BY uid ORDER BY ts) prev_price
             FROM price_log WHERE price IS NOT NULL)
           SELECT v.title, pl.prev_price - pl.price baisse, pl.ts
           FROM pl JOIN vehicles v ON v.uid=pl.uid AND v.kind='auto'
           WHERE pl.ts>=? AND pl.ts<? AND pl.prev_price > pl.price
           ORDER BY baisse DESC LIMIT 1""", (start, end)).fetchone()
    if drop:
        records.append({"label": f"Plus forte baisse de prix — {drop['title']}",
                        "value": "−" + _fr_money(drop["baisse"]),
                        "date": _to_date(drop["ts"]).isoformat()})
    busiest = con.execute(
        f"""SELECT dealer_name, COUNT(*) n FROM vehicles
           WHERE kind='auto' AND dealer_name<>'' AND NOT {PRIV}
             AND first_seen>=? AND first_seen<?
           GROUP BY dealer_name ORDER BY n DESC LIMIT 1""",
        (start, end)).fetchone()
    if busiest and busiest["n"]:
        records.append({"label": f"Concessionnaire le plus actif — {busiest['dealer_name']}",
                        "value": _fr_int(busiest["n"]) + " nouveautés",
                        "date": None})
    top_price = con.execute(
        "SELECT title, price FROM vehicles WHERE active=1 AND kind='auto'"
        " AND price IS NOT NULL ORDER BY price DESC LIMIT 1").fetchone()
    if top_price:
        records.append({"label": f"Véhicule le plus cher en vente — {top_price['title']}",
                        "value": _fr_money(top_price["price"]), "date": None})
    oldest = con.execute(
        "SELECT title, year FROM vehicles WHERE active=1 AND kind='auto'"
        " AND year IS NOT NULL ORDER BY year ASC LIMIT 1").fetchone()
    if oldest:
        records.append({"label": f"Doyen de l'inventaire — {oldest['title']}",
                        "value": f"année {oldest['year']}", "date": None})
    top_km = con.execute(
        "SELECT title, mileage_km FROM vehicles WHERE active=1 AND kind='auto'"
        " AND mileage_km IS NOT NULL ORDER BY mileage_km DESC LIMIT 1").fetchone()
    if top_km:
        records.append({"label": f"Odomètre le plus élevé — {top_km['title']}",
                        "value": _fr_km(top_km["mileage_km"]), "date": None})
    if has_ref and make_prev:
        make_cur = {r["lab"]: r["n"] for r in con.execute(
            "SELECT make lab, COUNT(*) n FROM vehicles WHERE active=1"
            " AND kind='auto' AND make<>'' GROUP BY lab HAVING n>=100")}
        gains = [(m, _delta(n, make_prev.get(m))[0]) for m, n in make_cur.items()]
        gains = [(m, p) for m, p in gains if p is not None]
        if gains:
            m, p = max(gains, key=lambda x: x[1])
            if p > 0:
                records.append({
                    "label": f"Marque en plus forte hausse — {m}",
                    "value": _fr_pct(p) + " d'inventaire", "date": None})
    top_region = con.execute(
        """SELECT region, COUNT(*) n FROM vehicles WHERE kind='auto'
           AND region<>'' AND first_seen>=? AND first_seen<?
           GROUP BY region ORDER BY n DESC LIMIT 1""", (start, end)).fetchone()
    if top_region and top_region["n"]:
        records.append({"label": f"Région la plus active — {top_region['region']}",
                        "value": _fr_int(top_region["n"]) + " nouveautés",
                        "date": None})

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


def dashboard(period: str = "30j", from_: str | None = None,
              to: str | None = None) -> dict:
    """Dashboard du contrat SPEC v2 — mis en cache 5 minutes par période."""
    if period not in PERIOD_LABELS and not (from_ and to):
        period = "30j"
    key = (period, from_ or "", to 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, from_, to)
    with _cache_lock:
        _cache[key] = (now, data)
    return data