# ============================================================================== # Author: Simon-Pierre Boucher # File: creaka/stats.py # Desc: Tableau de bord analytique /api/stats/dashboard (contrat ka-stats # SPEC.md v2) + rapport PDF Groupe-KA (/api/stats/report via kapdf.py, # 5 modes). AUCUNE stat inventée : tout est calculé depuis creators/ # accounts/sync_log (first_seen converti en date locale, cache 5 min). # v2 : sparklines KPI, jauges de complétude, multi-courbes par # plateforme, ajouts empilés par source, distributions (audiences, # comptes/créateur, confiance), heatmap horaire 7×24, 5 tableaux, # records enrichis. # ============================================================================== from __future__ import annotations import json import sqlite3 import time from datetime import date, datetime, timedelta, timezone from pathlib import Path from zoneinfo import ZoneInfo TZ = ZoneInfo("America/Toronto") ROOT = Path(__file__).resolve().parent.parent ECO_PATH = ROOT / "frontend" / "src" / "ka" / "ecosystem.json" CACHE_TTL = 300 # ≥ 5 min par période (SPEC §2) _CACHE: dict[tuple, tuple[float, dict]] = {} PERIOD_DAYS = {"auj": 1, "7j": 7, "30j": 30, "3m": 91, "6m": 182, "12m": 365} 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", } PLAT_LBL = { "instagram": "Instagram", "tiktok": "TikTok", "youtube": "YouTube", "twitch": "Twitch", "kick": "Kick", "x": "X (Twitter)", "facebook": "Facebook", "snapchat": "Snapchat", "substack": "Substack", "patreon": "Patreon", "onlyfans": "OnlyFans", "linkedin": "LinkedIn", "threads": "Threads", "podcast": "Balado", "site-web": "Site web", "spotify": "Spotify", "discord": "Discord", "fansly": "Fansly", "mym": "MYM", "autre": "Autre", } NICHE_LBL = { "humour": "Humour", "mode": "Mode", "beaute": "Beauté", "lifestyle": "Lifestyle", "famille-parentalite": "Famille & parentalité", "cuisine": "Cuisine", "gaming": "Gaming", "tech": "Tech", "sport-fitness": "Sport & fitness", "plein-air": "Plein air", "voyage": "Voyage", "musique": "Musique", "arts": "Arts", "danse": "Danse", "education": "Éducation", "finance-affaires": "Finance & affaires", "sante-mieux-etre": "Santé & mieux-être", "bouffe-resto": "Bouffe & resto", "actualite-opinion": "Actualité & opinion", "autre": "Autre", } TYPE_LBL = { "youtubeur": "Youtubeur", "influenceur": "Influenceur", "streamer": "Streamer", "podcasteur": "Podcasteur", "humoriste": "Humoriste", "createur-tiktok": "Créateur TikTok", "createur-ecrit": "Créateur écrit", "musicien": "Musicien", "artiste": "Artiste", "autre": "Autre", } # bornes réelles de normalize.audience_tier (§6.3) TIER_LBL = [("nano", "Nano (< 10 k)"), ("micro", "Micro (10 k – 100 k)"), ("macro", "Macro (100 k – 1 M)"), ("mega", "Méga (1 M et +)")] # tranches d'audience (distribution) — bornes en abonnés cumulés connus REACH_BINS = [(0, 1_000, "< 1 k"), (1_000, 10_000, "1 k – 10 k"), (10_000, 50_000, "10 k – 50 k"), (50_000, 100_000, "50 k – 100 k"), (100_000, 500_000, "100 k – 500 k"), (500_000, 1_000_000, "500 k – 1 M"), (1_000_000, None, "1 M et +")] def site_info() -> dict: """Identité Créa-Ka pour le PDF, lue dans ka/ecosystem.json (source commune).""" try: eco = json.loads(ECO_PATH.read_text(encoding="utf-8")) s = next(x for x in eco["sites"] if x["id"] == "crea-ka") return {"wordmark": s["wordmark"], "accent": s["accent"], "domain": s["domain"], "tagline": s.get("tagline", "")} except Exception: return {"wordmark": "Créa·Ka", "accent": "#7048e8", "domain": "www.crea-ka.com", "tagline": "Les créateurs d'ici, tous leurs liens"} # --- utilitaires ----------------------------------------------------------------- def _local_dt(iso: str) -> datetime | None: """ISO-8601 UTC (…Z) → datetime local (America/Toronto).""" if not iso: return None try: dt = datetime.strptime(iso[:19], "%Y-%m-%dT%H:%M:%S") return dt.replace(tzinfo=timezone.utc).astimezone(TZ) except ValueError: return None def _local_date(iso: str) -> date | None: dt = _local_dt(iso) if dt: return dt.date() try: return date.fromisoformat((iso or "")[:10]) except ValueError: return None def _parse_date(s: str) -> date | None: try: return date.fromisoformat(s.strip()[:10]) except (ValueError, AttributeError): return None def _pct(cur: float, prev: float) -> float | None: """Variation vs période précédente ; None si la base est nulle (pas de faux %).""" if not prev: return None return round(100.0 * (cur - prev) / prev, 1) def _delta(cur: float, prev: float) -> dict: p = _pct(cur, prev) if p is None: return {"delta_pct": None} return {"delta_pct": p, "direction": "up" if p >= 0 else "down"} def _fr_int(n: int) -> str: return f"{int(n):,}".replace(",", " ") def _spark(points: list[dict], keep: int = 20) -> list[dict]: """Sous-échantillonne une série pour la mini-tendance des KPI (≤ keep pts).""" if len(points) <= keep: return points step = (len(points) - 1) / (keep - 1) return [points[round(i * step)] for i in range(keep)] # --- construction du tableau de bord ---------------------------------------------- def _resolve_period(period: str, d_from: str, d_to: str, min_day: date | None) -> tuple[date, date, str]: today = datetime.now(TZ).date() f, t = _parse_date(d_from), _parse_date(d_to) if f and t: if t < f: f, t = t, f return f, t, f"du {f.isoformat()} au {t.isoformat()}" if period == "annee": return date(today.year, 1, 1), today, f"année {today.year}" if period == "tout": # depuis la première fiche (plancher 30 j pour des courbes lisibles) start = min(min_day or today, today - timedelta(days=29)) return start, today, PERIOD_LABELS["tout"] days = PERIOD_DAYS.get(period, 30) label = PERIOD_LABELS.get(period, PERIOD_LABELS["30j"]) return today - timedelta(days=days - 1), today, label def _build(con: sqlite3.Connection, period: str, d_from: str, d_to: str) -> dict: # fiches actives (mineurs & opt-out exclus, comme partout dans l'API) creators = con.execute( "SELECT id, display_name, first_seen, niches, audience_tier, region, " "city, bio, creator_type, total_reach, primary_platform, " "json_extract(doc,'$.source') AS src, " "json_extract(doc,'$.avatar_url') IS NOT NULL AS has_avatar " "FROM creators WHERE status='active' AND is_minor=0").fetchall() n_active = len(creators) dts_seen = [_local_dt(r["first_seen"]) for r in creators] days_seen = [d.date() for d in dts_seen if d] min_day = min(days_seen) if days_seen else None p_from, p_to, p_label = _resolve_period(period, d_from, d_to, min_day) span = (p_to - p_from).days + 1 prev_to = p_from - timedelta(days=1) prev_from = prev_to - timedelta(days=span - 1) # ajouts par jour (toute l'historique) — sert séries, heatmap, records adds_by_day: dict[date, int] = {} for d in days_seen: adds_by_day[d] = adds_by_day.get(d, 0) + 1 def added_between(a: date, b: date) -> int: return sum(v for d, v in adds_by_day.items() if a <= d <= b) def total_until(d: date) -> int: return sum(v for dd, v in adds_by_day.items() if dd <= d) # comptes reliés par plateforme (comptes « à vérifier » exclus, §12.1) + # agrégats abonnés/vérifiés pour le tableau plateformes plat_rows = con.execute( "SELECT a.platform, COUNT(*) c, " "SUM(CASE WHEN a.verified=1 THEN 1 ELSE 0 END) nverif, " "SUM(COALESCE(a.followers,0)) fol, " "COUNT(a.followers) nfol " "FROM accounts a JOIN creators c2 ON c2.id=a.creator_id " "WHERE a.needs_review=0 AND c2.status='active' AND c2.is_minor=0 " "GROUP BY a.platform ORDER BY c DESC").fetchall() n_accounts = sum(r["c"] for r in plat_rows) n_verified = sum(r["nverif"] or 0 for r in plat_rows) # comptes par créateur (multi-plateforme, distribution, record) acc_per_creator = con.execute( "SELECT a.creator_id, c2.display_name, COUNT(DISTINCT a.platform) np " "FROM accounts a JOIN creators c2 ON c2.id=a.creator_id " "WHERE a.needs_review=0 AND c2.status='active' AND c2.is_minor=0 " "GROUP BY a.creator_id").fetchall() n_multi = sum(1 for r in acc_per_creator if r["np"] >= 2) # confiance des rattachements (distribution) conf_rows = con.execute( "SELECT a.confidence FROM accounts a JOIN creators c2 ON c2.id=a.creator_id " "WHERE a.needs_review=0 AND c2.status='active' AND c2.is_minor=0").fetchall() # niches (multivaluées) — global + ajouts sur la période niche_total: dict[str, int] = {} niche_period: dict[str, int] = {} tier_total: dict[str, int] = {} tier_period: dict[str, int] = {} type_total: dict[str, int] = {} region_total: dict[str, int] = {} city_total: dict[str, int] = {} src_total: dict[str, int] = {} # ajouts par jour et par source (empilé) + par plateforme principale src_day: dict[tuple, int] = {} plat_day: dict[tuple, int] = {} reach_day: dict[date, int] = {} hour_cells: dict[tuple, int] = {} for r, dt_loc in zip(creators, dts_seen): d = dt_loc.date() if dt_loc else None in_p = bool(d and p_from <= d <= p_to) for n in (r["niches"] or "").split(","): if not n: continue niche_total[n] = niche_total.get(n, 0) + 1 if in_p: niche_period[n] = niche_period.get(n, 0) + 1 tier_total[r["audience_tier"]] = tier_total.get(r["audience_tier"], 0) + 1 if in_p: tier_period[r["audience_tier"]] = tier_period.get(r["audience_tier"], 0) + 1 if r["creator_type"]: type_total[r["creator_type"]] = type_total.get(r["creator_type"], 0) + 1 if r["region"]: region_total[r["region"]] = region_total.get(r["region"], 0) + 1 if r["city"]: city_total[r["city"]] = city_total.get(r["city"], 0) + 1 if r["src"]: src_total[r["src"]] = src_total.get(r["src"], 0) + 1 if d: src_day[(d, r["src"])] = src_day.get((d, r["src"]), 0) + 1 if d and r["primary_platform"]: plat_day[(d, r["primary_platform"])] = \ plat_day.get((d, r["primary_platform"]), 0) + 1 if d and r["total_reach"]: reach_day[d] = reach_day.get(d, 0) + r["total_reach"] if dt_loc: key = (dt_loc.weekday(), dt_loc.hour) # 0=lun … 6=dim (SPEC) hour_cells[key] = hour_cells.get(key, 0) + 1 # ---- KPI (deltas seulement quand ils sont réellement calculables) ---- added_cur = added_between(p_from, p_to) added_prev = added_between(prev_from, prev_to) total_end = total_until(p_to) total_start = total_until(prev_to) day_axis = [p_from + timedelta(days=i) for i in range(span)] pts_added = [{"t": d.isoformat(), "v": adds_by_day.get(d, 0)} for d in day_axis] running = total_until(p_from - timedelta(days=1)) pts_cumul = [] for d in day_axis: running += adds_by_day.get(d, 0) pts_cumul.append({"t": d.isoformat(), "v": running}) reach = sum(r["total_reach"] for r in creators if r["total_reach"]) kpis = [ {"id": "creators", "label": "Créateurs au répertoire", "value": total_end, "unit": "", **_delta(total_end, total_start), "spark": _spark(pts_cumul)}, {"id": "accounts", "label": "Comptes publics reliés", "value": n_accounts, "unit": "", "delta_pct": None}, {"id": "added", "label": "Créateurs ajoutés sur la période", "value": added_cur, "unit": "", **_delta(added_cur, added_prev), "spark": _spark(pts_added)}, {"id": "platforms", "label": "Plateformes couvertes", "value": len(plat_rows), "unit": "", "delta_pct": None}, {"id": "multi", "label": "Créateurs multi-plateformes", "value": n_multi, "unit": "", "delta_pct": None}, {"id": "verified", "label": "Comptes vérifiés (badge)", "value": n_verified, "unit": "", "delta_pct": None}, {"id": "niches", "label": "Niches couvertes", "value": len(niche_total), "unit": "", "delta_pct": None}, {"id": "regions", "label": "Régions représentées", "value": len(region_total), "unit": "", "delta_pct": None}, ] if reach: kpis.append({"id": "reach", "label": "Portée cumulée connue", "value": reach, "unit": "abonnés", "delta_pct": None}) if n_active: kpis.append({"id": "acc_avg", "label": "Comptes reliés par créateur (moy.)", "value": round(n_accounts / n_active, 2), "unit": "", "delta_pct": None}) # ---- jauges : couvertures & complétude (calculées, pas estimées) ---- def _cov(n: int) -> float: return round(100.0 * n / n_active, 1) if n_active else 0.0 n_bio = sum(1 for r in creators if r["bio"]) n_loc = sum(1 for r in creators if r["region"] or r["city"]) n_reach = sum(1 for r in creators if r["total_reach"]) n_avatar = sum(1 for r in creators if r["has_avatar"]) # complétude moyenne d'une fiche = moyenne des 4 champs clés remplis completeness = round((_cov(n_bio) + _cov(n_loc) + _cov(n_reach) + _cov(n_avatar)) / 4, 1) if n_active else 0.0 gauges = [ {"id": "multi", "label": "Créateurs multi-plateformes (2 comptes et +)", "value": _cov(n_multi), "max": 100, "unit": "%"}, {"id": "bio", "label": "Fiches avec biographie", "value": _cov(n_bio), "max": 100, "unit": "%"}, {"id": "loc", "label": "Fiches avec ville ou région déclarée", "value": _cov(n_loc), "max": 100, "unit": "%"}, {"id": "reach", "label": "Fiches avec audience connue", "value": _cov(n_reach), "max": 100, "unit": "%"}, {"id": "complete", "label": "Complétude moyenne des fiches " "(bio, lieu, audience, photo)", "value": completeness, "max": 100, "unit": "%"}, ] # ---- séries quotidiennes ---- cmp_added = [{"t": (prev_from + timedelta(days=i)).isoformat(), "v": adds_by_day.get(prev_from + timedelta(days=i), 0)} for i in range(span)] series = [ {"id": "added", "title": "Créateurs ajoutés par jour", "unit": "créateurs", "kind": "line", "points": pts_added, **({"compare": cmp_added} if any(c["v"] for c in cmp_added) else {})}, {"id": "cumul", "title": "Taille cumulative du répertoire", "unit": "créateurs", "kind": "area", "points": pts_cumul}, ] # portée cumulée découverte (somme des audiences connues des fiches ajoutées) if reach_day: run_r = sum(v for dd, v in reach_day.items() if dd < p_from) pts_reach = [] for d in day_axis: run_r += reach_day.get(d, 0) pts_reach.append({"t": d.isoformat(), "v": run_r}) if any(p["v"] for p in pts_reach): series.append({"id": "reach_cumul", "title": "Portée cumulée découverte (audiences connues)", "unit": "abonnés", "kind": "area", "points": pts_reach}) # journaux de sync : fiches ajoutées + mises à jour par les connecteurs sync_day: dict[date, int] = {} for r in con.execute("SELECT ts, COALESCE(added,0)+COALESCE(updated,0) n " "FROM sync_log").fetchall(): d = _local_date(r["ts"]) if d: sync_day[d] = sync_day.get(d, 0) + r["n"] if sync_day: pts_sync = [{"t": d.isoformat(), "v": sync_day.get(d, 0)} for d in day_axis] if any(p["v"] for p in pts_sync): series.append({"id": "sync", "title": "Fiches ajoutées ou mises à jour par les " "connecteurs (journaux de sync)", "unit": "fiches", "kind": "bar", "points": pts_sync}) # ---- multi-courbes : croissance par plateforme principale (top 4) ---- plat_totals: dict[str, int] = {} for (d, pl), v in plat_day.items(): plat_totals[pl] = plat_totals.get(pl, 0) + v top_plats = [p for p, _ in sorted(plat_totals.items(), key=lambda x: -x[1])[:4]] multiseries = [] if top_plats: mseries = [] for pl in top_plats: run = sum(v for (dd, p2), v in plat_day.items() if p2 == pl and dd < p_from) pts = [] for d in day_axis: run += plat_day.get((d, pl), 0) pts.append({"t": d.isoformat(), "v": run}) mseries.append({"label": PLAT_LBL.get(pl, pl), "points": pts}) if any(pt["v"] for s in mseries for pt in s["points"]): multiseries.append({ "id": "plat_growth", "title": "Croissance du répertoire par plateforme principale (top 4)", "unit": "créateurs", "series": mseries}) # ---- empilé : ajouts par source de découverte (top 5 + autres) ---- stacked = [] if src_day: top_src = [s for s, _ in sorted(src_total.items(), key=lambda x: -x[1])[:5]] keys = top_src + ["Autres"] pts = [] for d in day_axis: vals = [src_day.get((d, s), 0) for s in top_src] other = sum(v for (dd, s), v in src_day.items() if dd == d and s not in top_src) pts.append({"t": d.isoformat(), "values": vals + [other]}) if any(v for pt in pts for v in pt["values"]): stacked.append({"id": "sources", "title": "Créateurs ajoutés par source de découverte", "unit": "créateurs", "keys": keys, "points": pts}) # ---- répartitions (deltas = croissance réelle du stock sur la période) ---- def _growth(total: int, added: int) -> dict: base = total - added p = _pct(total, base) if added else None if p is None: return {} return {"delta_pct": p} breakdowns = [ {"id": "platforms", "title": "Comptes reliés par plateforme", "kind": "donut", "items": [{"label": PLAT_LBL.get(r["platform"], r["platform"]), "value": r["c"]} for r in plat_rows]}, {"id": "tiers", "title": "Créateurs par taille d'audience", "kind": "bar", "items": [{"label": lbl, "value": tier_total.get(t, 0), **_growth(tier_total.get(t, 0), tier_period.get(t, 0))} for t, lbl in TIER_LBL if tier_total.get(t)]}, {"id": "niches", "title": "Top niches", "kind": "bar", "items": [{"label": NICHE_LBL.get(n, n), "value": v, **_growth(v, niche_period.get(n, 0))} for n, v in sorted(niche_total.items(), key=lambda x: -x[1])[:12]]}, ] if type_total: breakdowns.append( {"id": "types", "title": "Par type de créateur", "kind": "bar", "items": [{"label": TYPE_LBL.get(t, t), "value": v} for t, v in sorted(type_total.items(), key=lambda x: -x[1])[:10]]}) # ---- distributions ---- distributions = [] reach_vals = [r["total_reach"] for r in creators if r["total_reach"]] if reach_vals: bins = [] for lo, hi, lbl in REACH_BINS: n = sum(1 for v in reach_vals if v >= lo and (hi is None or v < hi)) bins.append({"label": lbl, "value": n}) distributions.append({"id": "audiences", "title": "Distribution des audiences connues " "(abonnés cumulés)", "unit": "créateurs", "bins": bins}) if acc_per_creator: counts: dict[str, int] = {} for r in acc_per_creator: k = "5 et +" if r["np"] >= 5 else str(r["np"]) counts[k] = counts.get(k, 0) + 1 order = ["1", "2", "3", "4", "5 et +"] bins = [{"label": f"{k} compte{'s' if k != '1' else ''}", "value": counts[k]} for k in order if counts.get(k)] distributions.append({"id": "acc_per_creator", "title": "Comptes reliés par créateur", "unit": "créateurs", "bins": bins}) if conf_rows: conf_bins = [(0.0, 0.6, "< 60 %"), (0.6, 0.7, "60 – 70 %"), (0.7, 0.8, "70 – 80 %"), (0.8, 0.9, "80 – 90 %"), (0.9, 1.01, "90 – 100 %")] bins = [] for lo, hi, lbl in conf_bins: n = sum(1 for r in conf_rows if lo <= (r["confidence"] or 0) < hi) bins.append({"label": lbl, "value": n}) if any(b["value"] for b in bins): distributions.append({"id": "confidence", "title": "Confiance du rattachement des comptes", "unit": "comptes", "bins": bins}) # ---- géographie ---- geo = None if region_total: geo = {"title": "Par région déclarée (quand le créateur la rend publique)", "items": [{"label": k, "value": v} for k, v in sorted(region_total.items(), key=lambda x: -x[1])]} # ---- heatmaps : calendrier (ajouts/jour) + horaire 7×24 (découvertes) ---- heatmap = {"title": "Ajouts au répertoire", "cells": [{"date": d.isoformat(), "value": v} for d, v in sorted(adds_by_day.items())]} hourly = None if hour_cells: hourly = {"title": "Découvertes de créateurs par jour et heure " "(toute l'historique)", "cells": [{"dow": k[0], "hour": k[1], "value": v} for k, v in sorted(hour_cells.items())]} # ---- tableaux ---- top = sorted((r for r in creators if r["total_reach"]), key=lambda r: -r["total_reach"])[:100] tables = [] if top: tables.append({ "id": "top_creators", "title": "Top créateurs par audience connue", "columns": ["Créateur", "Taille", "Plateforme principale", "Abonnés cumulés", "Niches"], "rows": [[r["display_name"], dict(TIER_LBL).get(r["audience_tier"], r["audience_tier"]), PLAT_LBL.get(r["primary_platform"], r["primary_platform"]), r["total_reach"], ", ".join(NICHE_LBL.get(n, n) for n in (r["niches"] or "").split(",")[:2] if n)] for r in top]}) if plat_rows: tables.append({ "id": "platforms", "title": "Répartition par plateforme", "columns": ["Plateforme", "Comptes reliés", "Vérifiés", "Abonnés cumulés", "Audience moyenne / compte"], "rows": [[PLAT_LBL.get(r["platform"], r["platform"]), r["c"], r["nverif"] or 0, r["fol"] or 0, int(round((r["fol"] or 0) / r["nfol"])) if r["nfol"] else "—"] for r in plat_rows]}) tables.append({ "id": "niches", "title": "Répartition par niche", "columns": ["Niche", "Créateurs", "Ajoutés sur la période", "Part"], "rows": [[NICHE_LBL.get(n, n), v, niche_period.get(n, 0), f"{100 * v / max(1, n_active):.1f} %".replace(".", ",")] for n, v in sorted(niche_total.items(), key=lambda x: -x[1])]}) # connecteurs & dernière synchro (journaux réels) sync_rows = con.execute( "SELECT source, COUNT(*) runs, SUM(COALESCE(added,0)) a, " "SUM(COALESCE(updated,0)) u, SUM(COALESCE(errors,0)) e, MAX(ts) last " "FROM sync_log GROUP BY source ORDER BY a DESC").fetchall() if sync_rows: def _fmt_ts(ts: str) -> str: dt = _local_dt(ts) return dt.strftime("%Y-%m-%d %H:%M") if dt else (ts or "")[:16] tables.append({ "id": "connectors", "title": "Connecteurs & dernière synchronisation", "columns": ["Connecteur", "Synchros", "Fiches ajoutées", "Mises à jour", "Erreurs", "Dernière synchro (HE)"], "rows": [[r["source"], r["runs"], r["a"], r["u"], r["e"], _fmt_ts(r["last"])] for r in sync_rows]}) if city_total: n_city = sum(city_total.values()) tables.append({ "id": "cities", "title": "Créateurs par ville déclarée", "columns": ["Ville", "Créateurs", "Part des fiches localisées"], "rows": [[c, v, f"{100 * v / max(1, n_city):.1f} %".replace(".", ",")] for c, v in sorted(city_total.items(), key=lambda x: -x[1])[:50]]}) # ---- records & faits marquants (générés depuis les données) ---- records = [] in_period = {d: v for d, v in adds_by_day.items() if p_from <= d <= p_to} if in_period: best = max(in_period.items(), key=lambda x: x[1]) records.append({"label": "Jour record d'ajouts (période)", "value": f"{_fr_int(best[1])} créateurs", "date": best[0].isoformat()}) records.append({"label": "Moyenne d'ajouts par jour (période)", "value": f"{added_cur / span:.1f} créateurs".replace(".", ",")}) if adds_by_day: best_all = max(adds_by_day.items(), key=lambda x: x[1]) if not in_period or best_all[0] not in in_period: records.append({"label": "Jour record d'ajouts (toute l'historique)", "value": f"{_fr_int(best_all[1])} créateurs", "date": best_all[0].isoformat()}) if niche_period: bn = max(niche_period.items(), key=lambda x: x[1]) records.append({"label": "Niche la plus dynamique (ajouts sur la période)", "value": f"{NICHE_LBL.get(bn[0], bn[0])} — {_fr_int(bn[1])}"}) if plat_rows: records.append({"label": "Plateforme la plus reliée", "value": f"{PLAT_LBL.get(plat_rows[0]['platform'], plat_rows[0]['platform'])}" f" — {_fr_int(plat_rows[0]['c'])} comptes"}) if top: records.append({"label": "Plus grande portée connue", "value": f"{top[0]['display_name']} — " f"{_fr_int(top[0]['total_reach'])} abonnés"}) if acc_per_creator: bm = max(acc_per_creator, key=lambda r: r["np"]) if bm["np"] >= 2: records.append({"label": "Créateur le plus multi-plateforme", "value": f"{bm['display_name']} — {bm['np']} plateformes"}) if src_total: bs = max(src_total.items(), key=lambda x: x[1]) records.append({"label": "Source de découverte la plus productive", "value": f"{bs[0]} — {_fr_int(bs[1])} créateurs"}) if region_total: br = max(region_total.items(), key=lambda x: x[1]) records.append({"label": "Région la plus représentée (déclarée)", "value": f"{br[0]} — {_fr_int(br[1])} créateurs"}) if min_day: records.append({"label": "Première fiche au répertoire", "value": "ouverture du répertoire", "date": min_day.isoformat()}) last_sync = con.execute( "SELECT ts FROM sync_log ORDER BY id DESC LIMIT 1").fetchone() if last_sync and last_sync["ts"]: dt = _local_dt(last_sync["ts"]) if dt: records.append({"label": "Dernière synchronisation des connecteurs", "value": dt.strftime("%H:%M (heure de l'Est)"), "date": dt.date().isoformat()}) out = { "updated": datetime.now(TZ).isoformat(timespec="seconds"), "period": {"from": p_from.isoformat(), "to": p_to.isoformat(), "label": p_label}, "kpis": kpis, "gauges": gauges, "series": series, "breakdowns": breakdowns, "heatmap": heatmap, "tables": tables, "records": records[:12], } if multiseries: out["multiseries"] = multiseries if stacked: out["stacked"] = stacked if distributions: out["distributions"] = distributions if geo: out["geo"] = geo if hourly: out["hourly"] = hourly return out def dashboard(con: sqlite3.Connection, period: str = "30j", date_from: str = "", date_to: str = "") -> dict: """Point d'entrée avec cache mémoire (TTL 5 min par combinaison de période).""" key = (period, date_from, date_to) now = time.time() hit = _CACHE.get(key) if hit and now - hit[0] < CACHE_TTL: return hit[1] data = _build(con, period, date_from, date_to) if len(_CACHE) > 64: # borne dure (plages personnalisées illimitées) _CACHE.clear() _CACHE[key] = (now, data) return data # --- insights par créateur (fiche « légendaire ») ------------------------------ def creator_insights(con: sqlite3.Connection, doc: dict) -> dict: """Insights d'une fiche créateur : croissance (snapshots §13-14), engagement pondéré, rythme de publication et Ka Score composite /100. AUCUNE stat inventée : tout provient des comptes rattachés (accounts.metrics remplis par les acteurs Apify) et des snapshots quotidiens. Un champ absent reste absent — pas d'estimation. """ import math from .db import follower_history platforms = doc.get("platforms") or [] reach = doc.get("total_reach") or sum( p.get("followers") or 0 for p in platforms) or 0 # croissance : série totale quotidienne (report dernière valeur connue) hist = follower_history(con, doc["id"], days=95) total = hist["total"] def growth(days: int) -> dict | None: if len(total) < 2: return None last = total[-1] cutoff = (date.today() - timedelta(days=days)).isoformat() base = next((p for p in total if p["day"] >= cutoff), None) if base is None or base["day"] == last["day"] or not base["followers"]: return None delta = last["followers"] - base["followers"] return {"since": base["day"], "delta": delta, "pct": round(100 * delta / base["followers"], 2)} # engagement moyen pondéré par l'audience de chaque plateforme weighted = [(p["metrics"].get("engagement_rate_pct"), p.get("followers") or 1) for p in platforms if isinstance(p.get("metrics"), dict) and p["metrics"].get("engagement_rate_pct") is not None] engagement = (round(sum(e * w for e, w in weighted) / sum(w for _, w in weighted), 2) if weighted else None) # rythme de publication : somme des cadences hebdo déclarées par plateforme rates = [p["metrics"].get(k) for p in platforms if isinstance(p.get("metrics"), dict) for k in ("posts_per_week", "videos_per_week", "tweets_per_week") if isinstance(p["metrics"].get(k), (int, float))] pubs_week = round(sum(rates), 1) if rates else None # dernière activité publique connue, toutes plateformes confondues last_dates = [str(p["metrics"].get(k)) for p in platforms if isinstance(p.get("metrics"), dict) for k in ("last_post_at", "last_video_at", "last_tweet_at", "last_broadcast_at", "last_video_published") if p["metrics"].get(k)] last_activity = max(last_dates) if last_dates else None is_verified = any(p.get("verified") for p in platforms) live_now = any(isinstance(p.get("metrics"), dict) and p["metrics"].get("is_live_now") for p in platforms) # Ka Score /100 : audience 40 (log), engagement 25, présence 20, rythme 10, # vérification 5 — comparable d'un créateur à l'autre, jamais inventé : # une composante inconnue vaut simplement 0. parts = { "audience": round(40 * min(1.0, math.log10(max(reach, 1)) / 7), 1), "engagement": round(25 * min(1.0, (engagement or 0) / 10), 1), "presence": round(20 * min(1.0, len(platforms) / 5), 1), "rythme": round(10 * min(1.0, (pubs_week or 0) / 3), 1), "verification": 5.0 if is_verified else 0.0, } top = max(platforms, key=lambda p: p.get("followers") or 0, default=None) return { "ka_score": round(sum(parts.values()), 1), "ka_score_parts": parts, "total_reach": reach or None, "growth_7d": growth(7), "growth_30d": growth(30), "avg_engagement_pct": engagement, "publications_per_week": pubs_week, "last_activity": last_activity, "platforms_count": len(platforms), "is_verified_somewhere": is_verified, "is_live_now": live_now, "top_platform": (top or {}).get("platform"), "history": hist, }