spb/crea-ka

1# ==============================================================================2# Author: Simon-Pierre Boucher <contact@spboucher.ai>3# File:   creaka/connectors/twitch.py4# Desc:   Connecteur ENRICHISSEMENT Twitch — profil public par créateur SANS5#         clé via l'API GQL publique du site web officiel (Client-ID public),6#         abonnés inclus. Voie Helix officielle conservée si des clés7#         TWITCH_CLIENT_ID/TWITCH_CLIENT_SECRET apparaissent. Palier 2-3 (§9).8# ==============================================================================9"""Enrichissement Twitch, créateur par créateur — sans clé (vague 3).1011Le site web officiel de Twitch consomme ``gql.twitch.tv/gql`` avec un12Client-ID PUBLIC embarqué dans chaque page servie à n'importe quel visiteur13(kimne78kx3ncx6brgo4mv6wki5h1ko) — aucune authentification. On ne lit que les14champs publics du profil déjà rattaché (§15) : abonnés (followers.totalCount,15que Helix ne fournit plus sans jeton de modérateur depuis 2023), description,16avatar, statut partenaire/affilié, dernière diffusion (fraîcheur/dormance).17Requêtes REGROUPÉES (20 profils par POST) → ~2 requêtes pour le catalogue18Twitch actuel.1920Si des clés officielles apparaissent dans .env, la voie Helix (privilégiée21§10) reprend le dessus pour l'existence/badge, avec le même complément GQL22pour les abonnés.2324⚠ Empreinte TLS : gql.twitch.tv accepte python-requests (vérifié 2026-08-18,25HTTP 200) — contrairement à syndication.twitter.com (voir x_profil). Un repli26curl-binaire est prévu au cas où un blocage apparaîtrait.27"""28from __future__ import annotations2930import json31import os32import subprocess33import time3435from ..schema import Creator, now_iso36from .base import USER_AGENT, BaseConnector3738TOKEN_URL = "https://id.twitch.tv/oauth2/token"39USERS_URL = "https://api.twitch.tv/helix/users"40GQL_URL = "https://gql.twitch.tv/gql"4142# Client-ID PUBLIC du client web officiel de Twitch (embarqué dans les pages43# servies à tous les visiteurs — ce n'est PAS un secret)44WEB_CLIENT_ID = "kimne78kx3ncx6brgo4mv6wki5h1ko"4546# champs publics du profil (§15 : rien de privé, rien de contourné)47_GQL_FIELDS = ("id login displayName description profileImageURL(width: 300) "48               "followers { totalCount } roles { isPartner isAffiliate } "49               "lastBroadcast { startedAt game { displayName } } createdAt")505152def gql_user_query(handle: str) -> dict:53    """Requête GQL publique pour UN login (échappé via json.dumps)."""54    return {"query": "query { user(login: %s) { %s } }"55                     % (json.dumps(handle), _GQL_FIELDS)}565758def parse_gql_user(entry: dict) -> dict | None:59    """Une réponse GQL → dict plat {followers, partner, affiliate, bio,60    avatar, last_stream, game} — None si le compte n'existe pas/plus."""61    user = (entry.get("data") or {}).get("user")62    if not user:63        return None64    last = user.get("lastBroadcast") or {}65    roles = user.get("roles") or {}66    return {67        "followers": (user.get("followers") or {}).get("totalCount"),68        "partner": bool(roles.get("isPartner")),69        "affiliate": bool(roles.get("isAffiliate")),70        "bio": (user.get("description") or "").strip(),71        "avatar": user.get("profileImageURL") or "",72        "last_stream": last.get("startedAt"),73        "game": (last.get("game") or {}).get("displayName"),74    }757677class TwitchConnector(BaseConnector):78    source_id = "twitch"79    kind = "enrichment"80    request_delay = 1.081    max_profiles = 40082    batch_size = 20            # profils par POST GQL (le site en groupe autant)8384    def __init__(self) -> None:85        super().__init__()86        self.errors = 08788    # -- voie officielle Helix (si clés fournies) ------------------------------89    def _token(self, cid: str, secret: str) -> str:90        resp = self.post(TOKEN_URL, data={"client_id": cid,91                                          "client_secret": secret,92                                          "grant_type": "client_credentials"})93        return resp.json()["access_token"]9495    def _enrich_helix(self, pairs: list, cid: str, secret: str) -> list[Creator]:96        token = self._token(cid, secret)97        headers = {"Client-Id": cid, "Authorization": f"Bearer {token}"}98        enriched: list[Creator] = []99        for cr, tw in pairs:100            try:101                resp = self.get(USERS_URL, params={"login": tw.handle},102                                headers=headers)103            except Exception:104                self.errors += 1105                continue106            data = resp.json().get("data") or []107            if not data:108                continue109            user = data[0]110            tw.verified = user.get("broadcaster_type") == "partner"111            tw.confidence = max(tw.confidence or 0, 0.90)112            tw.signal = tw.signal or "api_officielle"113            tw.last_checked = now_iso()114            if not cr.bio and user.get("description"):115                cr.bio = user["description"]116            if not cr.avatar_url and user.get("profile_image_url"):117                cr.avatar_url = user["profile_image_url"]118            enriched.append(cr)119        return enriched120121    # -- voie publique GQL (sans clé) -------------------------------------------122    def _gql(self, handles: list[str]) -> list[dict]:123        """POST groupé — repli curl-binaire si l'empreinte TLS était refusée."""124        payload = [gql_user_query(h) for h in handles]125        try:126            resp = self.post(GQL_URL, json=payload,127                             headers={"Client-Id": WEB_CLIENT_ID})128            return resp.json()129        except Exception:130            pass  # 4xx/blocage → repli curl (même pattern que x-profil)131        self._throttle()132        try:133            proc = subprocess.run(134                ["curl", "-sS", "--fail", "--max-time", str(self.timeout),135                 "-A", USER_AGENT, "-H", f"Client-Id: {WEB_CLIENT_ID}",136                 "-H", "Content-Type: application/json",137                 "-d", json.dumps(payload), GQL_URL],138                capture_output=True, text=True, timeout=self.timeout + 5)139        finally:140            self._last_request = time.time()141        if proc.returncode != 0:142            raise RuntimeError(f"curl {proc.returncode}: gql.twitch.tv")143        return json.loads(proc.stdout)144145    def _enrich_gql(self, pairs: list) -> list[Creator]:146        enriched: list[Creator] = []147        for i in range(0, len(pairs), self.batch_size):148            batch = pairs[i:i + self.batch_size]149            try:150                replies = self._gql([tw.handle for _, tw in batch])151            except Exception:152                self.errors += 1153                continue154            if not isinstance(replies, list) or len(replies) != len(batch):155                self.errors += 1156                continue157            for (cr, tw), entry in zip(batch, replies):158                info = parse_gql_user(entry if isinstance(entry, dict) else {})159                tw.last_checked = now_iso()160                if info is None:   # compte disparu/renommé : horodaté, sans plus161                    enriched.append(cr)162                    continue163                if info["followers"] is not None:164                    tw.followers = int(info["followers"])165                if info["partner"]:166                    tw.verified = True167                tw.metrics.update({k: v for k, v in {168                    "affiliate": info["affiliate"] or None,169                    "last_stream": info["last_stream"],170                    "game": info["game"],171                }.items() if v})172                if not cr.bio and info["bio"]:173                    cr.bio = info["bio"]174                if not cr.avatar_url and info["avatar"].startswith("http"):175                    cr.avatar_url = info["avatar"]176                enriched.append(cr)177        return enriched178179    def enrich(self, creators: list[Creator]) -> list[Creator]:180        pairs = []181        for cr in creators:182            tw = next((a for a in cr.platforms if a.platform == "twitch"), None)183            if tw is not None:184                pairs.append((cr, tw))185        pairs = pairs[:self.max_profiles]186        if not pairs:187            return []188        cid = os.environ.get("TWITCH_CLIENT_ID", "")189        secret = os.environ.get("TWITCH_CLIENT_SECRET", "")190        if cid and secret:  # voie officielle privilégiée (§10) quand disponible191            enriched = self._enrich_helix(pairs, cid, secret)192            extra = self._enrich_gql(pairs)   # complément : abonnés (GQL public)193            seen = {id(c) for c in enriched}194            enriched += [c for c in extra if id(c) not in seen]195            return enriched196        return self._enrich_gql(pairs)197