climperor/web/fetch_streamer_live.py

"""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.
- Douyu: public ``www.douyu.com/betard/{room_id}``; ``room.show_status == 1``
  means live, ``== 2`` offline; ``room.videoLoop == 1`` (carousel) counts as
  offline.
- 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 clear
``is_live`` to False and drop ``live_probed_at`` (so consumers treat the
badge as stale/unknown) 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. Production live badges are owned by the
visit-triggered ``/api/live-status`` edge probe; this daily write is only a
``data.json`` fallback until that API returns.

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
DOUYU_SPACING = 0.3

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}"
DOUYU_BETARD_URL = "https://www.douyu.com/betard/{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 probe_douyu(room_id: str) -> bool:
    """show_status: 1 live, 2 offline; videoLoop carousel counts as offline."""
    payload = http_utils.http_json(
        DOUYU_BETARD_URL.format(room_id=room_id), timeout=TIMEOUT
    )
    if not isinstance(payload, dict):
        raise ValueError("douyu betard returned non-object")
    room = payload.get("room")
    if not isinstance(room, dict):
        raise ValueError("douyu betard returned no room")
    if int(room.get("videoLoop") or 0) == 1:
        return False
    status = room.get("show_status")
    try:
        return int(status) == 1
    except (TypeError, ValueError) as e:
        raise ValueError(f"douyu unexpected show_status {status!r}") from e


def room_ref_from_url(live_url: str) -> str | None:
    """First path segment of the live room URL (douyin/bilibili/douyu 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"
    if "douyu.com" in host:
        return "douyu"
    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)
            elif platform == "douyu":
                is_live = probe_douyu(ref)
                time.sleep(DOUYU_SPACING)
            else:
                print(f"skip {sid}: platform={platform!r} unsupported", flush=True)
                continue
        except (urllib.error.URLError, TimeoutError, OSError, ValueError) as e:
            # Align with /api/live-status and local serve_relations: unknown is
            # not live, and must not preserve a stale positive badge.
            row["is_live"] = False
            row.pop("live_probed_at", None)
            print(f"  FAIL {sid}: {e} (is_live=false, stale)", 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())