#!/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 # 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): """Mint a fresh master playlist URL in plain Python. No browser. 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 if hls: log_renditions(hls) return hls def log_renditions(master_url): """Record what the master offers. We have never seen one carrying both video and audio, so log it every time until we know how that looks.""" 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)] 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" log(f" {kind:11s} {int(bw.group(1))//1000 if bw else '?':>6} kbps " f"{res.group(1) if res else '-':>10} {codecs}") def find_manifest(timeout_s=90): """Load the live page headlessly and return the entry id, or None. 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. """ from playwright.sync_api import sync_playwright found = {"master": None, "media": 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=" in u and not found["media"]: found["media"] = u elif "hdnts=" in u and not found["master"]: found["master"] = u 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 and not found["media"]: 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() seen_url = found["media"] or found["master"] if not seen_url: return None m = re.search(r"/(\d+_[a-z0-9]+)/", seen_url, re.I) return m.group(1) if m else None def record(url, duration_min, prefer='best'): stamp = datetime.now().strftime("%Y%m%d-%H%M%S") out = os.path.join(RECORDINGS, f"brfc-{stamp}.ts") 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 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") 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 log(f"attempt {attempt}: looking for a live stream") try: entry_id = find_manifest() except Exception as exc: log(f" error: {type(exc).__name__}: {exc}") entry_id = None if entry_id: log(f" live entry: {entry_id}") try: url = mint_master(entry_id) except Exception as exc: log(f" mint failed: {type(exc).__name__}: {exc}") url = None if not url: log(" could not mint a manifest; retrying") time.sleep(args.poll) continue log(" minted fresh master playlist") proc = record(url, args.duration, args.prefer) log(f" downloader pid {proc.pid}; waiting for it to finish") t0 = time.time() proc.wait() ran = time.time() - t0 log(f" downloader exited rc={proc.returncode} after {ran:.0f}s") # A master playlist outlives its stream: after a match ends the # master is still served while its variants 404. So "found a # master" does not mean "stream is live" -- only a download that # actually ran does. Anything that dies immediately means the # stream is not up yet, so keep waiting rather than giving up. if proc.returncode == 0 or ran > 120: return 0 log(" died immediately -- stream not live yet, will retry") time.sleep(args.poll) continue log(f" nothing live yet; retrying in {args.poll}s") time.sleep(args.poll) log("gave up: no live stream appeared within the wait window") return 1 if __name__ == "__main__": sys.exit(main())