#!/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())