# ============================================================================== # Author: Simon-Pierre Boucher # File: creaka/connectors/twitch_decouverte.py # Desc: Connecteur DÉCOUVERTE — streamers Twitch québécois via l'API GQL # publique du site officiel (Client-ID public, sans clé) : recherche # de chaînes par termes QC + répertoire des diffusions en français. # Palier 3 (§9). Filtre conservateur : marqueur québécois OBLIGATOIRE # dans le nom, la bio ou le titre du stream. # ============================================================================== """Découverte des streamers Twitch québécois — sans clé. Deux voies, mêmes conventions que l'enrichisseur `twitch` (vague 3) : 1. ``searchFor`` (la recherche du site officiel) sur une liste de termes québécois (province, régions, villes) → chaînes dont le profil PUBLIC mentionne le Québec, avec abonnés, bio, avatar, statut partenaire ; 2. le répertoire des diffusions EN DIRECT en français (``streams``, languages: [FR]) → longue traîne de streamers QC actifs, paginé ; la couverture s'accumule à chaque passage (qui est en direct change chaque jour, comme youtube-recherche). On ne crée une fiche QUE si un marqueur québécois figure dans le nom, la description du profil ou le titre du stream (jamais d'inférence, §7/§15). Le compte Twitch est le compte SOURCE de la fiche → signal profil_source (0.98) ; l'enrichissement rattache ensuite ses autres plateformes. """ from __future__ import annotations import json import subprocess import sys import time from ..identity import account from ..normalize import slugify from ..schema import Creator from .base import USER_AGENT, BaseConnector from .twitch import GQL_URL, WEB_CLIENT_ID from .youtube_recherche import is_quebec # champs publics du profil (§15) — mêmes que l'enrichisseur twitch _USER_FIELDS = ("login displayName description createdAt " "followers { totalCount } profileImageURL(width: 300) " "roles { isPartner isAffiliate } " "lastBroadcast { startedAt game { displayName } }") # termes de recherche québécois (province, régions, villes, scène locale) SEARCH_TERMS = [ "québec", "québécois", "québécoise", "queb", "montréal", "mtl", "ville de québec", "rive-sud", "rive-nord", "saguenay", "gaspésie", "mauricie", "estrie", "outaouais", "abitibi", "témiscamingue", "lanaudière", "laurentides", "montérégie", "charlevoix", "beauce", "côte-nord", "bas-saint-laurent", "gatineau", "sherbrooke", "laval", "longueuil", "lévis", "trois-rivières", "chicoutimi", "rimouski", "drummondville", "granby", "victoriaville", "joliette", "terrebonne", "repentigny", "brossard", "shawinigan", "rouyn-noranda", "alma", "streamer québécois", "gamer québécois", "twitch québec", ] # jeu du stream → niche §6.1 (défaut : gaming — c'est Twitch) _GAME_NICHE = { "Just Chatting": "lifestyle", "Music": "musique", "Art": "arts", "Food & Drink": "cuisine", "Sports": "sport-fitness", "ASMR": "lifestyle", "Talk Shows & Podcasts": "actualite-opinion", "Software and Game Development": "tech", "Science & Technology": "tech", "Fitness & Health": "sport-fitness", "Travel & Outdoors": "plein-air", "Makers & Crafting": "arts", "Beauty & Body Art": "beaute", "Politics": "actualite-opinion", "DJs": "musique", } class TwitchDecouverteConnector(BaseConnector): source_id = "twitch-decouverte" kind = "discovery" request_delay = 1.0 search_limit = 25 # chaînes par page de recherche search_pages = 4 # pages par terme (cursor searchFor) stream_pages = 20 # pages du répertoire FR (30 streams/page) def __init__(self) -> None: super().__init__() self.errors = 0 # -- transport GQL (repli curl-binaire, même pattern que twitch.py) -------- def _gql(self, query: str) -> dict: payload = {"query": query} try: resp = self.post(GQL_URL, json=payload, headers={"Client-Id": WEB_CLIENT_ID}) return resp.json() except Exception: pass # 4xx/blocage TLS → repli curl self._throttle() try: proc = subprocess.run( ["curl", "-sS", "--fail", "--max-time", str(self.timeout), "-A", USER_AGENT, "-H", f"Client-Id: {WEB_CLIENT_ID}", "-H", "Content-Type: application/json", "-d", json.dumps(payload), GQL_URL], capture_output=True, text=True, timeout=self.timeout + 5) finally: self._last_request = time.time() if proc.returncode != 0: raise RuntimeError(f"curl {proc.returncode}: gql.twitch.tv") return json.loads(proc.stdout) # -- fabrication d'une fiche ------------------------------------------------- def _collect(self, user: dict, context: str, seen: set[str], creators: list[Creator], niche: str = "gaming") -> None: login = (user.get("login") or "").strip().lower() name = (user.get("displayName") or login).strip() bio = (user.get("description") or "").strip() if not login or login in seen: return # filtre conservateur : marqueur QC dans nom + bio + titre du stream if not is_quebec(f"{name} {bio} {context}"): return seen.add(login) last = user.get("lastBroadcast") or {} roles = user.get("roles") or {} game = (last.get("game") or {}).get("displayName") acc = account("twitch", login, "profil_source", followers=(user.get("followers") or {}).get("totalCount"), verified=bool(roles.get("isPartner")) or None) acc.metrics = {k: v for k, v in { "affiliate": bool(roles.get("isAffiliate")) or None, "last_stream": last.get("startedAt"), "game": game, }.items() if v} avatar = user.get("profileImageURL") or "" creators.append(Creator( source=self.source_id, external_id=slugify(name), display_name=name, bio=bio[:1200], niches=[_GAME_NICHE.get(game or "", niche)], creator_type="streamer", primary_platform="twitch", avatar_url=avatar if avatar.startswith("http") else None, platforms=[acc], source_ids=[f"{self.source_id}:{login}"], )) # -- voie 1 : recherche de chaînes par termes QC ----------------------------- def _search(self, seen: set[str], creators: list[Creator]) -> None: for term in SEARCH_TERMS: cursor = "" for _page in range(self.search_pages): cur = f', cursor: {json.dumps(cursor)}' if cursor else "" q = ("query { searchFor(userQuery: %s, platform: \"web\", " "target: {index: CHANNEL, limit: %d%s}) " "{ channels { cursor edges { item { ... on User { %s } } } } } }" % (json.dumps(term), self.search_limit, cur, _USER_FIELDS)) try: data = self._gql(q) channels = ((data.get("data") or {}).get("searchFor") or {}).get("channels") or {} except Exception: self.errors += 1 break edges = channels.get("edges") or [] for edge in edges: user = edge.get("item") or {} if user.get("login"): self._collect(user, "", seen, creators) cursor = channels.get("cursor") or "" if not cursor or not edges: break # -- voie 2 : répertoire des diffusions en français (longue traîne) ---------- def _directory(self, seen: set[str], creators: list[Creator]) -> None: after = "" for _page in range(self.stream_pages): aft = f', after: {json.dumps(after)}' if after else "" q = ("query { streams(first: 30%s, options: {languages: [FR]}) " "{ edges { cursor node { title broadcaster { %s } } } " "pageInfo { hasNextPage } } }" % (aft, _USER_FIELDS)) try: data = self._gql(q) streams = (data.get("data") or {}).get("streams") or {} except Exception: self.errors += 1 return edges = streams.get("edges") or [] for edge in edges: node = edge.get("node") or {} user = node.get("broadcaster") or {} if user.get("login"): self._collect(user, node.get("title") or "", seen, creators) if not edges or not (streams.get("pageInfo") or {}).get("hasNextPage"): return after = edges[-1].get("cursor") or "" if not after: return def fetch(self) -> list[Creator]: creators: list[Creator] = [] seen: set[str] = set() self._search(seen, creators) try: self._directory(seen, creators) except Exception as exc: # la voie 2 ne bloque jamais la voie 1 self.errors += 1 print(f"[crea-ka] ⚠ twitch-decouverte : répertoire FR en échec : " f"{exc}", file=sys.stderr) return creators