spb/crea-ka

1# ==============================================================================2# Author: Simon-Pierre Boucher <contact@spboucher.ai>3# File:   creaka/connectors/onlyqueb.py4# Desc:   Connecteur DÉCOUVERTE — OnlyQueb (onlyqueb.com), annuaire public de5#         créatrices/créateurs québécois sur abonnement. Mode d'accès : requêtes6#         directes (sitemap.xml + JSON-LD des pages profil, SSR). Palier 3 (§9).7#         Les liens `sameAs` sont AUTO-DÉCLARÉS par la personne sur son profil8#         → signal cross_link (0.90, §12.1).9# ==============================================================================10"""Découverte via le hub OnlyQueb.1112Chaque page profil expose un JSON-LD schema.org ProfilePage propre :13nom public, photo de profil publique et liste `sameAs` des comptes14(OnlyFans, Instagram, TikTok, X…). Adultes uniquement (plateforme 18+).15On n'agrège que ce que la personne affiche elle-même sur son profil (§15).16"""17from __future__ import annotations1819import json20import re2122from ..identity import account23from ..normalize import map_niche, platform_from_url, slugify24from ..schema import Creator25from .base import BaseConnector2627SITEMAP_URL = "https://onlyqueb.com/sitemap.xml"28_LOC_RE = re.compile(r"<loc>([^<]+)</loc>")29_LD_RE = re.compile(r'application/ld\+json"[^>]*>(.*?)</script>', re.S)30_EXCLUDED = ("/en", "/categories", "/explore", "/seo", "/signup", "/blog")31# fragments du payload Next.js/RSC embarqué dans la page (self.__next_f.push)32_PUSH_RE = re.compile(r'self\.__next_f\.push\(\[1,"((?:[^"\\]|\\.)*)"\]\)')3334# catégories OnlyQueb → niches §6.1 quand le lien de sens existe ; les35# descripteurs physiques (blonde, petite…) ne sont PAS des niches → « autre »36_CAT_NICHE = {"cosplay": "arts", "gamer": "gaming", "gaming": "gaming",37              "fitness": "sport-fitness", "athletic": "sport-fitness"}383940def parse_profile_ld(html: str) -> dict | None:41    """Page profil → {name, image, links[]} depuis le JSON-LD (None si absent)."""42    for blob in _LD_RE.findall(html):43        try:44            data = json.loads(blob)45        except Exception:46            continue47        if data.get("@type") != "ProfilePage":48            continue49        person = data.get("mainEntity") or {}50        if person.get("@type") != "Person":51            continue52        return {"name": (person.get("name") or "").strip(),53                "image": person.get("image") or None,54                "links": [u for u in (person.get("sameAs") or [])55                          if isinstance(u, str)]}56    return None575859def parse_profile_next(html: str) -> dict:60    """Page profil → objet `profile` du payload Next.js/RSC ({} si absent).6162    La même page expose, au-delà du JSON-LD, la fiche complète auto-déclarée :63    displayName réel, bio, headline, categories[], mymUrl/fanslyUrl/ofUrl,64    websiteUrl, updatedAt. Chaque fragment `self.__next_f.push([1,"…"])` est65    une chaîne JS → décodée comme chaîne JSON, puis l'objet `"profile":{…}`66    est extrait par décodage équilibré (raw_decode).67    """68    for m in _PUSH_RE.finditer(html):69        try:70            chunk = json.loads(f'"{m.group(1)}"')71        except Exception:72            continue73        i = chunk.find('"profile":{')74        if i < 0:75            continue76        try:77            obj, _ = json.JSONDecoder().raw_decode(chunk, i + len('"profile":'))78        except Exception:79            continue80        if isinstance(obj, dict) and obj.get("displayName") is not None:81            return obj82    return {}838485class OnlyQuebConnector(BaseConnector):86    source_id = "onlyqueb"87    kind = "discovery"88    request_delay = 1.089    max_profiles = 600       # garde-fou par passage9091    def _profile_urls(self) -> list[str]:92        xml = self.get(SITEMAP_URL).text93        urls = []94        for u in _LOC_RE.findall(xml):95            path = u.split("onlyqueb.com", 1)[-1]96            if path and path != "/" and path.count("/") == 1 \97                    and not any(path.startswith(e) for e in _EXCLUDED):98                urls.append(u)99        return urls[:self.max_profiles]100101    def fetch(self) -> list[Creator]:102        creators: list[Creator] = []103        for url in self._profile_urls():104            try:105                html = self.get(url).text106            except Exception:107                continue108            prof = parse_profile_ld(html)109            if not prof or not prof["name"]:110                continue111            extra = parse_profile_next(html)  # payload RSC : fiche complète112            links = list(prof["links"])113            for key in ("ofUrl", "mymUrl", "fanslyUrl", "instagramUrl",114                        "xUrl", "tiktokUrl", "snapchatUrl"):115                if (extra.get(key) or "").startswith("http"):116                    links.append(extra[key])117            accounts = []118            seen: set[str] = set()119            for link in links:120                hit = platform_from_url(link)121                if not hit:122                    continue123                platform, handle = hit124                if f"{platform}:{handle}" in seen:125                    continue126                seen.add(f"{platform}:{handle}")127                accounts.append(account(platform, handle, "cross_link", url=link))128            # site web personnel auto-déclaré → compte site-web (annuaire de liens)129            site = (extra.get("websiteUrl") or "").strip()130            if site.startswith("http") and not platform_from_url(site):131                domain = site.split("://", 1)[-1].split("/", 1)[0]132                domain = domain.removeprefix("www.")133                if domain and f"site-web:{domain}" not in seen:134                    seen.add(f"site-web:{domain}")135                    accounts.append(account("site-web", domain, "cross_link",136                                            url=site))137            if not accounts:138                continue  # sans lien vérifiable, pas de fiche (annuaire de liens)139            name = (extra.get("displayName") or "").strip() or prof["name"]140            bio = "\n".join(141                t.strip() for t in ((extra.get("headline") or ""),142                                    (extra.get("bio") or "")) if t.strip())143            cats = [c for c in (extra.get("categories") or [])144                    if isinstance(c, str)]145            niches = sorted({_CAT_NICHE.get(c, map_niche(c)) for c in cats})146            niches = [n for n in niches if n != "autre"] or ["autre"]147            slug = slugify(prof["name"])148            creators.append(Creator(149                source=self.source_id,150                external_id=slug,151                display_name=name,152                bio=bio,153                niches=niches,154                creator_type="influenceur",155                avatar_url=prof["image"],156                platforms=accounts,157                notes="Profil auto-déclaré sur l'annuaire OnlyQueb (créateurs "158                      "québécois sur abonnement, 18+).",159                source_ids=[f"{self.source_id}:{slug}"],160            ))161        return creators162