# =============================================================================
# Job·Ka — Groupe KA
# Auteur  : Simon-Pierre Boucher
# Contact : contact@spboucher.ai
# Fichier : jobka/statsdash.py
# Rôle    : Tableau de bord statistique — contrat commun Groupe KA v2
#   (voir frontend/src/ka/stats/SPEC.md). Construit le JSON du dashboard à
#   partir de requêtes SQL agrégées (jobs, sync_log, sources.json) avec un
#   cache mémoire de 5 minutes par clé de période. AUCUNE stat inventée : une
#   section sans donnée réelle est simplement absente du JSON (le front
#   affiche « Pas encore mesuré »). Les salaires publiés sont bornés
#   [SAL_MIN, SAL_MAX] $/an (et [SALH_MIN, SALH_MAX] $/h) pour écarter les
#   aberrations de lecture à la source.
#   v2 : sparklines KPI, jauges de complétude, série des retraits,
#   multi-courbes (top catégories), barres empilées (ajouts par source),
#   distributions (salaires annuels/horaires, âge des offres), heatmap
#   horaire 7×24 (observations des connecteurs), tableaux catégories +
#   sources, records enrichis. Deltas HONNÊTES : calculés seulement quand la
#   période précédente est couverte par nos observations.
# Créé    : 2026-08-17   Modifié : 2026-08-19
# =============================================================================
from __future__ import annotations

import json
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
_cache: dict[str, tuple[float, dict]] = {}
_cache_lock = threading.Lock()

SOURCES_PATH = Path(__file__).resolve().parent.parent / "data" / "sources.json"

# Bornes de plausibilité des salaires publiés : en dehors, c'est presque
# toujours une erreur de parsing à la source (taux horaire annualisé deux
# fois, montant par quart, etc.), pas un vrai salaire.
SAL_MIN, SAL_MAX = 25_000, 400_000        # $ CA / an
SALH_MIN, SALH_MAX = 12.0, 150.0          # $ CA / h

PERIODS = {
    "auj": ("Aujourd'hui", 0),
    "7j": ("7 jours", 6),
    "30j": ("30 jours", 29),
    "3m": ("3 mois", 89),
    "6m": ("6 mois", 179),
    "12m": ("12 mois", 364),
}

ATS_FR = {
    "workday": "Workday", "smartrecruiters": "SmartRecruiters",
    "workable": "Workable", "lever": "Lever", "ashby": "Ashby",
    "bamboohr": "BambooHR", "breezy": "Breezy", "recruitee": "Recruitee",
    "greenhouse": "Greenhouse", "custom": "Site employeur",
    "": "Autre / sur mesure",
}
MODE_FR = {"presentiel": "Présentiel", "hybride": "Hybride",
           "teletravail": "Télétravail", None: "Non précisé", "": "Non précisé"}
TYPE_FR = {"temps_plein": "Temps plein", "temps_partiel": "Temps partiel",
           "contractuel": "Contractuel", "stage": "Stage",
           "saisonnier": "Saisonnier", None: "Non précisé", "": "Non précisé"}

# Fourchettes salariales annualisées (bornes inférieures, $ / an)
SAL_BUCKETS = [
    (SAL_MIN, 40_000, "25 k$ – 40 k$"),
    (40_000, 60_000, "40 k$ – 60 k$"),
    (60_000, 80_000, "60 k$ – 80 k$"),
    (80_000, 100_000, "80 k$ – 100 k$"),
    (100_000, 130_000, "100 k$ – 130 k$"),
    (130_000, SAL_MAX + 1, "130 k$ et plus"),
]

# Fourchettes de taux horaires publiés ($ / h)
SALH_BUCKETS = [
    (SALH_MIN, 18.0, "12 $ – 18 $/h"),
    (18.0, 22.0, "18 $ – 22 $/h"),
    (22.0, 26.0, "22 $ – 26 $/h"),
    (26.0, 30.0, "26 $ – 30 $/h"),
    (30.0, 36.0, "30 $ – 36 $/h"),
    (36.0, 45.0, "36 $ – 45 $/h"),
    (45.0, SALH_MAX + 1, "45 $/h et plus"),
]

# Âge des offres actives (jours depuis la date de publication AFFICHÉE)
AGE_BUCKETS = [
    (0, 7, "0–7 j"),
    (7, 14, "8–14 j"),
    (14, 30, "15–30 j"),
    (30, 60, "31–60 j"),
    (60, 90, "61–90 j"),
    (90, 100_000, "Plus de 90 j"),
]

# Milieu de fourchette annualisé / horaire (expressions SQL réutilisées)
_SAL_MID = ("(COALESCE(salary_year_min, salary_year_max)"
            " + COALESCE(salary_year_max, salary_year_min)) / 2.0")
_SALH_MID = ("(COALESCE(salary_hour_min, salary_hour_max)"
             " + COALESCE(salary_hour_max, salary_hour_min)) / 2.0")


def _sal_where() -> str:
    """Clause : offre avec salaire annualisé publié ET plausible."""
    return (f" AND (salary_year_min IS NOT NULL OR salary_year_max IS NOT NULL)"
            f" AND {_SAL_MID} BETWEEN {SAL_MIN} AND {SAL_MAX}")


def _salh_where() -> str:
    """Clause : offre avec taux horaire publié ET plausible."""
    return (f" AND (salary_hour_min IS NOT NULL OR salary_hour_max IS NOT NULL)"
            f" AND {_SALH_MID} BETWEEN {SALH_MIN} AND {SALH_MAX}")


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


def _coverage(con) -> tuple[date | None, date | None]:
    """Première et dernière date OBSERVÉES par Job·Ka (first/last_seen)."""
    row = con.execute(
        "SELECT MIN(first_seen) a, MAX(last_seen) b FROM jobs"
        " WHERE dup_of IS NULL").fetchone()
    if row["a"] is None:
        return None, None
    return (datetime.fromtimestamp(row["a"], TZ).date(),
            datetime.fromtimestamp(row["b"], TZ).date())


def resolve_period(period: str, from_: str | None, to_: str | None,
                   cov_min: date, today: date) -> tuple[date, date, str]:
    """Bornes [from, to] (dates locales incluses) + libellé humain."""
    if from_ and to_:
        try:
            a = date.fromisoformat(from_)
            b = date.fromisoformat(to_)
            if a > b:
                a, b = b, a
            return max(a, cov_min), min(b, today), f"{a} → {b}"
        except ValueError:
            pass
    if period == "tout":
        return cov_min, today, "Toute la période"
    if period == "annee":
        return max(date(today.year, 1, 1), cov_min), today, "Année en cours"
    label, back = PERIODS.get(period, PERIODS["30j"])
    return max(today - timedelta(days=back), cov_min), today, label


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


def _days(a: date, b: date) -> list[date]:
    return [a + timedelta(days=i) for i in range((b - a).days + 1)]


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


def _fr_money(n: float) -> str:
    return _fr_int(n) + " $"


def _spark(points: list[dict], cap: int = 40) -> list[dict]:
    """Échantillonne une série pour la sparkline d'un KPI (≤ cap points)."""
    if len(points) <= cap:
        return points
    step = (len(points) - 1) / (cap - 1)
    idx = sorted({round(i * step) for i in range(cap)})
    return [points[i] for i in idx if i < len(points)]


def _source_names() -> dict[str, str]:
    """id de connecteur -> nom humain (registre data/sources.json)."""
    try:
        reg = json.loads(SOURCES_PATH.read_text(encoding="utf-8"))["sources"]
        return {s["id"]: s.get("name") or s["id"] for s in reg}
    except (OSError, ValueError, KeyError):
        return {}


# ------------------------------------------------------------------- dashboard
def compute(period: str = "30j", from_: str | None = None,
            to_: str | None = None) -> dict:
    key = f"{period}|{from_ or ''}|{to_ or ''}"
    now = time.time()
    with _cache_lock:
        hit = _cache.get(key)
        if hit and hit[0] > now:
            return hit[1]
    data = _compute(period, from_, to_)
    with _cache_lock:
        _cache[key] = (now + CACHE_TTL, data)
    return data


def _compute(period: str, from_: str | None, to_: str | None) -> dict:
    con = db.connect()
    try:
        return _build(con, period, from_, to_)
    finally:
        con.close()


def _build(con, period: str, from_: str | None, to_: str | None) -> dict:
    today = datetime.now(TZ).date()
    cov_min, _cov_max = _coverage(con)
    if cov_min is None:  # base vide
        return {"updated": datetime.now(TZ).isoformat(),
                "period": {"from": None, "to": None, "label": "—"},
                "kpis": [], "series": [], "breakdowns": [], "tables": [],
                "records": []}

    d_from, d_to, label = resolve_period(period, from_, to_, cov_min, today)
    ts_from = _day_start_ts(d_from)
    ts_to = _day_start_ts(d_to + timedelta(days=1))  # borne exclusive
    iso_from, iso_to = d_from.isoformat(), d_to.isoformat()
    BASE = " FROM jobs WHERE active=1 AND dup_of IS NULL"

    # ---- histogrammes journaliers (réutilisés partout) ----------------------
    starts = {r["d"]: r["n"] for r in con.execute(
        "SELECT date(first_seen,'unixepoch','localtime') d, COUNT(*) n"
        " FROM jobs WHERE dup_of IS NULL GROUP BY d")}
    ends = {r["d"]: r["n"] for r in con.execute(
        "SELECT date(last_seen,'unixepoch','localtime') d, COUNT(*) n"
        " FROM jobs WHERE dup_of IS NULL AND active=0 GROUP BY d")}

    def actives_at(d: date) -> int:
        """Reconstruction : cum(first_seen<=d) − cum(retraits<=d−1)."""
        iso = d.isoformat()
        prev = (d - timedelta(days=1)).isoformat()
        s = sum(n for dd, n in starts.items() if dd <= iso)
        e = sum(n for dd, n in ends.items() if dd <= prev)
        return s - e

    # ---- KPI -----------------------------------------------------------------
    actives_now = con.execute(f"SELECT COUNT(*) n{BASE}").fetchone()["n"]
    employers_now = con.execute(
        f"SELECT COUNT(DISTINCT employer) n{BASE} AND employer<>''"
    ).fetchone()["n"]
    cities_now = con.execute(
        f"SELECT COUNT(DISTINCT city) n{BASE} AND city<>''").fetchone()["n"]
    sources_now = con.execute(
        f"SELECT COUNT(DISTINCT source) n{BASE}").fetchone()["n"]
    new_in = sum(n for d, n in starts.items() if iso_from <= d <= iso_to)
    removed_in = sum(n for d, n in ends.items() if iso_from <= d <= iso_to)

    # période précédente de même longueur — deltas SEULEMENT si couverte
    span = (d_to - d_from).days + 1
    p_from, p_to = d_from - timedelta(days=span), d_from - timedelta(days=1)
    prev_ok = p_from >= cov_min
    new_prev = removed_prev = None
    if prev_ok:
        new_prev = sum(n for d, n in starts.items()
                       if p_from.isoformat() <= d <= p_to.isoformat())
        removed_prev = sum(n for d, n in ends.items()
                           if p_from.isoformat() <= d <= p_to.isoformat())
    actives_prev = actives_at(p_to) if p_to >= cov_min else None

    row = con.execute(
        f"SELECT AVG({_SAL_MID}) avg_s, COUNT(*) n_s{BASE}{_sal_where()}"
    ).fetchone()
    avg_sal, n_sal = row["avg_s"], row["n_s"]
    median_sal = None
    if n_sal:
        median_sal = con.execute(
            f"SELECT {_SAL_MID} m{BASE}{_sal_where()}"
            f" ORDER BY m LIMIT 1 OFFSET ?", (n_sal // 2,)).fetchone()["m"]

    remote_n = con.execute(
        f"SELECT COUNT(*) n{BASE} AND work_mode='teletravail'").fetchone()["n"]
    pct_remote = round(100.0 * remote_n / actives_now, 1) if actives_now else None

    # offres directes (pages carrières + dépôt direct) c. portails agrégateurs
    from .dedup import AGGREGATORS
    _agg = sorted(AGGREGATORS) or ["__aucun__"]
    _ph = ",".join("?" * len(_agg))
    direct_n = con.execute(
        f"SELECT COUNT(*) n{BASE} AND source NOT IN ({_ph})", _agg).fetchone()["n"]
    pct_direct = round(100.0 * direct_n / actives_now, 1) if actives_now else None

    days = _days(d_from, d_to)
    spark_act = _spark([{"t": d.isoformat(), "v": actives_at(d)} for d in days]) \
        if len(days) >= 2 else None
    spark_new = _spark([{"t": d.isoformat(), "v": starts.get(d.isoformat(), 0)}
                        for d in days]) if len(days) >= 2 else None
    spark_rem = _spark([{"t": d.isoformat(), "v": ends.get(d.isoformat(), 0)}
                        for d in days]) if len(days) >= 2 else None

    kpis = [
        {"id": "actives", "label": "Offres actives", "value": actives_now,
         "delta_pct": _pct(actives_now, actives_prev) if actives_prev else None,
         "direction": ("up" if actives_now >= actives_prev else "down")
                      if actives_prev else None,
         "spark": spark_act},
        {"id": "nouvelles", "label": "Nouvelles offres (période)",
         "value": new_in,
         "delta_pct": _pct(new_in, new_prev) if prev_ok else None,
         "direction": ("up" if new_in >= (new_prev or 0) else "down")
                      if prev_ok else None,
         "spark": spark_new},
        {"id": "retirees", "label": "Offres retirées ou expirées (période)",
         "value": removed_in,
         "delta_pct": _pct(removed_in, removed_prev) if prev_ok else None,
         "direction": ("down" if removed_in >= (removed_prev or 0) else "up")
                      if prev_ok else None,
         "spark": spark_rem},
        {"id": "employeurs", "label": "Employeurs avec offres actives",
         "value": employers_now},
        {"id": "villes", "label": "Villes couvertes (offres actives)",
         "value": cities_now},
        {"id": "sources", "label": "Sources avec offres actives",
         "value": sources_now},
    ]
    if median_sal:
        kpis.append({"id": "salaire_median",
                     "label": "Salaire annuel médian publié",
                     "value": round(median_sal), "unit": "$"})
    if avg_sal:
        kpis.append({"id": "salaire_moyen",
                     "label": "Salaire annuel moyen publié",
                     "value": round(avg_sal), "unit": "$"})
    if pct_remote is not None:
        kpis.append({"id": "teletravail", "label": "Offres en télétravail",
                     "value": pct_remote, "unit": "%"})
    if pct_direct is not None:
        kpis.append({"id": "directes", "label": "Offres directes employeur",
                     "value": pct_direct, "unit": "%"})
    kpis = [{k: v for k, v in kpi.items() if v is not None} for kpi in kpis]

    # ---- jauges (complétude des fiches actives) --------------------------------
    gauges = []
    if actives_now:
        comp = con.execute(
            f"""SELECT
                  SUM(CASE WHEN (salary_year_min IS NOT NULL
                                 OR salary_year_max IS NOT NULL
                                 OR salary_hour_min IS NOT NULL
                                 OR salary_hour_max IS NOT NULL)
                       THEN 1 ELSE 0 END) sal,
                  SUM(CASE WHEN work_mode IS NOT NULL AND work_mode<>''
                       THEN 1 ELSE 0 END) mode,
                  SUM(CASE WHEN lat IS NOT NULL AND lng IS NOT NULL
                       THEN 1 ELSE 0 END) geo,
                  SUM(CASE WHEN category<>'' THEN 1 ELSE 0 END) cat
                {BASE}""").fetchone()
        gauges = [
            {"id": "salaire", "label": "Offres avec salaire affiché",
             "value": round(100.0 * comp["sal"] / actives_now, 1), "max": 100,
             "unit": "%",
             "help": "Part des offres actives dont l'employeur publie un salaire"},
            {"id": "mode", "label": "Mode de travail précisé",
             "value": round(100.0 * comp["mode"] / actives_now, 1), "max": 100,
             "unit": "%",
             "help": "Présentiel, hybride ou télétravail explicitement indiqué"},
            {"id": "geoloc", "label": "Offres géolocalisées",
             "value": round(100.0 * comp["geo"] / actives_now, 1), "max": 100,
             "unit": "%",
             "help": "Offres positionnées sur la carte (lat/lng)"},
            {"id": "categorie", "label": "Offres catégorisées",
             "value": round(100.0 * comp["cat"] / actives_now, 1), "max": 100,
             "unit": "%",
             "help": "Offres rattachées à un secteur de la taxonomie Job·Ka"},
        ]

    # ---- séries temporelles ----------------------------------------------------
    series = []
    if len(days) >= 2:
        s_act = {"id": "actives_jour", "title": "Offres actives par jour",
                 "unit": "offres", "kind": "line",
                 "points": [{"t": d.isoformat(), "v": actives_at(d)} for d in days]}
        s_new = {"id": "nouvelles_jour", "title": "Nouvelles offres par jour",
                 "unit": "offres", "kind": "bar",
                 "points": [{"t": d.isoformat(),
                             "v": starts.get(d.isoformat(), 0)} for d in days]}
        s_rem = {"id": "retraits_jour",
                 "title": "Offres retirées ou expirées par jour",
                 "unit": "offres", "kind": "bar",
                 "points": [{"t": d.isoformat(),
                             "v": ends.get(d.isoformat(), 0)} for d in days]}
        if prev_ok:
            pdays = _days(p_from, p_to)
            s_act["compare"] = [{"t": d.isoformat(), "v": actives_at(d)}
                                for d in pdays]
        series = [s_act, s_new, s_rem]

    # dates de publication AFFICHÉES par les employeurs — données réelles qui
    # précèdent la mise en service de Job·Ka : la fenêtre demandée est prise
    # SANS la borner à la couverture d'observation (first_seen).
    rq_from, rq_to, _ = resolve_period(period, from_, to_, date(2000, 1, 1), today)
    rq_from = max(rq_from, today - timedelta(days=365))  # 12 mois max (lisibilité)
    posted_rows = con.execute(
        f"SELECT date_posted d, COUNT(*) n{BASE}"
        " AND date_posted IS NOT NULL AND date_posted>=? AND date_posted<=?"
        " GROUP BY date_posted ORDER BY date_posted",
        (rq_from.isoformat(), rq_to.isoformat())).fetchall()
    if len(posted_rows) >= 2:
        by_day = {r["d"]: r["n"] for r in posted_rows}
        series.append({"id": "publiees_jour",
                       "title": "Offres publiées par jour "
                                "(date affichée par l'employeur)",
                       "unit": "offres", "kind": "line",
                       "points": [{"t": d.isoformat(),
                                   "v": by_day.get(d.isoformat(), 0)}
                                  for d in _days(rq_from, rq_to)]})

    # ---- multi-courbes : publications par top catégories (≤ 4) -----------------
    multiseries = []
    top_cats = [r["c"] for r in con.execute(
        f"SELECT category c, COUNT(*) n{BASE} AND category<>''"
        " AND date_posted IS NOT NULL AND date_posted>=? AND date_posted<=?"
        " GROUP BY category ORDER BY n DESC LIMIT 4",
        (rq_from.isoformat(), rq_to.isoformat()))]
    if top_cats:
        rows = con.execute(
            f"SELECT category c, date_posted d, COUNT(*) n{BASE}"
            f" AND category IN ({','.join('?' * len(top_cats))})"
            " AND date_posted IS NOT NULL AND date_posted>=? AND date_posted<=?"
            " GROUP BY category, date_posted",
            (*top_cats, rq_from.isoformat(), rq_to.isoformat())).fetchall()
        grid: dict[str, dict[str, int]] = {c: {} for c in top_cats}
        for r in rows:
            grid[r["c"]][r["d"]] = r["n"]
        mdays = _days(rq_from, rq_to)
        ms = [{"label": c,
               "points": [{"t": d.isoformat(), "v": grid[c].get(d.isoformat(), 0)}
                          for d in mdays]}
              for c in top_cats]
        if len(mdays) >= 2 and any(sum(p["v"] for p in s["points"]) for s in ms):
            multiseries.append({
                "id": "cat_pub",
                "title": "Offres publiées par jour — top secteurs "
                         "(date affichée par l'employeur)",
                "unit": "offres", "series": ms})

    # ---- barres empilées : ajouts observés par source (sync_log) ---------------
    stacked = []
    add_rows = con.execute(
        "SELECT date(ts,'unixepoch','localtime') d, source, SUM(added) n"
        " FROM sync_log WHERE ts>=? AND ts<? AND added>0"
        " GROUP BY d, source", (ts_from, ts_to)).fetchall()
    if add_rows:
        names = _source_names()
        totals: dict[str, int] = {}
        for r in add_rows:
            totals[r["source"]] = totals.get(r["source"], 0) + r["n"]
        top_src = [s for s, _ in sorted(totals.items(), key=lambda kv: -kv[1])[:5]]
        keys = [names.get(s, s) for s in top_src]
        others = len(totals) > len(top_src)
        if others:
            keys.append("Autres")
        grid2: dict[str, list[int]] = {}
        for r in add_rows:
            vals = grid2.setdefault(r["d"], [0] * len(keys))
            if r["source"] in top_src:
                vals[top_src.index(r["source"])] += r["n"]
            elif others:
                vals[-1] += r["n"]
        pts = [{"t": d.isoformat(),
                "values": grid2.get(d.isoformat(), [0] * len(keys))}
               for d in days]
        if any(sum(p["values"]) for p in pts):
            stacked.append({"id": "ajouts_source",
                            "title": "Offres ajoutées par source (journal de "
                                     "synchronisation)",
                            "unit": "offres", "keys": keys, "points": pts})

    # ---- répartitions ----------------------------------------------------------
    # deltas de répartition : reconstruction des actifs à p_to par dimension,
    # seulement si la période précédente est couverte (deltas honnêtes)
    def _dim_delta(col: str) -> dict[str, float]:
        if p_to < cov_min:
            return {}
        ts_prev = _day_start_ts(p_to + timedelta(days=1))
        s_prev = {r["k"]: r["n"] for r in con.execute(
            f"SELECT {col} k, COUNT(*) n FROM jobs WHERE dup_of IS NULL"
            " AND first_seen<? GROUP BY k", (ts_prev,))}
        e_prev = {r["k"]: r["n"] for r in con.execute(
            f"SELECT {col} k, COUNT(*) n FROM jobs WHERE dup_of IS NULL"
            " AND active=0 AND last_seen<? GROUP BY k",
            (_day_start_ts(p_to),))}
        cur = {r["k"]: r["n"] for r in con.execute(
            f"SELECT {col} k, COUNT(*) n{BASE} GROUP BY k")}
        out = {}
        for k, n in cur.items():
            prev = s_prev.get(k, 0) - e_prev.get(k, 0)
            d = _pct(n, prev)
            if d is not None:
                out[k] = d
        return out

    breakdowns = []
    cat_delta = _dim_delta("category")
    cat_items = []
    for r in con.execute(
            f"SELECT category c, COUNT(*) n{BASE} AND category<>''"
            " GROUP BY category ORDER BY n DESC LIMIT 12"):
        it = {"label": r["c"], "value": r["n"]}
        if r["c"] in cat_delta:
            it["delta_pct"] = cat_delta[r["c"]]
        cat_items.append(it)
    if cat_items:
        breakdowns.append({"id": "categories",
                           "title": "Offres actives par secteur",
                           "kind": "bar", "items": cat_items})

    type_items = [{"label": TYPE_FR.get(r["t"], r["t"] or "Non précisé"),
                   "value": r["n"]}
                  for r in con.execute(
                      f"SELECT employment_type t, COUNT(*) n{BASE}"
                      " GROUP BY employment_type ORDER BY n DESC")]
    if type_items:
        breakdowns.append({"id": "types",
                           "title": "Temps plein, temps partiel, contrat…",
                           "kind": "donut", "items": type_items})

    mode_items = [{"label": MODE_FR.get(r["m"], r["m"] or "Non précisé"),
                   "value": r["n"]}
                  for r in con.execute(
                      f"SELECT work_mode m, COUNT(*) n{BASE}"
                      " GROUP BY work_mode ORDER BY n DESC")]
    if mode_items:
        breakdowns.append({"id": "modes",
                           "title": "Télétravail, hybride, présentiel",
                           "kind": "donut", "items": mode_items})

    ats_items = [{"label": ATS_FR.get(r["a"] or "", (r["a"] or "").title()),
                  "value": r["n"]}
                 for r in con.execute(
                     f"SELECT ats a, COUNT(*) n{BASE}"
                     " GROUP BY ats ORDER BY n DESC LIMIT 10")]
    if ats_items:
        breakdowns.append({"id": "ats",
                           "title": "Offres actives par plateforme ATS",
                           "kind": "donut", "items": ats_items})

    # ---- distributions -----------------------------------------------------------
    distributions = []
    sal_bins = []
    for lo, hi, blabel in SAL_BUCKETS:
        n = con.execute(
            f"SELECT COUNT(*) n{BASE}{_sal_where()}"
            f" AND {_SAL_MID} >= ? AND {_SAL_MID} < ?", (lo, hi)).fetchone()["n"]
        sal_bins.append({"label": blabel, "value": n})
    if any(b["value"] for b in sal_bins):
        distributions.append({"id": "salaires_annuels",
                              "title": "Distribution des salaires annuels "
                                       "publiés (offres actives)",
                              "unit": "offres", "bins": sal_bins})

    salh_bins = []
    for lo, hi, blabel in SALH_BUCKETS:
        n = con.execute(
            f"SELECT COUNT(*) n{BASE}{_salh_where()}"
            f" AND {_SALH_MID} >= ? AND {_SALH_MID} < ?", (lo, hi)).fetchone()["n"]
        salh_bins.append({"label": blabel, "value": n})
    if any(b["value"] for b in salh_bins):
        distributions.append({"id": "salaires_horaires",
                              "title": "Distribution des taux horaires "
                                       "publiés (offres actives)",
                              "unit": "offres", "bins": salh_bins})

    age_bins = []
    for lo, hi, blabel in AGE_BUCKETS:
        n = con.execute(
            f"""SELECT COUNT(*) n{BASE} AND date_posted IS NOT NULL
                AND CAST(julianday('now','localtime')
                         - julianday(date_posted) AS INTEGER) >= ?
                AND CAST(julianday('now','localtime')
                         - julianday(date_posted) AS INTEGER) < ?""",
            (lo, hi)).fetchone()["n"]
        age_bins.append({"label": blabel, "value": n})
    if any(b["value"] for b in age_bins):
        distributions.append({"id": "age_offres",
                              "title": "Âge des offres actives (jours depuis "
                                       "la publication affichée)",
                              "unit": "offres", "bins": age_bins})

    # ---- géographie (région administrative peu remplie -> villes) --------------
    geo_items = [{"label": r["c"], "value": r["n"]} for r in con.execute(
        f"SELECT city c, COUNT(*) n{BASE} AND city<>''"
        " GROUP BY city ORDER BY n DESC LIMIT 14")]
    geo = ({"title": "Top villes (offres actives)", "items": geo_items}
           if geo_items else None)

    # ---- heatmap calendrier (nouvelles offres observées par jour) --------------
    heat_cells = [{"date": d.isoformat(), "value": starts.get(d.isoformat(), 0)}
                  for d in days if starts.get(d.isoformat())]
    heatmap = ({"title": "Nouvelles offres par jour", "cells": heat_cells}
               if heat_cells else None)

    # ---- heatmap horaire 7×24 (ajouts observés par les connecteurs) ------------
    hourly = None
    hr_rows = con.execute(
        """SELECT CAST(strftime('%w', ts,'unixepoch','localtime') AS INTEGER) w,
                  CAST(strftime('%H', ts,'unixepoch','localtime') AS INTEGER) h,
                  SUM(added) n
           FROM sync_log WHERE ts>=? AND ts<? AND added>0
           GROUP BY w, h""", (ts_from, ts_to)).fetchall()
    hr_cells = [{"dow": (r["w"] + 6) % 7, "hour": r["h"], "value": r["n"]}
                for r in hr_rows if r["n"]]
    if hr_cells:
        hourly = {"title": "Offres ajoutées par heure d'observation",
                  "cells": hr_cells}

    # ---- tableaux ----------------------------------------------------------------
    tables = []
    top_emp = con.execute(
        f"""SELECT employer, COUNT(*) n, COUNT(DISTINCT city) nc,
                   AVG(CASE WHEN {_SAL_MID} BETWEEN {SAL_MIN} AND {SAL_MAX}
                        AND (salary_year_min IS NOT NULL
                             OR salary_year_max IS NOT NULL)
                        THEN {_SAL_MID} END) avg_s,
                   SUM(CASE WHEN first_seen>=? AND first_seen<? THEN 1 ELSE 0 END) new_n
            {BASE} AND employer<>''
            GROUP BY employer ORDER BY n DESC LIMIT 50""",
        (ts_from, ts_to)).fetchall()
    if top_emp:
        tables.append({
            "id": "top_employeurs", "title": "Top employeurs",
            "columns": ["Employeur", "Offres actives", "Villes",
                        "Salaire annuel moyen publié", "Nouvelles (période)"],
            "rows": [[r["employer"], r["n"], r["nc"],
                      _fr_money(r["avg_s"]) if r["avg_s"] else "—",
                      r["new_n"]] for r in top_emp]})

    top_cities = con.execute(
        f"""SELECT city, COUNT(*) n, COUNT(DISTINCT employer) ne,
                   AVG(CASE WHEN {_SAL_MID} BETWEEN {SAL_MIN} AND {SAL_MAX}
                        AND (salary_year_min IS NOT NULL
                             OR salary_year_max IS NOT NULL)
                        THEN {_SAL_MID} END) avg_s,
                   SUM(CASE WHEN first_seen>=? AND first_seen<? THEN 1 ELSE 0 END) new_n
            {BASE} AND city<>''
            GROUP BY city ORDER BY n DESC LIMIT 50""",
        (ts_from, ts_to)).fetchall()
    if top_cities:
        tables.append({
            "id": "top_villes", "title": "Top villes",
            "columns": ["Ville", "Offres actives", "Employeurs",
                        "Salaire annuel moyen publié", "Nouvelles (période)"],
            "rows": [[r["city"], r["n"], r["ne"],
                      _fr_money(r["avg_s"]) if r["avg_s"] else "—",
                      r["new_n"]] for r in top_cities]})

    cat_rows = con.execute(
        f"""SELECT category, COUNT(*) n,
                   AVG(CASE WHEN {_SAL_MID} BETWEEN {SAL_MIN} AND {SAL_MAX}
                        AND (salary_year_min IS NOT NULL
                             OR salary_year_max IS NOT NULL)
                        THEN {_SAL_MID} END) avg_s,
                   SUM(CASE WHEN first_seen>=? AND first_seen<? THEN 1 ELSE 0 END) new_n
            {BASE} AND category<>''
            GROUP BY category ORDER BY n DESC LIMIT 30""",
        (ts_from, ts_to)).fetchall()
    if cat_rows:
        def _dlt(c):
            d = cat_delta.get(c)
            return (("+" if d >= 0 else "") + f"{d:.1f}".replace(".", ",") + " %") \
                if d is not None else "—"
        tables.append({
            "id": "secteurs", "title": "Secteurs d'emploi",
            "columns": ["Secteur", "Offres actives",
                        "Salaire annuel moyen publié", "Nouvelles (période)",
                        "Δ actives"],
            "rows": [[r["category"], r["n"],
                      _fr_money(r["avg_s"]) if r["avg_s"] else "—",
                      r["new_n"], _dlt(r["category"])] for r in cat_rows]})

    # sources & connecteurs : offres actives + journal de synchronisation
    names = _source_names()
    src_jobs = {r["s"]: r["n"] for r in con.execute(
        f"SELECT source s, COUNT(*) n{BASE} GROUP BY source")}
    src_sync = {r["s"]: r for r in con.execute(
        """SELECT source s, SUM(CASE WHEN ts>=? AND ts<? THEN added ELSE 0 END) a,
                  MAX(ts) last_ts
           FROM sync_log GROUP BY source""", (ts_from, ts_to))}
    src_last_ok = {r["s"]: r["ok"] for r in con.execute(
        """SELECT source s, ok FROM sync_log
           WHERE id IN (SELECT MAX(id) FROM sync_log GROUP BY source)""")}
    src_rows = []
    for s, n in sorted(src_jobs.items(), key=lambda kv: -kv[1])[:60]:
        sy = src_sync.get(s)
        last = (datetime.fromtimestamp(sy["last_ts"], TZ)
                .strftime("%Y-%m-%d %H:%M") if sy and sy["last_ts"] else "—")
        src_rows.append([names.get(s, s), n,
                         sy["a"] if sy else 0, last,
                         "OK" if src_last_ok.get(s, 1) else "Échec"])
    if src_rows:
        tables.append({
            "id": "sources", "title": "Sources & connecteurs",
            "columns": ["Source", "Offres actives", "Ajouts (période)",
                        "Dernière synchro", "Statut"],
            "rows": src_rows})

    new_rows = con.execute(
        f"""SELECT title, employer, city, salary_label, date_posted
            {BASE} AND first_seen>=? AND first_seen<?
            ORDER BY date_posted IS NULL, date_posted DESC, first_seen DESC
            LIMIT 50""", (ts_from, ts_to)).fetchall()
    if new_rows:
        tables.append({
            "id": "nouvelles_offres", "title": "Nouvelles offres (période)",
            "columns": ["Titre", "Employeur", "Ville", "Salaire publié",
                        "Publiée le"],
            "rows": [[r["title"][:70], r["employer"], r["city"] or "—",
                      r["salary_label"] or "—", r["date_posted"] or "—"]
                     for r in new_rows]})

    # ---- records & faits marquants -------------------------------------------------
    records = []
    in_period = {d: n for d, n in starts.items() if iso_from <= d <= iso_to}
    if in_period:
        best = max(in_period, key=in_period.get)
        records.append({"label": "Jour record de nouvelles offres",
                        "value": _fr_int(in_period[best]) + " offres",
                        "date": best})
    rem_period = {d: n for d, n in ends.items() if iso_from <= d <= iso_to}
    if rem_period:
        worst = max(rem_period, key=rem_period.get)
        records.append({"label": "Jour record de retraits d'offres",
                        "value": _fr_int(rem_period[worst]) + " offres",
                        "date": worst})
    top_sal = con.execute(
        f"""SELECT title, employer, salary_year_max s
            {BASE} AND salary_year_max BETWEEN {SAL_MIN} AND {SAL_MAX}
            ORDER BY salary_year_max DESC LIMIT 1""").fetchone()
    if top_sal:
        records.append({
            "label": "Salaire publié le plus élevé (borné, offres actives)",
            "value": f"{_fr_money(top_sal['s'])} / an — "
                     f"{top_sal['title'][:40]} ({top_sal['employer']})"})
    if top_emp:
        emp = max(top_emp, key=lambda r: r["new_n"])
        if emp["new_n"]:
            records.append({"label": "Employeur le plus actif "
                                     "(nouvelles offres, période)",
                            "value": f"{emp['employer']} — "
                                     f"{_fr_int(emp['new_n'])} offres"})
        records.append({"label": "Employeur au plus grand nombre "
                                 "d'offres actives",
                        "value": f"{top_emp[0]['employer']} — "
                                 f"{_fr_int(top_emp[0]['n'])} offres"})
    if top_cities:
        dyn = max(top_cities, key=lambda r: r["new_n"])
        if dyn["new_n"]:
            records.append({"label": "Ville la plus dynamique "
                                     "(nouvelles offres, période)",
                            "value": f"{dyn['city']} — "
                                     f"{_fr_int(dyn['new_n'])} offres"})
    if cat_rows:
        records.append({"label": "Secteur dominant (offres actives)",
                        "value": f"{cat_rows[0]['category']} — "
                                 f"{_fr_int(cat_rows[0]['n'])} offres"})
    if src_rows:
        records.append({"label": "Source la plus fournie (offres actives)",
                        "value": f"{src_rows[0][0]} — "
                                 f"{_fr_int(src_rows[0][1])} offres"})
    oldest = con.execute(
        f"""SELECT title, employer, date_posted{BASE}
            AND date_posted IS NOT NULL AND date_posted >= '2000-01-01'
            ORDER BY date_posted ASC LIMIT 1""").fetchone()
    if oldest:
        records.append({"label": "Offre active la plus ancienne "
                                 "(date de publication affichée)",
                        "value": f"{oldest['title'][:40]} "
                                 f"({oldest['employer']})",
                        "date": oldest["date_posted"]})
    total_obs = con.execute(
        "SELECT COUNT(*) n FROM jobs WHERE dup_of IS NULL").fetchone()["n"]
    records.append({"label": "Offres observées depuis le lancement",
                    "value": _fr_int(total_obs) + " offres",
                    "date": cov_min.isoformat()})
    dups = con.execute(
        "SELECT COUNT(*) n FROM jobs WHERE active=1"
        " AND dup_of IS NOT NULL").fetchone()["n"]
    if dups:
        records.append({"label": "Doublons inter-sources masqués",
                        "value": _fr_int(dups) + " offres"})

    out = {
        "updated": datetime.now(TZ).isoformat(),
        "period": {"from": iso_from, "to": iso_to, "label": label},
        "kpis": kpis,
        "series": series,
        "breakdowns": breakdowns,
        "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 geo:
        out["geo"] = geo
    if heatmap:
        out["heatmap"] = heatmap
    if hourly:
        out["hourly"] = hourly
    return out