# ============================================================================== # Author: Simon-Pierre Boucher # File: src/main.py (ka-x) # Desc: Profils X (Twitter) SANS clé via le service OFFICIEL de syndication # des widgets (syndication.twitter.com/srv/timeline-profile) : objet # user complet + timeline récente avec métriques par tweet. Conçu pour # être appelé sans authentification ; l'empreinte TLS Chrome de # curl_cffi évite les 429 réservés aux clients python-requests. # ============================================================================== from __future__ import annotations import asyncio import json import re import time from apify import Actor from .net import Fetcher SYNDICATION_URL = ("https://syndication.twitter.com/srv/timeline-profile/" "screen-name/{u}") _NEXT_DATA_RE = re.compile( r'', re.S) def _find_user(node, handle: str): """Parcourt le JSON Next.js : l'objet user dont screen_name == handle.""" if isinstance(node, dict): if str(node.get("screen_name", "")).lower() == handle and \ "followers_count" in node: return node for v in node.values(): hit = _find_user(v, handle) if hit is not None: return hit elif isinstance(node, list): for v in node: hit = _find_user(v, handle) if hit is not None: return hit return None def parse_tweets(data: dict, handle: str, cap: int) -> list[dict]: entries = (((data.get("props") or {}).get("pageProps") or {}) .get("timeline") or {}).get("entries") or [] tweets: list[dict] = [] for entry in entries: t = (entry.get("content") or {}).get("tweet") or {} if not t.get("id_str"): continue u = (t.get("user") or {}) ent = t.get("entities") or {} media = (t.get("extended_entities") or ent).get("media") or [] tweets.append({ "id": t.get("id_str"), "url": f"https://x.com/{u.get('screen_name', handle)}" f"/status/{t.get('id_str')}", "text": (t.get("full_text") or t.get("text") or "")[:500], "likes": t.get("favorite_count"), "retweets": t.get("retweet_count"), "replies": t.get("reply_count"), "quotes": t.get("quote_count"), "created_at": t.get("created_at"), "is_retweet": bool(t.get("retweeted_status")), "is_reply": bool(t.get("in_reply_to_status_id_str")), "is_quote": bool(t.get("is_quote_status")), "lang": t.get("lang"), "hashtags": [h.get("text") for h in (ent.get("hashtags") or []) if h.get("text")][:8], "mentions": [m.get("screen_name") for m in (ent.get("user_mentions") or []) if m.get("screen_name")][:8], "media_type": (media[0].get("type") if media else None), "media_count": len(media) or None, "media_urls": [x.get("media_url_https") for x in media if x.get("media_url_https")][:4], "links": [x.get("expanded_url") for x in (ent.get("urls") or []) if x.get("expanded_url")][:4], }) if len(tweets) >= cap: break return tweets def _ts(created_at: str) -> float | None: """« Tue Sep 12 23:56:24 +0000 2023 » → epoch (None si illisible).""" try: from datetime import datetime return datetime.strptime(created_at, "%a %b %d %H:%M:%S %z %Y").timestamp() 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()] max_tweets = int(inp.get("maxTweets", 20)) 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 4)) async def one(u: str) -> None: async with sem: miss = {"kind": "profile", "platform": "x", "username": u, "found": False} try: resp = await fetcher.get(SYNDICATION_URL.format(u=u), session_id=u, impersonate=None) # 429 : retries avec backoff + NOUVELLE session proxy # (nouvelle IP résidentielle) — une seule relance ne # suffisait pas (13/40 handles en 429, constat 2026-08-21) for i, imp in enumerate(("chrome", None, "chrome")): if resp.status_code != 429: break await asyncio.sleep(5 * (i + 1)) resp = await fetcher.get(SYNDICATION_URL.format(u=u), session_id=f"{u}r{i}", impersonate=imp) except Exception as exc: await Actor.push_data({**miss, "error": str(exc)[:200]}) return m = _NEXT_DATA_RE.search(resp.text or "") if resp.status_code != 200 or not m: await Actor.push_data( {**miss, "error": f"http_{resp.status_code}"}) return try: data = json.loads(m.group(1)) except Exception: await Actor.push_data({**miss, "error": "bad_json"}) return user = _find_user(data, u) if not user: await Actor.push_data({**miss, "error": "user_not_found"}) return tweets = parse_tweets(data, u, max_tweets) own = [t for t in tweets if not t["is_retweet"]] likes = [t["likes"] for t in own if isinstance(t["likes"], int)] rts = [t["retweets"] for t in own if isinstance(t["retweets"], int)] reps = [t["replies"] for t in own if isinstance(t["replies"], int)] followers = user.get("followers_count") avg_likes = round(sum(likes) / len(likes)) if likes else None avg_rts = round(sum(rts) / len(rts)) if rts else None engagement = None if followers and avg_likes is not None: engagement = round((avg_likes + (avg_rts or 0)) / followers * 100, 3) stamps = sorted(s for s in (_ts(t["created_at"]) for t in own) if s) tweets_per_week = None if len(stamps) >= 3 and stamps[-1] > stamps[0]: tweets_per_week = round( (len(stamps) - 1) / ((stamps[-1] - stamps[0]) / 604_800), 2) top = max((t for t in own if isinstance(t["likes"], int)), key=lambda t: t["likes"], default=None) hashtags: dict[str, int] = {} mentions: dict[str, int] = {} for t in own: for h in t["hashtags"]: hashtags[h.lower()] = hashtags.get(h.lower(), 0) + 1 for mn in t["mentions"]: mentions[mn.lower()] = mentions.get(mn.lower(), 0) + 1 quotes = [t["quotes"] for t in own if isinstance(t["quotes"], int)] created_ts = _ts(user.get("created_at")) await Actor.push_data({ "kind": "profile", "platform": "x", "found": True, "id": user.get("id_str"), "username": user.get("screen_name"), "full_name": user.get("name"), "biography": user.get("description"), "location": user.get("location"), "website": ((user.get("entities") or {}).get("url") or {}) .get("urls", [{}])[0].get("expanded_url") if (user.get("entities") or {}).get("url") else None, "followers": followers, "following": user.get("friends_count"), "tweets_count": user.get("statuses_count"), "listed_count": user.get("listed_count"), "likes_given": user.get("favourites_count"), "is_verified": bool(user.get("verified") or user.get("is_blue_verified")), "is_blue_verified": user.get("is_blue_verified"), "verified_type": user.get("verified_type"), "is_protected": user.get("protected"), "created_at": user.get("created_at"), "account_age_days": (round((time.time() - created_ts) / 86_400) if created_ts else None), "avatar": (user.get("profile_image_url_https") or "") .replace("_normal.", "_400x400."), "banner": user.get("profile_banner_url"), # agrégats sur les tweets récents (hors retweets) "avg_likes": avg_likes, "avg_retweets": avg_rts, "avg_replies": (round(sum(reps) / len(reps)) if reps else None), "avg_quotes": (round(sum(quotes) / len(quotes)) if quotes else None), "engagement_rate_pct": engagement, "tweets_per_week": tweets_per_week, "last_tweet_at": own[0]["created_at"] if own else None, "reply_share_pct": (round(sum(t["is_reply"] for t in own) / len(own) * 100) if own else None), "media_share_pct": (round(sum(bool(t["media_count"]) for t in own) / len(own) * 100) if own else None), "link_share_pct": (round(sum(bool(t["links"]) for t in own) / len(own) * 100) if own else None), "retweet_share_pct": (round(sum(t["is_retweet"] for t in tweets) / len(tweets) * 100) if tweets else None), "top_hashtags": sorted(hashtags, key=hashtags.get, reverse=True)[:8], # comptes les plus mentionnés → signal d'identité §12.1 "top_mentions": sorted(mentions, key=mentions.get, reverse=True)[:8], "top_tweet": ({"url": top["url"], "likes": top["likes"], "retweets": top["retweets"]} if top else None), "recent_tweets": tweets, }) await asyncio.gather(*[one(u) for u in usernames]) Actor.log.info(f"terminé : {len(usernames)} profils")