# -----------------------------------------------------------------------------
# Fabri-Ka — Agrégateur de produits québécois
# Auteur : Simon-Pierre Boucher — contact@spboucher.ai
# statsdash.py : tableau de bord /api/stats/dashboard (contrat ka-stats SPEC v2)
# -----------------------------------------------------------------------------
"""Toutes les valeurs viennent de la base réelle (products / stores / sync_log) :
- volumes et nouveautés par jour via `first_seen` (epoch) ;
- prix moyen/médian sur les produits actifs avec prix (> 0, plafond 500 000 $
  pour la moyenne, même garde-fou que /api/stats/extended) ;
- deltas des KPI de stock = état à la fin vs état au DÉBUT de la période
  (reconstruit avec first_seen) ; delta des nouveautés = fenêtre précédente
  de même longueur ; deltas des répartitions = même reconstruction.
- jauges = couvertures mesurées (prix, image, dispo, géoloc, boutiques) ;
- activité horaire = heures réelles de détection (first_seen) ;
- sources & fraîcheur = journal sync_log.
Aucun chiffre inventé : indisponible => champ omis / None (le front masque).
Cache mémoire 5 min par période."""
from __future__ import annotations

import sqlite3
import time
from datetime import date, datetime, timedelta
from zoneinfo import ZoneInfo

from . import db
from .schema import CATEGORIES

TZ = ZoneInfo("America/Toronto")
PRICE_CAP = 500_000  # prix aberrants exclus de la moyenne
CACHE_TTL = 300      # ≥ 5 min (SPEC)

PLATFORM_LABELS = {
    "shopify": "Shopify", "woocommerce": "WooCommerce", "wix": "Wix",
    "squarespace": "Squarespace", "lightspeed": "Lightspeed",
    "prestashop": "PrestaShop", "snipcart": "Snipcart",
    "wordpress": "WordPress", "generic": "Site générique", "": "Inconnue",
}

ORIGIN_LABELS = {
    "A": "A — Fabriqué au Québec", "B": "B — Conçu au Québec",
    "C": "C — Détaillant québécois", "D": "D — Mixte", "E": "E — À vérifier",
}

BUCKETS = [
    ("0-10", "Moins de 10 $"), ("10-25", "10 – 25 $"), ("25-50", "25 – 50 $"),
    ("50-100", "50 – 100 $"), ("100-250", "100 – 250 $"),
    ("250-1000", "250 – 1 000 $"), ("1000+", "1 000 $ et plus"),
]

SIZE_BUCKETS = [
    ("1-10", "1 – 10"), ("11-50", "11 – 50"), ("51-200", "51 – 200"),
    ("201-500", "201 – 500"), ("501-1000", "501 – 1 000"), ("1000+", "> 1 000"),
]


def _cat_label(key: str | None) -> str:
    return CATEGORIES.get(key or "", (key or "Autre", []))[0]


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


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


def _delta(cur, prev) -> float | None:
    """% de variation, None si la base de comparaison est vide (rien d'inventé)."""
    if prev is None or cur is None or prev <= 0:
        return None
    return round(100.0 * (cur - prev) / prev, 1)


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


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


def _bounds(con: sqlite3.Connection, period: str, from_: str | None,
            to: str | None) -> tuple[date, date, str]:
    today = datetime.now(TZ).date()
    if from_ and to:
        start, end = date.fromisoformat(from_), date.fromisoformat(to)
        if end < start:
            start, end = end, start
        return start, min(end, today), f"du {start} au {min(end, today)}"
    days = {"7j": (7, "7 jours"), "30j": (30, "30 jours"), "3m": (90, "3 mois"),
            "6m": (180, "6 mois"), "12m": (365, "12 mois")}
    if period == "auj":
        return today, today, "aujourd'hui"
    if period == "annee":
        return date(today.year, 1, 1), today, f"année {today.year}"
    if period == "tout":
        row = con.execute("SELECT MIN(first_seen) FROM products").fetchone()
        start = (datetime.fromtimestamp(row[0], TZ).date()
                 if row and row[0] else today)
        return start, today, "toute la période"
    n, label = days.get(period, days["30j"])
    return today - timedelta(days=n - 1), today, label


def _daily(con, t0: float, t1: float) -> dict[str, int]:
    """Nouveaux produits par jour (date locale) dans [t0, t1)."""
    rows = con.execute(
        """SELECT date(first_seen,'unixepoch','localtime') AS d, COUNT(*) AS n
           FROM products WHERE first_seen>=? AND first_seen<? GROUP BY d""",
        (t0, t1)).fetchall()
    return {r[0]: r[1] for r in rows}


def _median(con, extra_where: str = "", args: tuple = ()) -> float | None:
    n = con.execute(
        f"SELECT COUNT(*) FROM products WHERE active=1 AND listing_status='published' AND price>0{extra_where}",
        args).fetchone()[0]
    if not n:
        return None
    row = con.execute(
        f"""SELECT price FROM products WHERE active=1 AND listing_status='published' AND price>0{extra_where}
            ORDER BY price LIMIT 1 OFFSET ?""", args + (n // 2,)).fetchone()
    return round(row[0], 2) if row else None


def _build(period: str, from_: str | None, to: str | None) -> dict:
    con = db.connect()
    try:
        start, end, label = _bounds(con, period, from_, to)
        ndays = (end - start).days + 1
        t0, t1 = _ts(start), _ts(end + timedelta(days=1))
        p_start, p_end = start - timedelta(days=ndays), start - timedelta(days=1)
        pt0, pt1 = _ts(p_start), _ts(p_end + timedelta(days=1))
        one = lambda sql, a=(): con.execute(sql, a).fetchone()[0]  # noqa: E731
        days = [start + timedelta(days=i) for i in range(min(ndays, 400))]
        iso_days = [d.isoformat() for d in days]

        # ----- KPI : état courant vs état au début de la période -------------
        total = one("SELECT COUNT(*) FROM products WHERE active=1 AND listing_status='published'")
        total_t0 = one("SELECT COUNT(*) FROM products WHERE active=1 AND listing_status='published' AND first_seen<?", (t0,))
        stores_live = one("SELECT COUNT(*) FROM stores WHERE product_count>0")
        stores_t0 = one("""SELECT COUNT(DISTINCT store_id) FROM products
                           WHERE active=1 AND listing_status='published' AND first_seen<?""", (t0,))
        stores_reg = one("SELECT COUNT(*) FROM stores")
        new_cur = one("SELECT COUNT(*) FROM products WHERE first_seen>=? AND first_seen<?", (t0, t1))
        new_prev = one("SELECT COUNT(*) FROM products WHERE first_seen>=? AND first_seen<?", (pt0, pt1))
        avg_now = one("""SELECT ROUND(AVG(price),2) FROM products
                         WHERE active=1 AND listing_status='published' AND price>0 AND price<=?""", (PRICE_CAP,))
        avg_t0 = one("""SELECT ROUND(AVG(price),2) FROM products
                        WHERE active=1 AND listing_status='published' AND price>0 AND price<=? AND first_seen<?""",
                     (PRICE_CAP, t0))
        med_now = _median(con)
        med_t0 = _median(con, " AND first_seen<?", (t0,))
        cats = one("SELECT COUNT(DISTINCT category) FROM products WHERE active=1 AND listing_status='published' AND category<>''")
        cats_t0 = one("""SELECT COUNT(DISTINCT category) FROM products
                         WHERE active=1 AND listing_status='published' AND category<>'' AND first_seen<?""", (t0,))
        regions = one("""SELECT COUNT(DISTINCT region) FROM stores
                         WHERE region<>'' AND product_count>0""")
        regions_t0 = one("""SELECT COUNT(DISTINCT s.region) FROM stores s
                            JOIN products p ON p.store_id=s.id
                            WHERE s.region<>'' AND p.active=1 AND p.listing_status='published' AND p.first_seen<?""", (t0,))
        img_where = " AND images IS NOT NULL AND images<>'' AND images<>'[]'"
        with_img = one(f"SELECT COUNT(*) FROM products WHERE active=1 AND listing_status='published'{img_where}")
        with_img_t0 = one(f"""SELECT COUNT(*) FROM products
                              WHERE active=1 AND listing_status='published' AND first_seen<?{img_where}""", (t0,))

        # ----- séries quotidiennes (aussi utilisées comme sparklines) --------
        cur_daily = _daily(con, t0, t1)
        prev_daily = _daily(con, pt0, pt1)
        new_points = [{"t": d, "v": cur_daily.get(d, 0)} for d in iso_days]
        compare = None
        if sum(prev_daily.values()):
            pdays = [(p_start + timedelta(days=i)).isoformat()
                     for i in range(min(ndays, 400))]
            compare = [{"t": d, "v": prev_daily.get(d, 0)} for d in pdays]
        base = one("SELECT COUNT(*) FROM products WHERE first_seen<?", (t0,))
        cum_points, acc = [], base
        for pt in new_points:
            acc += pt["v"]
            cum_points.append({"t": pt["t"], "v": acc})

        # prix moyen cumulatif du catalogue détecté (reconstruit via first_seen)
        pr = con.execute("""SELECT date(first_seen,'unixepoch','localtime') AS d,
                                   SUM(price), COUNT(*) FROM products
                            WHERE price>0 AND price<=? GROUP BY d""",
                         (PRICE_CAP,)).fetchall()
        psum = pcnt = 0.0
        by_day = {}
        for d, s, n in pr:
            by_day[d] = (s, n)
        start_iso = start.isoformat()
        for d, (s, n) in by_day.items():
            if d < start_iso:
                psum += s
                pcnt += n
        avg_points = []
        for d in iso_days:
            s, n = by_day.get(d, (0.0, 0))
            psum += s
            pcnt += n
            if pcnt:
                avg_points.append({"t": d, "v": round(psum / pcnt, 2)})

        # boutiques cumulées (première détection d'un produit par boutique)
        st_first = con.execute("""SELECT date(first_seen,'unixepoch','localtime') AS d,
                                         COUNT(*) FROM (
                                    SELECT store_id, MIN(first_seen) AS first_seen
                                    FROM products GROUP BY store_id) GROUP BY d""").fetchall()
        st_by_day = {r[0]: r[1] for r in st_first}
        st_acc = sum(v for d, v in st_by_day.items() if d < start_iso)
        st_points = []
        for d in iso_days:
            st_acc += st_by_day.get(d, 0)
            st_points.append({"t": d, "v": st_acc})

        def kpi(id_, lab, value, prev=None, unit="", spark=None):
            d = _delta(value, prev)
            out = {"id": id_, "label": lab, "value": value, "unit": unit,
                   "delta_pct": d,
                   "direction": None if d is None else ("up" if d >= 0 else "down")}
            sp = _spark(spark or [])
            if sp:
                out["spark"] = sp
            return out

        kpis = [
            kpi("produits", "Produits actifs au catalogue", total, total_t0,
                spark=cum_points),
            kpi("boutiques", "Boutiques en ligne avec produits", stores_live,
                stores_t0, spark=st_points),
            kpi("nouveautes", f"Nouveaux produits ({label})", new_cur, new_prev,
                spark=new_points),
            kpi("prix_moyen", "Prix moyen (produits actifs)", avg_now, avg_t0,
                unit="$", spark=avg_points),
            kpi("prix_median", "Prix médian (produits actifs)", med_now, med_t0,
                unit="$"),
            kpi("avec_image", "Produits actifs avec image", with_img, with_img_t0),
            kpi("categories", "Catégories couvertes", cats, cats_t0),
            kpi("regions", "Régions avec boutiques actives", regions, regions_t0),
            kpi("registre", "Boutiques au registre", stores_reg),
        ]

        # ----- jauges : couvertures mesurées ----------------------------------
        with_price = one("SELECT COUNT(*) FROM products WHERE active=1 AND listing_status='published' AND price>0")
        avail = one("SELECT COUNT(*) FROM products WHERE active=1 AND listing_status='published' AND available=1")
        live_geo = one("""SELECT COUNT(*) FROM stores
                          WHERE product_count>0 AND region<>'' AND region IS NOT NULL""")
        pct = lambda a, b: round(100.0 * a / b, 1) if b else None  # noqa: E731
        gauges = []
        for gid, glabel, val, help_ in [
            ("prix", "Produits avec prix affiché", pct(with_price, total),
             "part des produits actifs dont le prix est connu"),
            ("image", "Produits avec image", pct(with_img, total),
             "part des produits actifs avec au moins une image"),
            ("dispo", "Produits en stock", pct(avail, total),
             "part des produits actifs marqués disponibles à la source"),
            ("boutiques_actives", "Boutiques du registre avec produits",
             pct(stores_live, stores_reg),
             "boutiques dont le catalogue public est agrégé"),
            ("geoloc", "Boutiques actives géolocalisées", pct(live_geo, stores_live),
             "boutiques actives avec région administrative connue"),
        ]:
            if val is not None:
                gauges.append({"id": gid, "label": glabel, "value": val,
                               "max": 100, "unit": "%", "help": help_})

        # ----- séries ----------------------------------------------------------
        series = [
            {"id": "nouveautes_jour", "title": "Nouveaux produits détectés par jour",
             "unit": "produits", "kind": "line" if compare else "bar",
             "points": new_points, **({"compare": compare} if compare else {})},
            {"id": "cumul", "title": "Produits détectés — cumul du catalogue",
             "unit": "produits", "kind": "area", "points": cum_points},
        ]
        if len(avg_points) > 1:
            series.append(
                {"id": "prix_moyen", "title": "Prix moyen du catalogue détecté (cumul)",
                 "unit": "$", "kind": "line", "points": avg_points})

        # ----- multi-courbes : nouveautés des 4 premières catégories ----------
        top4 = [r[0] for r in con.execute(
            """SELECT category FROM products WHERE first_seen>=? AND first_seen<?
               AND category<>'' GROUP BY category ORDER BY COUNT(*) DESC LIMIT 4""",
            (t0, t1)).fetchall()]
        multiseries = []
        if top4 and ndays > 1:
            qmarks = ",".join("?" * len(top4))
            rows = con.execute(
                f"""SELECT category, date(first_seen,'unixepoch','localtime') AS d,
                           COUNT(*) FROM products
                    WHERE first_seen>=? AND first_seen<? AND category IN ({qmarks})
                    GROUP BY category, d""", (t0, t1, *top4)).fetchall()
            grid: dict[str, dict[str, int]] = {c: {} for c in top4}
            for c, d, n in rows:
                grid[c][d] = n
            multiseries.append({
                "id": "cats_nouveautes",
                "title": "Nouveaux produits par jour — top catégories",
                "unit": "produits",
                "series": [{"label": _cat_label(c),
                            "points": [{"t": d, "v": grid[c].get(d, 0)}
                                       for d in iso_days]} for c in top4]})

        # ----- barres empilées : ajouts par plateforme e-commerce -------------
        plat_rows = con.execute(
            """SELECT COALESCE(s.platform,''), date(p.first_seen,'unixepoch','localtime') AS d,
                      COUNT(*) FROM products p JOIN stores s ON s.id=p.store_id
               WHERE p.first_seen>=? AND p.first_seen<? GROUP BY 1, d""",
            (t0, t1)).fetchall()
        stacked = []
        if plat_rows and ndays > 1:
            tot_by_plat: dict[str, int] = {}
            for pl, d, n in plat_rows:
                tot_by_plat[pl] = tot_by_plat.get(pl, 0) + n
            top_pl = sorted(tot_by_plat, key=tot_by_plat.get, reverse=True)[:5]
            others = [pl for pl in tot_by_plat if pl not in top_pl]
            keys = [PLATFORM_LABELS.get(pl, pl or "Inconnue") for pl in top_pl]
            if others:
                keys.append("Autres")
            pgrid: dict[str, dict[str, int]] = {}
            for pl, d, n in plat_rows:
                k = (PLATFORM_LABELS.get(pl, pl or "Inconnue")
                     if pl in top_pl else "Autres")
                pgrid.setdefault(d, {})
                pgrid[d][k] = pgrid[d].get(k, 0) + n
            stacked.append({
                "id": "ajouts_plateforme",
                "title": "Nouveaux produits par jour, par plateforme e-commerce",
                "unit": "produits", "keys": keys,
                "points": [{"t": d, "values": [pgrid.get(d, {}).get(k, 0)
                                               for k in keys]} for d in iso_days]})

        # ----- répartitions ---------------------------------------------------
        plat = con.execute("""SELECT platform, COUNT(*) FROM stores
                              WHERE product_count>0 GROUP BY platform
                              ORDER BY 2 DESC""").fetchall()
        top_cats = con.execute("""SELECT category, COUNT(*) FROM products
                                  WHERE active=1 AND listing_status='published' GROUP BY category
                                  ORDER BY 2 DESC LIMIT 12""").fetchall()
        cats_t0_rows = dict(con.execute(
            """SELECT category, COUNT(*) FROM products
               WHERE active=1 AND listing_status='published' AND first_seen<? GROUP BY category""", (t0,)).fetchall())
        # donut catégories : top 7 + « Autres »
        donut_cats = [{"label": _cat_label(c), "value": n} for c, n in top_cats[:7]]
        rest = total - sum(n for _, n in top_cats[:7])
        if rest > 0:
            donut_cats.append({"label": "Autres", "value": rest})
        origin_rows = con.execute(
            """SELECT COALESCE(s.origin_class,''), COUNT(p.uid) FROM stores s
               JOIN products p ON p.store_id=s.id AND p.active=1 AND p.listing_status='published'
               GROUP BY 1 ORDER BY 2 DESC""").fetchall()
        region_now = con.execute(
            """SELECT s.region, COUNT(p.uid) FROM stores s
               JOIN products p ON p.store_id=s.id AND p.active=1 AND p.listing_status='published'
               WHERE s.region<>'' GROUP BY s.region ORDER BY 2 DESC""").fetchall()
        region_t0 = dict(con.execute(
            """SELECT s.region, COUNT(p.uid) FROM stores s
               JOIN products p ON p.store_id=s.id AND p.active=1 AND p.listing_status='published'
                 AND p.first_seen<?
               WHERE s.region<>'' GROUP BY s.region""", (t0,)).fetchall())
        breakdowns = [
            {"id": "categories", "title": "Produits actifs par catégorie",
             "kind": "donut", "items": donut_cats},
            {"id": "plateformes", "title": "Boutiques par plateforme e-commerce",
             "kind": "donut",
             "items": [{"label": PLATFORM_LABELS.get(p or "", p or "Inconnue"),
                        "value": n} for p, n in plat]},
            {"id": "regions_delta", "title": "Produits actifs par région (Δ période)",
             "kind": "bars",
             "items": [{"label": r, "value": n,
                        "delta_pct": _delta(n, region_t0.get(r))}
                       for r, n in region_now[:14]]},
            {"id": "origine", "title": "Produits par classe d'origine (A–E)",
             "kind": "bars",
             "items": [{"label": ORIGIN_LABELS.get(o, o or "Non classée"),
                        "value": n} for o, n in origin_rows]},
        ]

        # ----- distributions --------------------------------------------------
        buckets = dict(con.execute("""SELECT CASE
                WHEN price < 10 THEN '0-10' WHEN price < 25 THEN '10-25'
                WHEN price < 50 THEN '25-50' WHEN price < 100 THEN '50-100'
                WHEN price < 250 THEN '100-250' WHEN price < 1000 THEN '250-1000'
                ELSE '1000+' END AS b, COUNT(*) FROM products
                WHERE active=1 AND listing_status='published' AND price>0 GROUP BY b""").fetchall())
        sizes = dict(con.execute("""SELECT CASE
                WHEN product_count <= 10 THEN '1-10'
                WHEN product_count <= 50 THEN '11-50'
                WHEN product_count <= 200 THEN '51-200'
                WHEN product_count <= 500 THEN '201-500'
                WHEN product_count <= 1000 THEN '501-1000'
                ELSE '1000+' END AS b, COUNT(*) FROM stores
                WHERE product_count>0 GROUP BY b""").fetchall())
        distributions = [
            {"id": "prix", "title": "Distribution des prix (produits actifs)",
             "unit": "produits",
             "bins": [{"label": lab, "value": buckets[k]}
                      for k, lab in BUCKETS if buckets.get(k)]},
            {"id": "taille_boutiques",
             "title": "Boutiques par taille de catalogue (nb de produits)",
             "unit": "boutiques",
             "bins": [{"label": lab, "value": sizes[k]}
                      for k, lab in SIZE_BUCKETS if sizes.get(k)]},
        ]

        geo = {"title": "Produits actifs par région",
               "items": [{"label": r, "value": n} for r, n in region_now]}

        # ----- heatmap : nouveautés/jour sur 26 semaines ----------------------
        h_start = datetime.now(TZ).date() - timedelta(days=181)
        hm = _daily(con, _ts(h_start), _ts(datetime.now(TZ).date() + timedelta(days=1)))
        heatmap = {"title": "Nouveaux produits par jour",
                   "cells": [{"date": d, "value": v} for d, v in sorted(hm.items())]}

        # ----- heatmap horaire 7×24 : heures réelles de détection -------------
        hr = con.execute(
            """SELECT strftime('%w', first_seen,'unixepoch','localtime') AS w,
                      strftime('%H', first_seen,'unixepoch','localtime') AS h,
                      COUNT(*) FROM products
               WHERE first_seen>=? AND first_seen<? GROUP BY w, h""",
            (t0, t1)).fetchall()
        hourly = None
        if hr:
            hourly = {"title": "Produits détectés par heure (période)",
                      "cells": [{"dow": (int(w) + 6) % 7, "hour": int(h), "value": n}
                                for w, h, n in hr]}

        # ----- tableaux --------------------------------------------------------
        # une seule passe agrégée sur les produits des 50 boutiques (les
        # sous-requêtes corrélées balayaient les 3 900+ boutiques : > 2 min
        # sur la DB de 1,9 Go)
        shop_rows = con.execute("""
            WITH top AS (SELECT id, name, region, product_count FROM stores
                         WHERE product_count>0
                         ORDER BY product_count DESC LIMIT 50)
            SELECT t.name, t.region, t.product_count,
                   COALESCE(a.nouv, 0) AS nouv, a.pavg
            FROM top t LEFT JOIN (
                SELECT store_id,
                       SUM(CASE WHEN first_seen>=? AND first_seen<?
                                THEN 1 ELSE 0 END) AS nouv,
                       ROUND(AVG(CASE WHEN active=1
                                       AND listing_status='published'
                                       AND price>0 AND price<=?
                                      THEN price END), 2) AS pavg
                FROM products WHERE store_id IN (SELECT id FROM top)
                GROUP BY store_id) a ON a.store_id=t.id
            ORDER BY t.product_count DESC""", (t0, t1, PRICE_CAP)).fetchall()
        cat_rows = con.execute("""
            SELECT category, COUNT(*) AS n, COUNT(DISTINCT store_id) AS st,
                   ROUND(AVG(CASE WHEN price>0 AND price<=? THEN price END),2)
            FROM products WHERE active=1 AND listing_status='published' GROUP BY category
            ORDER BY n DESC""", (PRICE_CAP,)).fetchall()
        region_tbl = con.execute("""
            SELECT s.region, COUNT(DISTINCT s.id) AS st, COUNT(p.uid) AS n,
                   ROUND(AVG(CASE WHEN p.price>0 AND p.price<=? THEN p.price END),2),
                   SUM(CASE WHEN p.first_seen>=? AND p.first_seen<? THEN 1 ELSE 0 END)
            FROM stores s JOIN products p ON p.store_id=s.id AND p.active=1 AND p.listing_status='published'
            WHERE s.region<>'' GROUP BY s.region ORDER BY n DESC""",
            (PRICE_CAP, t0, t1)).fetchall()
        nouv_rows = con.execute("""
            SELECT p.title, s.name, p.category, p.price,
                   date(p.first_seen,'unixepoch','localtime')
            FROM products p JOIN stores s ON s.id=p.store_id
            WHERE p.first_seen>=? AND p.first_seen<?
            ORDER BY p.first_seen DESC LIMIT 50""", (t0, t1)).fetchall()
        sync_rows = con.execute("""
            SELECT datetime(ts,'unixepoch','localtime'), store_id, found, added,
                   updated, removed, status
            FROM sync_log WHERE ts>=? AND ts<? ORDER BY ts DESC LIMIT 50""",
            (t0, t1)).fetchall()
        tables = [
            {"id": "top_boutiques", "title": "Top boutiques",
             "columns": ["Boutique", "Région", "Produits", "Nouveautés (période)", "Prix moyen"],
             "rows": [[name or "—", reg or "—", n, nouv, _price(pavg)]
                      for name, reg, n, nouv, pavg in shop_rows]},
            {"id": "categories", "title": "Catégories en détail",
             "columns": ["Catégorie", "Produits", "Boutiques", "Prix moyen", "Prix médian"],
             "rows": [[_cat_label(c), n, st, _price(pavg),
                       _price(_median(con, " AND category=?", (c,)))]
                      for c, n, st, pavg in cat_rows]},
            {"id": "regions", "title": "Régions en détail",
             "columns": ["Région", "Boutiques", "Produits", "Prix moyen", "Nouveautés (période)"],
             "rows": [[r, st, n, _price(pavg), nouv]
                      for r, st, n, pavg, nouv in region_tbl]},
        ]
        if nouv_rows:
            tables.append(
                {"id": "nouveautes", "title": "Nouveautés récentes (période)",
                 "columns": ["Produit", "Boutique", "Catégorie", "Prix", "Détecté le"],
                 "rows": [[(t or "—")[:60], (s or "—")[:36], _cat_label(c),
                           _price(p if p and p > 0 else None), d]
                          for t, s, c, p, d in nouv_rows]})
        if sync_rows:
            tables.append(
                {"id": "syncs", "title": "Sources & fraîcheur — dernières synchronisations",
                 "columns": ["Date", "Boutique", "Trouvés", "Ajoutés", "Mis à jour",
                             "Retirés", "Statut"],
                 "rows": [[d, sid, f or 0, a or 0, u or 0, rm or 0, st or "—"]
                          for d, sid, f, a, u, rm, st in sync_rows]})

        # ----- records & faits marquants --------------------------------------
        records = []
        if cur_daily:
            day, v = max(cur_daily.items(), key=lambda kv: kv[1])
            records.append({"label": "Jour record de nouveautés (période)",
                            "value": _int(v) + " produits", "date": day})
        all_time = con.execute(
            """SELECT date(first_seen,'unixepoch','localtime') AS d, COUNT(*) AS n
               FROM products GROUP BY d ORDER BY n DESC LIMIT 1""").fetchone()
        if all_time:
            records.append({"label": "Jour record de nouveautés (depuis le début)",
                            "value": _int(all_time[1]) + " produits",
                            "date": all_time[0]})
        top_shop = con.execute("""SELECT s.name, COUNT(*) FROM products p
            JOIN stores s ON s.id=p.store_id WHERE p.first_seen>=? AND p.first_seen<?
            GROUP BY p.store_id ORDER BY 2 DESC LIMIT 1""", (t0, t1)).fetchone()
        if top_shop:
            records.append({"label": f"Boutique la plus prolifique (période) — {(top_shop[0] or '')[:40]}",
                            "value": _int(top_shop[1]) + " nouveautés"})
        big_shop = con.execute("""SELECT name, product_count FROM stores
            ORDER BY product_count DESC LIMIT 1""").fetchone()
        if big_shop and big_shop[1]:
            records.append({"label": f"Boutique la mieux garnie — {(big_shop[0] or '')[:40]}",
                            "value": _int(big_shop[1]) + " produits"})
        top_region = con.execute("""SELECT s.region, COUNT(*) FROM products p
            JOIN stores s ON s.id=p.store_id WHERE s.region<>''
            AND p.first_seen>=? AND p.first_seen<? GROUP BY s.region
            ORDER BY 2 DESC LIMIT 1""", (t0, t1)).fetchone()
        if top_region:
            records.append({"label": "Région la plus active (période)",
                            "value": f"{top_region[0]} — " + _int(top_region[1]) + " nouveautés"})
        if top_cats:
            records.append({"label": "Catégorie la plus fournie",
                            "value": f"{_cat_label(top_cats[0][0])} — "
                                     + _int(top_cats[0][1]) + " produits"})
        rich_cat = con.execute("""SELECT category,
            ROUND(AVG(CASE WHEN price>0 AND price<=? THEN price END),2) AS a
            FROM products WHERE active=1 AND listing_status='published' GROUP BY category
            HAVING COUNT(*)>=100 AND a IS NOT NULL
            ORDER BY a DESC LIMIT 1""", (PRICE_CAP,)).fetchone()
        if rich_cat:
            records.append({"label": f"Catégorie au prix moyen le plus élevé — {_cat_label(rich_cat[0])}",
                            "value": _price(rich_cat[1])})
        if plat:
            records.append({"label": "Plateforme e-commerce dominante",
                            "value": f"{PLATFORM_LABELS.get(plat[0][0] or '', plat[0][0] or 'Inconnue')} — "
                                     + _int(plat[0][1]) + " boutiques"})
        dear = con.execute("""SELECT p.title, p.price, s.name FROM products p
            JOIN stores s ON s.id=p.store_id WHERE p.active=1 AND p.listing_status='published' AND p.price>0
            AND p.price<=? ORDER BY p.price DESC LIMIT 1""", (PRICE_CAP,)).fetchone()
        if dear:
            records.append({"label": f"Produit le plus cher au catalogue — {(dear[0] or '')[:34]} ({dear[2]})",
                            "value": _price(dear[1])})
        best_sync = con.execute("""SELECT store_id, added,
            date(ts,'unixepoch','localtime') FROM sync_log
            WHERE ts>=? AND ts<? AND added>0
            ORDER BY added DESC LIMIT 1""", (t0, t1)).fetchone()
        if best_sync:
            records.append({"label": f"Synchronisation la plus fructueuse (période) — {best_sync[0]}",
                            "value": _int(best_sync[1]) + " ajouts",
                            "date": best_sync[2]})

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


_cache: dict[tuple, tuple[float, dict]] = {}


def dashboard(period: str = "30j", from_: str | None = None,
              to: str | None = None) -> dict:
    key = (period, from_ or "", to or "")
    hit = _cache.get(key)
    if hit and time.time() - hit[0] < CACHE_TTL:
        return hit[1]
    data = _build(period, from_, to)
    if len(_cache) > 64:
        _cache.clear()
    _cache[key] = (time.time(), data)
    return data