# ============================================================================== # Author: Simon-Pierre Boucher # File: src/main.py (ka-tiktok) # Desc: Profils TikTok publics ULTRA DÉTAILLÉS via la page @handle : le JSON # SSR __UNIVERSAL_DATA_FOR_REHYDRATION__ contient user + stats complètes # (abonnés, cœurs, vidéos) ; vidéos récentes avec COVERS (miniatures), # durée, hashtags, musique, épinglés + cadence (videos_per_week) et # top hashtags quand TikTok les inclut dans le SSR (ItemModule/itemList). # ============================================================================== from __future__ import annotations import asyncio import json import re from apify import Actor from .net import Fetcher PROFILE_URL = "https://www.tiktok.com/@{u}" _STATE_RE = re.compile( r'', re.S) _ITEM_RE = re.compile(r'"ItemModule"\s*:\s*(\{.*?\})\s*,\s*"[A-Z]', re.S) def parse_videos(html: str, data: dict) -> list[dict]: """Vidéos récentes si le SSR les inclut (souvent absent — best effort).""" items: list[dict] = [] raw: dict = {} scope = data.get("__DEFAULT_SCOPE__") or {} detail = scope.get("webapp.user-detail") or {} for it in (detail.get("itemList") or []): raw[it.get("id", str(len(raw)))] = it if not raw: m = _ITEM_RE.search(html) if m: try: raw = json.loads(m.group(1)) except Exception: raw = {} for vid in list(raw.values())[:20]: stats = vid.get("stats") or vid.get("statsV2") or {} author = vid.get("author") handle = author if isinstance(author, str) else \ (author or {}).get("uniqueId", "") video = vid.get("video") or {} music = vid.get("music") or {} items.append({ "id": vid.get("id"), "url": f"https://www.tiktok.com/@{handle}/video/{vid.get('id')}", "caption": (vid.get("desc") or "")[:500], "views": _num(stats.get("playCount")), "likes": _num(stats.get("diggCount")), "comments": _num(stats.get("commentCount")), "shares": _num(stats.get("shareCount")), "saves": _num(stats.get("collectCount")), "timestamp": vid.get("createTime"), "cover": (video.get("cover") or video.get("dynamicCover") or video.get("originCover")), "duration_s": _num(video.get("duration")), "width": _num(video.get("width")), "height": _num(video.get("height")), "is_pinned": bool(vid.get("isPinnedItem")), "hashtags": [c.get("hashtagName") for c in (vid.get("textExtra") or []) if c.get("hashtagName")][:8], "music": ({"title": music.get("title"), "author": music.get("authorName"), "original": music.get("original")} if music.get("title") else None), }) return items def _num(v): try: return int(v) except (TypeError, ValueError): return None async def main() -> None: async with Actor: inp = await Actor.get_input() or {} usernames = [u.strip().lstrip("@").lower() for u in (inp.get("usernames") or []) if u and u.strip()] proxy = await Actor.create_proxy_configuration( actor_proxy_input=inp.get("proxyConfiguration")) fetcher = Fetcher(proxy, delay=float(inp.get("requestDelay") or 1)) sem = asyncio.Semaphore(int(inp.get("concurrency") or 3)) async def one(u: str) -> None: async with sem: miss = {"kind": "profile", "platform": "tiktok", "username": u, "found": False} resp = None m = None for attempt in range(3): # coquille sans SSR → nouvelle IP try: resp = await fetcher.get(PROFILE_URL.format(u=u), session_id=f"{u}t{attempt}") except Exception as exc: await Actor.push_data({**miss, "error": str(exc)[:200]}) return if resp.status_code == 404: await Actor.push_data({**miss, "error": "http_404"}) return m = _STATE_RE.search(resp.text or "") if m and "webapp.user-detail" in (resp.text or ""): break info = None data: dict = {} if m: try: data = json.loads(m.group(1)) info = ((data.get("__DEFAULT_SCOPE__") or {}) .get("webapp.user-detail") or {}).get("userInfo") except Exception: info = None if not info or not (info.get("user") or {}).get("uniqueId"): await Actor.push_data( {**miss, "error": f"shell_page_{resp.status_code}"}) return user = info.get("user") or {} stats = info.get("statsV2") or info.get("stats") or {} if user.get("privateAccount"): await Actor.push_data({**miss, "error": "private"}) return followers = _num(stats.get("followerCount")) videos = parse_videos(resp.text, data) views = [v["views"] for v in videos if isinstance(v["views"], int)] likes = [v["likes"] for v in videos if isinstance(v["likes"], int)] comments = [v["comments"] for v in videos if isinstance(v["comments"], int)] shares = [v["shares"] for v in videos if isinstance(v["shares"], int)] saves = [v["saves"] for v in videos if isinstance(v["saves"], int)] avg_views = round(sum(views) / len(views)) if views else None avg_likes = round(sum(likes) / len(likes)) if likes else None engagement = None if followers and avg_likes is not None: engagement = round( (avg_likes + (round(sum(comments) / len(comments)) if comments else 0) + (round(sum(shares) / len(shares)) if shares else 0)) / followers * 100, 3) top_video = max( (v for v in videos if isinstance(v["views"], int)), key=lambda v: v["views"], default=None) stamps = sorted(_num(v["timestamp"]) for v in videos if _num(v["timestamp"])) videos_per_week = None if len(stamps) >= 3 and stamps[-1] > stamps[0]: videos_per_week = round( (len(stamps) - 1) / ((stamps[-1] - stamps[0]) / 604_800), 2) last_video_at = None if stamps: from datetime import datetime, timezone last_video_at = datetime.fromtimestamp( stamps[-1], tz=timezone.utc) \ .strftime("%Y-%m-%dT%H:%M:%SZ") tag_counts: dict[str, int] = {} for v in videos: for h in v.get("hashtags") or []: tag_counts[h.lower()] = tag_counts.get(h.lower(), 0) + 1 commerce = user.get("commerceUserInfo") or {} bio_link = ((user.get("bioLink") or {}).get("link") or "").strip() total_likes = _num(stats.get("heartCount") or stats.get("heart")) videos_count = _num(stats.get("videoCount")) await Actor.push_data({ "kind": "profile", "platform": "tiktok", "found": True, "id": user.get("id"), "sec_uid": user.get("secUid"), "username": user.get("uniqueId"), "full_name": user.get("nickname"), "biography": user.get("signature"), "bio_link": bio_link, "followers": followers, "following": _num(stats.get("followingCount")), "friends": _num(stats.get("friendCount")), "likes_given": _num(stats.get("diggCount")), "total_likes": total_likes, "videos_count": videos_count, "avg_likes_per_video_lifetime": ( round(total_likes / videos_count) if total_likes and videos_count else None), "is_verified": bool(user.get("verified")), "is_organization": bool(user.get("isOrganization")), "is_seller": bool(user.get("ttSeller")), "is_live_now": bool(user.get("roomId")), # drapeau mineur déclaré par TikTok → régime restreint §15 "is_under_18": (bool(user.get("isUnderAge18")) if "isUnderAge18" in user else None), "is_embed_banned": (bool(user.get("isEmbedBanned")) if "isEmbedBanned" in user else None), "commerce_category": (commerce.get("category") or None if commerce.get("commerceUser") else None), "account_created_at": user.get("createTime"), "region": user.get("region"), "language": user.get("language"), "avatar": user.get("avatarLarger") or user.get("avatarMedium"), # agrégats sur les vidéos présentes dans le SSR "avg_views": avg_views, "avg_likes": avg_likes, "avg_comments": (round(sum(comments) / len(comments)) if comments else None), "avg_shares": (round(sum(shares) / len(shares)) if shares else None), "avg_saves": (round(sum(saves) / len(saves)) if saves else None), "pinned_videos_count": (sum(1 for v in videos if v["is_pinned"]) or None), "engagement_rate_pct": engagement, "videos_per_week": videos_per_week, "last_video_at": last_video_at, "top_hashtags": sorted(tag_counts, key=tag_counts.get, reverse=True)[:8], "top_video": ({"url": top_video["url"], "views": top_video["views"], "likes": top_video["likes"], "cover": top_video.get("cover")} if top_video else None), "recent_videos": videos, }) await asyncio.gather(*[one(u) for u in usernames]) Actor.log.info(f"terminé : {len(usernames)} profils")