"""Independent emergency programme for OBS and YouTube backup ingest.

This path is intentionally much smaller than the normal broadcast:

* ``build`` produces one deterministic 1920x1080 H.264/AAC holding clip from vendored branding;
* OBS plays that local file as a looping media source; and
* ``stream`` loops the same file to an explicitly configured *backup* RTMPS endpoint.

It does not import or contact the producer, Cloud SQL, telemetry, Chromium, or the read API.  The
only runtime dependencies are the local MP4, FFmpeg, DNS/network, and the distribution endpoint.
The destination is read from the environment and is always redacted from logs.
"""
from __future__ import annotations

import argparse
import json
import os
import signal
import subprocess
import sys
import time
from pathlib import Path
from urllib.parse import urlsplit


WIDTH = 1920
HEIGHT = 1080
FPS = 30
BITRATE_K = 4_500
CARD_SECONDS = 12
HEALTH_MAX_AGE_S = 15


class EmergencyStreamError(RuntimeError):
    """A configuration or media error that must prevent the standby from claiming readiness."""


def _repo_root() -> Path:
    return Path(__file__).resolve().parents[3]


def default_logo() -> Path:
    """The vendored raster source for the canonical mark; never a network URL."""
    return _repo_root() / "v3" / "broadcast" / "core" / "web" / "assets" / "logo-source.png"


def default_font() -> Path:
    """A concrete local face, so a broken fontconfig cannot hang the safety-card build."""
    candidates = (
        _repo_root() / "brand" / "fonts" / "Roboto-Regular.ttf",
        Path("C:/Windows/Fonts/arial.ttf"),
        Path("C:/Windows/Fonts/Arial.ttf"),
        Path("/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf"),
        Path("/usr/share/fonts/truetype/liberation2/LiberationSans-Regular.ttf"),
    )
    for candidate in candidates:
        if candidate.is_file():
            return candidate
    raise EmergencyStreamError(
        "no deterministic broadcast font found; install fonts-dejavu-core or vendor Roboto-Regular.ttf"
    )


def _filter_path(path: Path) -> str:
    """Escape a filesystem path for FFmpeg's filter expression (not for the process argv)."""
    return str(path.resolve()).replace("\\", "/").replace(":", r"\:").replace("'", r"\'")


def build_card_cmd(*, output: Path, logo: Path, font: Path, ffmpeg: str = "ffmpeg",
                   duration_s: int = CARD_SECONDS, bitrate_k: int = BITRATE_K) -> list[str]:
    """FFmpeg command for the local, self-contained technical-interruption programme.

    The source PNG has a white background. ``colorkey`` removes it before the mark is placed over the
    navy card, retaining the canonical red/green sails without needing Pillow, Chromium, SVG support,
    or a second asset origin. The encoder carries silent AAC deliberately: a video-only backup can
    produce a disruptive audio reconfiguration at distribution failover.
    """
    if duration_s < 4:
        raise ValueError("the holding clip must be at least four seconds")
    if bitrate_k < 1_000:
        raise ValueError("the holding clip bitrate is implausibly low for 1080p")
    face = _filter_path(font)
    graph = (
        "[1:v]scale=260:260,format=rgba,colorkey=0xFFFFFF:0.12:0.03[mark];"
        "[0:v][mark]overlay=(W-w)/2:125[base];"
        "[base]"
        f"drawtext=fontfile='{face}':text='VIEWREGATTA':fontcolor=white:fontsize=48:"
        "x=(w-text_w)/2:y=405,"
        f"drawtext=fontfile='{face}':text='TEMPORARY INTERRUPTION':fontcolor=white:fontsize=72:"
        "x=(w-text_w)/2:y=505,"
        f"drawtext=fontfile='{face}':text='We are sorry. Coverage will resume shortly.':"
        "fontcolor=white@0.92:fontsize=42:x=(w-text_w)/2:y=630,"
        f"drawtext=fontfile='{face}':text='Thank you for your patience.':"
        "fontcolor=white@0.72:fontsize=30:x=(w-text_w)/2:y=705,"
        "drawbox=x=660:y=795:w=280:h=8:color=0xE51F1F:t=fill,"
        "drawbox=x=980:y=795:w=280:h=8:color=0x13A83C:t=fill[outv]"
    )
    return [
        ffmpeg, "-hide_banner", "-y",
        "-f", "lavfi", "-i", f"color=c=0x071522:s={WIDTH}x{HEIGHT}:r={FPS}:d={duration_s}",
        "-loop", "1", "-i", str(logo),
        "-f", "lavfi", "-i", "anullsrc=r=48000:cl=stereo",
        "-filter_complex", graph,
        "-map", "[outv]", "-map", "2:a:0", "-t", str(duration_s),
        "-c:v", "libx264", "-preset", "medium", "-tune", "stillimage",
        "-profile:v", "high", "-level:v", "4.1", "-pix_fmt", "yuv420p",
        "-r", str(FPS), "-g", str(FPS * 2), "-keyint_min", str(FPS * 2), "-sc_threshold", "0",
        "-b:v", f"{bitrate_k}k", "-minrate", f"{bitrate_k}k", "-maxrate", f"{bitrate_k}k",
        "-bufsize", f"{bitrate_k * 2}k",
        "-x264-params", "nal-hrd=cbr:force-cfr=1:filler=1",
        "-c:a", "aac", "-b:a", "128k", "-ar", "48000", "-ac", "2",
        "-movflags", "+faststart", str(output),
    ]


def build_card(*, output: Path, logo: Path | None = None, font: Path | None = None,
               ffmpeg: str = "ffmpeg",
               duration_s: int = CARD_SECONDS, bitrate_k: int = BITRATE_K) -> dict:
    """Build atomically and verify before replacing the on-air file."""
    output = output.resolve()
    logo = (logo or default_logo()).resolve()
    font = (font or default_font()).resolve()
    if not logo.is_file():
        raise EmergencyStreamError(f"brand mark not found: {logo}")
    if not font.is_file():
        raise EmergencyStreamError(f"font not found: {font}")
    output.parent.mkdir(parents=True, exist_ok=True)
    temporary = output.with_name(f".{output.stem}.building{output.suffix}")
    try:
        subprocess.run(
            build_card_cmd(output=temporary, logo=logo, font=font, ffmpeg=ffmpeg,
                           duration_s=duration_s, bitrate_k=bitrate_k),
            check=True,
        )
        details = verify_media(temporary)
        os.replace(temporary, output)
        return {**details, "path": str(output), "bytes": output.stat().st_size}
    except subprocess.CalledProcessError as e:
        raise EmergencyStreamError(f"FFmpeg could not build the emergency programme: {e}") from e
    finally:
        temporary.unlink(missing_ok=True)


def _fraction(raw: str) -> float:
    numerator, denominator = raw.split("/", 1)
    return float(numerator) / float(denominator)


def probe_media(path: Path, *, ffprobe: str = "ffprobe") -> dict:
    """Read the fields that must match OBS/YouTube; no human-formatted FFprobe parsing."""
    try:
        result = subprocess.run(
            [ffprobe, "-v", "error", "-show_streams", "-show_format", "-of", "json", str(path)],
            check=True, capture_output=True, text=True,
        )
    except (OSError, subprocess.CalledProcessError) as e:
        raise EmergencyStreamError(f"cannot probe emergency media {path}: {e}") from e
    return json.loads(result.stdout)


def verify_media(path: Path, *, ffprobe: str = "ffprobe", min_duration_s: float = 4.0) -> dict:
    """Refuse media that would change format or leave the output without audio.

    ONE opinion about what a playable clip is, shared by everything that installs one — the emergency
    card and the day cutter's replays both go into the same OBS at the same canvas, and two checks
    would eventually disagree about the same file. Only the MINIMUM LENGTH differs and so it is the
    only parameter: four seconds is about a holding card that has to survive being looped, while a
    replay of a single mark rounding is legitimately brief.
    """
    if not path.is_file() or path.stat().st_size == 0:
        raise EmergencyStreamError(f"media is absent or empty: {path}")
    doc = probe_media(path, ffprobe=ffprobe)
    streams = doc.get("streams") or []
    video = next((s for s in streams if s.get("codec_type") == "video"), None)
    audio = next((s for s in streams if s.get("codec_type") == "audio"), None)
    faults: list[str] = []
    if not video:
        faults.append("no video stream")
    else:
        if (video.get("width"), video.get("height")) != (WIDTH, HEIGHT):
            faults.append(f"video is {video.get('width')}x{video.get('height')}, expected {WIDTH}x{HEIGHT}")
        if video.get("codec_name") != "h264":
            faults.append(f"video codec is {video.get('codec_name')}, expected h264")
        if video.get("pix_fmt") != "yuv420p":
            faults.append(f"pixel format is {video.get('pix_fmt')}, expected yuv420p")
        try:
            actual_fps = _fraction(video.get("avg_frame_rate") or "0/1")
        except (ValueError, ZeroDivisionError):
            actual_fps = 0
        if abs(actual_fps - FPS) > 0.01:
            faults.append(f"frame rate is {actual_fps:g}, expected {FPS}")
    if not audio:
        faults.append("no audio stream")
    else:
        if audio.get("codec_name") != "aac":
            faults.append(f"audio codec is {audio.get('codec_name')}, expected aac")
        if int(audio.get("sample_rate") or 0) != 48_000:
            faults.append(f"audio sample rate is {audio.get('sample_rate')}, expected 48000")
    duration = float((doc.get("format") or {}).get("duration") or 0)
    if duration < min_duration_s:
        faults.append(f"duration is {duration:.2f}s, expected at least {min_duration_s:g}s")
    if faults:
        raise EmergencyStreamError("; ".join(faults))
    return {
        "path": str(path.resolve()),
        "bytes": path.stat().st_size,
        "durationS": round(duration, 3),
        "video": f"h264 {WIDTH}x{HEIGHT} {FPS}fps yuv420p",
        "audio": "aac 48000Hz",
    }


def destination_from_env(env: dict[str, str] | os._Environ[str] = os.environ) -> tuple[str, str]:
    """Return ``(full_url, redaction_token)`` without guessing YouTube's backup address.

    ``YOUTUBE_BACKUP_RTMPS_URL`` may be the complete secret URL. Prefer the split form in systemd:
    ``YOUTUBE_BACKUP_INGEST`` (the backup base copied from Live Control Room) plus
    ``YOUTUBE_STREAM_KEY``. Requiring the word ``backup`` or the conventional ``b.`` hostname is a
    guard against accidentally sending a second encoder to the primary input.
    """
    full = (env.get("YOUTUBE_BACKUP_RTMPS_URL") or "").strip()
    key = (env.get("YOUTUBE_STREAM_KEY") or "").strip()
    if not full:
        base = (env.get("YOUTUBE_BACKUP_INGEST") or "").strip().rstrip("/")
        if not base or not key:
            raise EmergencyStreamError(
                "set YOUTUBE_BACKUP_INGEST and YOUTUBE_STREAM_KEY (or YOUTUBE_BACKUP_RTMPS_URL)"
            )
        full = f"{base}/{key}"
    parsed = urlsplit(full)
    if parsed.scheme != "rtmps" or not parsed.hostname:
        raise EmergencyStreamError("the backup destination must be an rtmps:// URL")
    marker = f"{parsed.hostname}{parsed.path}".lower()
    if "backup" not in marker and not parsed.hostname.lower().startswith("b."):
        raise EmergencyStreamError(
            "the destination does not look like a backup ingest; copy the backup RTMPS URL, not primary"
        )
    token = key or parsed.path.rstrip("/").rsplit("/", 1)[-1]
    if not token:
        raise EmergencyStreamError("the backup destination has no stream key")
    return full, token


def stream_cmd(*, media: Path, destination: str, ffmpeg: str = "ffmpeg") -> list[str]:
    """Loop a pre-validated programme in real time without decoding or re-encoding it."""
    return [
        ffmpeg, "-hide_banner", "-nostdin", "-loglevel", "warning",
        "-re", "-stream_loop", "-1", "-i", str(media),
        "-map", "0:v:0", "-map", "0:a:0", "-c", "copy",
        "-progress", "pipe:1", "-stats_period", "2",
        "-f", "flv", destination,
    ]


def _write_health(path: Path, doc: dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_suffix(path.suffix + ".tmp")
    temporary.write_text(json.dumps(doc, sort_keys=True) + "\n", encoding="utf-8")
    os.replace(temporary, path)


def stream(*, media: Path, health_path: Path, ffmpeg: str = "ffmpeg") -> int:
    """Run FFmpeg once. systemd owns restart policy; this owns validation, health and redaction."""
    details = verify_media(media)
    destination, secret = destination_from_env()
    safe_destination = destination.replace(secret, "***") if secret else "<redacted>"
    print(f"[emergency] streaming {details['video']} + {details['audio']} -> {safe_destination}", flush=True)
    process = subprocess.Popen(  # noqa: S603 - fixed FFmpeg argv, no shell
        stream_cmd(media=media, destination=destination, ffmpeg=ffmpeg),
        stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1,
    )

    def stop(_signum, _frame) -> None:  # noqa: ANN001 - signal handler contract
        if process.poll() is None:
            process.terminate()

    signal.signal(signal.SIGTERM, stop)
    signal.signal(signal.SIGINT, stop)
    progress: dict[str, str] = {}
    _write_health(health_path, {
        "status": "starting", "updatedUtcMs": int(time.time() * 1000), "pid": process.pid,
    })
    assert process.stdout is not None
    for raw in process.stdout:
        line = raw.rstrip("\r\n")
        if "=" in line and line.split("=", 1)[0].replace("_", "").isalnum():
            key, value = line.split("=", 1)
            progress[key] = value
            if key == "progress":
                _write_health(health_path, {
                    "status": "streaming" if value == "continue" else value,
                    "updatedUtcMs": int(time.time() * 1000),
                    "pid": process.pid,
                    "frame": int(progress.get("frame") or 0),
                    "fps": progress.get("fps"),
                    "bitrate": progress.get("bitrate"),
                    "outTime": progress.get("out_time"),
                })
                progress.clear()
        elif line:
            print(line.replace(destination, safe_destination).replace(secret, "***"), flush=True)
    code = process.wait()
    _write_health(health_path, {
        "status": "stopped", "updatedUtcMs": int(time.time() * 1000),
        "pid": process.pid, "exitCode": code,
    })
    return code if code else 1  # an endless standby exiting cleanly is still a fault for systemd


def status(health_path: Path, *, now_ms: int | None = None,
           max_age_s: int = HEALTH_MAX_AGE_S) -> dict:
    """Readiness from encoder progress, not merely from the existence of a process."""
    try:
        doc = json.loads(health_path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as e:
        raise EmergencyStreamError(f"no valid encoder health at {health_path}: {e}") from e
    now = now_ms if now_ms is not None else int(time.time() * 1000)
    age_ms = now - int(doc.get("updatedUtcMs") or 0)
    if doc.get("status") != "streaming":
        raise EmergencyStreamError(f"encoder status is {doc.get('status')!r}, not 'streaming'")
    if age_ms < 0 or age_ms > max_age_s * 1000:
        raise EmergencyStreamError(f"encoder progress is stale by {age_ms / 1000:.1f}s")
    return {**doc, "ageMs": age_ms}


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    sub = parser.add_subparsers(dest="command", required=True)
    build = sub.add_parser("build", help="build the local technical-interruption MP4 atomically")
    build.add_argument("--output", type=Path, required=True)
    build.add_argument("--logo", type=Path, default=default_logo())
    build.add_argument("--font", type=Path, default=None,
                       help="concrete TTF path; defaults to vendored Roboto or a known system face")
    build.add_argument("--ffmpeg", default="ffmpeg")
    build.add_argument("--duration", type=int, default=CARD_SECONDS)
    build.add_argument("--bitrate", type=int, default=BITRATE_K)
    verify = sub.add_parser("verify", help="verify the exact on-air media contract")
    verify.add_argument("media", type=Path)
    verify.add_argument("--ffprobe", default="ffprobe")
    run = sub.add_parser("stream", help="loop the verified MP4 to the configured backup RTMPS input")
    run.add_argument("--input", type=Path, required=True)
    run.add_argument("--health", type=Path, required=True)
    run.add_argument("--ffmpeg", default="ffmpeg")
    check = sub.add_parser("status", help="fail unless recent FFmpeg progress proves output is moving")
    check.add_argument("--health", type=Path, required=True)
    check.add_argument("--max-age", type=int, default=HEALTH_MAX_AGE_S)
    args = parser.parse_args()
    try:
        if args.command == "build":
            result = build_card(output=args.output, logo=args.logo, font=args.font, ffmpeg=args.ffmpeg,
                                duration_s=args.duration, bitrate_k=args.bitrate)
        elif args.command == "verify":
            result = verify_media(args.media, ffprobe=args.ffprobe)
        elif args.command == "stream":
            return stream(media=args.input, health_path=args.health, ffmpeg=args.ffmpeg)
        else:
            result = status(args.health, max_age_s=args.max_age)
        print(json.dumps(result, indent=2, sort_keys=True))
        return 0
    except EmergencyStreamError as e:
        print(f"emergency standby: {e}", file=sys.stderr)
        return 2


if __name__ == "__main__":
    raise SystemExit(main())