brfc-stream-pirate/catcher.py

#!/usr/bin/env python3
"""Receive a manifest URL from the Tampermonkey sniffer and record the match.

Runs on localhost only. The userscript POSTs whatever manifest URL the player
actually requested; we hand it to yt-dlp.

Reload resilience is the whole point of the liveness check below. A page
reload mid-match is expected (network blips, tab crashes), and it must be
*safe* -- it must not start a duplicate recording, and it must restart a
recording that has died. So we dedupe on "is a downloader for this match
still alive", not "have I seen this match before".
"""
import json
import os
import re
import subprocess
import sys
import threading
from datetime import datetime
from http.server import BaseHTTPRequestHandler, HTTPServer

HOST, PORT = "127.0.0.1", 8787
ROOT = os.path.dirname(os.path.abspath(__file__))
RECORDINGS = os.path.join(ROOT, "recordings")
LOGS = os.path.join(ROOT, "logs")

# yt-dlp lifts session cookies from here. Firefox is required on Windows:
# Chrome/Edge lock their cookie DB while running (yt-dlp issue 7271).
BROWSER = os.environ.get("BRFC_BROWSER", "firefox")

# Hard stop, so an unattended run cannot record forever if the stream never
# ends cleanly. Pad past 90 min for half-time, stoppage and post-match.
DURATION_MIN = int(os.environ.get("BRFC_DURATION_MIN", "150"))

# Kaltura entry id (e.g. 0_xm5vrrw6). Every manifest for one game -- master and
# each variant, each with its own token -- shares it. It names the *feed*, not
# the match: the same id came back three weeks apart (see README). That is fine
# here, because the dedupe below is on liveness, not on "seen this before".
ASSET_RE = re.compile(r"/(\d+_[a-z0-9]+)/", re.I)

_procs = {}                 # asset key -> Popen of the running downloader
_lock = threading.Lock()


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, "catcher.log"), "a", encoding="utf-8") as fh:
        fh.write(line + "\n")


def asset_key(url):
    m = ASSET_RE.search(url)
    return m.group(1) if m else url


def start_recording(key, url, user_agent, page):
    stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
    out = os.path.join(RECORDINGS, f"brfc-{key}-{stamp}.ts")
    os.makedirs(RECORDINGS, exist_ok=True)

    cmd = [
        sys.executable, "-m", "yt_dlp",
        "--hls-use-mpegts",          # truncated file still plays
        "--no-part",
        "--cookies-from-browser", BROWSER,
        "--user-agent", user_agent or "Mozilla/5.0",
        "--referer", page or "https://www.bristolrovers.co.uk/",
        # yt-dlp's own retry flags do NOT reach the ffmpeg downloader, which is
        # what actually pulls live HLS. These are ffmpeg's own reconnect knobs.
        "--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,
        url,
    ]
    logfile = os.path.join(LOGS, f"yt-dlp-{stamp}.log")
    fh = open(logfile, "w", encoding="utf-8")
    proc = subprocess.Popen(cmd, stdout=fh, stderr=subprocess.STDOUT, cwd=ROOT)
    log(f"START  {key}  -> {os.path.basename(out)}  (pid {proc.pid}, cap {DURATION_MIN}min)")
    return proc


class Handler(BaseHTTPRequestHandler):
    def do_POST(self):
        if self.path not in ("/stream", "/trace"):
            self.send_error(404)
            return
        n = int(self.headers.get("Content-Length", 0))
        try:
            body = json.loads(self.rfile.read(n) or b"{}")
        except json.JSONDecodeError:
            self.send_error(400, "bad json")
            return

        if self.path == "/trace":
            self.do_TRACE_post(body)
            self.send_response(200)
            self.send_header("Access-Control-Allow-Origin", "*")
            self.end_headers()
            self.wfile.write(b'{"ok":true}')
            return

        url = body.get("url", "")
        if not url:
            self.send_error(400, "no url")
            return

        key = asset_key(url)
        action = "ignored"
        with _lock:
            proc = _procs.get(key)
            if proc is not None and proc.poll() is None:
                action = "already recording"       # healthy: reload is a no-op
            else:
                if proc is not None:
                    log(f"RESTART {key}: downloader had exited (rc={proc.returncode})")
                _procs[key] = start_recording(key, url, body.get("userAgent"),
                                              body.get("page"))
                action = "started"

        # Always 200 the browser; the player re-requests manifests constantly.
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Access-Control-Allow-Origin", "*")
        self.end_headers()
        self.wfile.write(json.dumps({"ok": True, "action": action}).encode())

    def do_TRACE_post(self, body):
        """Discovery mode: append the request/response the page made.

        Used to find which API call mints the stream token, so the chain can
        run without a GUI browser. Written as JSONL for easy grepping.
        """
        os.makedirs(LOGS, exist_ok=True)
        with open(os.path.join(LOGS, "trace.jsonl"), "a", encoding="utf-8") as fh:
            fh.write(json.dumps(body, ensure_ascii=False) + chr(10))

    def do_OPTIONS(self):
        self.send_response(204)
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Access-Control-Allow-Headers", "Content-Type")
        self.end_headers()

    def log_message(self, *args):
        pass


if __name__ == "__main__":
    log(f"catcher listening on http://{HOST}:{PORT}/stream  (cookies: {BROWSER})")
    HTTPServer((HOST, PORT), Handler).serve_forever()