spb/qc-election
Public
Python 66.6%
HTML 24.8%
CSS 4.9%
JavaScript 3.6%
1# ==============================================================================2# Author: Simon-Pierre Boucher <contact@spboucher.ai>3# File: src/main.py (qc-social-pulse)4# Desc: Pouls social de l'élection québécoise 2026 — acteur maison5# multi-plateformes, sans connexion, proxy résidentiel CA :6# * instagram : comptes des partis/chef·fe·s (web_profile_info) →7# 12 derniers posts (légende, likes, commentaires) + profil;8# * facebook : pages publiques → abonnés (best-effort, FB mure);9# * youtube : RECHERCHE de vidéos récentes par parti (ytInitialData,10# videoRenderer + lockupViewModel) puis COMMENTAIRES PUBLICS via11# l'API interne youtubei (opinion citoyenne réelle);12# * tiktok : profils (__UNIVERSAL_DATA_FOR_REHYDRATION__) → stats;13# * x : best-effort syndication (souvent mort — found=False).14# Sortie : items {platform, party, kind: post|comment|profile|video,15# text, url, created_at, engagement, meta} consommés par qc-election.16# ==============================================================================17from __future__ import annotations1819import asyncio20import json21import re22import urllib.parse23from datetime import datetime, timezone2425from apify import Actor2627from .net import Fetcher2829IG_PROFILE = ("https://i.instagram.com/api/v1/users/web_profile_info/"30 "?username={u}")31IG_HDRS = {"x-ig-app-id": "936619743392459", "Accept": "application/json"}32FB_PAGE = "https://www.facebook.com/{u}"33TT_PAGE = "https://www.tiktok.com/@{u}"34YT_SEARCH = ("https://www.youtube.com/results?search_query={q}&sp=EgQIAxAB"35 "&hl=fr&gl=CA") # sp=EgQIAxAB : téléversées cette semaine36YT_NEXT = "https://www.youtube.com/youtubei/v1/next?prettyPrint=false"37YT_CTX = {"context": {"client": {"clientName": "WEB",38 "clientVersion": "2.20250101.00.00",39 "hl": "fr", "gl": "CA"}}}40X_SYND = ("https://cdn.syndication.twimg.com/timeline/profile"41 "?screen_name={u}&showReplies=false")4243_INITIAL_RE = re.compile(r"var ytInitialData\s*=\s*(\{.*?\});</script>", re.S)44_FB_FOLLOWERS = (re.compile(r'"follower_count"\s*:\s*(\d+)'),45 re.compile(r'"global_likers_count"\s*:\s*(\d+)'))46_TT_UNIVERSAL = re.compile(47 r'<script id="__UNIVERSAL_DATA_FOR_REHYDRATION__"[^>]*>(\{.*?\})</script>',48 re.S)495051def _iso(ts) -> str | None:52 try:53 return datetime.fromtimestamp(int(ts), tz=timezone.utc) \54 .strftime("%Y-%m-%dT%H:%M:%SZ")55 except (TypeError, ValueError, OSError):56 return None575859def _walk(node, key: str):60 """Tous les objets portant `key` dans un arbre JSON (générateur)."""61 if isinstance(node, dict):62 if key in node:63 yield node[key]64 for v in node.values():65 yield from _walk(v, key)66 elif isinstance(node, list):67 for v in node:68 yield from _walk(v, key)697071def _runs_text(node) -> str:72 if not isinstance(node, dict):73 return ""74 if "simpleText" in node:75 return str(node["simpleText"])76 return "".join(str(r.get("text", "")) for r in (node.get("runs") or []))777879# ---------------------------------------------------------------- instagram --80IG_PROFILE_ALT = ("https://www.instagram.com/api/v1/users/web_profile_info/"81 "?username={u}")828384async def do_instagram(f: Fetcher, task: dict) -> list[dict]:85 u = task["target"]86 resp = None87 # IG répond parfois 400 (challenge) sur une IP donnée : on tourne les88 # sessions proxy et on alterne les deux hôtes de l'endpoint web.89 for attempt in range(4):90 url = (IG_PROFILE if attempt % 2 == 0 else IG_PROFILE_ALT)91 resp = await f.get(url.format(u=urllib.parse.quote(u)),92 headers={**IG_HDRS,93 "Referer": f"https://www.instagram.com/{u}/"},94 session_id=f"ig{u}a{attempt}", retries=2)95 if resp.status_code == 200:96 break97 if resp.status_code == 404: # compte inexistant : inutile d'insister98 break99 await asyncio.sleep(1.5)100 if resp is None or resp.status_code != 200:101 return [{"platform": "instagram", "party": task["party"],102 "kind": "profile", "target": u, "found": False,103 "status": resp.status_code if resp is not None else None}]104 user = ((resp.json().get("data") or {}).get("user") or {})105 if not user:106 return [{"platform": "instagram", "party": task["party"],107 "kind": "profile", "target": u, "found": False}]108 items: list[dict] = []109 followers = (user.get("edge_followed_by") or {}).get("count")110 posts = [(e.get("node") or {}) for e in111 ((user.get("edge_owner_to_timeline_media") or {}).get("edges") or [])]112 for n in posts:113 cap = (n.get("edge_media_to_caption") or {}).get("edges") or []114 likes = (n.get("edge_liked_by") or {}).get("count") or 0115 comments = (n.get("edge_media_to_comment") or {}).get("count") or 0116 items.append({117 "platform": "instagram", "party": task["party"], "kind": "post",118 "target": u, "found": True,119 "text": ((cap[0].get("node") or {}).get("text", "") if cap else "")[:600],120 "url": f"https://www.instagram.com/p/{n.get('shortcode')}/",121 "created_at": _iso(n.get("taken_at_timestamp")),122 "engagement": likes + comments,123 "meta": {"likes": likes, "comments": comments,124 "video_views": n.get("video_view_count")}})125 items.append({126 "platform": "instagram", "party": task["party"], "kind": "profile",127 "target": u, "found": True,128 "url": f"https://www.instagram.com/{u}/",129 "engagement": followers or 0,130 "meta": {"followers": followers,131 "full_name": user.get("full_name"),132 "n_posts": (user.get("edge_owner_to_timeline_media") or {}).get("count")}})133 return items134135136# ----------------------------------------------------------------- facebook --137_FB_TEXT_COUNT = re.compile(138 r"([\d][\d\s .,]*)\s*([KkMm]?)\s*"139 r"(?:followers|abonnés|mentions\s+J[’']aime|J[’']aime|likes)", re.I)140_FB_OGDESC = re.compile(141 r'<meta (?:property="og:description"|name="description") content="([^"]*)"')142143144def _fb_text_count(text: str) -> int | None:145 m = _FB_TEXT_COUNT.search(text or "")146 if not m:147 return None148 num = re.sub(r"[\s ]", "", m.group(1)).replace(",", ".")149 try:150 val = float(num)151 except ValueError:152 return None153 unit = (m.group(2) or "").lower()154 return int(val * (1_000 if unit == "k" else 1_000_000 if unit == "m" else 1))155156157async def do_facebook(f: Fetcher, task: dict) -> list[dict]:158 u = task["target"]159 followers, status = None, None160 for url in (FB_PAGE.format(u=u), f"https://mbasic.facebook.com/{u}"):161 resp = await f.get(url, session_id=f"fb{u}", retries=3)162 status = resp.status_code163 txt = resp.text or ""164 for rx in _FB_FOLLOWERS:165 m = rx.search(txt)166 if m:167 followers = int(m.group(1))168 break169 if followers is None: # texte « N abonnés » de l'og:description170 og = _FB_OGDESC.search(txt)171 if og:172 followers = _fb_text_count(og.group(1))173 if followers is not None:174 break175 return [{"platform": "facebook", "party": task["party"], "kind": "profile",176 "target": u, "found": followers is not None,177 "url": FB_PAGE.format(u=u), "engagement": followers or 0,178 "meta": {"followers": followers, "status": status}}]179180181# ------------------------------------------------------------------- tiktok --182async def do_tiktok(f: Fetcher, task: dict) -> list[dict]:183 u = task["target"]184 resp = await f.get(TT_PAGE.format(u=urllib.parse.quote(u)),185 session_id=f"tt{u}", retries=3)186 txt = resp.text or ""187 m = _TT_UNIVERSAL.search(txt)188 stats, user = {}, {}189 if m:190 try:191 data = json.loads(m.group(1))192 ui = (((data.get("__DEFAULT_SCOPE__") or {})193 .get("webapp.user-detail") or {}).get("userInfo") or {})194 stats, user = ui.get("stats") or {}, ui.get("user") or {}195 except (json.JSONDecodeError, AttributeError):196 pass197 if not user: # repli : ancien état SIGI_STATE198 m2 = re.search(r'<script id="SIGI_STATE"[^>]*>(\{.*?\})</script>',199 txt, re.S)200 if m2:201 try:202 sigi = json.loads(m2.group(1))203 um = (sigi.get("UserModule") or {})204 user = next(iter((um.get("users") or {}).values()), {})205 stats = next(iter((um.get("stats") or {}).values()), {})206 except (json.JSONDecodeError, AttributeError):207 pass208 if not user:209 return [{"platform": "tiktok", "party": task["party"],210 "kind": "profile", "target": u, "found": False,211 "status": resp.status_code}]212 return [{"platform": "tiktok", "party": task["party"], "kind": "profile",213 "target": u, "found": bool(user),214 "url": TT_PAGE.format(u=u),215 "engagement": stats.get("followerCount") or 0,216 "meta": {"followers": stats.get("followerCount"),217 "hearts": stats.get("heartCount"),218 "videos": stats.get("videoCount"),219 "verified": user.get("verified")}}]220221222# ------------------------------------------------------------------ youtube --223def _yt_videos(data: dict, cap: int) -> list[dict]:224 """videoRenderer (classique) + lockupViewModel (moderne) → vidéos."""225 vids: list[dict] = []226 for vr in _walk(data, "videoRenderer"):227 vid = vr.get("videoId")228 if not vid:229 continue230 vids.append({"videoId": vid,231 "title": _runs_text(vr.get("title") or {})[:200],232 "published": _runs_text(vr.get("publishedTimeText") or {}),233 "views": _runs_text(vr.get("viewCountText") or {}),234 "channel": _runs_text(vr.get("ownerText") or {})})235 if len(vids) >= cap:236 return vids237 for lv in _walk(data, "lockupViewModel"):238 vid = ((lv.get("rendererContext") or {}).get("commandContext") or {})239 vid = (((vid.get("onTap") or {}).get("innertubeCommand") or {})240 .get("watchEndpoint") or {}).get("videoId") \241 or lv.get("contentId")242 if not vid or any(v["videoId"] == vid for v in vids):243 continue244 md = (lv.get("metadata") or {}).get("lockupMetadataViewModel") or {}245 title = ((md.get("title") or {}).get("content") or "")[:200]246 vids.append({"videoId": vid, "title": title, "published": "",247 "views": "", "channel": ""})248 if len(vids) >= cap:249 break250 return vids251252253def _yt_comments(data: dict, cap: int) -> list[dict]:254 """Réponse youtubei/next (continuation) → commentaires publics.255 Format moderne : frameworkUpdates → commentEntityPayload; repli sur256 l'ancien commentRenderer."""257 out: list[dict] = []258 for payload in _walk(data, "commentEntityPayload"):259 props = payload.get("properties") or {}260 content = ((props.get("content") or {}).get("content") or "").strip()261 toolbar = payload.get("toolbar") or {}262 likes_txt = str(toolbar.get("likeCountNotliked") or263 toolbar.get("likeCountLiked") or "0").strip()264 m = re.search(r"[\d,.\s]+", likes_txt.replace(" ", " "))265 likes = 0266 if m:267 digits = re.sub(r"[^\d]", "", m.group())268 likes = int(digits) if digits else 0269 if "k" in likes_txt.lower():270 likes *= 1000271 if content:272 out.append({"text": content[:600], "likes": likes,273 "published": str(props.get("publishedTime") or "")})274 if len(out) >= cap:275 return out276 if not out: # ancien format277 for cr in _walk(data, "commentRenderer"):278 content = _runs_text(cr.get("contentText") or {}).strip()279 if content:280 out.append({"text": content[:600],281 "likes": int(cr.get("likeCount") or 0),282 "published": _runs_text(283 cr.get("publishedTimeText") or {})})284 if len(out) >= cap:285 break286 return out287288289async def do_youtube(f: Fetcher, task: dict, max_comments: int) -> list[dict]:290 q = task.get("query") or task.get("target") or ""291 limit = int(task.get("limit") or 3)292 resp = await f.get(YT_SEARCH.format(q=urllib.parse.quote(q)),293 session_id=f"yt{abs(hash(q)) % 99999}")294 m = _INITIAL_RE.search(resp.text or "")295 if not m:296 return [{"platform": "youtube", "party": task["party"],297 "kind": "video", "found": False, "query": q,298 "status": resp.status_code}]299 try:300 data = json.loads(m.group(1))301 except json.JSONDecodeError:302 return [{"platform": "youtube", "party": task["party"],303 "kind": "video", "found": False, "query": q}]304 items: list[dict] = []305 for v in _yt_videos(data, limit):306 items.append({"platform": "youtube", "party": task["party"],307 "kind": "video", "found": True, "query": q,308 "text": v["title"],309 "url": f"https://www.youtube.com/watch?v={v['videoId']}",310 "created_at": None,311 "engagement": 0,312 "meta": {"published": v["published"], "views": v["views"],313 "channel": v["channel"]}})314 if max_comments <= 0:315 continue316 # 1er appel next → jeton de continuation de la section commentaires317 body = dict(YT_CTX); body["videoId"] = v["videoId"]318 r1 = await f.post(YT_NEXT, data=json.dumps(body),319 headers={"Content-Type": "application/json"},320 session_id=f"ytc{v['videoId']}")321 token = None322 try:323 d1 = r1.json()324 for isr in _walk(d1, "itemSectionRenderer"):325 if isr.get("sectionIdentifier") == "comment-item-section":326 for cont in _walk(isr, "continuationCommand"):327 token = cont.get("token")328 break329 if token:330 break331 if not token: # certains rendus placent le jeton ailleurs332 for cont in _walk(d1, "continuationCommand"):333 tok = cont.get("token") or ""334 if tok.startswith("Eg"):335 token = tok336 break337 except (json.JSONDecodeError, AttributeError):338 token = None339 if not token:340 continue341 body2 = dict(YT_CTX); body2["continuation"] = token342 r2 = await f.post(YT_NEXT, data=json.dumps(body2),343 headers={"Content-Type": "application/json"},344 session_id=f"ytc{v['videoId']}")345 try:346 comments = _yt_comments(r2.json(), max_comments)347 except (json.JSONDecodeError, AttributeError):348 comments = []349 for c in comments:350 items.append({"platform": "youtube", "party": task["party"],351 "kind": "comment", "found": True, "query": q,352 "text": c["text"],353 "url": f"https://www.youtube.com/watch?v={v['videoId']}",354 "created_at": None,355 "engagement": c["likes"],356 "meta": {"video_title": v["title"],357 "published": c["published"]}})358 return items359360361# ------------------------------------------------------------------------ x --362async def do_x(f: Fetcher, task: dict) -> list[dict]:363 u = task["target"]364 resp = await f.get(X_SYND.format(u=urllib.parse.quote(u)),365 session_id=f"x{u}", retries=2)366 ok = resp.status_code == 200 and "timeline" in (resp.text or "")[:2000].lower()367 return [{"platform": "x", "party": task["party"], "kind": "profile",368 "target": u, "found": ok, "status": resp.status_code}]369370371HANDLERS = {"instagram": do_instagram, "facebook": do_facebook,372 "tiktok": do_tiktok, "x": do_x}373374375async def main() -> None:376 async with Actor:377 inp = await Actor.get_input() or {}378 tasks = inp.get("tasks") or []379 max_comments = int(inp.get("maxCommentsPerVideo") or 20)380 country = (inp.get("proxyCountry") or "CA").upper()381 proxy_conf = await Actor.create_proxy_configuration(382 groups=["RESIDENTIAL"], country_code=country)383 f = Fetcher(proxy_conf, delay=1.2)384 pushed = 0385 for task in tasks:386 platform = (task.get("platform") or "").lower().strip()387 try:388 if platform == "youtube":389 items = await do_youtube(f, task, max_comments)390 elif platform in HANDLERS:391 items = await HANDLERS[platform](f, task)392 else:393 items = [{"platform": platform, "found": False,394 "error": "plateforme inconnue"}]395 except Exception as exc: # une tâche ne tue jamais le run396 Actor.log.exception(f"tâche {task}: {exc}")397 items = [{"platform": platform,398 "party": task.get("party"), "found": False,399 "error": str(exc)[:200]}]400 fetched = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")401 for it in items:402 it["fetched_at"] = fetched403 await Actor.push_data(items)404 pushed += len(items)405 Actor.log.info(f"terminé : {pushed} items pour {len(tasks)} tâches")406