"""Cut a piece of the day's broadcast out of the recording OBS is already making.

The producer's own replay took an hour and a half per race, because it RE-RENDERED: headless Chrome
on the CPU at six to eight frames a second, for a picture the A4000 beside it draws in real time. But
that picture has already been drawn, and OBS has already written it to disk at exactly the quality
that went out — the local recording is configured "same as stream" precisely so the per-race clips
cost no second NVENC session (`docs/plans/trwc-obs-production-workflow.md` §7).

So a replay is not a render. It is a **stream copy**: no encoder, no GPU, under a second for forty
seconds, and stream quality by construction rather than by settings anyone has to keep in step.

WHAT MAKES IT POSSIBLE AT ALL. Nothing in the picture is machine-readable — the stamp is text for a
human and the 52-cell strip that used to ride on a boat-cam is gone — so the UTC time of frame zero
cannot be recovered from the video afterwards. `recording_anchor.py` writes it down while it is still
knowable, once a minute, and its readers hand it back. That module's docstring has named "the day
cutter" since it was written; this is that file.

    python -m v3.broadcast.live.day_cutter --anchor D:/trwc/recording_anchor.jsonl
        --from-utc-ms 1755612345000 --to-utc-ms 1755612385000 --out D:/vr/replay/replay.mp4

Three things are the whole of it, and each was a defect waiting rather than a detail:

* **A race can straddle a file.** OBS splits the recording every 60 minutes and each new file starts a
  new zero, which is why the anchor log groups by path. A cut that crosses the boundary is made from
  both files and concatenated.
* **A stream copy cuts on keyframes.** At a ~2 s GOP the clip can open up to two seconds early. That
  is acceptable for a replay and is said here rather than discovered on air; the alternative is
  re-encoding, which is the cost this module exists to avoid.
* **The file is being written while it is read.** Fragmented MP4 is readable up to the write point —
  that is why the format was chosen — but a cut must never reach past what has actually been written,
  or it silently returns short.

Installation is `emergency_video_cache.capture_once`'s pattern, verbatim and deliberately: candidate
file, then `verify_media`, then an atomic replace. A half-written file must never be what OBS opens.
"""
from __future__ import annotations

import argparse
import json
import subprocess
import sys
import time
from dataclasses import dataclass
from pathlib import Path

_HERE = Path(__file__).resolve().parent
if str(_HERE) not in sys.path:                        # so the siblings resolve either way
    sys.path.insert(0, str(_HERE))

from emergency_stream import EmergencyStreamError, verify_media  # noqa: E402
from recording_anchor import read_samples  # noqa: E402

#: How far past the newest anchor sample a cut may reach in the file still being written. The sampler
#: ticks once a minute, so the recording is always further along than the last sample says; with no
#: allowance every cut of the last minute would be refused, and that is exactly the minute a replay is
#: usually of. One tick and a little, and never applied to a file that is closed.
OPEN_TAIL_MS = 75_000

#: How large a hole between one file's last sample and the next file's zero still counts as an OBS
#: SPLIT rather than a stop and a restart. A split is seamless — file N ends exactly where N+1 begins —
#: so its "gap" is only the sampling interval, while a restart leaves however long somebody took. The
#: distinction is what lets a race straddle a split and still refuse to be concatenated across an
#: interruption, which would produce a clip that jumps with nothing on screen to explain it.
SPLIT_TOLERANCE_MS = 75_000

#: A replay shorter than this is not a replay. Deliberately below the emergency card's four seconds: a
#: single mark rounding is legitimately brief, and that constant is about a holding card that has to
#: survive being looped.
MIN_CLIP_S = 1.5


class DayCutterError(RuntimeError):
    """A cut that cannot be made, said in terms an operator can act on."""


@dataclass(frozen=True)
class Segment:
    """One recording FILE and the wall-clock window it holds."""

    path: str
    anchor_utc_ms: int
    end_utc_ms: int                                   # the last instant proven to be in the file

    def offset_s(self, utc_ms: float) -> float:
        return (utc_ms - self.anchor_utc_ms) / 1000.0


@dataclass(frozen=True)
class Piece:
    """One seek-and-length cut out of one file. A cut across a split is two of these."""

    path: str
    offset_s: float
    duration_s: float


def segments(samples: list[dict], *, now_utc_ms: float | None = None) -> list[Segment]:
    """One `Segment` per recording file, oldest first.

    The anchor is the MEDIAN of that file's samples, for the reason `anchor_for` uses one: each sample
    carries the jitter of a round trip to OBS, and the median discards the outliers without a
    threshold anybody has to justify. The end is the newest sample's own wall clock — what the file is
    PROVEN to contain — extended by `OPEN_TAIL_MS` for the newest file only, which is still open.
    """
    rows = [s for s in samples if s.get("anchorUtcMs") is not None and s.get("path")]
    if not rows:
        return []
    rows.sort(key=lambda s: int(s.get("tUtcMs", 0)))
    by_path: dict[str, list[dict]] = {}
    for row in rows:
        by_path.setdefault(str(row["path"]), []).append(row)

    now_ms = time.time() * 1000.0 if now_utc_ms is None else now_utc_ms
    newest = str(rows[-1]["path"])
    partial: list[tuple[str, int, int]] = []                      # (path, anchor, last proven sample)
    for path, group in by_path.items():
        anchors = sorted(int(s["anchorUtcMs"]) for s in group)
        partial.append((path, anchors[len(anchors) // 2], int(group[-1].get("tUtcMs", 0))))
    partial.sort(key=lambda row: row[1])

    out: list[Segment] = []
    for i, (path, anchor, proven) in enumerate(partial):
        if path == newest:
            # Still being written. The sampler ticks once a minute, so the file always reaches further
            # than its last sample says — but only so far, and never past now.
            end = int(min(proven + OPEN_TAIL_MS, now_ms))
        elif i + 1 < len(partial) and partial[i + 1][1] - proven <= SPLIT_TOLERANCE_MS:
            # A SPLIT: the next file's zero is where this one ended, so the recording is continuous
            # across the boundary and a race may be cut out of both halves. Taking the last sample as
            # the end instead would leave up to a minute of the day unreachable at every split — and
            # a race that happened to straddle one would be refused as a gap.
            end = int(partial[i + 1][1])
        else:
            end = int(proven)                                     # closed, and nothing followed it
        out.append(Segment(path=path, anchor_utc_ms=anchor, end_utc_ms=end))
    return out


def plan_cut(samples: list[dict], from_utc_ms: float, to_utc_ms: float,
             *, now_utc_ms: float | None = None) -> list[Piece]:
    """The pieces that together are the window, in order — one per recording file it spans.

    Refuses rather than returning something short. A replay that quietly opens in the wrong place, or
    ends before the moment it was taken for, is worse than one that did not happen: the operator sees
    a picture and believes it.
    """
    if to_utc_ms <= from_utc_ms:
        raise DayCutterError("the cut ends before it starts")
    if (to_utc_ms - from_utc_ms) / 1000.0 < MIN_CLIP_S:
        raise DayCutterError(f"a cut of {(to_utc_ms - from_utc_ms) / 1000.0:.1f}s is not a replay "
                             f"(minimum {MIN_CLIP_S}s)")
    segs = segments(samples, now_utc_ms=now_utc_ms)
    if not segs:
        raise DayCutterError("no recording anchor samples — was recording_anchor.py running?")

    pieces: list[Piece] = []
    cursor = from_utc_ms
    for seg in segs:
        if seg.end_utc_ms <= cursor or seg.anchor_utc_ms >= to_utc_ms:
            continue
        if seg.anchor_utc_ms > cursor:
            # A gap between two files — the recording was stopped and restarted, or a split lost
            # time. Say so: the alternative is a clip that jumps, with nothing to explain the jump.
            raise DayCutterError(
                f"the recording has a gap at {int(cursor)} (the next file starts {seg.anchor_utc_ms})")
        piece_end = min(seg.end_utc_ms, to_utc_ms)
        pieces.append(Piece(path=seg.path, offset_s=round(seg.offset_s(cursor), 3),
                            duration_s=round((piece_end - cursor) / 1000.0, 3)))
        cursor = piece_end
        if cursor >= to_utc_ms:
            break
    if not pieces:
        raise DayCutterError(f"nothing recorded covers {int(from_utc_ms)} to {int(to_utc_ms)}")
    if cursor < to_utc_ms:
        raise DayCutterError(
            f"the recording only reaches {int(cursor)}, {(to_utc_ms - cursor) / 1000.0:.1f}s short of "
            f"the requested end — the moment may not be written to disk yet")
    return pieces


def _cut_cmd(piece: Piece, out: Path, *, ffmpeg: str) -> list[str]:
    """The seek goes BEFORE the input: that is the fast seek, and for a stream copy it is also the
    keyframe-accurate one."""
    return [ffmpeg, "-y", "-hide_banner", "-loglevel", "error",
            "-ss", f"{piece.offset_s:.3f}", "-i", piece.path, "-t", f"{piece.duration_s:.3f}",
            "-c", "copy", "-movflags", "+faststart", str(out)]


def _concat_cmd(parts: list[Path], out: Path, *, ffmpeg: str, list_file: Path) -> list[str]:
    body = "".join("file " + repr(p.as_posix()) + "\n" for p in parts)
    list_file.write_text(body, encoding="utf-8")
    return [ffmpeg, "-y", "-hide_banner", "-loglevel", "error",
            "-f", "concat", "-safe", "0", "-i", str(list_file),
            "-c", "copy", "-movflags", "+faststart", str(out)]


def cut(samples: list[dict], from_utc_ms: float, to_utc_ms: float, *, output: Path,
        ffmpeg: str = "ffmpeg", ffprobe: str = "ffprobe",
        now_utc_ms: float | None = None) -> dict:
    """Make the clip and install it atomically. Returns what was installed.

    The old file survives every failure, which is the property that lets an operator take a replay
    twice in a row without a bad take blanking the scene in between.
    """
    pieces = plan_cut(samples, from_utc_ms, to_utc_ms, now_utc_ms=now_utc_ms)
    output = output.resolve()
    output.parent.mkdir(parents=True, exist_ok=True)
    candidate = output.with_name(f".{output.stem}.candidate{output.suffix}")
    listing = output.with_name(f".{output.stem}.concat.txt")
    parts: list[Path] = []
    try:
        for i, piece in enumerate(pieces):
            single = len(pieces) == 1
            target = candidate if single else output.with_name(f".{output.stem}.p{i}{output.suffix}")
            _run(_cut_cmd(piece, target, ffmpeg=ffmpeg))
            parts.append(target)
        if len(pieces) > 1:
            _run(_concat_cmd(parts, candidate, ffmpeg=ffmpeg, list_file=listing))
        details = verify_media(candidate, ffprobe=ffprobe, min_duration_s=MIN_CLIP_S)
        candidate.replace(output)                     # atomic: OBS holds this path open
        return {**details, "path": str(output), "fromUtcMs": int(from_utc_ms),
                "toUtcMs": int(to_utc_ms), "pieces": [p.path for p in pieces]}
    except EmergencyStreamError as e:
        raise DayCutterError(f"the cut is not a usable clip: {e}") from e
    finally:
        candidate.unlink(missing_ok=True)
        for part in parts:
            if part != candidate:
                part.unlink(missing_ok=True)
        listing.unlink(missing_ok=True)


def _run(cmd: list[str]) -> None:
    try:
        subprocess.run(cmd, check=True, capture_output=True, text=True)
    except FileNotFoundError as e:
        raise DayCutterError(f"{cmd[0]} is not on PATH") from e
    except subprocess.CalledProcessError as e:
        raise DayCutterError(f"ffmpeg refused the cut: {(e.stderr or '').strip()[:300]}") from e


def main(argv=None) -> int:                                       # noqa: ANN001
    p = argparse.ArgumentParser(description="Cut a window of the day's broadcast out of the recording.")
    p.add_argument("--anchor", required=True, type=Path,
                   help="the recording_anchor.jsonl written beside the show")
    p.add_argument("--from-utc-ms", required=True, type=int)
    p.add_argument("--to-utc-ms", required=True, type=int)
    p.add_argument("--out", required=True, type=Path, help="where to install the clip")
    p.add_argument("--ffmpeg", default="ffmpeg")
    p.add_argument("--ffprobe", default="ffprobe")
    a = p.parse_args(argv)
    try:
        details = cut(read_samples(a.anchor), a.from_utc_ms, a.to_utc_ms, output=a.out,
                      ffmpeg=a.ffmpeg, ffprobe=a.ffprobe)
    except DayCutterError as e:
        print(f"[day-cutter] {e}", file=sys.stderr)
        return 1
    print(json.dumps(details, indent=1))
    return 0


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