spb/crea-ka

1# ==============================================================================2# Author: Simon-Pierre Boucher <contact@spboucher.ai>3# File:   creaka/connectors/x_profil.py4# Desc:   Connecteur ENRICHISSEMENT « x-profil » — abonnés X (Twitter) SANS clé5#         via le service public de syndication (widgets d'intégration officiels)6#         syndication.twitter.com/srv/timeline-profile/screen-name/{handle},7#         repli page publique x.com/{handle}. Palier 3 (§9).8#         Facebook sondé aussi : page publique bloquée sans clé (HTTP 400)9#         → aucune voie fiable, on passe (documenté dans data/sources.json).10# ==============================================================================11"""Enrichissement X (Twitter) : nombre d'abonnés par créateur, sans clé.1213L'API officielle de lecture est payante depuis 2023 et l'ancien widget14followbutton (cdn.syndication.twimg.com/widgets/followbutton) est mort15(vérifié 2026-08-18 : réponse vide). Deux voies publiques restent :16171. ``syndication.twitter.com/srv/timeline-profile/screen-name/{h}`` — le18   service OFFICIEL des widgets d'intégration : son __NEXT_DATA__ contient19   l'objet user complet (followers_count, verified, avatar). Fiable et conçu20   pour être appelé sans authentification. VOIE PRINCIPALE.212. la page ``x.com/{handle}`` rendue côté serveur expose22   ``UserRelationshipCounts",followers:N`` — mais mur de connexion après23   ~15-20 requêtes par IP (constaté 2026-08-18). REPLI seulement.2425⚠ Ces services 429-ent l'empreinte TLS de python-requests (constaté262026-08-18 : curl 200, requests 429, même IP/UA/en-têtes) → le fetch passe27par le binaire système ``curl`` (toujours présent sur macOS), avec le même28User-Agent CreaKaBot et le même throttling poli que les autres connecteurs.2930On ne lit QUE les compteurs publics du profil déjà rattaché (§15).31Cap 150 profils/passage + re-visite 7 j (~180 comptes X → rotation douce).32"""33from __future__ import annotations3435import json36import re37import subprocess38import time39from datetime import datetime, timedelta, timezone4041from ..schema import Creator, now_iso42from .base import USER_AGENT, BaseConnector4344SYNDICATION_URL = ("https://syndication.twitter.com/srv/timeline-profile/"45                   "screen-name/{h}")46PAGE_URL = "https://x.com/{h}"4748_NEXT_DATA_RE = re.compile(49    r'<script id="__NEXT_DATA__"[^>]*>(.*?)</script>', re.S)5051# repli x.com : état intégré du rendu serveur (clés non citées, pas du JSON)52_COUNT_RES = (53    re.compile(r'UserRelationshipCounts",followers:(\d+),following:(\d+)'),54    re.compile(r"followers:(\d+),following:(\d+)"),55)565758def _find_user(node, handle: str) -> dict | None:59    """Parcourt le JSON Next.js : l'objet user dont screen_name == handle."""60    if isinstance(node, dict):61        sn = node.get("screen_name")62        if isinstance(sn, str) and sn.lower() == handle \63                and "followers_count" in node:64            return node65        for v in node.values():66            hit = _find_user(v, handle)67            if hit:68                return hit69    elif isinstance(node, list):70        for item in node:71            hit = _find_user(item, handle)72            if hit:73                return hit74    return None757677def parse_syndication(html: str, handle: str) -> dict | None:78    """Page timeline-profile → {followers, following, verified, avatar} ou None."""79    m = _NEXT_DATA_RE.search(html)80    if not m:81        return None82    try:83        data = json.loads(m.group(1))84    except ValueError:85        return None86    user = _find_user(data, handle.lower())87    if not user:88        return None89    return {90        "followers": int(user.get("followers_count") or 0),91        "following": user.get("friends_count"),92        "verified": bool(user.get("verified")) or None,93        "avatar": (user.get("profile_image_url_https") or "").replace(94            "_normal.", "_400x400."),95    }969798def parse_counts(html: str) -> tuple[int, int] | None:99    """(repli) HTML x.com → (followers, following) ou None (mur de connexion)."""100    for rx in _COUNT_RES:101        m = rx.search(html)102        if m:103            return int(m.group(1)), int(m.group(2))104    return None105106107class XProfilConnector(BaseConnector):108    source_id = "x-profil"109    kind = "enrichment"110    request_delay = 2.0           # poli — service d'intégration public111    timeout = 15112    max_profiles = 150113    revisit_days = 7114115    def __init__(self) -> None:116        super().__init__()117        self.errors = 0118        self._fallback_left = 10  # x.com bloque vite : repli très limité119120    def _needs_visit(self, metrics: dict) -> bool:121        raw = metrics.get("x_checked")122        if not raw:123            return True124        try:125            checked = datetime.fromisoformat(raw.replace("Z", "+00:00"))126        except ValueError:127            return True128        return (datetime.now(timezone.utc) - checked129                > timedelta(days=self.revisit_days))130131    def _curl(self, url: str) -> str:132        """GET via le binaire curl (voir docstring : requests est 429-é),133        throttling poli hérité du connecteur ; lève en cas d'échec HTTP."""134        self._throttle()135        try:136            proc = subprocess.run(137                ["curl", "-sS", "--fail", "--max-time", str(self.timeout),138                 "-A", USER_AGENT, "-H", "Accept-Language: en", url],139                capture_output=True, text=True, timeout=self.timeout + 5)140        finally:141            self._last_request = time.time()142        if proc.returncode != 0:143            raise RuntimeError(f"curl {proc.returncode}: {url}")144        return proc.stdout145146    def _lookup(self, handle: str) -> dict | None:147        info = parse_syndication(self._curl(SYNDICATION_URL.format(h=handle)),148                                 handle)149        if info:150            return info151        if self._fallback_left <= 0:152            return None153        self._fallback_left -= 1154        counts = parse_counts(self._curl(PAGE_URL.format(h=handle)))155        if counts is None:156            return None157        return {"followers": counts[0], "following": counts[1],158                "verified": None, "avatar": ""}159160    def enrich(self, creators: list[Creator]) -> list[Creator]:161        # jamais lus d'abord, puis par portée (l'appelant trie déjà par reach)162        candidates = []163        for cr in creators:164            acc = next((a for a in cr.platforms if a.platform == "x"), None)165            if acc is not None and self._needs_visit(acc.metrics):166                candidates.append((cr, acc))167        candidates.sort(key=lambda t: bool(t[1].metrics.get("x_checked")))168169        enriched: list[Creator] = []170        fetched = 0171        for cr, acc in candidates:172            if fetched >= self.max_profiles:173                break174            fetched += 1          # le cap borne les TENTATIVES réseau175            try:176                info = self._lookup(acc.handle)177            except Exception:178                self.errors += 1179                continue180            if info is None:181                # profil sans tweet public / suspendu : rien à lire — mais on182                # horodate pour ne pas re-consommer le cap avant 7 jours183                acc.metrics["x_checked"] = now_iso()184                enriched.append(cr)185                continue186            acc.followers = info["followers"]187            if info.get("verified") and acc.verified is None:188                acc.verified = True189            if info.get("following") is not None:190                acc.metrics["following"] = info["following"]191            acc.metrics["x_checked"] = now_iso()192            acc.last_checked = now_iso()193            if info.get("avatar", "").startswith("http") and not cr.avatar_url:194                cr.avatar_url = info["avatar"]195            enriched.append(cr)196        return enriched197