"""Probe real live status for streamers into data/streamers.json. For every streamer row with ``live_url`` this script writes: - ``is_live`` — True only when the platform confirms the room is live - ``live_probed_at`` — UTC ISO timestamp of the successful probe Probe backend follows the ``live_url`` host (may differ from profile ``platform``, e.g. Douyin profile + Bilibili room). Approach (verified 2026-07): - Bilibili: public API ``api.live.bilibili.com/room/v1/Room/get_info`` with the numeric room id taken from the ``live_url`` path; ``data.live_status == 1`` means live (0 offline, 2 replay — replay is treated as offline). No login. - Douyin: one shared cookie session is warmed up (www.douyin.com + live.douyin.com), then per room we GET ``live.douyin.com/{web_rid}`` with a browser UA (the numeric ``live_url`` path segment is the ``web_rid``). The SSR page embeds ``roomStore.roomInfo.room.status`` inside the streaming pace chunks as escaped JSON: ``status == 2`` means live, ``status == 4`` offline; the embedded ``web_rid`` must match the requested one. (The ``webcast/room/web/enter`` API was considered but returns empty bodies without request signing, so the SSR page is the source of truth.) Everything is soft-fail: network errors, empty or non-JSON responses keep the previous ``is_live`` value and never abort a refresh tier (exit code is always 0). Only the two probe fields are touched; all other keys (including ``live_url``) are preserved. Note the ``daily`` tier means the badge trails reality by up to a day — truly real-time would need a higher-frequency job. Preview only — do not merge into relations/heroes or recommend. Usage: python fetch_streamer_live.py # probe all rows with live_url python fetch_streamer_live.py --ids shawang,xiaowang python fetch_streamer_live.py --dry-run # probe + print, do not write """ from __future__ import annotations import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1])) import argparse import http.cookiejar import json import re import time import urllib.error import urllib.parse import urllib.request from datetime import datetime, timezone from shared import http_utils from shared.paths import DATA from fetch_streamers import BROWSER_UA OUT = DATA / "streamers.json" TIMEOUT = 20 # Douyin rate-limits aggressively; keep ~1s spacing between its requests. DOUYIN_SPACING = 1.0 BILIBILI_SPACING = 0.5 DOUYIN_HOME = "https://www.douyin.com/" DOUYIN_LIVE_HOME = "https://live.douyin.com/" BILIBILI_INFO_URL = "https://api.live.bilibili.com/room/v1/Room/get_info?room_id={room_id}" # Escaped JSON inside the SSR pace chunks: \"roomStore\":{\"roomInfo\":{\"room\":{ DOUYIN_ROOMSTORE_RE = re.compile( r'\\"roomStore\\":\s*\{\\"roomInfo\\":\s*\{\\"room\\":\s*\{' ) DOUYIN_STATUS_RE = re.compile(r'\\"status\\":\s*(\d)') DOUYIN_WEBRID_RE = re.compile(r'\\"web_rid\\":\s*\\"(\d+)\\"') def _now_iso() -> str: return datetime.now(timezone.utc).isoformat() def _douyin_opener() -> urllib.request.OpenerDirector: jar = http.cookiejar.CookieJar() return urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar)) def _douyin_get( opener: urllib.request.OpenerDirector, url: str, *, referer: str ) -> bytes: req = urllib.request.Request( url, headers={ "User-Agent": BROWSER_UA, "Accept": "*/*", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", "Referer": referer, }, ) with opener.open(req, timeout=TIMEOUT) as resp: return resp.read() def warm_douyin(opener: urllib.request.OpenerDirector) -> None: """Seed cookies once so subsequent webcast calls are not blocked.""" for url in (DOUYIN_HOME, DOUYIN_LIVE_HOME): try: _douyin_get(opener, url, referer=DOUYIN_HOME) except (urllib.error.URLError, TimeoutError, OSError) as e: print(f" douyin warm-up {url} failed: {e}", flush=True) time.sleep(DOUYIN_SPACING) def probe_douyin(opener: urllib.request.OpenerDirector, rid: str) -> bool: """Parse roomStore status from the SSR live room page (2 live / 4 offline).""" page_url = f"{DOUYIN_LIVE_HOME}{rid}" raw = _douyin_get(opener, page_url, referer=DOUYIN_LIVE_HOME) if not raw: raise ValueError("empty room page") html = raw.decode("utf-8", "replace") store = DOUYIN_ROOMSTORE_RE.search(html) if not store: raise ValueError("no roomStore in page (blocked or layout changed)") # The room object opens with id_str/status; a short window is enough. window = html[store.end() : store.end() + 3000] status_m = DOUYIN_STATUS_RE.search(window) if not status_m: raise ValueError("roomStore has no status field") embedded = DOUYIN_WEBRID_RE.search(html) if not embedded or embedded.group(1) != rid: raise ValueError("page resolved to a different room (stale web_rid?)") status = int(status_m.group(1)) if status == 2: return True if status == 4: return False raise ValueError(f"unexpected room status {status}") def probe_bilibili(room_id: str) -> bool: """live_status: 0 offline, 1 live, 2 replay (replay counts as offline).""" payload = http_utils.http_json( BILIBILI_INFO_URL.format(room_id=room_id), timeout=TIMEOUT ) if not isinstance(payload, dict) or payload.get("code") != 0: raise ValueError(f"bilibili api error: code={payload.get('code')!r}") data = payload.get("data") if not isinstance(data, dict): raise ValueError("bilibili api returned no data") return data.get("live_status") == 1 def room_ref_from_url(live_url: str) -> str | None: """First path segment of the live room URL (douyin web_rid / bilibili room id).""" path = urllib.parse.urlparse(live_url.strip()).path.strip("/") if not path: return None return path.split("/")[0] or None def live_platform_from_url(live_url: str, fallback: str = "") -> str: """Probe backend follows the live room host (may differ from profile platform).""" host = urllib.parse.urlparse(live_url.strip()).netloc.lower() if "bilibili.com" in host: return "bilibili" if "douyin.com" in host: return "douyin" return (fallback or "").strip().lower() def probe_streamers( payload: dict, *, ids: set[str] | None = None ) -> tuple[int, int, int]: """Probe rows with live_url in place. Returns (live, offline, fail).""" rows = payload.get("streamers") if not isinstance(rows, list): raise SystemExit("streamers.json: missing streamers array") targets = [] for row in rows: if not isinstance(row, dict): continue sid = str(row.get("id") or "").strip() if not sid or (ids is not None and sid not in ids): continue live_url = str(row.get("live_url") or "").strip() if not live_url: continue fallback = str(row.get("platform") or "").strip().lower() platform = live_platform_from_url(live_url, fallback) ref = room_ref_from_url(live_url) if not ref: print(f"skip {sid}: cannot parse room ref from {live_url!r}", flush=True) continue targets.append((row, sid, platform, ref)) opener = None if any(platform == "douyin" for _, _, platform, _ in targets): opener = _douyin_opener() warm_douyin(opener) live = offline = fail = 0 for row, sid, platform, ref in targets: try: if platform == "douyin": assert opener is not None is_live = probe_douyin(opener, ref) time.sleep(DOUYIN_SPACING) elif platform == "bilibili": is_live = probe_bilibili(ref) time.sleep(BILIBILI_SPACING) else: print(f"skip {sid}: platform={platform!r} unsupported", flush=True) continue except (urllib.error.URLError, TimeoutError, OSError, ValueError) as e: print(f" FAIL {sid}: {e} (keeping previous is_live)", flush=True) fail += 1 continue row["is_live"] = is_live row["live_probed_at"] = _now_iso() state = "LIVE" if is_live else "offline" print(f" {state} {sid} ({platform} {ref})", flush=True) if is_live: live += 1 else: offline += 1 return live, offline, fail def main() -> int: ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) ap.add_argument("--out", type=Path, default=OUT) ap.add_argument( "--ids", default=None, help="comma-separated streamer ids to probe (default: all with live_url)", ) ap.add_argument( "--dry-run", action="store_true", help="probe and print only; do not write streamers.json", ) args = ap.parse_args() raw = args.out.read_text(encoding="utf-8") if args.out.is_file() else None if raw is None: raise SystemExit(f"{args.out} not found") payload = json.loads(raw) if not isinstance(payload, dict): raise SystemExit(f"{args.out}: expected object") id_set = None if args.ids: id_set = {s.strip() for s in args.ids.split(",") if s.strip()} live, offline, fail = probe_streamers(payload, ids=id_set) summary = f"live={live} offline={offline} fail={fail}" if args.dry_run: print(f"dry-run: would write {args.out} ({summary})", flush=True) return 0 http_utils.write_json_atomic(args.out, payload) print(f"wrote {args.out} {summary}", flush=True) # Soft-fail for CI/refresh_web: always exit 0 after writing. return 0 if __name__ == "__main__": raise SystemExit(main())