#!/usr/bin/env python3
"""Supply a verified English subtitle for files waiting too long on Bazarr.

`consolidate-subs.py` records files with no English candidate in
`var/state/needs-english.json` (first_seen/last_seen, keyed by resolved path). Once a
file has waited past `SUB_FALLBACK_AGE_HOURS` (default 24), this script tries subliminal
(via `media_stack.subtitles.fetch_via_subliminal`), then local whisper transcription,
and accepts only a result verified to belong to THIS file. Working files live in a
per-file temp directory under `WORKDIR_ROOT` (local disk, never the media pool).

VERIFICATION
  - Subliminal can return the WRONG episode: `media_stack.subverify.overlap()` needs BOTH
    a unigram word ratio AND an ordered-bigram ratio (calibration in its docstring).
  - A file named `<episode>.en.srt` may not be English: `media_stack.sublang` scores the TEXT.
  - An UNTAGGED/`und` audio stream may not be English: `detect_language` (a MULTILINGUAL
    model, never `.en`) must confirm it first.
  If subliminal's candidate fails, the whole file is transcribed. An ASR of the episode
  itself is always the right episode, so only "did it produce speech" applies. A Whisper
  transcript is published as `<stem>.en.generated.srt` (subliminal's stays `<stem>.en.srt`)
  so consolidate-subs.py prefers any real subtitle later and names the muxed track
  "English (generated)".

NO-CLOBBER PLACEMENT (`_place_sidecar`): right before writing, the file is re-probed; if
CONSOLIDATED_SUBS appeared or any English sidecar now exists, this run backs off
(ALREADY_CONSOLIDATED / SIDECAR_APPEARED). The sidecar goes to a hidden, UNIQUE-per-attempt
temp file (`tempfile.mkstemp`) and is published with `os.link()`, never a rename, so an
existing file is never touched and no truncated subtitle reaches the library. A video that
already has a main English sidecar is skipped outright (SIDECAR_PRESENT): the watcher or
the 05:00 `consolidate-subs.py --scan` owns it.

SAME-STEM SIBLING VIDEOS: with `Episode.mp4` and `Episode.mkv` both present, a sidecar is
ambiguous (`consolidate-watch.sh` tries `.mkv` first). Such a candidate is skipped
(SIBLING_COLLISION) before any work and re-checked before publishing.

PLAYBACK: FAILS CLOSED, BOUNDED. `None` from `JellyfinClient.now_playing_sessions` means
"someone is watching"; every ffmpeg extraction is pause-aware (`media_stack.subverify`
`should_abort`). A wait past `SUB_FALLBACK_MAX_WAIT_MIN` ends the WHOLE run cleanly (exit 0;
the next hourly tick retries). A real run without `JELLYFIN_API_KEY` refuses to start
(exit 1); `--dry-run` is exempt. The transcription child gets this process's validated
key/URL and refuses to extract without one.

RUNTIME BOUNDS: full-file transcription (the only unbounded step) runs as `python3 -m
media_stack.subverify transcribe ...` under a hard wall-clock timeout
(`SUB_FALLBACK_ASR_TIMEOUT_MIN`) so it can be killed (ASR_TIMEOUT). One is never STARTED with
under 10 minutes of the run's `SUB_FALLBACK_BUDGET_MIN` left (DEFERRED_BUDGET; the tracker
entry is untouched).

ONE FILE AT A TIME: `consolidate-watch.sh` has no concurrency cap on sidecar arrivals, so
this places one sidecar, waits for that file's pipeline run to finish or time out, then
starts the next.

`--dub` MODE (own cron line and lock file, so a long dub backlog never blocks the NEEDS_BAZARR
run; same budget, playback gate and logs): a different job.
A CONSOLIDATED_SUBS anime file (`is_anime`: Sonarr `seriesType`, else the path keywords) that
keeps both Japanese and English audio (`is_dual_audio`) but has no dub-like English text
subtitle (`has_dub_subtitle`) and no `.en.dub.srt` gets a whole-file Whisper transcript of its
English audio, published as `<stem>.en.dub.srt` (same no-clobber placement); consolidate-subs.py
then adds it as the "English (dub)" track. Targets come from walking `MEDIA_ROOT`'s `tv/`
`.mkv` files (a Radarr movie is never anime); a file whose tag, audio and subtitles settle it
as no target is remembered by `[ino, size, mtime_ns]` in `var/state/sub-fallback-dub.json` so
the next run skips the probe.

Exit codes:
  0  ok, including nothing to do, a second instance finding the lock held, and a
     playback wait that exceeded SUB_FALLBACK_MAX_WAIT_MIN
  1  blinded: needs-english.json unreadable, or a real run with no JELLYFIN_API_KEY
"""

from __future__ import annotations

import argparse
import json
import os
import signal
import subprocess
import sys
import tempfile
import time
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Callable, NamedTuple

sys.path.insert(0, str(Path(__file__).resolve().parent))

from media_stack.audio import is_anime, is_dual_audio  # noqa: E402
from media_stack.clients.jellyfin import JellyfinClient  # noqa: E402
from media_stack.envconfig import env_float, env_int  # noqa: E402
from media_stack.fileid import identity_matches  # noqa: E402
from media_stack.locking import (  # noqa: E402
    acquire_file_lock, acquire_singleton_lock, handle_sigterm as _handle_sigterm,
)
from media_stack.paths import (  # noqa: E402
    MEDIA_LAN_IP, MEDIA_ROOT, MEDIA_STACK_ROOT, VAR_DIR, VAR_LOG, VAR_RUN, VAR_STATE,
    ensure_var_dirs, load_env_file,
)
from media_stack.playback import (  # noqa: E402
    PLAYBACK_POLL_S, PlaybackWaitBudget, PlaybackWaitTimeout, call_with_playback_pause,
    wait_while_playing as _shared_wait_while_playing,
)
from media_stack.probe import already_processed, probe  # noqa: E402
from media_stack.state import load_state, load_state_locked, mutate_state  # noqa: E402
from media_stack.sublang import score as sublang_score  # noqa: E402
from media_stack.subtitles import (  # noqa: E402
    fetch_via_subliminal, find_dub_sidecars, find_sidecar_subs, has_dub_subtitle,
)
from media_stack.subverify import (  # noqa: E402
    CHILD_PLAYBACK_WAIT_REPORT_FILENAME, MIN_ASR_WORDS, MIN_BIGRAM_RATIO,
    MIN_WORD_RATIO, asr_window, detect_language, english_audio_stream,
    is_audio_explicitly_english, overlap,
)

ensure_var_dirs()  # writer: lock file + jsonl log

NEEDS_ENGLISH_STATE_FILE = VAR_STATE / "needs-english.json"   # written by consolidate-subs.py
DUB_CHECKED_FILE = VAR_STATE / "sub-fallback-dub.json"        # --dub: settled non-targets
LOCK_FILE = VAR_RUN / "sub-fallback.lock"
DUB_LOCK_FILE = VAR_RUN / "sub-fallback-dub.lock"
JSONL_FILE = VAR_LOG / "sub-fallback.jsonl"

# Local disk: a temp file on the pool would compete for the I/O the playback gate protects.
WORKDIR_ROOT = VAR_DIR / "tmp"

MIN_WHISPER_CUES = 1   # fewer cues than this means no speech

LID_PROBABILITY_FLOOR = 0.6
LID_WINDOW_S = 30.0
# 25% first (avoids cold-open music/silence); a low-confidence result retries at 50%, then 10%.
LID_FRACTIONS = (0.25, 0.50, 0.10)

# Never START a full transcription with less than this left of the run budget.
DEFER_TRANSCRIPTION_IF_BUDGET_UNDER_S = 10 * 60

PIPELINE_POLL_S = 20
SUBPROCESS_EXIT_POLL_S = 0.2

# mkstemp's 0600 would survive the hard link and hide the file from Jellyfin/Bazarr.
SIDECAR_MODE = 0o644
# ".generated" lets a real subtitle outrank a Whisper one; find_sidecar_subs and
# consolidate-watch.sh treat it as a tag beside ".en".
PLAIN_SIDECAR_SUFFIX = ".en.srt"
GENERATED_SIDECAR_SUFFIX = ".en.generated.srt"
# The subtitle for the English dub audio (consolidate-subs.py muxes it as "English (dub)").
DUB_SIDECAR_SUFFIX = ".en.dub.srt"


class ASRTimeout(Exception):
    """Transcription child exceeded `Config.asr_timeout_min` and was killed."""


class MissingJellyfinKey(Exception):
    """Transcription child exited 3 (no key); the parent always passes one, so it is a bug signal."""


@dataclass
class Config:
    jellyfin_url: str
    jellyfin_key: str
    age_hours: float
    budget_min: float
    pipeline_timeout_min: float
    max_wait_min: float
    asr_timeout_min: float
    whisper_model: str
    whisper_verify_model: str
    whisper_lid_model: str
    whisper_threads: int


# Past these a value is a typo that overflows on conversion (`* 60` -> inf, `timedelta`).
_MAX_MINUTES_SETTING = 1440.0
_MAX_AGE_HOURS = 8760.0


def load_config(env: dict[str, str]) -> Config:
    # The default must respect the bound `env_int` enforces on overrides.
    cpu_bound = os.cpu_count() or 64
    default_threads = min(6, cpu_bound)
    return Config(
        jellyfin_url=env.get("JELLYFIN_URL") or f"http://{MEDIA_LAN_IP}:8096",
        jellyfin_key=env.get("JELLYFIN_API_KEY", ""),
        age_hours=env_float(env, "SUB_FALLBACK_AGE_HOURS", 24.0, _MAX_AGE_HOURS, log=log),
        budget_min=env_float(env, "SUB_FALLBACK_BUDGET_MIN", 50.0, _MAX_MINUTES_SETTING, log=log),
        pipeline_timeout_min=env_float(
            env, "SUB_FALLBACK_PIPELINE_TIMEOUT_MIN", 60.0, _MAX_MINUTES_SETTING, log=log),
        max_wait_min=env_float(
            env, "SUB_FALLBACK_MAX_WAIT_MIN", 30.0, _MAX_MINUTES_SETTING, log=log),
        asr_timeout_min=env_float(
            env, "SUB_FALLBACK_ASR_TIMEOUT_MIN", 90.0, _MAX_MINUTES_SETTING, log=log),
        whisper_model=env.get("WHISPER_MODEL") or "medium.en",
        whisper_verify_model=env.get("WHISPER_VERIFY_MODEL") or "small.en",
        whisper_lid_model=env.get("WHISPER_LID_MODEL") or "small",
        whisper_threads=env_int(env, "WHISPER_THREADS", default_threads, cpu_bound, log=log),
    )


def log(msg: str) -> None:
    """Timestamped stdout line (cron redirects to var/log/sub-fallback.log)."""
    print(f"{time.strftime('%Y-%m-%d %H:%M:%S')} {msg}", flush=True)


def _utc_now_iso() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def _parse_iso(value: object) -> datetime | None:
    """Parse both formats `needs-english.json` holds: `isoformat()` (`+00:00`; 3.10's
    `fromisoformat` rejects `Z`) and the old `%Y-%m-%dT%H:%M:%SZ`. The file is never migrated."""
    if not isinstance(value, str) or not value:
        return None
    try:
        dt = datetime.fromisoformat(value)
        if dt.tzinfo is None:
            dt = dt.replace(tzinfo=timezone.utc)
        return dt
    except ValueError:
        pass
    try:
        return datetime.strptime(value, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc)
    except ValueError:
        return None


def load_needs_english(path: Path) -> dict | None:
    """Tracker dict, `{}` if absent, `None` if unreadable or not an object (main() exits 1).
    Not `load_state`, which maps "unreadable" to `{}`: a corrupt tracker must be loud."""
    if not path.exists():
        return {}
    try:
        data = json.loads(path.read_text())
    except Exception:
        return None
    return data if isinstance(data, dict) else None


def select_candidates(
    state: dict,
    now: datetime,
    age_hours: float,
    file_exists: Callable[[Path], bool],
    is_consolidated: Callable[[Path], bool],
) -> tuple[list[Path], list[str]]:
    """`(candidates, stale_keys)`: files waiting >= `age_hours` that exist and lack
    CONSOLIDATED_SUBS, oldest first; and gone/consolidated/malformed entries to drop."""
    aged: list[tuple[datetime, Path]] = []
    stale: list[str] = []
    for key, entry in state.items():
        path = Path(key)
        if not file_exists(path) or is_consolidated(path):
            stale.append(key)
            continue
        first_seen = _parse_iso(entry.get("first_seen")) if isinstance(entry, dict) else None
        if first_seen is None:
            stale.append(key)
            continue
        if now - first_seen >= timedelta(hours=age_hours):
            aged.append((first_seen, path))
    aged.sort(key=lambda t: t[0])
    return [p for _, p in aged], stale


def _has_consolidated_tag(path: Path) -> bool:
    info = probe(path)
    return bool(info) and already_processed(info)


def drop_stale_entries(stale_entries: dict[str, object]) -> int:
    """Remove a key only if its CURRENT entry still equals the snapshot that judged it
    stale (`consolidate-subs.py` may have upserted a FRESH one), compared inside
    `mutate_state`'s lock. Returns the number removed."""
    if not stale_entries:
        return 0
    removed_count = 0

    def _drop(state: dict) -> dict:
        nonlocal removed_count
        removed_count = 0
        for k, snapshot_entry in stale_entries.items():
            if state.get(k) == snapshot_entry:
                state.pop(k, None)
                removed_count += 1
        return state

    mutate_state(NEEDS_ENGLISH_STATE_FILE, _drop)
    return removed_count


def wait_while_playing(jellyfin: JellyfinClient, cfg: Config,
                       budget: PlaybackWaitBudget) -> None:
    """`media_stack.playback.wait_while_playing` bound to `cfg`."""
    _shared_wait_while_playing(jellyfin, cfg.jellyfin_key, cfg.max_wait_min, budget,
                               poll_s=PLAYBACK_POLL_S, log=log)


def _call_with_playback_pause(fn: Callable, jellyfin: JellyfinClient, cfg: Config,
                              budget: PlaybackWaitBudget, *args, **kwargs):
    """`media_stack.playback.call_with_playback_pause` bound to `cfg`."""
    return call_with_playback_pause(fn, jellyfin, cfg.jellyfin_key, cfg.max_wait_min, budget,
                                    *args, poll_s=PLAYBACK_POLL_S, log=log, **kwargs)


def _read_child_playback_wait_report(workdir: Path) -> float:
    """Seconds the child spent waiting, from its report file; missing or bad is `0.0`
    (bookkeeping must never fail a successful transcription)."""
    report = workdir / CHILD_PLAYBACK_WAIT_REPORT_FILENAME
    try:
        return float(report.read_text().strip())
    except (OSError, ValueError):
        return 0.0


def _wait_for_exit_unreaped(pid: int, timeout_s: float) -> bool:
    """Poll for `pid`'s exit WITHOUT reaping it (`os.waitid` + `WNOWAIT`): reaping frees
    the pid and process-group id for reuse, so a later `killpg` could hit an unrelated
    group. The zombie keeps the id reserved until the caller kills the group, then reaps.
    `True` if the wait timed out (caller must kill)."""
    t0 = time.monotonic()
    while True:
        try:
            exited = os.waitid(os.P_PID, pid, os.WEXITED | os.WNOHANG | os.WNOWAIT) is not None
        except ChildProcessError:
            exited = True
        if exited:
            return False
        if time.monotonic() - t0 >= timeout_s:
            return True
        time.sleep(SUBPROCESS_EXIT_POLL_S)


class _ChildOutcome(NamedTuple):
    returncode: int
    timed_out: bool
    stdout: str
    stderr: str


def _run_child_in_own_group(
    cmd: list[str], workdir: Path, role: str, timeout_s: float, *,
    cwd: str | None = None, env: dict[str, str] | None = None,
    after_exit: Callable[[], None] = lambda: None,
) -> _ChildOutcome:
    """Run `cmd` in its own session under a hard wall-clock timeout; never raises for the
    child's own status (the caller maps `returncode`).

    `start_new_session=True` puts the child and everything it spawns in one process group,
    which the `finally` ALWAYS kills (timeout, Ctrl-C, SIGTERM's `SystemExit`, clean exit).
    stdout/stderr go to FILES (`<role>.stdout`/`.stderr` in `workdir`), never PIPEs, and the
    wait uses `_wait_for_exit_unreaped`: kill the group FIRST, reap SECOND, leader unreaped
    between, which makes `killpg` safe against pid reuse. `after_exit` runs last in that
    `finally`, after the output is read, on EVERY exit path.
    """
    stdout_path = workdir / f"{role}.stdout"
    stderr_path = workdir / f"{role}.stderr"
    with open(stdout_path, "wb") as out_fh, open(stderr_path, "wb") as err_fh:
        proc = subprocess.Popen(
            cmd, cwd=cwd, env=env,
            stdout=out_fh, stderr=err_fh,
            start_new_session=True,
        )
    pgid = proc.pid  # new session: pgid == pid
    timed_out = False
    try:
        timed_out = _wait_for_exit_unreaped(pgid, timeout_s)
    finally:
        try:
            os.killpg(pgid, signal.SIGKILL)
        except ProcessLookupError:
            pass
        proc.wait()
        stdout = stdout_path.read_text(errors="replace")
        stderr = stderr_path.read_text(errors="replace")
        after_exit()
    return _ChildOutcome(proc.returncode, timed_out, stdout, stderr)


def _transcribe_in_subprocess(video: Path, audio_idx: int, cfg: Config,
                              budget: PlaybackWaitBudget, workdir: Path,
                              timeout_s: float) -> str:
    """Run the transcribe child (`TMPDIR=workdir`) under a hard wall-clock timeout.

    The child gets `cfg`'s key/URL and `SUB_FALLBACK_MAX_WAIT_MIN` = `budget`'s CURRENT
    remaining allowance; the seconds it spent waiting are deducted from `budget` on EVERY
    exit path. Only a timeout becomes `ASRTimeout`; exit 2 maps to `PlaybackWaitTimeout`,
    exit 3 to `MissingJellyfinKey`. Process-group discipline: `_run_child_in_own_group`.
    """
    cmd = [sys.executable, "-m", "media_stack.subverify", "transcribe",
           str(video), str(audio_idx), cfg.whisper_model, str(cfg.whisper_threads)]
    remaining_wait_min = max(0.0, budget.remaining_s) / 60.0
    env = {
        **os.environ,
        "TMPDIR": str(workdir),
        "JELLYFIN_API_KEY": cfg.jellyfin_key,
        "JELLYFIN_URL": cfg.jellyfin_url,
        "SUB_FALLBACK_MAX_WAIT_MIN": f"{remaining_wait_min:g}",
    }
    outcome = _run_child_in_own_group(
        cmd, workdir, "_transcribe", timeout_s, cwd=str(MEDIA_STACK_ROOT), env=env,
        after_exit=lambda: budget.spend(_read_child_playback_wait_report(workdir)))
    if outcome.timed_out:
        raise ASRTimeout(f"transcription exceeded {timeout_s:g}s: {video}")
    if outcome.returncode == 2:
        raise PlaybackWaitTimeout(
            f"transcription subprocess hit its own playback-wait bound: {video}")
    if outcome.returncode == 3:
        raise MissingJellyfinKey(
            f"transcription subprocess had no JELLYFIN_API_KEY: {video}")
    if outcome.returncode != 0:
        raise RuntimeError(
            f"transcribe subprocess failed ({video}): {(outcome.stderr or '').strip()[:300]}")
    return outcome.stdout


def _pipeline_lock_is_free(path: Path) -> bool:
    """The tag is written BEFORE the inline normalization that holds this lock, so the tag
    alone does not mean the pipeline is done."""
    with acquire_file_lock(path) as acquired:
        return acquired


def wait_for_pipeline(video: Path, timeout_s: float) -> tuple[bool, int]:
    """Poll until `video` has the CONSOLIDATED_SUBS tag AND its per-file lock is free
    (pipeline FINISHED). Returns `(reached, waited_seconds)`.

    Consolidation can replace an `.mp4` with a same-stem `.mkv`, so the `.mkv` is checked,
    but ONLY once `video` no longer exists: a `.mkv` beside a present `.mp4` is a sibling
    collision, and crediting its tag here would be as wrong as placing a sidecar for it.
    """
    mkv_alt = video.with_suffix(".mkv")
    accept_mkv_alt = mkv_alt != video
    t0 = time.monotonic()
    while time.monotonic() - t0 < timeout_s:
        info = probe(video)
        if info and already_processed(info) and _pipeline_lock_is_free(video):
            return True, int(time.monotonic() - t0)
        if accept_mkv_alt and not video.exists():
            info = probe(mkv_alt)
            if info and already_processed(info) and _pipeline_lock_is_free(mkv_alt):
                return True, int(time.monotonic() - t0)
        time.sleep(PIPELINE_POLL_S)
    return False, int(time.monotonic() - t0)


def _video_identity(video: Path) -> tuple[int, int, int, int, int] | None:
    """`(st_dev, st_ino, st_size, st_mtime_ns, st_ctime_ns)`; ctime exposes a same-size
    rewrite with mtime restored. `None` if gone."""
    try:
        st = video.stat()
    except OSError:
        return None
    return (st.st_dev, st.st_ino, st.st_size, st.st_mtime_ns, st.st_ctime_ns)


def _video_identity_sans_dev(video: Path) -> list[int] | None:
    """`_video_identity` minus `st_dev`, as a list, to compare with the tracker's stored
    identity, which means only "same release" (see `media_stack.fileid.identity_matches`)."""
    identity = _video_identity(video)
    return list(identity[1:]) if identity else None


# `consolidate-watch.sh` matches a sidecar's stem only against these (`.mkv` first).
_WATCHER_STEM_MATCH_EXTS = (".mkv", ".mp4")


def _sibling_video_collision(video: Path) -> Path | None:
    """A same-stem sibling with another of `_WATCHER_STEM_MATCH_EXTS` (`Episode.mp4`
    beside `Episode.mkv`), else `None`; no sidecar may be placed while one exists."""
    if video.suffix.lower() not in _WATCHER_STEM_MATCH_EXTS:
        return None
    stem = video.stem
    try:
        entries = list(video.parent.iterdir())
    except OSError:
        return None
    for entry in entries:
        if entry == video:
            continue
        suffix = entry.suffix.lower()
        if suffix not in _WATCHER_STEM_MATCH_EXTS or suffix == video.suffix.lower():
            continue
        if entry.stem != stem:
            continue
        try:
            if entry.is_file():
                return entry
        except OSError:
            continue
    return None


def _main_sidecars(video: Path) -> list[tuple[Path, bool, bool]]:
    """Non-forced `find_sidecar_subs(video)`; callers check `_has_forced_sidecar` first."""
    return [s for s in find_sidecar_subs(video) if not s[1]]


def _has_forced_sidecar(video: Path) -> bool:
    """ANY forced English sidecar exists. consolidate-subs.py owns those (content-gates,
    embeds, removes), so this script leaves such a file to it and never touches one."""
    return any(is_forced for _, is_forced, _ in find_sidecar_subs(video))


def _run_consolidate_subs_directly(video: Path, timeout_s: float, workdir: Path) -> tuple[bool, str]:
    """Run `consolidate-subs.py` as the watcher's inotify trigger would, when the
    `close_write` event could not be generated. Safe alongside a real watcher run (the
    per-file flock makes the loser a harmless SKIP).

    Same process-group discipline as `_transcribe_in_subprocess`
    (`_run_child_in_own_group`). Returns `(ran_ok, detail)`, informational only:
    `wait_for_pipeline` verifies the tag landed.
    """
    script = MEDIA_STACK_ROOT / "consolidate-subs.py"
    outcome = _run_child_in_own_group(
        [sys.executable, str(script), str(video)], workdir, "_consolidate_direct", timeout_s)
    if outcome.timed_out:
        return False, f"timed out after {timeout_s:g}s"
    detail = (outcome.stdout or outcome.stderr or "").strip()[-300:]
    return outcome.returncode == 0, detail


def _sidecar_dest(video: Path, source: str) -> Path:
    """`<stem>.en.generated.srt` for Whisper, `<stem>.en.dub.srt` for a dub transcript,
    `<stem>.en.srt` otherwise."""
    suffix = {"whisper": GENERATED_SIDECAR_SUFFIX,
              "dub": DUB_SIDECAR_SUFFIX}.get(source, PLAIN_SIDECAR_SUFFIX)
    return video.with_name(video.stem + suffix)


_SIDECAR_TEMP_SUFFIX = ".sfb-tmp"


def _sidecar_temp_glob(dest: Path) -> str:
    """Glob for every `_place_sidecar` temp name for `dest`."""
    return "." + dest.name + ".*" + _SIDECAR_TEMP_SUFFIX


def _cleanup_stale_sidecar_temps(dest: Path) -> None:
    """Unlink temps from a crashed attempt; safe even if hard-linked to `dest` (drops a link)."""
    try:
        for stale in dest.parent.glob(_sidecar_temp_glob(dest)):
            stale.unlink(missing_ok=True)
    except OSError:
        pass


def _place_sidecar(video: Path, dest: Path, text: str,
                   *, video_identity: tuple[int, int, int, int, int] | None,
                   pipeline_timeout_s: float, rec: dict, workdir: Path,
                   dub: bool = False) -> str | None:
    """Publish `text` as `dest` atomically, never exposing partial content or clobbering.

    On this mergerfs pool `renameat2(RENAME_NOREPLACE)` returns `EINVAL` but `os.link()`
    works and raises `EEXIST` on an existing target. So write the FULL text to a hidden,
    UNIQUE temp file in `dest`'s directory (not a subtitle extension, not starting with the
    stem, so neither `find_sidecar_subs` nor the watcher sees it), fsync, THEN `os.link()`.
    The name is unique per attempt: with a fixed one, a run dying after the link but before
    its `finally` leaves two names for one inode and the next `O_TRUNC` would truncate the
    published sidecar. Leftovers are swept first.

    Right before the link: re-probe (`ALREADY_CONSOLIDATED`), any forced sidecar
    (`FORCED_SIDECAR_PRESENT`), any main sidecar (`SIDECAR_APPEARED`), `video_identity`
    (`VIDEO_CHANGED`), same-stem sibling (`SIBLING_COLLISION`); any leaves `dest` untouched.
    With `dub` the file must instead BE consolidated (`NOT_CONSOLIDATED`) with no dub subtitle
    (`DUB_PRESENT`) and no dub sidecar (`SIDECAR_APPEARED`); the rest is the same.
    `FileExistsError` is the real no-clobber guarantee (`SIDECAR_APPEARED`); any other
    `OSError` is `PLACE_FAILED`, NEVER a fallback to a clobbering rename.

    After the link, `dest` is opened `O_WRONLY|O_APPEND` and closed unwritten: `os.link()`
    fires only `IN_CREATE`, so this generates the `close_write` the watcher reacts to. A failing
    nudge is recoverable: fall back to `_run_consolidate_subs_directly` (bounded by
    `pipeline_timeout_s`), record `rec["event_fallback_*"]`, still return `None`;
    `wait_for_pipeline` verifies the outcome.
    """
    _cleanup_stale_sidecar_temps(dest)
    tmp_fd, tmp_name = tempfile.mkstemp(
        prefix="." + dest.name + ".", suffix=_SIDECAR_TEMP_SUFFIX,
        dir=str(dest.parent))
    tmp = Path(tmp_name)
    try:
        with os.fdopen(tmp_fd, "w", encoding="utf-8") as fh:
            # fchmod is umask-independent.
            os.fchmod(fh.fileno(), SIDECAR_MODE)
            fh.write(text)
            fh.flush()
            os.fsync(fh.fileno())

        info = probe(video)
        if info is None:
            return "PROBE_FAILED"
        if dub:
            if not already_processed(info):
                return "NOT_CONSOLIDATED"
            if has_dub_subtitle(info):
                return "DUB_PRESENT"
            if find_dub_sidecars(video):
                return "SIDECAR_APPEARED"
        else:
            if already_processed(info):
                return "ALREADY_CONSOLIDATED"
            if _has_forced_sidecar(video):
                return "FORCED_SIDECAR_PRESENT"
            if _main_sidecars(video):
                return "SIDECAR_APPEARED"
        if _video_identity(video) != video_identity:
            return "VIDEO_CHANGED"
        if _sibling_video_collision(video) is not None:
            return "SIBLING_COLLISION"

        try:
            os.link(tmp, dest)
        except FileExistsError:
            return "SIDECAR_APPEARED"
        except OSError:
            return "PLACE_FAILED"

        try:
            append_fd = os.open(dest, os.O_WRONLY | os.O_APPEND)
            os.close(append_fd)
        except OSError as exc:
            rec["event_fallback_used"] = True
            rec["event_fallback_error"] = f"{type(exc).__name__}: {exc}"
            try:
                ran_ok, detail = _run_consolidate_subs_directly(video, pipeline_timeout_s, workdir)
            except Exception as helper_exc:
                # A failing recovery must not turn a published sidecar into ERROR.
                rec["event_fallback_helper_error"] = f"{type(helper_exc).__name__}: {helper_exc}"
            else:
                rec["event_fallback_pipeline_ran_ok"] = ran_ok
                rec["event_fallback_pipeline_detail"] = detail
        return None
    finally:
        tmp.unlink(missing_ok=True)


def _verify_untagged_audio_english(
    video: Path, audio_idx: int, duration: float, cfg: Config,
    jellyfin: JellyfinClient, budget: PlaybackWaitBudget, workdir: Path, rec: dict,
) -> bool:
    """Whisper LID (MULTILINGUAL `cfg.whisper_lid_model`) says an UNTAGGED/`und` stream is
    English; retries other windows while probability < `LID_PROBABILITY_FLOOR` (e.g. a
    music bed). Records the final language/probability in `rec`."""
    lang: str | None = None
    prob = 0.0
    for frac in LID_FRACTIONS:
        start = max(0.0, duration * frac)
        lang, prob = _call_with_playback_pause(
            detect_language, jellyfin, cfg, budget,
            video, audio_idx, start, LID_WINDOW_S,
            cfg.whisper_lid_model, cfg.whisper_threads, workdir,
        )
        if prob >= LID_PROBABILITY_FLOOR:
            break
    rec["audio_lid_language"] = lang
    rec["audio_lid_probability"] = round(prob, 3)
    return lang == "en" and prob >= LID_PROBABILITY_FLOOR


def _verify_subliminal_candidate(
    video: Path, audio_idx: int, srt_text: str, start: float, window: float,
    cfg: Config, jellyfin: JellyfinClient, budget: PlaybackWaitBudget,
    workdir: Path, rec: dict,
) -> bool:
    """Accept `srt_text` only if it scores English AND an ASR window overlaps it on BOTH
    ratios; a second, later window is tried when the first has too little speech.
    Records every number in `rec`."""
    lang = sublang_score(srt_text)
    rec["subliminal_lang_verdict"] = lang["verdict"]
    rec["subliminal_lang_eng_ratio"] = lang["eng"]
    if lang["verdict"] != "eng":
        return False

    asr_text = _call_with_playback_pause(
        asr_window, jellyfin, cfg, budget, video, audio_idx, start, window,
        cfg.whisper_verify_model, cfg.whisper_threads, workdir)
    word_ratio, bigram_ratio, nwords = overlap(asr_text, srt_text, start, window)
    if nwords < MIN_ASR_WORDS:
        start2 = start + window
        asr_text2 = _call_with_playback_pause(
            asr_window, jellyfin, cfg, budget, video, audio_idx, start2, window,
            cfg.whisper_verify_model, cfg.whisper_threads, workdir)
        word_ratio, bigram_ratio, nwords = overlap(asr_text2, srt_text, start2, window)
    rec["subliminal_word_ratio"] = round(word_ratio, 3)
    rec["subliminal_bigram_ratio"] = round(bigram_ratio, 3)
    rec["subliminal_asr_words"] = nwords
    if nwords < MIN_ASR_WORDS:
        rec["subliminal_reject_reason"] = "too little speech to verify"
        return False
    if bigram_ratio < MIN_BIGRAM_RATIO or word_ratio < MIN_WORD_RATIO:
        rec["subliminal_reject_reason"] = "wrong episode (low overlap)"
        return False
    return True


def _revalidate_candidate(video: Path, cfg: Config) -> str | None:
    """Re-apply `select_candidates`'s rules to the CURRENT tracker entry: earlier
    candidates take minutes, during which consolidate-subs.py may have cleared it (FIXED)
    or re-recorded it with a fresh `first_seen` (file replaced). `load_state_locked`
    avoids a torn read. `None` if still valid, else a `REVALIDATION_SKIP` reason."""
    try:
        state = load_state_locked(NEEDS_ENGLISH_STATE_FILE)
    except OSError as exc:
        return f"tracker unreadable: {exc}"
    entry = state.get(str(video.resolve()))
    if not isinstance(entry, dict):
        return "entry no longer present"
    first_seen = _parse_iso(entry.get("first_seen"))
    if first_seen is None:
        return "entry malformed (no first_seen)"
    if datetime.now(timezone.utc) - first_seen < timedelta(hours=cfg.age_hours):
        return "first_seen no longer past the grace period (entry was reset)"
    if not video.is_file():
        return "file no longer exists"
    if _has_consolidated_tag(video):
        return "already consolidated"
    stored_identity = entry.get("identity")
    if stored_identity is not None:
        current = _video_identity_sans_dev(video)
        # Tracker identity has three fields (ino, size, mtime_ns).
        if not identity_matches(stored_identity, current, allow_legacy=True):
            return "file identity no longer matches the tracked entry"
    return None


def process_candidate(video: Path, cfg: Config, jellyfin: JellyfinClient,
                      workdir: Path, budget_deadline: float) -> dict:
    """Give `video` a verified English sidecar. Raises only `PlaybackWaitTimeout`, which
    ends the run cleanly.

    `budget_deadline` is an ABSOLUTE `time.monotonic()` run deadline, checked when a
    transcription would actually start. The local `PlaybackWaitBudget` is per candidate,
    so `SUB_FALLBACK_MAX_WAIT_MIN` bounds their TOTAL.
    """
    rec: dict = {"path": str(video)}
    t0 = time.monotonic()
    try:
        revalidation_reason = _revalidate_candidate(video, cfg)
        if revalidation_reason is not None:
            rec["result"] = "REVALIDATION_SKIP"
            rec["reason"] = revalidation_reason
            return rec

        if _sibling_video_collision(video) is not None:
            rec["result"] = "SIBLING_COLLISION"
            return rec

        initial_identity = _video_identity(video)  # before the wait: a long wait is when a file gets replaced
        budget = PlaybackWaitBudget(cfg.max_wait_min)  # cumulative across every wait below
        wait_while_playing(jellyfin, cfg, budget)

        info = probe(video)
        if info is None:
            rec["result"] = "PROBE_FAILED"
            return rec
        if already_processed(info):
            rec["result"] = "ALREADY_CONSOLIDATED"
            return rec

        if _has_forced_sidecar(video):
            rec["result"] = "FORCED_SIDECAR_PRESENT"
            return rec

        if _main_sidecars(video):
            # Never touch an existing sidecar. If the watcher missed its close_write, the
            # 05:00 `consolidate-subs.py --scan` picks it up.
            rec["result"] = "SIDECAR_PRESENT"
            return rec

        duration = float((info.get("format") or {}).get("duration") or 0)

        audio_idx = english_audio_stream(info)
        if audio_idx is None:
            rec["result"] = "NO_ENGLISH_AUDIO"
            return rec

        if not is_audio_explicitly_english(info, audio_idx):
            if not _verify_untagged_audio_english(video, audio_idx, duration, cfg,
                                                  jellyfin, budget, workdir, rec):
                rec["result"] = "AUDIO_LANGUAGE_UNVERIFIED"
                return rec

        start = min(300.0, max(60.0, duration * 0.25))
        window = min(150.0, max(60.0, duration * 0.2))

        chosen_text: str | None = None
        source: str | None = None

        fetched = fetch_via_subliminal(video, workdir)
        if fetched is not None:
            txt = fetched.read_text(encoding="utf-8", errors="replace")
            if _verify_subliminal_candidate(video, audio_idx, txt, start, window,
                                            cfg, jellyfin, budget, workdir, rec):
                chosen_text, source = txt, "subliminal"
            fetched.unlink(missing_ok=True)

        if chosen_text is None:
            remaining_budget_s = budget_deadline - time.monotonic()
            if remaining_budget_s < DEFER_TRANSCRIPTION_IF_BUDGET_UNDER_S:
                rec["result"] = "DEFERRED_BUDGET"
                return rec
            if budget.remaining_s <= 0:
                # Defensive, normally unreachable: an exhausted allowance would make the child time out at once.
                raise PlaybackWaitTimeout(
                    f"cumulative playback wait for this candidate exceeded "
                    f"{cfg.max_wait_min:g} min")
            try:
                srt_text = _transcribe_in_subprocess(
                    video, audio_idx, cfg, budget, workdir, cfg.asr_timeout_min * 60)
            except ASRTimeout:
                rec["result"] = "ASR_TIMEOUT"
                return rec
            except MissingJellyfinKey as exc:
                rec["result"] = "MISSING_JELLYFIN_KEY"
                rec["error"] = str(exc)
                return rec
            cue_count = srt_text.count(" --> ")
            rec["whisper_cues"] = cue_count
            if cue_count >= MIN_WHISPER_CUES:
                chosen_text, source = srt_text, "whisper"
            else:
                rec["result"] = "TOO_FEW_CUES"
                return rec

        rec["source"] = source
        wait_while_playing(jellyfin, cfg, budget)  # ASR/subliminal take minutes; re-check before placing
        dest = _sidecar_dest(video, source)
        block_reason = _place_sidecar(video, dest, chosen_text,
                                      video_identity=initial_identity,
                                      pipeline_timeout_s=cfg.pipeline_timeout_min * 60,
                                      rec=rec, workdir=workdir)
        if block_reason is not None:
            rec["result"] = block_reason
            return rec

        reached, waited_s = wait_for_pipeline(video, cfg.pipeline_timeout_min * 60)
        rec["pipeline_reached"] = reached
        rec["pipeline_wait_s"] = waited_s
        rec["result"] = "FIXED" if reached else "PIPELINE_TIMEOUT"
        return rec
    except PlaybackWaitTimeout:
        raise
    except Exception as exc:
        rec["result"] = "ERROR"
        rec["error"] = f"{type(exc).__name__}: {exc}"
        return rec
    finally:
        rec["elapsed_s"] = round(time.monotonic() - t0, 1)


# --- --dub mode: a transcript of the English dub for dual-audio files ---------------------

# Settled by the file's own contents, so remembered per identity; a pending dub sidecar is not.
_DUB_SETTLED = frozenset({"NOT_CONSOLIDATED", "NOT_DUAL_AUDIO", "DUB_PRESENT"})
# Neither logged nor recorded in the jsonl. NOT_ANIME comes from Sonarr, not the file, so it is
# never remembered (a series type can change) but is cheap: it is judged before any probe.
_DUB_QUIET = _DUB_SETTLED | {"DUB_SIDECAR_PRESENT", "NOT_ANIME"}


def _dub_skip_reason(video: Path, info: dict) -> str | None:
    """Why `video` (its `probe()` result) needs no dub transcript, else None."""
    if not already_processed(info):
        return "NOT_CONSOLIDATED"
    if not is_dual_audio([s for s in info.get("streams", []) if s.get("codec_type") == "audio"]):
        return "NOT_DUAL_AUDIO"
    if has_dub_subtitle(info):
        return "DUB_PRESENT"
    if find_dub_sidecars(video):
        return "DUB_SIDECAR_PRESENT"
    return None


def _dub_files(media_root: Path) -> list[tuple[Path, list[int]]]:
    """`.mkv` files under `tv/` (a Radarr movie is never anime; hidden names skipped; only
    Matroska carries the pipeline tag) as `(path, [ino, size, mtime_ns])`, newest first."""
    found: list[tuple[Path, list[int]]] = []
    for dirpath, dirnames, filenames in os.walk(media_root / "tv"):
        dirnames[:] = [d for d in dirnames if not d.startswith(".")]
        for name in filenames:
            lower = name.lower()
            if name.startswith(".") or not lower.endswith(".mkv") or lower.endswith(".tmp.mkv"):
                continue
            path = Path(dirpath) / name
            try:
                st = path.stat()
            except OSError:
                continue
            found.append((path, [st.st_ino, st.st_size, st.st_mtime_ns]))
    found.sort(key=lambda t: t[1][2], reverse=True)
    return found


def wait_for_dub(video: Path, dest: Path, timeout_s: float) -> tuple[bool, int]:
    """Poll until consolidate-subs.py has taken (or quarantined) the sidecar `dest` AND
    released `video`'s lock. Returns `(reached, waited_seconds)`."""
    t0 = time.monotonic()
    while time.monotonic() - t0 < timeout_s:
        if not dest.exists() and _pipeline_lock_is_free(video):
            return True, int(time.monotonic() - t0)
        time.sleep(PIPELINE_POLL_S)
    return False, int(time.monotonic() - t0)


def process_dub_candidate(video: Path, cfg: Config, jellyfin: JellyfinClient,
                          workdir: Path, budget_deadline: float) -> dict:
    """Give the dual-audio `video` a Whisper transcript of its English audio as
    `<stem>.en.dub.srt`. `rec["result"]` is a `_dub_skip_reason` for a non-target, else the
    outcome. Raises only `PlaybackWaitTimeout`, which ends the run cleanly."""
    rec: dict = {"path": str(video), "mode": "dub"}
    t0 = time.monotonic()
    try:
        if _sibling_video_collision(video) is not None:
            rec["result"] = "SIBLING_COLLISION"
            return rec
        if not is_anime(video):
            rec["result"] = "NOT_ANIME"
            return rec
        initial_identity = _video_identity(video)
        budget = PlaybackWaitBudget(cfg.max_wait_min)
        wait_while_playing(jellyfin, cfg, budget)   # before ANY read of the file

        info = probe(video)
        if info is None:
            rec["result"] = "PROBE_FAILED"
            return rec
        reason = _dub_skip_reason(video, info)
        if reason is not None:
            rec["result"] = reason
            return rec
        # `is_dual_audio` guarantees an explicitly English stream, so no language-ID pass.
        audio_idx = english_audio_stream(info)
        if audio_idx is None:
            rec["result"] = "NO_ENGLISH_AUDIO"
            return rec

        if budget_deadline - time.monotonic() < DEFER_TRANSCRIPTION_IF_BUDGET_UNDER_S:
            rec["result"] = "DEFERRED_BUDGET"
            return rec
        try:
            srt_text = _transcribe_in_subprocess(
                video, audio_idx, cfg, budget, workdir, cfg.asr_timeout_min * 60)
        except ASRTimeout:
            rec["result"] = "ASR_TIMEOUT"
            return rec
        except MissingJellyfinKey as exc:
            rec["result"] = "MISSING_JELLYFIN_KEY"
            rec["error"] = str(exc)
            return rec
        rec["whisper_cues"] = srt_text.count(" --> ")
        if rec["whisper_cues"] < MIN_WHISPER_CUES:
            rec["result"] = "TOO_FEW_CUES"
            return rec

        wait_while_playing(jellyfin, cfg, budget)   # the transcript took a while
        dest = _sidecar_dest(video, "dub")
        block_reason = _place_sidecar(video, dest, srt_text,
                                      video_identity=initial_identity,
                                      pipeline_timeout_s=cfg.pipeline_timeout_min * 60,
                                      rec=rec, workdir=workdir, dub=True)
        if block_reason is not None:
            rec["result"] = block_reason
            return rec

        reached, waited_s = wait_for_dub(video, dest, cfg.pipeline_timeout_min * 60)
        rec["pipeline_reached"] = reached
        rec["pipeline_wait_s"] = waited_s
        final = probe(video) if reached else None
        rec["result"] = ("PIPELINE_TIMEOUT" if not reached
                         else "FIXED" if final and has_dub_subtitle(final) else "DUB_NOT_ADDED")
        return rec
    except PlaybackWaitTimeout:
        raise
    except Exception as exc:
        rec["result"] = "ERROR"
        rec["error"] = f"{type(exc).__name__}: {exc}"
        return rec
    finally:
        rec["elapsed_s"] = round(time.monotonic() - t0, 1)


def run_dub(cfg: Config, dry_run: bool) -> int:
    """The `--dub` run: walk the library newest first and give each target a dub transcript.
    Files settled as non-targets are remembered, so the walk stays cheap after the first run."""
    files = _dub_files(MEDIA_ROOT)
    checked = load_state(DUB_CHECKED_FILE)
    todo = [(video, ident) for video, ident in files if checked.get(str(video)) != ident]

    if dry_run:
        log(f"DRY RUN: {len(todo)} of {len(files)} file(s) not yet checked "
            f"(nothing transcribed, nothing written)")
        for video, _ident in todo:
            info = probe(video) if is_anime(video) else None
            if info is not None and _dub_skip_reason(video, info) is None:
                log(f"  would transcribe the English dub of: {video}")
        return 0

    jellyfin = JellyfinClient(cfg.jellyfin_url)
    WORKDIR_ROOT.mkdir(parents=True, exist_ok=True)
    budget_deadline = time.monotonic() + cfg.budget_min * 60
    settled: dict[str, list[int]] = {}
    handled = 0
    try:
        for video, ident in todo:
            if time.monotonic() >= budget_deadline:
                log(f"budget ({cfg.budget_min:g} min) reached; the rest waits for the next run")
                break
            with tempfile.TemporaryDirectory(prefix="sub-fallback_", dir=WORKDIR_ROOT) as workdir:
                rec = process_dub_candidate(video, cfg, jellyfin, Path(workdir), budget_deadline)
            result = rec.get("result")
            if result in _DUB_SETTLED:
                settled[str(video)] = ident
            if result in _DUB_QUIET:
                continue
            handled += 1
            rec["ts"] = _utc_now_iso()
            _append_jsonl(JSONL_FILE, rec)
            log(f"dub {result} ({rec.get('elapsed_s', '?')}s): {video}")
    except PlaybackWaitTimeout as exc:
        log(f"{exc}; ending run cleanly, the next hourly run retries")
        return 0
    finally:
        live = {str(video) for video, _ident in files}
        try:
            mutate_state(DUB_CHECKED_FILE, lambda cur: {
                k: v for k, v in {**cur, **settled}.items() if k in live})
        except OSError as exc:
            log(f"WARNING: could not save {DUB_CHECKED_FILE}: {exc}")
    log(f"done: {handled} dub candidate(s) handled this run")
    return 0


def _append_jsonl(path: Path, rec: dict) -> None:
    with path.open("a", encoding="utf-8") as fh:
        fh.write(json.dumps(rec) + "\n")


def main() -> int:
    signal.signal(signal.SIGTERM, _handle_sigterm)

    ap = argparse.ArgumentParser(
        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--dry-run", action="store_true",
                    help="list candidates and what would be attempted; fetch/mutate nothing")
    ap.add_argument("--dub", action="store_true",
                    help="instead: transcribe the English dub of consolidated dual-audio files "
                         "that lack a dub subtitle (writes <stem>.en.dub.srt)")
    args = ap.parse_args()

    lock_fp = acquire_singleton_lock(DUB_LOCK_FILE if args.dub else LOCK_FILE)
    if lock_fp is None:
        return 0  # another instance is running; exit quietly, not an error

    env = {**load_env_file(MEDIA_STACK_ROOT / ".env"), **os.environ}
    cfg = load_config(env)

    if cfg.whisper_lid_model.endswith(".en"):
        log(f"ERROR: WHISPER_LID_MODEL={cfg.whisper_lid_model!r} is an English-only model "
            f"and cannot detect language -- it would always report 'en' with high "
            f"confidence, silently defeating the whole point of verifying an untagged "
            f"audio stream. Use a multilingual model (e.g. 'small', 'base', 'medium').")
        if not args.dry_run:
            return 1

    if not args.dry_run and not cfg.jellyfin_key:
        log("ERROR: JELLYFIN_API_KEY is not set; a real run cannot fail closed on "
            "playback without it -- refusing to start blinded. --dry-run still works.")
        return 1

    if args.dub:
        return run_dub(cfg, args.dry_run)

    state = load_needs_english(NEEDS_ENGLISH_STATE_FILE)
    if state is None:
        log(f"ERROR: {NEEDS_ENGLISH_STATE_FILE} exists but could not be read/parsed; run blinded")
        return 1

    now = datetime.now(timezone.utc)
    candidates, stale = select_candidates(
        state, now, cfg.age_hours, lambda p: p.is_file(), _has_consolidated_tag)

    if args.dry_run:
        log(f"DRY RUN: {len(candidates)} candidate(s) due, {len(stale)} stale "
            f"entries would be dropped (nothing fetched, nothing written)")
        for p in candidates:
            log(f"  would attempt: {p}")
        return 0

    if stale:
        # `drop_stale_entries` skips keys re-recorded since the snapshot.
        removed = drop_stale_entries({k: state[k] for k in stale})
        kept = len(stale) - removed
        msg = f"dropped {removed} stale tracker entries (file gone or already consolidated)"
        if kept:
            msg += f"; {kept} re-recorded since the snapshot and kept"
        log(msg)

    if not candidates:
        log("nothing to do (0 candidates past the age threshold)")
        return 0

    jellyfin = JellyfinClient(cfg.jellyfin_url)
    WORKDIR_ROOT.mkdir(parents=True, exist_ok=True)
    budget_s = cfg.budget_min * 60
    t_run_start = time.monotonic()
    budget_deadline = t_run_start + budget_s
    started = 0

    for video in candidates:
        if time.monotonic() >= budget_deadline:
            log(f"budget ({cfg.budget_min:g} min) reached; "
                f"deferring {len(candidates) - started} remaining file(s) to the next run")
            break
        started += 1
        log(f"start ({started}/{len(candidates)}): {video}")
        try:
            with tempfile.TemporaryDirectory(prefix="sub-fallback_", dir=WORKDIR_ROOT) as workdir:
                rec = process_candidate(video, cfg, jellyfin, Path(workdir), budget_deadline)
        except PlaybackWaitTimeout as exc:
            log(f"{exc}; ending run cleanly, the next hourly run retries")
            return 0
        rec["ts"] = _utc_now_iso()
        _append_jsonl(JSONL_FILE, rec)
        log(f"{rec.get('result')} ({rec.get('source', '-')}, {rec.get('elapsed_s', '?')}s): {video}")

    log(f"done: {started}/{len(candidates)} file(s) started this run")
    return 0


if __name__ == "__main__":
    sys.exit(main())