brfc-stream-pirate/record_live.py

#!/usr/bin/env python3
"""Unattended recorder: find the live stream, then record it.

Why this needs a headless browser at all: auth and the token mint are plain
HTTP (see discover.py), but the *entry id* for a given match only appears in
the page at runtime -- the static HTML carries no id when nothing is live. So
we render the page headlessly just long enough to learn the manifest URL, then
hand it to ffmpeg and close the browser.

Nothing here needs your password. It rides the Firefox session cookies, which
outlive kickoff by months.

    # start now, keep looking for up to 45 min, then record 150 min
    python record_live.py --max-wait 45 --duration 150

    # take both feeds at once, or pin one when you know which you hold
    python record_live.py --feed both
    python record_live.py --entry 0_xm5vrrw6

Two different choices, often confused. **--feed** picks which *entry* to record
when the page exposes more than one for the same match, typically an audio feed
and a video feed. **--prefer** picks a rendition *inside* one entry's master.
Getting --feed wrong cannot be rescued by --prefer, and it fails quietly: you
get a complete, playable, perfectly healthy recording of the wrong thing.

    # or wait until a wall-clock time first (03:15 before an 03:45 kickoff)
    python record_live.py --at "2026-09-02 03:15" --max-wait 60

Recording stops on the duration cap. That is deliberate and load-bearing: the
feed does not emit #EXT-X-ENDLIST -- after full time it switches to a static
slate and streams forever.
"""
import argparse
import contextlib
import io
import os
import re
import subprocess
import sys
import time
from datetime import datetime

ROOT = os.path.dirname(os.path.abspath(__file__))
RECORDINGS = os.path.join(ROOT, "recordings")
LOGS = os.path.join(ROOT, "logs")
LIVE_URL = "https://www.bristolrovers.co.uk/video/live"
UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:131.0) "
      "Gecko/20100101 Firefox/131.0")
COOKIE_DOMAINS = ("bristolrovers", "streamamg", "kaltura", "akamaized")


def log(msg):
    line = f"[{datetime.now():%Y-%m-%d %H:%M:%S}] {msg}"
    print(line, flush=True)
    os.makedirs(LOGS, exist_ok=True)
    with open(os.path.join(LOGS, "record_live.log"), "a", encoding="utf-8") as fh:
        fh.write(line + chr(10))


def firefox_cookies():
    from yt_dlp.cookies import extract_cookies_from_browser
    with contextlib.redirect_stderr(io.StringIO()):
        jar = extract_cookies_from_browser("firefox")
    out = []
    for c in jar:
        dom = c.domain or ""
        if not any(k in dom.lower() for k in COOKIE_DOMAINS):
            continue
        out.append({"name": c.name, "value": c.value or "", "domain": dom,
                    "path": c.path or "/",
                    "expires": float(c.expires) if c.expires else -1,
                    "httpOnly": False, "secure": bool(c.secure),
                    "sameSite": "Lax"})
    return out


TOKEN_URL = ("https://www.bristolrovers.co.uk/oauth/browser-token"
             "?client_id=streamamg&token_type=urn%3Aietf%3Aparams%3Aoauth%3Atoken-type%3Aid_token")
# Public client key, shipped to every browser -- not a secret.
API_KEY = "50ivbQ3IXpF2mwT4tPAJ2V6cVm4CGRT2YKmDK5v6"


def mint_master(entry_id, quiet=False):
    """Mint a fresh master playlist URL in plain Python. No browser.

    Returns (master_url, rendition_kinds). The master token lives ~25 seconds,
    so mint immediately before you use the URL -- classify with one call, then
    mint again for the recording.

    Handing yt-dlp the *media* playlist (hdntl, path token) fails: Python
    clients get 403 on that URL form, and yt-dlp fetches the playlist with
    urllib before ffmpeg ever runs. The *master* (hdnts, query token) is fetched
    fine by Python, so we mint one here and pass that instead. It lives only
    ~25 seconds, which is why it is minted immediately before recording.
    """
    import http.cookiejar, json, re, urllib.request
    from yt_dlp.cookies import extract_cookies_from_browser

    with contextlib.redirect_stderr(io.StringIO()):
        jar = extract_cookies_from_browser("firefox")
    cj = http.cookiejar.CookieJar()
    for c in jar:
        if "bristolrovers" in (c.domain or "").lower():
            cj.set_cookie(c)
    op = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cj))
    hdrs = {"User-Agent": UA, "Accept": "application/json, text/plain, */*"}

    body = op.open(urllib.request.Request(TOKEN_URL, headers=hdrs),
                   timeout=20).read().decode("utf-8", "replace")
    m = re.search(r"(eyJ[\w-]+\.eyJ[\w-]+\.[\w-]+)", body)
    if not m:
        log("  mint: no JWT in browser-token response (session expired?)")
        return None, []
    jwt = m.group(1)

    req = urllib.request.Request(
        f"https://api.playback.streamamg.com/v1/entry/{entry_id}",
        headers={**hdrs, "Authorization": "Bearer " + jwt, "x-api-key": API_KEY,
                 "Origin": "https://www.bristolrovers.co.uk",
                 "Referer": "https://www.bristolrovers.co.uk/"})
    data = json.loads(urllib.request.urlopen(req, timeout=20)
                      .read().decode("utf-8", "replace"))
    hls = (data.get("media") or {}).get("hls") or None
    kinds = log_renditions(hls, quiet=quiet) if hls else []
    return hls, kinds


def log_renditions(master_url, quiet=False):
    """Log what the master offers, and return the kind of each rendition.

    The return value is what feed_kind() classifies an entry on. Codecs are the
    only trustworthy signal: entry ids look alike, and the same id comes back
    week after week for the same feed.
    """
    import urllib.request
    try:
        req = urllib.request.Request(master_url, headers={"User-Agent": UA,
              "Referer": "https://www.bristolrovers.co.uk/"})
        body = urllib.request.urlopen(req, timeout=15).read().decode("utf-8", "replace")
    except Exception as exc:
        log(f"  (could not read master for rendition list: {exc})")
        return []
    PREFIX = "#EXT-X-STREAM-INF:"
    streams = [ln[len(PREFIX):] for ln in body.splitlines()
               if ln.startswith(PREFIX)]
    kinds = []
    if not quiet:
        log(f"  master offers {len(streams)} rendition(s):")
    for st in streams:
        bw = re.search(r"BANDWIDTH=(\d+)", st)
        res = re.search(r"RESOLUTION=([\dx]+)", st)
        cod = re.search(r'CODECS="([^"]+)"', st)
        codecs = cod.group(1) if cod else "?"
        kind = "audio-only" if ("avc" not in codecs and "hvc" not in codecs) else "video"
        kinds.append(kind)
        if not quiet:
            log(f"    {kind:11s} {int(bw.group(1))//1000 if bw else '?':>6} kbps  "
                f"{res.group(1) if res else '-':>10}  {codecs}")
    return kinds


def feed_kind(kinds):
    """'video' if the master carries any video rendition, else 'audio'."""
    if not kinds:
        return "unknown"
    return "video" if any(k == "video" for k in kinds) else "audio"


def find_manifest(timeout_s=90, settle_s=15):
    """Load the live page headlessly and return every entry id it exposes.

    The browser is needed only for this: the entry id appears at runtime, and
    the static HTML carries none when nothing is live. Everything after this
    is plain HTTP.

    Returns a list, not one id, because the page can expose more than one feed
    for the same match -- an audio entry and a video entry. Which manifest
    loads first is a race, and it has been observed to come out differently on
    the same night. Returning the first one silently records the wrong feed.
    So keep watching for `settle_s` after the first hit, and let the caller
    choose.
    """
    from playwright.sync_api import sync_playwright

    entries = []                # discovery order, de-duplicated
    first_seen = [None]

    with sync_playwright() as p:
        browser = p.firefox.launch(headless=True)
        ctx = browser.new_context(user_agent=UA,
                                  viewport={"width": 1440, "height": 900})
        ctx.add_cookies(firefox_cookies())
        page = ctx.new_page()

        def on_response(resp):
            u = resp.url
            if ".m3u8" not in u.split("?")[0].lower():
                return
            if "hdntl=" not in u and "hdnts=" not in u:
                return
            m = re.search(r"/(\d+_[a-z0-9]+)/", u, re.I)
            if not m or m.group(1) in entries:
                return
            entries.append(m.group(1))
            if first_seen[0] is None:
                first_seen[0] = time.time()

        page.on("response", on_response)
        try:
            page.goto(LIVE_URL, wait_until="domcontentloaded", timeout=60000)
        except Exception as exc:
            log(f"  page load failed: {exc}")
            browser.close()
            return None

        # The manifest is fetched on load, before any click. We still click,
        # because a stream that has only just gone live may need the nudge.
        deadline = time.time() + timeout_s
        clicked = False
        while time.time() < deadline:
            # Stop once the page has settled: a feed the player never requests
            # will not appear no matter how long we wait.
            if first_seen[0] and time.time() - first_seen[0] > settle_s:
                break
            if not clicked:
                for sel in (".bmpui-ui-hugeplaybacktogglebutton",
                            ".bmpui-ui-playbacktogglebutton",
                            "button[aria-label*='play' i]"):
                    try:
                        el = page.query_selector(sel)
                        if el and el.is_visible():
                            el.click()
                            clicked = True
                            break
                    except Exception:
                        pass
                with contextlib.suppress(Exception):
                    page.evaluate("() => { const v=document.querySelector('video');"
                                  " if (v) { v.muted = true; v.play(); } }")
            page.wait_for_timeout(2000)

        browser.close()

    return entries


def record(url, duration_min, prefer='best', tag=None):
    stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
    name = f"brfc-{stamp}-{tag}.ts" if tag else f"brfc-{stamp}.ts"
    out = os.path.join(RECORDINGS, name)
    os.makedirs(RECORDINGS, exist_ok=True)
    logfile = os.path.join(LOGS, f"yt-dlp-{stamp}.log")
    cmd = [
        sys.executable, "-m", "yt_dlp",
        # Force ffmpeg. yt-dlp's native HLS downloader fetches variant
        # playlists with Python and gets 403/404 on the path-token URLs;
        # ffmpeg fetches them fine. This is not a preference, it is required.
        "--downloader", "ffmpeg",
        "--hls-use-mpegts",        # a killed process still leaves a playable file
        "--no-part",
        "--cookies-from-browser", "firefox",
        "--user-agent", UA,
        "--referer", "https://www.bristolrovers.co.uk/",
        "--downloader-args",
        # ffmpeg 4.3.1 has no -reconnect_on_network_error.
        "ffmpeg_i:-reconnect 1 -reconnect_streamed 1 "
        "-reconnect_delay_max 30",
        "--downloader-args", f"ffmpeg_o:-t {duration_min * 60}",
        "-o", out,
    ]
    if prefer == "audio":
        # Pin audio when the entitlement is audio-only: if a video rendition
        # ever appears in the master, taking it would 403 on the segments.
        cmd += ["-f", "bestaudio/best"]
    elif prefer == "video":
        cmd += ["-f", "bestvideo*+bestaudio/best"]
    cmd.append(url)
    log(f"recording -> {os.path.basename(out)}  (cap {duration_min} min)")
    log(f"  yt-dlp log -> {os.path.basename(logfile)}")
    with open(logfile, "w", encoding="utf-8") as fh:
        return subprocess.Popen(cmd, stdout=fh, stderr=subprocess.STDOUT, cwd=ROOT)


def classify_entries(entries):
    """Label each discovered entry video or audio, by minting its master once.

    Returns [(entry_id, kind)] in discovery order, skipping anything that will
    not mint. The master minted here is thrown away -- its token lives ~25
    seconds, so the recording mints its own.
    """
    out = []
    for entry in entries:
        try:
            url, kinds = mint_master(entry)
        except Exception as exc:
            log(f"  {entry}: mint failed: {type(exc).__name__}: {exc}")
            continue
        if not url:
            continue
        kind = feed_kind(kinds)
        log(f"  {entry} is a {kind} feed")
        out.append((entry, kind))
    return out


def choose_feeds(chosen, want):
    """Order classified entries by preference. 'both' and 'any' take them all.

    A preference is not a filter: the non-preferred feeds stay on the list as
    fallbacks, because a feed you hold no pass for still resolves a master and
    only fails later, on the segments.
    """
    if want in ("both", "any"):
        return chosen
    first = [c for c in chosen if c[1] == want]
    rest = [c for c in chosen if c[1] != want]
    if not first and rest:
        log(f"  no {want} feed on offer -- falling back to "
            f"{', '.join(k for _, k in rest)}")
    return first + rest


def run_recorders(targets, args):
    """Record, and report whether a recording actually ran.

    'both' starts every feed at once: concurrency is not limited and a second
    copy is cheap insurance. Otherwise take the first choice and fall through
    to the next only when the downloader dies immediately -- which is what an
    unentitled feed looks like, the master resolving and the segments 403ing.
    """
    def start(entry, kind):
        try:
            url, _ = mint_master(entry, quiet=True)
        except Exception as exc:
            log(f"  {entry}: mint failed: {type(exc).__name__}: {exc}")
            return None
        if not url:
            log(f"  {entry}: could not re-mint a master; skipping")
            return None
        proc = record(url, args.duration, args.prefer, tag=kind)
        log(f"  {entry} ({kind}): downloader pid {proc.pid}")
        return proc, time.time()

    def finish(entry, kind, proc, t0):
        proc.wait()
        ran = time.time() - t0
        log(f"  {entry} ({kind}): rc={proc.returncode} after {ran:.0f}s")
        # A master outlives its stream, so minting one proves nothing about
        # whether the match is up. Only a download that actually ran does.
        if proc.returncode == 0 or ran > 120:
            return True
        log("  died immediately -- not live yet, or no pass for this feed")
        return False

    if args.feed == "both":
        started = []
        for entry, kind in targets:
            s = start(entry, kind)
            if s is not None:
                started.append((entry, kind) + s)
        # Every one must be waited on, so collect results before reducing --
        # any() would short-circuit and orphan a running downloader.
        results = [finish(*row) for row in started]
        return any(results)

    # One feed at a time, falling through on immediate death so an unentitled
    # first choice does not cost the match.
    for entry, kind in targets:
        s = start(entry, kind)
        if s is None:
            continue
        if finish(entry, kind, *s):
            return True
    return False


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--at", help='wall-clock start, e.g. "2026-09-02 03:15"')
    ap.add_argument("--max-wait", type=int, default=45,
                    help="minutes to keep looking for a live stream")
    ap.add_argument("--duration", type=int, default=150,
                    help="recording cap in minutes")
    ap.add_argument("--poll", type=int, default=60, help="seconds between attempts")
    ap.add_argument("--prefer", choices=("best", "audio", "video"), default="best",
                    help="which rendition to take when the master offers both; "
                         "'best' is yt-dlp's default and prefers video")
    ap.add_argument("--entry", help="skip discovery and record this entry id, "
                    "bypassing --feed entirely")
    ap.add_argument("--feed", choices=("video", "audio", "any", "both"),
                    default="video",
                    help="which FEED to record when the page exposes more than "
                         "one entry. 'video' prefers a video feed and falls "
                         "back to audio; 'both' records every feed in parallel. "
                         "Distinct from --prefer, which picks a rendition "
                         "inside one feed and cannot rescue a wrong feed.")
    args = ap.parse_args()

    if args.at:
        target = datetime.strptime(args.at, "%Y-%m-%d %H:%M")
        wait = (target - datetime.now()).total_seconds()
        if wait > 0:
            log(f"sleeping {wait/3600:.2f}h until {target:%Y-%m-%d %H:%M}")
            time.sleep(wait)

    deadline = time.time() + args.max_wait * 60
    attempt = 0
    while time.time() < deadline:
        attempt += 1
        if args.entry:
            entries = [args.entry]
            log(f"attempt {attempt}: pinned entry {args.entry}")
        else:
            log(f"attempt {attempt}: looking for a live stream")
            try:
                entries = find_manifest()
            except Exception as exc:
                log(f"  error: {type(exc).__name__}: {exc}")
                entries = []

        if not entries:
            log(f"  nothing live yet; retrying in {args.poll}s")
            time.sleep(args.poll)
            continue

        log(f"  live entries: {', '.join(entries)}")
        chosen = classify_entries(entries)
        if not chosen:
            log("  could not mint a manifest for any entry; retrying")
            time.sleep(args.poll)
            continue

        targets = choose_feeds(chosen, args.feed)
        if not targets:
            log(f"  no {args.feed} feed among {[k for _, k in chosen]}; retrying")
            time.sleep(args.poll)
            continue

        if run_recorders(targets, args) or time.time() >= deadline:
            return 0
        time.sleep(args.poll)

    log("gave up: no live stream appeared within the wait window")
    return 1


if __name__ == "__main__":
    sys.exit(main())