spb/crea-ka

1# ==============================================================================2# Author: Simon-Pierre Boucher <contact@spboucher.ai>3# File:   src/main.py (ka-snapchat)4# Desc:   Profils Snapchat publics via le JSON __NEXT_DATA__ de la page5#         snapchat.com/add/<user> : publicProfileInfo (abonnés, bio, badge,6#         catégorie, avatar) + présence story/spotlight + comptes reliés.7#         Structure validée le 2026-08-18 (pageProps.userProfile.$case).8# ==============================================================================9from __future__ import annotations1011import asyncio12import json13import re1415from apify import Actor1617from .net import Fetcher1819PROFILE_URL = "https://www.snapchat.com/add/{u}"20_NEXT_DATA_RE = re.compile(21    r'<script id="__NEXT_DATA__"[^>]*>(.*?)</script>', re.S)222324def _num(v):25    try:26        return int(v)27    except (TypeError, ValueError):28        return None293031async def main() -> None:32    async with Actor:33        inp = await Actor.get_input() or {}34        usernames = [u.strip().lstrip("@").lower()35                     for u in (inp.get("usernames") or []) if u and u.strip()]36        proxy = await Actor.create_proxy_configuration(37            actor_proxy_input=inp.get("proxyConfiguration"))38        fetcher = Fetcher(proxy, delay=float(inp.get("requestDelay") or 1))39        sem = asyncio.Semaphore(int(inp.get("concurrency") or 3))4041        async def one(u: str) -> None:42            async with sem:43                miss = {"kind": "profile", "platform": "snapchat",44                        "username": u, "found": False}45                try:46                    resp = await fetcher.get(PROFILE_URL.format(u=u),47                                             session_id=u)48                except Exception as exc:49                    await Actor.push_data({**miss, "error": str(exc)[:200]})50                    return51                m = _NEXT_DATA_RE.search(resp.text or "")52                if resp.status_code != 200 or not m:53                    await Actor.push_data(54                        {**miss, "error": f"http_{resp.status_code}"})55                    return56                try:57                    props = (json.loads(m.group(1)).get("props") or {}) \58                        .get("pageProps") or {}59                except Exception:60                    await Actor.push_data({**miss, "error": "bad_json"})61                    return62                up = props.get("userProfile") or {}63                info = up.get("publicProfileInfo") or {}64                story = (props.get("story") or {})65                snaps = story.get("snapList") or []66                lenses = props.get("lenses") or []67                spotlight = props.get("spotlightHighlights") or []68                if not info.get("username"):69                    await Actor.push_data({**miss, "error": "no_profile"})70                    return71                related = [72                    {"username": r.get("username"), "title": r.get("title")}73                    for r in ((info.get("relatedAccountsInfo") or [])74                              if isinstance(info.get("relatedAccountsInfo"),75                                            list) else [])76                    if isinstance(r, dict) and r.get("username")]77                story_previews = []78                for s in snaps[:6]:79                    if not isinstance(s, dict):80                        continue81                    urls = s.get("snapUrls") or {}82                    prev = (urls.get("mediaPreviewUrl") or {}) \83                        if isinstance(urls.get("mediaPreviewUrl"), dict) \84                        else {"value": urls.get("mediaPreviewUrl")}85                    url = prev.get("value") or urls.get("mediaUrl")86                    if url:87                        story_previews.append(url)88                spotlight_previews = [89                    h.get("thumbnailUrl")90                    for h in spotlight[:6]91                    if isinstance(h, dict) and h.get("thumbnailUrl")]92                subs = _num(info.get("subscriberCount"))93                await Actor.push_data({94                    "kind": "profile",95                    "platform": "snapchat",96                    "found": True,97                    "username": info.get("username"),98                    "full_name": info.get("title"),99                    "biography": info.get("bio"),100                    "followers": subs if subs else None,  # 0 = masqué101                    "is_verified": str(info.get("badge")) == "1",102                    "category": info.get("categoryStringId"),103                    "website": info.get("websiteUrl"),104                    "address": info.get("address"),105                    "avatar": info.get("profilePictureUrl"),106                    "banner": (info.get("squareHeroImageUrl")107                               or info.get("heroImageUrl") or None),108                    "snapcode": info.get("snapcodeImageUrl") or None,109                    "has_story": bool(info.get("hasStory")),110                    "has_spotlight": bool(info.get("hasSpotlightHighlights")),111                    "has_curated_highlights": bool(112                        info.get("hasCuratedHighlights")),113                    "story_snaps_count": len(snaps) or None,114                    "story_previews": story_previews or None,115                    "lenses_count": len(lenses) or None,116                    "spotlight_highlights_count": len(spotlight) or None,117                    "spotlight_previews": spotlight_previews or None,118                    "publisher_type": info.get("publisherType") or None,119                    "subcategory": info.get("subcategoryStringId") or None,120                    "related_accounts": related,121                })122123        await asyncio.gather(*[one(u) for u in usernames])124        Actor.log.info(f"terminé : {len(usernames)} profils")125