spb/crea-ka

1# ==============================================================================2# Author: Simon-Pierre Boucher <contact@spboucher.ai>3# File:   creaka/connectors/youtube.py4# Desc:   Connecteur ENRICHISSEMENT YouTube — stats de chaîne par créateur.5#         Mode d'accès : YouTube Data API v3 (OFFICIELLE, privilégiée §10) si6#         YOUTUBE_API_KEY est définie ; sinon voie SANS clé complète (vague 3) :7#         page publique de la chaîne + InnerTube « browse » (l'API interne que8#         la page elle-même appelle, clé publique embarquée). Palier 2-3 (§9).9# ==============================================================================10"""Enrichissement YouTube, créateur par créateur.1112Voie privilégiée : l'API officielle `channels?forHandle=` (fiable, conforme,13quota 10 000 unités/jour). Sans clé, voie publique complète (vague 3) :14151. page ``youtube.com/@handle`` (ytInitialData) → abonnés, nb de vidéos,16   description (→ bio si vide), bannière, avatar, id de chaîne UC…, et le17   jeton du panneau « À propos » ;182. InnerTube ``browse`` avec ce jeton (la requête que le navigateur fait en19   cliquant « …plus ») → vues totales, nb de vidéos exact, pays affiché ;203. page ``/videos`` → date approximative de la dernière vidéo (« il y a X »)21   → fraîcheur/dormance (metrics.last_video + dormant >12 mois).2223Cap 300 chaînes/passage avec ROTATION (metrics.yt_checked, re-visite 14 j,24jamais-lues d'abord) : les ~4 200 comptes YouTube convergent en ~2 semaines.25Signal : `api_officielle` (0.95) quand confirmé par l'API — le handle26interrogé est celui déjà rattaché ; la voie publique ne touche pas au score.27"""28from __future__ import annotations2930import json31import os32import re33from datetime import datetime, timedelta, timezone3435from ..normalize import parse_count36from ..schema import Creator, now_iso37from .base import BaseConnector3839API_URL = ("https://www.googleapis.com/youtube/v3/channels"40           "?part=statistics,snippet&forHandle={h}&key={key}")41PAGE_URL = "https://www.youtube.com/@{h}"42CHANNEL_URL = "https://www.youtube.com/channel/{h}"43BROWSE_URL = "https://www.youtube.com/youtubei/v1/browse"4445_YTID_RE = re.compile(r"var ytInitialData = (\{.*?\});</script>", re.S)46_KEY_RE = re.compile(r'"INNERTUBE_API_KEY":"([^"]+)"')47_VER_RE = re.compile(r'"INNERTUBE_CONTEXT_CLIENT_VERSION":"([^"]+)"')48_UCID_RE = re.compile(r"UC[0-9A-Za-z_-]{22}")49_SUBS_RES = (50    re.compile(r'"subscriberCountText"[^}]*?"(?:simpleText|content)"\s*:\s*"([^"]+)"'),51    re.compile(r'"subscriberCount"\s*:\s*"?(\d+)"?'),52    re.compile(r'"content"\s*:\s*"([\d.,\s]+[kKmMbB]?\s*(?:subscribers|abonn[ée]s))"'),53)54_VIDEOS_RE = re.compile(r'"content"\s*:\s*"([\d.,\s]+[kKmM]?)\s*videos?"')55# première vidéo de l'onglet /videos (tri chronologique) : « 3 weeks ago »56_AGO_RE = re.compile(r'"(?:content|simpleText)":"(?:Streamed )?'57                     r'(\d+ (?:second|minute|hour|day|week|month|year)s? ago)"')58_AGO_UNITS = {"second": 1, "minute": 60, "hour": 3600, "day": 86400,59              "week": 7 * 86400, "month": 30 * 86400, "year": 365 * 86400}6061# au-delà de 12 mois sans vidéo : chaîne dormante (métrique publique, comme62# balados-itunes — la fiche reste active, le créateur vit peut-être ailleurs)63DORMANT_AFTER_DAYS = 365646566def ago_to_iso(text: str, now: datetime | None = None) -> str | None:67    """« 3 weeks ago » / « Streamed 1 year ago » → date ISO approximative."""68    m = re.search(r"(\d+)\s+(second|minute|hour|day|week|month|year)s?\s+ago",69                  text or "")70    if not m:71        return None72    seconds = int(m.group(1)) * _AGO_UNITS[m.group(2)]73    at = (now or datetime.now(timezone.utc)) - timedelta(seconds=seconds)74    return at.strftime("%Y-%m-%d")757677def _walk_token(node) -> str | None:78    """Premier jeton continuationCommand d'un sous-arbre JSON."""79    if isinstance(node, dict):80        cmd = node.get("continuationCommand")81        if isinstance(cmd, dict) and cmd.get("token"):82            return cmd["token"]83        for v in node.values():84            tok = _walk_token(v)85            if tok:86                return tok87    elif isinstance(node, list):88        for item in node:89            tok = _walk_token(item)90            if tok:91                return tok92    return None939495def parse_channel_page(html: str) -> dict | None:96    """Page publique de la chaîne → {followers, videos, description, avatar,97    banner, channel_id, about_token, key, ver} (champs manquants = None)."""98    m = _YTID_RE.search(html)99    if not m:100        return None101    try:102        data = json.loads(m.group(1))103    except ValueError:104        return None105    meta = (data.get("metadata") or {}).get("channelMetadataRenderer") or {}106    out: dict = {"description": (meta.get("description") or "").strip(),107                 "channel_id": meta.get("externalId"),108                 "followers": None, "videos": None,109                 "avatar": None, "banner": None, "about_token": None}110    thumbs = ((meta.get("avatar") or {}).get("thumbnails")) or []111    if thumbs:112        out["avatar"] = max(thumbs, key=lambda t: t.get("width") or 0).get("url")113    header = json.dumps((data.get("header") or {}), ensure_ascii=False)114    for rx in _SUBS_RES:115        mm = rx.search(header)116        if mm:117            out["followers"] = parse_count(mm.group(1))118            break119    mm = _VIDEOS_RE.search(header)120    if mm:121        out["videos"] = parse_count(mm.group(1))122    vm = ((((data.get("header") or {}).get("pageHeaderRenderer") or {})123           .get("content") or {}).get("pageHeaderViewModel") or {})124    sources = ((((vm.get("banner") or {}).get("imageBannerViewModel") or {})125                .get("image") or {}).get("sources")) or []126    if sources:127        out["banner"] = max(sources, key=lambda s: s.get("width") or 0).get("url")128    # jeton du panneau « À propos » : sous l'aperçu de description du header129    # (c'est LA requête que fait le navigateur au clic sur « …plus »)130    out["about_token"] = _walk_token(vm.get("description")) or \131        _walk_token(vm.get("attribution"))132    key_m, ver_m = _KEY_RE.search(html), _VER_RE.search(html)133    out["key"] = key_m.group(1) if key_m else None134    out["ver"] = ver_m.group(1) if ver_m else None135    return out136137138def parse_about(text: str) -> dict | None:139    """Réponse browse du panneau À propos → {views, videos, country} ou None."""140    if "aboutChannelViewModel" not in (text or ""):141        return None142    out: dict = {}143    mm = re.search(r'"viewCountText"\s*:\s*"([^"]+)"', text)144    if mm:145        out["views"] = parse_count(mm.group(1))146    mm = re.search(r'"videoCountText"\s*:\s*"([^"]+)"', text)147    if mm:148        out["videos"] = parse_count(mm.group(1))149    mm = re.search(r'"country"\s*:\s*(?:\{"content":)?"([^"]+)"', text)150    if mm:151        out["country"] = mm.group(1)152    return out153154155class YouTubeConnector(BaseConnector):156    source_id = "youtube"157    kind = "enrichment"158    request_delay = 1.0159    max_profiles = 1500        # voie API officielle (1 requête légère/chaîne)160    max_profiles_page = 300    # voie publique (2-3 requêtes/chaîne) — rotation161    revisit_days = 14162163    def __init__(self) -> None:164        super().__init__()165        self.errors = 0166167    # -- voie API officielle -----------------------------------------------------168    def _via_api(self, handle: str, key: str) -> dict | None:169        resp = self.get(API_URL.format(h=handle, key=key))170        items = resp.json().get("items") or []171        if not items:172            return None173        stats = items[0].get("statistics") or {}174        return {"followers": parse_count(stats.get("subscriberCount")),175                "views": parse_count(stats.get("viewCount")),176                "videos": parse_count(stats.get("videoCount")),177                "confirmed": True}178179    # -- voie publique sans clé (vague 3) -----------------------------------------180    def _channel_url(self, acc) -> str:181        if _UCID_RE.fullmatch(acc.handle):182            return CHANNEL_URL.format(h=acc.handle)183        return PAGE_URL.format(h=acc.handle)184185    def _browse_about(self, key: str, ver: str, token: str) -> dict | None:186        resp = self.post(187            BROWSE_URL, params={"key": key},188            json={"context": {"client": {"clientName": "WEB",189                                         "clientVersion": ver,190                                         "hl": "en", "gl": "CA"}},191                  "continuation": token},192            headers={"Accept-Language": "en-CA,en;q=0.8"})193        return parse_about(resp.text)194195    def _last_video(self, base_url: str) -> str | None:196        html = self.get(base_url.rstrip("/") + "/videos",197                        headers={"Accept-Language": "en-CA,en;q=0.8"}).text198        m = _YTID_RE.search(html)199        if not m:200            return None201        mm = _AGO_RE.search(m.group(1))202        return ago_to_iso(mm.group(1)) if mm else None203204    def _via_page(self, acc) -> dict | None:205        base = self._channel_url(acc)206        html = self.get(base, headers={"Accept-Language": "en-CA,en;q=0.8"}).text207        page = parse_channel_page(html)208        if page is None:209            return None210        info: dict = {"followers": page["followers"], "videos": page["videos"],211                      "description": page["description"], "avatar": page["avatar"],212                      "banner": page["banner"], "confirmed": False}213        if page["about_token"] and page["key"] and page["ver"]:214            try:  # vues totales + nb de vidéos exact « si affichés »215                about = self._browse_about(page["key"], page["ver"],216                                           page["about_token"]) or {}217                info.update({k: v for k, v in about.items() if v is not None})218            except Exception:219                self.errors += 1220        if info.get("videos"):  # date de la dernière vidéo (fraîcheur/dormance)221            try:222                info["last_video"] = self._last_video(base)223            except Exception:224                self.errors += 1225        return info226227    def _needs_visit(self, metrics: dict) -> bool:228        raw = metrics.get("yt_checked")229        if not raw:230            return True231        try:232            checked = datetime.fromisoformat(raw.replace("Z", "+00:00"))233        except ValueError:234            return True235        return (datetime.now(timezone.utc) - checked236                > timedelta(days=self.revisit_days))237238    def enrich(self, creators: list[Creator]) -> list[Creator]:239        key = os.environ.get("YOUTUBE_API_KEY", "")240        if key:241            return self._enrich_api(creators, key)242        return self._enrich_page(creators)243244    def _enrich_api(self, creators: list[Creator], key: str) -> list[Creator]:245        enriched: list[Creator] = []246        fetched = 0247        for cr in creators:248            yt = next((a for a in cr.platforms if a.platform == "youtube"), None)249            if yt is None or fetched >= self.max_profiles:250                continue251            try:252                info = self._via_api(yt.handle, key)253            except Exception:254                self.errors += 1255                continue256            fetched += 1257            if not info or info.get("followers") is None:258                continue259            yt.followers = info["followers"]260            yt.last_checked = now_iso()261            extra = {k: info.get(k) for k in ("views", "videos")}262            yt.metrics.update({k: v for k, v in extra.items() if v is not None})263            # existence confirmée par l'API officielle → signal renforcé264            yt.confidence = max(yt.confidence or 0, 0.90)265            yt.signal = yt.signal or "api_officielle"266            enriched.append(cr)267        return enriched268269    def _enrich_page(self, creators: list[Creator]) -> list[Creator]:270        # rotation : jamais-lues d'abord, puis les plus anciennes re-visites ;271        # l'appelant (ingest.load_active) trie déjà par portée décroissante272        candidates = []273        for cr in creators:274            yt = next((a for a in cr.platforms if a.platform == "youtube"), None)275            if yt is not None and self._needs_visit(yt.metrics):276                candidates.append((cr, yt))277        candidates.sort(key=lambda t: bool(t[1].metrics.get("yt_checked")))278279        enriched: list[Creator] = []280        fetched = 0281        for cr, yt in candidates:282            if fetched >= self.max_profiles_page:283                break284            fetched += 1          # le cap borne les TENTATIVES réseau285            try:286                info = self._via_page(yt)287            except Exception:288                self.errors += 1289                continue290            yt.metrics["yt_checked"] = now_iso()291            if info is None:      # chaîne fermée/renommée : horodatée, sans plus292                enriched.append(cr)293                continue294            if info.get("followers") is not None:295                yt.followers = info["followers"]296            yt.last_checked = now_iso()297            dormant = None298            if info.get("last_video"):299                try:300                    last = datetime.strptime(info["last_video"], "%Y-%m-%d")301                    dormant = (datetime.now(timezone.utc)302                               - last.replace(tzinfo=timezone.utc)303                               > timedelta(days=DORMANT_AFTER_DAYS)) or None304                except ValueError:305                    pass306            yt.metrics.update({k: v for k, v in {307                "videos": info.get("videos"),308                "views": info.get("views"),309                "last_video": info.get("last_video"),310                "dormant": dormant,311                "country": info.get("country"),312                "banner": info.get("banner"),313            }.items() if v is not None})314            if not cr.bio and info.get("description"):315                cr.bio = info["description"][:1200]316            if not cr.avatar_url and (info.get("avatar") or "").startswith("http"):317                cr.avatar_url = info["avatar"]318            enriched.append(cr)319        return enriched320