spb/crea-ka

1# ==============================================================================2# Author: Simon-Pierre Boucher <contact@spboucher.ai>3# File:   creaka/connectors/tiktok.py4# Desc:   Connecteur ENRICHISSEMENT TikTok — profil public par créateur :5#         abonnés, badge vérifié, bio, lien de bio (découverte croisée).6#         Mode d'accès : Scrapfly (plateforme hostile au scraping, API7#         officielle limitée aux comptes connectés) — palier 4 (§9), §15.8# ==============================================================================9"""Enrichissement TikTok, créateur par créateur.1011La page publique tiktok.com/@handle embarque un JSON d'hydratation12(__UNIVERSAL_DATA_FOR_REHYDRATION__) contenant stats.followerCount, la bio13(`signature`), le badge vérifié et le lien de bio (`bioLink`). On le lit via14Scrapfly. Lien de bio → page link-in-bio ou compte croisé (signal cross_link).15"""16from __future__ import annotations1718from datetime import datetime, timedelta, timezone1920import json21import re2223from ..identity import account, merge_accounts24from ..normalize import parse_count, platform_from_url25from ..schema import Creator, now_iso26from .base import BaseConnector27from .linkinbio import is_supported as is_linkinbio2829PROFILE_URL = "https://www.tiktok.com/@{h}"30_STATE_RE = re.compile(31    r'<script id="__UNIVERSAL_DATA_FOR_REHYDRATION__"[^>]*>(.*?)</script>', re.S)323334def parse_profile(html: str) -> dict | None:35    """HTML de profil TikTok → dict userInfo {user:…, stats:…} (None si absent)."""36    m = _STATE_RE.search(html)37    if not m:38        return None39    try:40        data = json.loads(m.group(1))41        return (data.get("__DEFAULT_SCOPE__", {})42                    .get("webapp.user-detail", {})43                    .get("userInfo")) or None44    except Exception:45        return None46474849def _fresh(last_checked: str, hours: int = 72) -> bool:50    """Compte déjà rafraîchi récemment (acteur Apify) — épargner Scrapfly."""51    if not last_checked:52        return False53    try:54        seen = datetime.fromisoformat(last_checked.replace("Z", "+00:00"))55    except ValueError:56        return False57    return datetime.now(timezone.utc) - seen < timedelta(hours=hours)5859class TikTokConnector(BaseConnector):60    source_id = "tiktok-profil"61    kind = "enrichment"62    request_delay = 1.563    max_profiles = 7006465    def enrich(self, creators: list[Creator]) -> list[Creator]:66        enriched: list[Creator] = []67        fetched = 068        for cr in creators:69            tk = next((a for a in cr.platforms if a.platform == "tiktok"), None)70            if tk is None or fetched >= self.max_profiles:71                continue72            if _fresh(tk.last_checked):73                continue74            try:75                result = self.scrapfly(PROFILE_URL.format(h=tk.handle),76                                       render_js=False)77            except Exception:78                continue79            fetched += 180            if result.get("status_code") != 200:81                continue82            info = parse_profile(result.get("content") or "")83            if not info:84                continue85            user = info.get("user") or {}86            stats = info.get("stats") or {}87            if user.get("privateAccount"):88                continue  # §15 : uniquement du public89            tk.followers = parse_count(stats.get("followerCount")) or tk.followers90            tk.verified = bool(user.get("verified"))91            tk.last_checked = now_iso()92            extra = {93                "following": parse_count(stats.get("followingCount")),94                "likes": parse_count(stats.get("heartCount")),95                "videos": parse_count(stats.get("videoCount")),96            }97            tk.metrics.update({k: v for k, v in extra.items() if v is not None})98            if not cr.avatar_url and user.get("avatarLarger"):99                cr.avatar_url = user["avatarLarger"]100            if not cr.bio and user.get("signature"):101                cr.bio = user["signature"]102            url = ((user.get("bioLink") or {}).get("link") or "").strip()103            if url and not url.startswith("http"):104                url = "https://" + url105            if url:106                if is_linkinbio(url) and not cr.link_in_bio_url:107                    cr.link_in_bio_url = url108                else:109                    hit = platform_from_url(url)110                    if hit and hit[0] != "tiktok":111                        cr.platforms = merge_accounts(cr.platforms, [112                            account(hit[0], hit[1], "cross_link",113                                    url=url).finalize()])114            enriched.append(cr)115        return enriched116