#!/usr/bin/env python3 """Fusionne les extractions (data/extracted/*.json) en un dataset unique dédupliqué. Sorties : data/qc_influenceurs.json + data/qc_influenceurs.csv """ import csv import glob import json import os import re import unicodedata BASE = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) EXTRACTED = os.path.join(BASE, "data", "extracted") OUT_JSON = os.path.join(BASE, "data", "qc_influenceurs.json") OUT_CSV = os.path.join(BASE, "data", "qc_influenceurs.csv") PLATFORMS = ["instagram", "tiktok", "youtube", "facebook", "x", "twitch"] METRIC_KEYS = { "instagram": "instagram_followers", "tiktok": "tiktok_followers", "youtube": "youtube_subscribers", "facebook": "facebook_followers", "x": "x_followers", "twitch": "twitch_followers", } # variantes connues du même créateur (clé normalisée -> clé canonique) MANUAL_ALIASES = { "carl is cooking": "carl arsenault", } CATEGORY_FIXES = { "tennis / sport": "sport", "art / miniatures": "autre", "docuséries / arts": "autre", } def norm_name(name: str) -> str: s = re.sub(r"\(.*?\)", " ", name or "") # "Nabil Lahrech (Aiekillu)" == "Nabil Lahrech" s = unicodedata.normalize("NFKD", s) s = "".join(c for c in s if not unicodedata.combining(c)) s = re.sub(r"[^a-z0-9]+", " ", s.lower()).strip() return s def name_aliases(name: str): """Clés supplémentaires : contenu des parenthèses (pseudo) et nom complet.""" keys = {norm_name(name)} for inner in re.findall(r"\((.*?)\)", name or ""): k = norm_name(inner) if k: keys.add(k) return keys def norm_handle(h): if not h: return None h = str(h).strip().lstrip("@").lower() h = re.sub(r"^https?://(www\.)?(instagram|tiktok|youtube|facebook|x|twitter|twitch)\.(com|tv)/(@)?", "", h) h = h.strip("/").split("/")[0].split("?")[0] return h or None def merge_record(dst, src): for p in PLATFORMS: h = norm_handle((src.get("handles") or {}).get(p)) if h and not dst["handles"].get(p): dst["handles"][p] = h for mk in METRIC_KEYS.values(): v = (src.get("metrics") or {}).get(mk) if isinstance(v, (int, float)) and v > 0: cur = dst["metrics"].get(mk) # garder la valeur la plus élevée (les listes datées sous-estiment) if not cur or v > cur: dst["metrics"][mk] = int(v) if src.get("category") and (not dst.get("category") or dst["category"] == "autre"): dst["category"] = src["category"] if src.get("location") and dst.get("location") in (None, "", "Québec"): dst["location"] = src["location"] note = (src.get("notes") or "").strip() if note and note not in dst["notes"]: dst["notes"] = (dst["notes"] + " | " + note).strip(" |") for s in src.get("sources") or []: if s not in dst["sources"]: dst["sources"].append(s) def main(): records = [] for f in sorted(glob.glob(os.path.join(EXTRACTED, "*.json"))): try: data = json.load(open(f)) except Exception as e: print(f"ERREUR lecture {f}: {e}") continue if isinstance(data, dict): data = data.get("influencers") or data.get("data") or [] print(f"{os.path.basename(f)}: {len(data)} records") records.extend(data) by_key = {} handle_index = {} # (platform, handle) -> key alias_index = {} # alias normalisé -> key NON_QC = ("toronto", "vancouver", "calgary", "winnipeg", "scarborough", "ajax", "ontario", "terre-neuve", "alberta", "colombie-britannique") for r in records: if not r.get("name"): continue loc = (r.get("location") or "").lower() if any(x in loc for x in NON_QC): continue # hors Québec (listes canadiennes mal filtrées) key = norm_name(r["name"]) if not key: continue key = MANUAL_ALIASES.get(key, key) # match par handle ou alias (pseudo entre parenthèses) si le nom diffère matched = None for p in PLATFORMS: h = norm_handle((r.get("handles") or {}).get(p)) if h and (p, h) in handle_index: matched = handle_index[(p, h)] break if not matched: for a in name_aliases(r["name"]): if a in alias_index: matched = alias_index[a] break key = matched or key if key not in by_key: by_key[key] = { "name": r["name"].strip(), "handles": {p: None for p in PLATFORMS}, "metrics": {mk: None for mk in METRIC_KEYS.values()}, "category": None, "location": None, "notes": "", "sources": [], } merge_record(by_key[key], r) for p in PLATFORMS: h = by_key[key]["handles"].get(p) if h: handle_index[(p, h)] = key for a in name_aliases(r["name"]) | name_aliases(by_key[key]["name"]): alias_index[a] = key out = list(by_key.values()) for r in out: vals = [v for v in r["metrics"].values() if v] r["total_followers"] = sum(vals) if vals else None r["platform_count"] = sum(1 for p in PLATFORMS if r["handles"].get(p) or r["metrics"].get(METRIC_KEYS[p])) t = r["total_followers"] or 0 r["tier"] = "méga (1M+)" if t >= 1_000_000 else "macro (100K+)" if t >= 100_000 else "micro (10K+)" if t >= 10_000 else "nano/inconnu" r["category"] = CATEGORY_FIXES.get(r["category"] or "autre", r["category"] or "autre") r["location"] = r["location"] or "Québec" out.sort(key=lambda r: -(r["total_followers"] or 0)) json.dump(out, open(OUT_JSON, "w"), ensure_ascii=False, indent=2) with open(OUT_CSV, "w", newline="") as fh: w = csv.writer(fh) w.writerow(["nom", "categorie", "localisation", "tier", "total_followers", "instagram_handle", "instagram_followers", "tiktok_handle", "tiktok_followers", "youtube_handle", "youtube_subscribers", "facebook_handle", "facebook_followers", "x_handle", "x_followers", "twitch_handle", "twitch_followers", "nb_plateformes", "notes", "sources"]) for r in out: w.writerow([r["name"], r["category"], r["location"], r["tier"], r["total_followers"] or "", r["handles"]["instagram"] or "", r["metrics"]["instagram_followers"] or "", r["handles"]["tiktok"] or "", r["metrics"]["tiktok_followers"] or "", r["handles"]["youtube"] or "", r["metrics"]["youtube_subscribers"] or "", r["handles"]["facebook"] or "", r["metrics"]["facebook_followers"] or "", r["handles"]["x"] or "", r["metrics"]["x_followers"] or "", r["handles"]["twitch"] or "", r["metrics"]["twitch_followers"] or "", r["platform_count"], r["notes"], ";".join(r["sources"])]) print(f"\nTotal dédupliqué : {len(out)} influenceurs") print(f"Avec au moins une métrique : {sum(1 for r in out if r['total_followers'])}") print(f"Méga (1M+) : {sum(1 for r in out if r['tier'].startswith('méga'))}") print(f"Macro (100K+) : {sum(1 for r in out if r['tier'].startswith('macro'))}") if __name__ == "__main__": main()