#!/usr/bin/env python3 """Pre-match check: is the session still good, and can this host record? Run it a few hours before kickoff, not five minutes before. The one thing it cannot fix for you is a dead login: refreshing that means opening Firefox and signing in by hand, which takes a human (see README, "Decision: do not automate login"). python preflight.py # session + host checks python preflight.py --entry 0_z236v0e7 # also exercise the entry API Exit code is 0 only if the session minted a JWT. Anything else means tonight's scheduled run would fail, and it would fail at kickoff with nobody watching. Prints no secrets. Cookie values and tokens are reported by length only. """ import argparse import base64 import contextlib import http.cookiejar import io import json import os import re import shutil import sys import time import urllib.error import urllib.request from datetime import datetime, timezone ROOT = os.path.dirname(os.path.abspath(__file__)) RECORDINGS = os.path.join(ROOT, "recordings") UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:131.0) " "Gecko/20100101 Firefox/131.0") 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" OK, BAD, WARN = "ok ", "FAIL", "warn" def say(status, msg): print(f"[{status}] {msg}", flush=True) def check_deps(): """Everything record_live.py reaches for, checked before kickoff.""" good = True for mod, hint in (("yt_dlp", "pip install yt-dlp"), ("playwright", "pip install playwright && " "python -m playwright install firefox")): try: __import__(mod) say(OK, f"{mod} importable") except ImportError: say(BAD, f"{mod} missing -- {hint}") good = False if shutil.which("ffmpeg"): say(OK, "ffmpeg on PATH") else: say(BAD, "ffmpeg not on PATH -- the downloader is forced to ffmpeg") good = False return good def check_disk(need_gb=8): """A video match is ~4.7 GB. Audio is ~200 MB. Budget for video.""" os.makedirs(RECORDINGS, exist_ok=True) free_gb = shutil.disk_usage(RECORDINGS).free / 2**30 if free_gb >= need_gb: say(OK, f"{free_gb:.0f} GB free on the recordings volume") return True say(BAD, f"only {free_gb:.1f} GB free, want >= {need_gb} GB for a video match") return False def firefox_jar(): from yt_dlp.cookies import extract_cookies_from_browser with contextlib.redirect_stderr(io.StringIO()): return extract_cookies_from_browser("firefox") def check_cookies(jar): """Cookie *present and unexpired* is necessary, not sufficient. The server can invalidate a session while the browser still holds the cookie, which is exactly what check_session() is for.""" brfc = [c for c in jar if "bristolrovers" in (c.domain or "").lower()] say(OK if brfc else BAD, f"{len(brfc)} bristolrovers cookies in the Firefox jar") sess = [c for c in brfc if c.name.startswith(("SSESS", "SESS"))] if not sess: say(BAD, "no SSESS/SESS cookie -- Firefox is not logged in") return brfc, False for c in sess: if not c.expires: say(WARN, f"{c.name[:16]}... is a session cookie (dies with Firefox)") continue days = (c.expires - time.time()) / 86400 say(OK if days > 1 else BAD, f"{c.name[:16]}... len={len(c.value or '')} expires in {days:.0f}d") return brfc, True def check_session(brfc): """The real login test: does the site still mint a JWT off this cookie?""" cj = http.cookiejar.CookieJar() for c in brfc: cj.set_cookie(c) op = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cj)) hdrs = {"User-Agent": UA, "Accept": "application/json, text/plain, */*"} try: resp = op.open(urllib.request.Request(TOKEN_URL, headers=hdrs), timeout=20) body = resp.read().decode("utf-8", "replace") except urllib.error.HTTPError as exc: detail = exc.read().decode("utf-8", "replace")[:200] say(BAD, f"browser-token returned HTTP {exc.code}: {detail}") if exc.code in (401, 403): say(BAD, "session is dead server-side -- sign in again in Firefox") return None except Exception as exc: say(BAD, f"browser-token unreachable: {type(exc).__name__}: {exc}") return None m = re.search(r"(eyJ[\w-]+\.eyJ[\w-]+\.[\w-]+)", body) if not m: say(BAD, "no JWT in the browser-token response -- session is stale") return None jwt = m.group(1) seg = jwt.split(".")[1] try: claims = json.loads(base64.urlsafe_b64decode(seg + "=" * (-len(seg) % 4))) except Exception: claims = {} exp = claims.get("exp") when = (f", expires {datetime.fromtimestamp(exp, timezone.utc):%H:%M:%S}Z " f"({(exp - time.time()) / 60:+.0f} min)" if exp else "") say(OK, f"session mints a JWT ({len(jwt)} chars{when})") return jwt def check_entry(jwt, entry_id): """Optional: exercise the rest of the chain against a known entry id. A master playlist outlives its stream, so a 200 here does not prove the match is live -- it proves the entitlement still resolves. """ req = urllib.request.Request( f"https://api.playback.streamamg.com/v1/entry/{entry_id}", headers={"User-Agent": UA, "Authorization": "Bearer " + jwt, "x-api-key": API_KEY, "Origin": "https://www.bristolrovers.co.uk", "Referer": "https://www.bristolrovers.co.uk/"}) try: data = json.loads(urllib.request.urlopen(req, timeout=20) .read().decode("utf-8", "replace")) except urllib.error.HTTPError as exc: say(WARN, f"entry {entry_id}: HTTP {exc.code} " f"(stale entry id, or not entitled)") return except Exception as exc: say(WARN, f"entry {entry_id}: {type(exc).__name__}: {exc}") return hls = (data.get("media") or {}).get("hls") if not hls: say(WARN, f"entry {entry_id} resolved but carries no HLS master") return say(OK, f"entry {entry_id} resolves to a master playlist") try: body = urllib.request.urlopen(urllib.request.Request( hls, headers={"User-Agent": UA, "Referer": "https://www.bristolrovers.co.uk/"}), timeout=15).read().decode("utf-8", "replace") except Exception as exc: say(WARN, f"could not read the master: {type(exc).__name__}: {exc}") return PREFIX = "#EXT-X-STREAM-INF:" streams = [ln[len(PREFIX):] for ln in body.splitlines() if ln.startswith(PREFIX)] say(OK, 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" print(f" {kind:11s} {int(bw.group(1)) // 1000 if bw else '?':>6} kbps " f"{res.group(1) if res else '-':>10} {codecs}") def main(): ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) ap.add_argument("--entry", help="entry id to exercise the entry API against") args = ap.parse_args() print(f"preflight {datetime.now():%Y-%m-%d %H:%M:%S}\n") host_ok = check_deps() & check_disk() try: jar = firefox_jar() except Exception as exc: say(BAD, f"cannot read Firefox cookies: {type(exc).__name__}: {exc}") return 1 brfc, have_cookie = check_cookies(jar) jwt = check_session(brfc) if have_cookie else None if jwt and args.entry: check_entry(jwt, args.entry) print() if not jwt: say(BAD, "NOT READY -- sign in at https://www.bristolrovers.co.uk/user/login " "in Firefox, then run this again") return 1 if not host_ok: say(BAD, "session is good but this host cannot record -- fix the above") return 1 say(OK, "READY") return 0 if __name__ == "__main__": sys.exit(main())