#!/usr/bin/env python3
"""Consolidate English subtitles in a media file: pick the best text track, ffsubsync it,
clean it, and remux it as the only English subtitle (a real Forced track is kept; a file
keeping both Japanese and English audio, if anime, also gets an "English (dub)" subtitle).

Usage: consolidate-subs.py <file> | --scan <dir> [--jobs N]
Exit codes: 0 ok/skip, 1 hard fail.
"""
from __future__ import annotations
import argparse, hashlib, json, os, re, shutil, signal, subprocess, sys, tempfile, time
from typing import NamedTuple
from concurrent.futures import ProcessPoolExecutor, as_completed
from contextlib import nullcontext
from datetime import datetime, timezone
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
from media_stack.paths import (  # noqa: E402
    VAR_DIR, VAR_LOG, VAR_QUARANTINE_SUBTITLES, VAR_STATE, ensure_var_dirs,
)
ensure_var_dirs()
from media_stack.config import (  # noqa: E402
    ASS_CODECS, IMAGE_CODECS, PIPELINE_VERSION, NORMALIZE_TMP_DIR,
    SUBLIMINAL_MIN_COVERAGE, TEXT_CODECS, TMP_MKV_RE,
)
from media_stack.lang import is_english_tag  # noqa: E402
from media_stack.probe import (  # noqa: E402
    already_processed, duration_shortfall, file_key, parse_duration, probe,
    real_video_streams,
)
from media_stack.fileid import (  # noqa: E402
    Identity, file_identity, identity_from_stat, identity_matches,
)
from media_stack.state import load_state, mutate_state, update_state_entry  # noqa: E402
from media_stack.audio import (  # noqa: E402
    confident_default_audio, get_audio_lang_pref, is_anime, is_dual_audio, select_keep_audio,
)
from media_stack.arrlang import original_language_for  # noqa: E402
from media_stack.subtitles import (  # noqa: E402
    ass_to_srt, clean_srt_text, count_subtitle_lines, extract_track,
    decode_subtitle_bytes, decode_subtitle_text, fetch_via_subliminal, find_dub_sidecars,
    find_sidecar_subs, has_dub_subtitle, is_dub_track, is_forced_track, is_generated_sidecar,
    is_generated_track, is_sdh_track, run_ffsubsync, score_text_track, strip_hi_annotations,
)
from media_stack.sublang import classify_candidate  # noqa: E402
from media_stack.mux import (  # noqa: E402
    ENG_DUB_TRACK_NAME, ENG_GENERATED_TRACK_NAME, ENG_TRACK_NAME, add_dub_subtitle, remux,
)
from media_stack.loudness import normalize_budget_s  # noqa: E402
from media_stack.locking import acquire_file_lock  # noqa: E402
from media_stack.tags import set_consolidated_tag  # noqa: E402
from media_stack import sublang, subverify  # noqa: E402
from media_stack.subverify import (  # noqa: E402
    UPGRADE_MIN_PASSING_WINDOWS, generated_track_windows, judge_sidecar_against_windows,
)
from media_stack.sweeps import sweep_orphans as _sweep_orphans  # noqa: E402

STATE_FILE = VAR_STATE / "consolidate-subs.state.json"
LOG_FILE   = VAR_LOG / "consolidate-subs.log"
QUARANTINE_LOG = VAR_LOG / "sub-quarantine.jsonl"

# Separate from STATE_FILE: that cache is keyed on (size, mtime, v), which a landing
# sidecar changes neither of, so it would skip the file forever. Read by sub-fallback.py.
NEEDS_ENGLISH_STATE_FILE = VAR_STATE / "needs-english.json"


def sweep_orphans(root: Path, dry_run: bool = False) -> dict:
    return _sweep_orphans(root, script_basename=Path(__file__).name, dry_run=dry_run)


def _log_sweep_result(res: dict, dry: bool) -> None:
    prefix = "SWEEP-DRY" if dry else "SWEEP"
    for path, pid in res["deleted_files"]:
        log(f"{prefix}: tmp file (pid={pid}): {path}")
    for path in res["deleted_dirs"]:
        log(f"{prefix}: workdir: {path}")
    for kind, path, info in res["skipped"]:
        log(f"{prefix}-SKIP {kind}: {path} ({info})")
    log(f"{prefix} summary: {len(res['deleted_files'])} files, {len(res['deleted_dirs'])} dirs, {len(res['skipped'])} skipped")


def log(msg: str) -> None:
    line = f"{time.strftime('%Y-%m-%d %H:%M:%S')} {msg}"
    print(line, flush=True)
    try:
        with LOG_FILE.open("a") as fh:
            fh.write(line + "\n")
    except Exception:
        pass


def _utc_now_iso() -> str:
    """Microsecond precision: whole seconds would defeat sub-fallback's snapshot-vs-current check."""
    return datetime.now(timezone.utc).isoformat(timespec="microseconds")


def _needs_english_safe(op_name: str, path_for_log: Path, fn) -> None:
    """Swallow any failure: this bookkeeping must never turn a real result into FAIL."""
    try:
        fn()
    except Exception as exc:
        log(f"WARNING: needs-english.json {op_name} failed for {path_for_log} "
            f"({type(exc).__name__}: {exc}); consolidate result unaffected")


def _file_identity(path: Path) -> list[int] | None:
    """`[st_ino, st_size, st_mtime_ns]`. No ctime (unlike media_stack.fileid): this means
    "same release", and a hardlink change would needlessly reset the grace period."""
    try:
        st = path.stat()
    except OSError:
        return None
    return [st.st_ino, st.st_size, st.st_mtime_ns]


def _clear_needs_english(path: Path, dry_run: bool) -> None:
    """Drop `path`'s needs-english entry once FIXED."""
    if dry_run:
        return
    key = str(path.resolve())

    def _drop(state: dict) -> dict:
        state.pop(key, None)
        return state

    _needs_english_safe("clear", path, lambda: mutate_state(NEEDS_ENGLISH_STATE_FILE, _drop))


def _apply_cleanup_needs_english_update(
    path: Path, pre_identity: list[int] | None, result_path: Path, dry_run: bool,
) -> None:
    """Record NEEDS_BAZARR after the cleanup remux. `first_seen` resets only if the entry
    at `path` stores an identity other than `pre_identity` (a new release replaced the
    file); the entry moves to `result_path` (the suffix may change)."""
    if dry_run:
        return
    old_key = str(path.resolve())
    new_key = str(result_path.resolve())
    now = _utc_now_iso()
    current_identity = _file_identity(result_path)

    def _update(state: dict) -> dict:
        old_entry = state.pop(old_key, None)
        first_seen = None
        if isinstance(old_entry, dict):
            stored_identity = old_entry.get("identity")
            if stored_identity is None or stored_identity == pre_identity:
                first_seen = old_entry.get("first_seen")
        state[new_key] = {
            "first_seen": first_seen or now,
            "last_seen": now,
            "identity": current_identity,
        }
        return state

    _needs_english_safe(
        "cleanup-upsert", result_path, lambda: mutate_state(NEEDS_ENGLISH_STATE_FILE, _update))


def process_file(path_str: str, dry_run: bool = False) -> dict:
    """Returns {path, status: SKIP/FIXED/NEEDS_BAZARR/FAIL, detail}; never raises. dry_run
    mutates nothing (a would-be FIXED returns SKIP with detail prefix "dry-run:")."""
    try:
        return _process_file_inner(path_str, dry_run=dry_run)
    except Exception as e:
        return {"path": path_str, "status": "FAIL",
                "detail": f"unhandled: {type(e).__name__}: {str(e)[:120]}"}


def _process_file_inner(path_str: str, dry_run: bool = False) -> dict:
    path = Path(path_str)
    if not path.is_file():
        return {"path": path_str, "status": "FAIL", "detail": "not a file"}
    if TMP_MKV_RE.search(path.name):
        return {"path": path_str, "status": "SKIP",
                "detail": "tmp file (orphan candidate; run --sweep-only to clean)"}
    # Single-file entry points bypass walk()'s prune. Check BEFORE acquire_file_lock (it
    # creates the .lock file). Resolve first (a relative path inside the workdir would
    # slip past `parts`); on failure REFUSE, since the unverified spelling fails open.
    try:
        _resolved = path.resolve()
    except OSError as e:
        return {"path": path_str, "status": "FAIL",
                "detail": f"cannot resolve path: {e}"}
    if NORMALIZE_TMP_DIR in _resolved.parts:
        return {"path": path_str, "status": "SKIP",
                "detail": "normalize-audio workdir intermediate"}

    with acquire_file_lock(path) as acquired:
        if not acquired:
            return {"path": path_str, "status": "SKIP",
                    "detail": "locked by other runner"}
        return _process_locked(path, path_str, dry_run=dry_run)


def _kill_process_group(proc: subprocess.Popen, grace_s: float = 15.0) -> None:
    """Kill `proc`'s group (start_new_session=True, so pid == pgid): SIGTERM, `grace_s` for
    EVERY member (a grandchild may ignore it), then SIGKILL. The leader is reaped as soon
    as it exits (a zombie still counts as a member). Never raises."""
    pgid = proc.pid
    try:
        os.killpg(pgid, signal.SIGTERM)
        deadline = time.monotonic() + grace_s
        while time.monotonic() < deadline:
            proc.poll()
            os.killpg(pgid, 0)   # ProcessLookupError once the group is empty
            time.sleep(0.1)
        os.killpg(pgid, signal.SIGKILL)
    except OSError:
        pass  # group gone (ProcessLookupError) or no longer signalable
    try:
        proc.communicate(timeout=15)
    except subprocess.TimeoutExpired:
        pass


def _raise_on_sigterm(signum, frame) -> None:
    raise SystemExit(128 + signum)


def _normalize_audio_inline(path: Path, locked_path: Path | None = None) -> None:
    """Run normalize-audio.py in a subprocess (its process pool and locks must not mix with
    ours). `locked_path` becomes NORMALIZE_INHERIT_LOCK_PATH so the child inherits our lock.
    Errors never propagate: consolidation is already committed. The cap is
    `normalize_budget_s(size)`; on expiry or interrupt the child's whole process group is
    killed, leaving no orphan ffmpeg."""
    script = Path(__file__).parent / "normalize-audio.py"
    env = {**os.environ}
    if locked_path is not None:
        env["NORMALIZE_INHERIT_LOCK_PATH"] = str(locked_path)
    try:
        budget_s = normalize_budget_s(path.stat().st_size)
    except OSError:
        budget_s = 7200
    try:
        proc = None
        try:   # else SIGTERM to us leaves the child (own session) running
            previous = signal.signal(signal.SIGTERM, _raise_on_sigterm)
        except ValueError:   # not the main thread
            previous = None
        try:
            proc = subprocess.Popen(
                [sys.executable, str(script), str(path)],
                stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env=env,
                start_new_session=True,
            )
            stdout, stderr = proc.communicate(timeout=budget_s)
        except subprocess.TimeoutExpired:
            _kill_process_group(proc)
            log(f"audio-normalize TIMEOUT ({_hms(budget_s)} budget), process group killed: {path}")
            return
        except BaseException:
            if proc is not None:
                _kill_process_group(proc)
            raise
        finally:
            if previous is not None:
                signal.signal(signal.SIGTERM, previous)
        # rc=0 is not enough: a "locked" SKIP also exits 0 and is a no-op.
        stdout_tail = (stdout or "").strip().splitlines()[-1:] or [""]
        last = stdout_tail[0]
        if proc.returncode == 0 and "locked by other pipeline" in last:
            log(f"audio-normalize SKIPPED (lock conflict): {path} — {last[:160]}")
        elif proc.returncode == 0:
            log(f"audio-normalize OK: {path}")
        else:
            tail = (stderr or "").strip().splitlines()[-1:] or [""]
            log(f"audio-normalize FAIL rc={proc.returncode}: {path} -- {tail[0][:200]}")
    except Exception as e:
        log(f"audio-normalize ERR: {path} -- {type(e).__name__}: {e}")


def _expose_tag_normalize(out: Path, path: Path, replacement: Path,
                          path_str: str, tag: bool = True,
                          normalize: bool = True) -> tuple[dict | None, bool]:
    """Swap the muxed `out` into `replacement`, stamp CONSOLIDATED_SUBS, optionally run
    inline normalization, all under the destination .mkv's lock (on .mp4 -> .mkv `path`'s
    lock does not cover the new file).

    The no-main-English cleanup passes `tag=False, normalize=False`: it must not mark the
    file consolidated, and normalize-audio could re-encode audio or let ffmpeg auto-flag a
    default, bypassing `confident_default_audio`.

    Returns `(early_result, tag_written)`; `early_result` is a SKIP (destination locked) or
    FAIL (sibling .mkv appeared, or `out` is shorter than the source) dict for the caller to
    return, else None. A failed tag is
    logged, not rolled back (state.json backstops idempotency).
    """
    dest_lock = (acquire_file_lock(replacement) if replacement != path
                 else nullcontext(True))
    with dest_lock as dest_acquired:
        if replacement != path and not dest_acquired:
            out.unlink(missing_ok=True)
            return ({"path": path_str, "status": "SKIP",
                     "detail": f"destination {replacement.name} locked by other runner"},
                    False)
        # Re-check under the lock (TOCTOU with _process_locked's check).
        if replacement != path and replacement.exists():
            out.unlink(missing_ok=True)
            return ({"path": path_str, "status": "FAIL",
                     "detail": f"collision: sibling .mkv appeared for {path.name}"},
                    False)
        # A short remux output once replaced full episodes: compare durations before the swap.
        problem = duration_shortfall(probe(path), probe(out), log)
        if problem:
            out.unlink(missing_ok=True)
            return ({"path": path_str, "status": "FAIL",
                     "detail": f"duration verify: {problem}"}, False)
        # atomic same-filesystem rename: readers see the old file or the finished mux, never a partial one
        os.replace(out, replacement)
        if replacement != path and path.exists():
            path.unlink()
        tag_written = False
        if tag:
            # An uncaught raise would skip the state update and sidecar cleanup.
            try:
                tag_written = set_consolidated_tag(replacement, PIPELINE_VERSION)
            except Exception as exc:
                log(f"ERROR: CONSOLIDATED_SUBS tag write raised for {replacement}: "
                    f"{type(exc).__name__}: {exc} (consolidation done; tag missing)")
            else:
                if not tag_written:
                    log(f"ERROR: CONSOLIDATED_SUBS tag write failed for {replacement} "
                        f"(consolidation done + state recorded; MKV idempotency marker missing)")
        if normalize:
            _normalize_audio_inline(replacement, locked_path=replacement)
        return (None, tag_written)


# "Positively foreign": base subtag is a 2-3 letter ASCII code that is not English, not
# und/mul/mis/zxx, not in the qaa-qtz private-use range. Anything else is KEPT.
# ASCII-only: str.isalpha() would accept e.g. "éé".
_SUBTITLE_SPECIAL_LANG_CODES = frozenset({"und", "mul", "mis", "zxx"})
_SUBTITLE_LANG_BASE_RE = re.compile(r"[a-z]{2,3}")


def _is_positively_foreign_sub(s: dict) -> bool:
    """The cleanup drops only these."""
    lang = ((s.get("tags") or {}).get("language") or "").strip().lower()
    base = lang.split("-", 1)[0]
    if not _SUBTITLE_LANG_BASE_RE.fullmatch(base):
        return False
    if base in _SUBTITLE_SPECIAL_LANG_CODES:
        return False
    if "qaa" <= base <= "qtz":
        return False
    return not is_english_tag(lang)


# Content gate: an English tag or filename is a claim, not proof, so `classify_candidate`
# checks the text. A rejected EMBEDDED candidate is never chosen; a rejected SIDECAR is
# moved to VAR_QUARANTINE_SUBTITLES, never deleted.

_QUARANTINE_NAME_TRIES = 1000


class _Unclassifiable(OSError):
    """Could not classify (over the bound, changing while read, or not text). An OSError so
    candidate loops count it as an extraction failure; never used on its tag or name."""


class _ForcedGate(NamedTuple):
    state: str            # "accepted" | "rejected" | "unreadable"
    text: "str | None"    # cleaned SRT to embed
    info: dict


def _gate_embedded_forced(path: Path, forced: dict) -> _ForcedGate:
    """Gate the embedded forced English track BEFORE any decision depends on it (a foreign
    track tagged English must not make the file "only forced English"); the gate sees the
    bytes before ASS/SSA conversion.

      - accepted: `text` is the cleaned SRT; None if conversion failed.
      - rejected: proven foreign; dropped.
      - unreadable: proves nothing; track is kept.
    """
    idx = forced["index"]
    scratch = Path(tempfile.mkdtemp(prefix=f"consolsub_{os.getpid()}_forced_", dir=path.parent))
    try:
        ext = ".ass" if forced.get("codec_name") in ASS_CODECS else ".srt"
        ftmp = _workfile(scratch, "forced", ext)
        if not extract_track(path, idx, forced.get("codec_name"), ftmp):
            log(f"embedded forced track {idx} in {path} could not be extracted; "
                f"not counted as forced English")
            return _ForcedGate("unreadable", None, {})
        try:
            ok, info = _gate_local_file(ftmp)
        except _Unclassifiable as exc:
            log(f"content-gate SKIP embedded forced track {idx} in {path}: cannot be "
                f"classified ({exc}); not used as the forced track")
            return _ForcedGate("unreadable", None, {})
        except OSError as exc:
            log(f"WARNING: cannot read embedded forced track {idx} in {path}: "
                f"{type(exc).__name__}: {exc}")
            return _ForcedGate("unreadable", None, {})
        if not ok:
            log(f"content-gate REJECT embedded forced track {idx} in {path}: tagged English, "
                f"text scored {info.get('verdict')} (reason={info.get('reason')}) "
                f"-- proven foreign, not used as the forced track")
            return _ForcedGate("rejected", None, info)
        try:
            if ftmp.suffix == ".ass":
                fsrt = _workfile(scratch, "forced-srt", ".srt")
                if not ass_to_srt(ftmp, fsrt):
                    log(f"embedded forced track {idx} in {path} passed the content gate but "
                        f"could not be converted to SRT; not used as the forced track")
                    return _ForcedGate("accepted", None, info)
                ftmp = fsrt
            return _ForcedGate(
                "accepted", clean_srt_text(ftmp.read_text(encoding="utf-8", errors="replace")), info)
        except OSError as exc:
            log(f"WARNING: embedded forced track {idx} in {path} could not be prepared: "
                f"{type(exc).__name__}: {exc}; not used as the forced track")
            return _ForcedGate("accepted", None, info)
    finally:
        shutil.rmtree(scratch, ignore_errors=True)


def _snapshot_text(p: Path) -> "tuple[_Snapshot, str]":
    """Raises `_Unclassifiable` if over the size bound, changing while read, or not text."""
    snap = _read_snapshot(p)
    if snap is None:
        raise _Unclassifiable(f"{p.name}: over the {UPGRADE_MAX_SIDECAR_BYTES}-byte bound "
                              f"or changing while it was read")
    text = decode_subtitle_text(snap.data)
    if text is None:
        raise _Unclassifiable(f"{p.name}: not decodable as text")
    return snap, text


def _gate_local_file(p: Path) -> tuple[bool, dict]:
    """Content gate for a scratch file: `(is_english, info)`. An accepted file is rewritten
    as DECODED UTF-8 so UTF-16/cp1252 never reaches clean, ffsubsync or mkvmerge. Raises
    `_Unclassifiable`."""
    _snap, text = _snapshot_text(p)
    is_eng, info = classify_candidate(text)
    if is_eng:
        p.write_bytes(text.encode("utf-8"))
    return is_eng, info


def _last_cue_start(srt: Path) -> float | None:
    """Start (s) of the latest cue in a UTF-8 SRT; None if none or unreadable."""
    try:
        cues = subverify._sidecar_cues(srt.read_text(encoding="utf-8", errors="replace"))
    except OSError:
        return None
    return max((start for start, _ in cues), default=None)


def _hms(seconds: float) -> str:
    s = int(seconds)
    return f"{s // 3600}:{s % 3600 // 60:02d}:{s % 60:02d}"


def _late_sidecar_detail(base: str, foreign: int, redundant: int, dry_run: bool) -> str:
    verb = "dry-run: would quarantine" if dry_run else "quarantined"
    parts = []
    if foreign:
        parts.append(f"{verb} {foreign} sidecar(s)")
    if redundant:
        parts.append(f"{verb} {redundant} redundant sidecar(s)")
    return "; ".join([base, *parts])


def _quarantine_stamp() -> str:
    """UTC timestamp without colons, safe in a filename."""
    return datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S.%fZ")


def _log_quarantine(original: Path, dest: Path, verdict_info: dict, reason: str) -> None:
    """Append to QUARANTINE_LOG. Never raises: must not undo a completed move."""
    rec = {
        "timestamp": _utc_now_iso(),
        "original_path": str(original),
        "quarantine_path": str(dest),
        "verdict": verdict_info.get("verdict"),
        "scores": verdict_info,
        "reason": reason,
    }
    try:
        QUARANTINE_LOG.parent.mkdir(parents=True, exist_ok=True)
        with QUARANTINE_LOG.open("a") as fh:
            fh.write(json.dumps(rec) + "\n")
    except Exception as exc:
        log(f"WARNING: sub-quarantine.jsonl write failed for {dest}: "
            f"{type(exc).__name__}: {exc}")


def _quarantine_fsync_chain(quarantine_dir: Path) -> list[Path]:
    """Directories to fsync before the original goes: the copy's and each ancestor up to
    `var/`, on EVERY move (finding a directory existing proves nothing about disk). A
    quarantine dir outside `VAR_DIR` is assumed to use `var/quarantine/subtitles`."""
    top = VAR_DIR if VAR_DIR in quarantine_dir.parents else quarantine_dir.parent.parent
    chain = [quarantine_dir]
    p = quarantine_dir
    while p != top and p.parent != p:
        p = p.parent
        chain.append(p)
    return chain


def _fsync_dir(path: Path) -> None:
    fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY)
    try:
        os.fsync(fd)
    finally:
        os.close(fd)


def _quarantine_sidecar(sidecar: Path, snap: _Snapshot, verdict_info: dict, reason: str) -> bool:
    """Move a content-rejected sidecar into VAR_QUARANTINE_SUBTITLES instead of deleting it
    (`classify_candidate` is a lexical heuristic).

    The copy is written from `snap.data` (the bytes classified, no re-read) with
    O_CREAT|O_EXCL (new suffix on collision; an existing file is never touched), fsynced and
    size-checked, then the directory chain is fsynced, and only then is the original
    unlinked, and only if it still has `snap.identity` (a sidecar replaced meanwhile by
    Bazarr or sub-fallback.py survives and our copy is removed). Nothing is renamed: the
    library and `var/` are different filesystems. Any failure is logged, removes our partial
    copy, keeps the original and never raises.

    Returns True iff the sidecar was moved (a failed jsonl write still returns True).
    """
    dest: Path | None = None
    try:
        VAR_QUARANTINE_SUBTITLES.mkdir(parents=True, exist_ok=True)
        stamp = _quarantine_stamp()
        fd = -1
        for n in range(_QUARANTINE_NAME_TRIES):
            name = f"{stamp}-{sidecar.name}" if n == 0 else f"{stamp}-{n}-{sidecar.name}"
            candidate = VAR_QUARANTINE_SUBTITLES / name
            try:
                fd = os.open(candidate, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o660)
            except FileExistsError:
                continue
            dest = candidate
            break
        if dest is None:
            raise OSError(f"no free quarantine name after {_QUARANTINE_NAME_TRIES} tries")
        with os.fdopen(fd, "wb") as fh:
            fh.write(snap.data)
            fh.flush()
            os.fsync(fh.fileno())
            written = os.fstat(fh.fileno()).st_size
        if written != len(snap.data):
            raise OSError(f"quarantine copy is {written} bytes, expected {len(snap.data)}")
        for d in _quarantine_fsync_chain(VAR_QUARANTINE_SUBTITLES):
            _fsync_dir(d)
        now = file_identity(sidecar)
        if now is None:
            raise OSError("sidecar vanished before it could be removed")
        if now != snap.identity:
            raise OSError("sidecar changed after it was classified "
                          "(replaced or modified); left in place")
        sidecar.unlink()
    except Exception as exc:
        log(f"WARNING: quarantine failed for {sidecar}: {type(exc).__name__}: {exc}; "
            f"original left in place")
        if dest is not None:
            try:
                dest.unlink(missing_ok=True)
            except Exception:
                pass
        return False
    _log_quarantine(sidecar, dest, verdict_info, reason)
    return True


def _gate_forced_sidecars(sidecars: list, accepted: dict,
                          unclassified: "list[Path] | None" = None) -> list:
    """FULL-path content gate for FORCED English-named sidecars (a bogus one would block
    `sub-fallback.py` and show in Jellyfin).

    A foreign one is quarantined and dropped from the returned list (also when the
    quarantine failed). An accepted one (short forced files score TOO-LITTLE-TEXT, so only
    clearly foreign ones are rejected) is recorded in `accepted` as `path -> snapshot`;
    FIXED may embed it and only then counts it consumed. An unclassifiable one is left on
    disk and appended to `unclassified` for the caller to report as an extraction failure."""
    kept = []
    for entry in sidecars:
        sc_path, is_forced = entry[0], entry[1]
        if not is_forced:
            kept.append(entry)
            continue
        try:
            snap = _read_snapshot(sc_path)
        except OSError as exc:
            log(f"WARNING: cannot read forced sidecar {sc_path}: {type(exc).__name__}: {exc}; "
                f"left in place, not counted as forced English")
            if unclassified is not None:
                unclassified.append(sc_path)
            continue
        text = None if snap is None else decode_subtitle_text(snap.data)
        if snap is None or text is None:
            log(f"content-gate SKIP forced sidecar {sc_path}: cannot be classified "
                f"({'over the size bound or changing while it was read' if snap is None else 'not decodable as text'}); "
                f"left in place, not counted as forced English")
            if unclassified is not None:
                unclassified.append(sc_path)
            continue
        is_eng, verdict_info = classify_candidate(text)
        if is_eng:
            accepted[sc_path] = snap
            kept.append(entry)
            continue
        log(f"content-gate REJECT forced sidecar {sc_path}: named English, text scored "
            f"{verdict_info.get('verdict')} (reason={verdict_info.get('reason')}) -- quarantining")
        # Out of every decision whether or not the move succeeds.
        if not _quarantine_sidecar(sc_path, snap, verdict_info,
                                   "forced sidecar rejected during consolidation"):
            log(f"content-gate: rejected forced sidecar {sc_path} could not be quarantined; "
                f"left on disk, not counted as forced English")
    return kept


_REDUNDANT_REASON = "redundant: embedded English subtitle present"
_LATE_FOREIGN_REASON = "late sidecar beside already-processed file"


def _embedded_english_text_tracks(info: dict | None) -> list[dict]:
    """Embedded TEXT subtitle streams tagged English, except dub ones (they follow the dub
    audio and never stand in for a main track)."""
    return [s for s in (info or {}).get("streams", [])
            if s.get("codec_type") == "subtitle"
            and s.get("codec_name") in TEXT_CODECS
            and is_english_tag((s.get("tags") or {}).get("language"))
            and not is_dub_track(s)]


def _eligible_embedded(tracks: list[dict], is_forced: bool) -> list[dict]:
    """A MAIN sidecar is retired only by a non-forced track, a FORCED one only by a forced
    track (a main track does not cover signs and foreign speech)."""
    return [t for t in tracks if is_forced_track(t) == is_forced]


def _late_english_sidecar_verdict(media: Path, sc_path: Path, is_forced: bool,
                                  tracks: list[dict]) -> str:
    """"redundant" (quarantine), "keep" (the file's only English subtitle), or "upgrade"
    (the only embedded English is a generated Whisper transcript: left to the generated ->
    real upgrade check). Tracks are by TAG; the caller verifies content."""
    eligible = _eligible_embedded(tracks, is_forced)
    if not eligible:
        return "keep"
    # Whisper never makes a forced track, so "upgrade" is main-only.
    if (not is_forced and not is_generated_sidecar(media, sc_path)
            and all(is_generated_track(t) for t in eligible)):
        return "upgrade"
    return "redundant"


_MAX_EMBEDDED_CONTENT_CHECKS = 3
# (start_s, duration_s): ~10 minutes from 5 minutes in (skips intros), else from the start.
_EMBEDDED_SAMPLE_WINDOWS = ((300.0, 600.0), (0.0, 600.0))


def _embedded_track_is_english(path: Path, track: dict, forced: bool = False) -> bool:
    """Does an embedded track tagged English POSITIVELY look like English? Must score `eng`
    (the content gate's benefit of the doubt is not enough: this retires a sidecar); a
    FORCED track may also be too short to score. Unreadable or empty is False (the caller
    keeps the sidecar)."""
    idx = track.get("index")
    text = None
    for start, dur in _EMBEDDED_SAMPLE_WINDOWS:
        try:
            text = subverify.embedded_text_window(
                path, idx, track.get("codec_name") or "", start, dur)
        except Exception as exc:
            log(f"WARNING: cannot sample embedded subtitle {idx} of {path}: "
                f"{type(exc).__name__}: {exc}")
            return False
        if text and text.strip():
            break
    if not text or not text.strip():
        log(f"embedded subtitle {idx} of {path} could not be read for verification")
        return False
    verdict = sublang.score(text)["verdict"]
    if verdict == "eng" or (forced and verdict == "TOO-LITTLE-TEXT"):
        return True
    log(f"embedded subtitle {idx} of {path} is tagged English but its text is not positive "
        f"English evidence (language check: {verdict})")
    return False


def _upgrade_rejection_reasons(path: Path, snaps: dict[Path, _Snapshot | None],
                               sidecars: list, record: object) -> dict[str, str] | None:
    """The upgrade check's decision on the non-generated main English sidecars, from its
    rejection `record`. None if no record still holds for THIS video and THESE sidecars;
    else `{name: reason}` for those judged wrong (possibly empty: a sidecar with no verdict
    must never be given up). A verdict holds per sidecar while its fingerprint and the
    video's identity match. A valid record means generated-only, so no probe is needed."""
    video_identity = file_identity(path)
    if video_identity is None:
        return None
    real = [sc for sc, forced, _sdh in sidecars
            if not forced and not is_generated_sidecar(path, sc)]
    known = {sc: snaps[sc] for sc in real if snaps.get(sc) is not None}
    verdicts = _recorded_verdicts(record, video_identity, known)
    if verdicts is None:
        return None
    return {name: why for name, why in verdicts.items() if why}


_UNSET: object = object()  # no baseline taken (None: tried and failed)


def _quarantine_foreign_sidecars(path: Path, dry_run: bool, info: dict | None = None,
                                 upgrade_rejection: object = None,
                                 video_identity: object = _UNSET) -> tuple[int, int]:
    """Late-sidecar sweep for an ALREADY-PROCESSED video, run after
    `_generated_upgrade_check` declined to upgrade (a late sidecar changes neither the
    cache key nor the tag, so it would otherwise sit there forever). Quarantines, never
    deletes:
      - sidecars that score non-English ("foreign");
      - English sidecars made REDUNDANT by an embedded English text subtitle whose TEXT
        (a ~10-minute sample, not just its tag) is verified English;
      - a non-generated sidecar beside a generated-only file once the upgrade check
        REJECTED it for this exact video and sidecar (`upgrade_rejection`), reason
        "rejected upgrade: <why>". Without that rejection it is kept and the check retries.
    An English sidecar is kept when the video has no embedded English text subtitle, the
    probe fails, or the embedded text is non-English or unreadable.
    Returns `(foreign, redundant)` counts (what WOULD be, in `dry_run`).

    `video_identity` (taken before the caller probed, else here) is re-checked before EACH
    move: on change the sweep stops, keeping everything not yet moved. If the caller's
    baseline could not be read (None) nothing is moved.
    """
    try:
        sidecars = find_sidecar_subs(path)
    except OSError as exc:
        log(f"WARNING: late-sidecar scan failed for {path}: {type(exc).__name__}: {exc}")
        return 0, 0
    if video_identity is _UNSET:
        video_identity = file_identity(path)
    if video_identity is None:
        log(f"WARNING: late-sidecar sweep of {path}: the video's identity could not be read "
            f"before it was probed; keeping everything this call")
        return 0, 0
    snaps: dict[Path, _Snapshot | None] = {}
    for sc_path, _forced, _sdh in sidecars:
        try:
            snaps[sc_path] = _read_snapshot(sc_path)
        except OSError as exc:
            log(f"WARNING: cannot read sidecar {sc_path} for late-sidecar "
                f"classification: {type(exc).__name__}: {exc}")
            snaps[sc_path] = None
            continue
        if snaps[sc_path] is None:
            log(f"WARNING: sidecar {sc_path} too large or changing; left unclassified")

    tracks: list[dict] | None = None   # embedded English text tracks (None: unknown)
    probed = False

    def _embedded_tracks() -> list[dict] | None:
        nonlocal tracks, probed
        if not probed:
            probed = True
            data = info if info is not None else probe(path)
            if data is None:
                log(f"WARNING: cannot probe {path} to judge late English sidecars "
                    f"redundant; keeping them")
            else:
                tracks = _embedded_english_text_tracks(data)
        return tracks

    rejection = _upgrade_rejection_reasons(path, snaps, sidecars, upgrade_rejection)
    content_ok: dict = {}   # (embedded stream index, sidecar is forced) -> positive English
    foreign = redundant = 0
    for sc_path, is_forced, _is_sdh in sidecars:
        snap = snaps[sc_path]
        if snap is None:
            continue
        text = decode_subtitle_text(snap.data)
        if text is None:
            log(f"WARNING: sidecar {sc_path} is not decodable as text; left unclassified")
            continue
        is_eng, verdict_info = classify_candidate(text)
        if is_eng and rejection is not None and not is_forced \
                and not is_generated_sidecar(path, sc_path):
            # A sidecar the upgrade check REJECTED is quarantined; no verdict: kept, retried.
            why = rejection.get(sc_path.name)
            if why is None:
                log(f"kept: possible upgrade for generated subtitle: {sc_path}")
                continue
            reason, is_redundant = f"rejected upgrade: {why}", True
        elif is_eng:
            embedded = _embedded_tracks()
            if embedded is None:
                continue
            verdict = _late_english_sidecar_verdict(path, sc_path, is_forced, embedded)
            if verdict == "keep":
                continue
            if verdict == "upgrade":
                log(f"kept: possible upgrade for generated subtitle: {sc_path}")
                continue
            # CONTENT, not just the tag: a French track tagged `eng` must not cost the file
            # its only real subtitle.
            candidates = sorted(_eligible_embedded(embedded, is_forced),
                                key=is_generated_track)
            verified = False
            for track in candidates[:_MAX_EMBEDDED_CONTENT_CHECKS]:
                key_ = (track.get("index"), is_forced)
                if key_ not in content_ok:
                    content_ok[key_] = _embedded_track_is_english(path, track, is_forced)
                if content_ok[key_]:
                    verified = True
                    break
            if not verified:
                log(f"kept: embedded English subtitle of {path} not verified as English; "
                    f"keeping late sidecar {sc_path}")
                continue
            reason, is_redundant = _REDUNDANT_REASON, True
        else:
            reason, is_redundant = _LATE_FOREIGN_REASON, False
        if dry_run:
            log(f"DRY-RUN: would quarantine late sidecar {sc_path} "
                f"({reason if is_redundant else 'verdict=' + str(verdict_info.get('verdict'))})")
            ok = True
        else:
            if file_identity(path) != video_identity:
                log(f"WARNING: {path} changed while its late sidecars were being judged; "
                    f"keeping the rest (the next scan starts over)")
                break
            ok = _quarantine_sidecar(sc_path, snap, verdict_info, reason)
        if ok:
            if is_redundant:
                redundant += 1
            else:
                foreign += 1
    return foreign, redundant


def _late_sidecar_sweep(path: Path, dry_run: bool, info: dict | None,
                upgrade_rejection: object,
                video_identity: object = _UNSET) -> tuple[int, int]:
    """`_quarantine_foreign_sidecars` that never changes the SKIP result it decorates."""
    try:
        return _quarantine_foreign_sidecars(path, dry_run, info, upgrade_rejection,
                                            video_identity)
    except Exception as exc:
        log(f"WARNING: late-sidecar sweep failed for {path}: {type(exc).__name__}: {exc}")
        return 0, 0


def _cleanup_no_main_english(
    path: Path, path_str: str, status: str, base_detail: str,
    sub_streams: list[dict], audio_streams: list[dict], video_streams: list[dict],
    forced_text: dict | None, eng_image: list[dict],
    dry_run: bool,
    proven_foreign_sub_indices: frozenset[int] = frozenset(),
) -> dict:
    """On NEEDS_BAZARR, record the file in the needs-english tracker for `sub-fallback.py`.
    `pre_identity` is captured BEFORE the impl so a new release still resets `first_seen`
    while the impl's own remux does not."""
    pre_identity = _file_identity(path)
    result = _cleanup_no_main_english_impl(
        path, path_str, status, base_detail,
        sub_streams, audio_streams, video_streams,
        forced_text, eng_image, dry_run, proven_foreign_sub_indices,
    )
    if result.get("status") == "NEEDS_BAZARR":
        _apply_cleanup_needs_english_update(path, pre_identity, Path(result["path"]), dry_run)
    return result


def _cleanup_no_main_english_impl(
    path: Path, path_str: str, status: str, base_detail: str,
    sub_streams: list[dict], audio_streams: list[dict], video_streams: list[dict],
    forced_text: dict | None, eng_image: list[dict],
    dry_run: bool,
    proven_foreign_sub_indices: frozenset[int] = frozenset(),
) -> dict:
    """Cleanup remux for a file with no main English subtitle yet (called from the four
    `_process_locked` NEEDS_BAZARR / FAIL branches), so a MULTI release does not keep every
    foreign subtitle and a foreign default audio for weeks while Bazarr works.

    Never delete a track that might be wanted and never silently change a flag: a wrong
    default is recoverable, a deleted track is not.
      - Video is not touched (cover-art removal is the full path's job).
      - Audio is never dropped or reordered; only the `default` flag moves, and only when
        `confident_default_audio` is confident (never a bare `default`, never a guess).
        `None` leaves audio as is, so an Arr outage cannot re-remux back and forth.
      - Subtitles: only a POSITIVELY foreign track (`_is_positively_foreign_sub`) or one
        proven foreign by the content gate is dropped; ambiguous ones, the forced-English
        text track and every English image subtitle (`eng_image`) are kept.
      - Every kept track gets explicit flags from the SOURCE dispositions
        (`cleanup_flags`); the confident audio pick is the only deliberate change.
      - No loudness normalization (it can re-encode audio and ffmpeg auto-flags a default).

    Does NOT stamp CONSOLIDATED_SUBS or touch sidecars: the file still gets the full pass
    once an English subtitle appears.

    The output is re-probed before the swap: track counts and the default-flag pattern
    must match the intent, else FAIL with the original untouched. Returns the caller's
    `status` (unless remux/verify failed); `detail` gains a cleanup summary.
    """
    original_language = original_language_for(path)
    confident_idx = confident_default_audio(
        path, audio_streams, original_language=original_language)

    default_flagged = {
        s["index"] for s in audio_streams
        if (s.get("disposition") or {}).get("default")
    }
    audio_default_change = confident_idx is not None and default_flagged != {confident_idx}

    keep_video_idx = [s["index"] for s in video_streams]
    keep_audio_idx = [s["index"] for s in audio_streams]

    keep_sub_indices = {s["index"] for s in eng_image}
    if forced_text is not None:
        keep_sub_indices.add(forced_text["index"])
    # forced_text/eng_image are never dropped: they are not gated.
    for s in sub_streams:
        if s["index"] in keep_sub_indices:
            continue
        if s["index"] in proven_foreign_sub_indices or _is_positively_foreign_sub(s):
            continue
        keep_sub_indices.add(s["index"])
    kept_sub_streams = [s for s in sub_streams if s["index"] in keep_sub_indices]

    dropped_subs = sum(1 for s in sub_streams if s["index"] not in keep_sub_indices)

    something_to_clean = audio_default_change or dropped_subs > 0

    if not something_to_clean:
        return {"path": path_str, "status": status,
                "detail": f"{base_detail}; nothing to clean"}

    if dry_run:
        return {"path": path_str, "status": status,
                "detail": (f"{base_detail}; dry-run: would clean "
                           f"audio_default_change={audio_default_change} "
                           f"dropped_subs={dropped_subs}")}

    cleanup_flags: dict[int, dict[str, bool]] = {}
    for s in video_streams:
        cleanup_flags[s["index"]] = {"default": bool((s.get("disposition") or {}).get("default"))}
    for s in audio_streams:
        is_default = (s["index"] == confident_idx) if audio_default_change \
            else bool((s.get("disposition") or {}).get("default"))
        cleanup_flags[s["index"]] = {"default": is_default}
    for s in kept_sub_streams:
        disp = s.get("disposition") or {}
        cleanup_flags[s["index"]] = {
            "default": bool(disp.get("default")),
            "forced": bool(disp.get("forced")),
        }

    out = path.with_suffix(path.suffix + f".consol.{os.getpid()}.tmp.mkv")
    out.unlink(missing_ok=True)
    ok, detail = remux(
        path, out, keep_video_idx, keep_audio_idx, None, None,
        keep_subtitle_indices=sorted(keep_sub_indices),
        cleanup_flags=cleanup_flags,
    )
    if not ok:
        out.unlink(missing_ok=True)
        return {"path": path_str, "status": "FAIL", "detail": f"cleanup remux: {detail}"}

    # Verify BEFORE the swap; never touch the original on a mismatch.
    verify_info = probe(out)
    if verify_info is None:
        out.unlink()
        return {"path": path_str, "status": "FAIL", "detail": "cleanup verify: probe failed"}
    verify_streams = verify_info.get("streams", [])
    out_video_streams = [s for s in verify_streams if s.get("codec_type") == "video"]
    out_audio_streams = [s for s in verify_streams if s.get("codec_type") == "audio"]
    out_subs = sum(1 for s in verify_streams if s.get("codec_type") == "subtitle")
    expected_audio = len(audio_streams)
    expected_subs = len(keep_sub_indices)
    expected_video = len(video_streams)
    if (len(out_audio_streams) != expected_audio or out_subs != expected_subs
            or len(out_video_streams) != expected_video):
        out.unlink()
        return {"path": path_str, "status": "FAIL",
                "detail": (f"cleanup verify: expected audio={expected_audio} "
                           f"subs={expected_subs} video={expected_video}, got "
                           f"audio={len(out_audio_streams)} subs={out_subs} "
                           f"video={len(out_video_streams)}")}

    # Streams of each type keep their relative order: compare by position.
    intended_video_defaults = [cleanup_flags[i]["default"] for i in keep_video_idx]
    intended_audio_defaults = [cleanup_flags[i]["default"] for i in keep_audio_idx]
    out_video_defaults = [bool((s.get("disposition") or {}).get("default")) for s in out_video_streams]
    out_audio_defaults = [bool((s.get("disposition") or {}).get("default")) for s in out_audio_streams]
    if out_video_defaults != intended_video_defaults or out_audio_defaults != intended_audio_defaults:
        out.unlink()
        return {"path": path_str, "status": "FAIL",
                "detail": (f"cleanup verify: default-flag mismatch -- "
                           f"video expected={intended_video_defaults} got={out_video_defaults}, "
                           f"audio expected={intended_audio_defaults} got={out_audio_defaults}")}

    replacement = path.with_suffix(".mkv")
    if replacement != path and replacement.exists():
        out.unlink(missing_ok=True)
        return {"path": path_str, "status": "FAIL",
                "detail": f"collision: sibling .mkv exists for {path.name}"}

    # tag=False: stamping would stop this branch from ever running again.
    early_result, _tag_written = _expose_tag_normalize(
        out, path, replacement, path_str, tag=False, normalize=False)
    if early_result is not None:
        return early_result

    return {"path": str(replacement), "status": status,
            "detail": (f"{base_detail}; cleaned: audio_default_change={audio_default_change} "
                       f"dropped_subs={dropped_subs}")}


def _workfile(workdir: Path, role: str, suffix: str) -> Path:
    """Every workdir temp file comes from here, never from a name derived from the video or
    a sidecar (a video called `forced` with a `forced.srt` sidecar would collide)."""
    fd, name = tempfile.mkstemp(prefix=f"{role}-", suffix=suffix, dir=workdir)
    os.close(fd)
    return Path(name)


def _sidecar_as_srt(sc_path: Path, workdir: Path, data: bytes | None = None) -> Path | None:
    """Copy sidecar `sc_path` (or write `data`) into `workdir`, converted to SRT if
    `.ass`/`.ssa`/`.vtt`. None if conversion failed."""
    local = _workfile(workdir, "sidecar", sc_path.suffix)
    if data is None:
        shutil.copy2(sc_path, local)
    else:
        local.write_bytes(data)
    suffix = local.suffix.lower()
    if suffix in (".ass", ".ssa"):
        srt_p = _workfile(workdir, "sidecar-srt", ".srt")
        if not ass_to_srt(local, srt_p):
            return None
        local = srt_p
    elif suffix == ".vtt":
        srt_p = _workfile(workdir, "sidecar-srt", ".srt")
        r = subprocess.run(["ffmpeg", "-y", "-v", "error", "-i", str(local), str(srt_p)],
                           capture_output=True, timeout=60)
        if r.returncode != 0:
            return None
        local = srt_p
    return local


class _SubCandidate(NamedTuple):
    score: int
    path: Path
    source: str       # "embedded" | "sidecar" | "subliminal"
    generated: bool   # Whisper output (".generated" sidecar / "(generated)" title)
    sdh: bool         # hearing-impaired / SDH / CC variant

    @property
    def rank(self) -> tuple[bool, bool, int]:
        """Higher wins: real beats generated; within a class non-SDH beats SDH (used only
        when it is the sole candidate in its class); score only breaks ties."""
        return (not self.generated, not self.sdh, self.score)


def _embedded_candidate(path: Path, s: dict,
                        workdir: Path) -> tuple[_SubCandidate | None, str]:
    """Extract embedded text track `s`, content-gate it and convert it to SRT. Returns
    `(candidate, outcome)`; outcome is "ok", "failed" (could not be extracted, read or
    converted) or "rejected" (the content gate proved it foreign)."""
    ext = ".ass" if s.get("codec_name") in ASS_CODECS else ".srt"
    tmp = _workfile(workdir, f"embedded{s['index']}", ext)
    if not extract_track(path, s["index"], s.get("codec_name"), tmp):
        return None, "failed"
    # Gate the extracted file before any ASS/SSA conversion.
    try:
        is_eng, verdict_info = _gate_local_file(tmp)
    except OSError as exc:
        if isinstance(exc, _Unclassifiable):
            log(f"content-gate SKIP embedded track {s['index']} in {path}: cannot be "
                f"classified ({exc}); not a candidate")
        return None, "failed"
    if not is_eng:
        log(f"content-gate REJECT embedded track {s['index']} in {path}: "
            f"tagged English, text scored {verdict_info.get('verdict')} "
            f"(reason={verdict_info.get('reason')}) -- treated as foreign")
        return None, "rejected"
    if tmp.suffix == ".ass":
        srt_p = _workfile(workdir, f"embedded{s['index']}-srt", ".srt")
        if not ass_to_srt(tmp, srt_p):
            return None, "failed"
        tmp = srt_p
    lc = count_subtitle_lines(tmp)
    return (_SubCandidate(score_text_track(s, lc), tmp, "embedded",
                          is_generated_track(s), is_sdh_track(s)), "ok")


# "English (dub)": the subtitle that matches the English DUB audio of a file that keeps both
# Japanese and English audio (the main track translates the Japanese). Never ffsubsync'd: a
# release's CC track is timed to its release and a Whisper one to the dub audio itself.

def _finish_dub(srt: Path) -> bool:
    """Clean the UTF-8 SRT `srt` in place and strip its hearing-impaired annotations; False
    (file unchanged) if no cue is left."""
    try:
        text = strip_hi_annotations(
            clean_srt_text(srt.read_text(encoding="utf-8", errors="replace")))
        if "-->" not in text:
            return False
        srt.write_text(text, encoding="utf-8")
    except OSError:
        return False
    return True


def _dub_from_sidecar(sc_path: Path, workdir: Path) -> tuple[Path, Identity] | None:
    """Read the `.en.dub.srt` `sc_path` ONCE, content-gate it and return its cleaned SRT
    working copy with the identity it was read at. A foreign one is quarantined; one that
    cannot be read or converted, or has no dialogue left, stays in place. None then."""
    try:
        snap, text = _snapshot_text(sc_path)
        is_eng, verdict_info = classify_candidate(text)
        if not is_eng:
            log(f"content-gate REJECT dub sidecar {sc_path}: named English, text scored "
                f"{verdict_info.get('verdict')} (reason={verdict_info.get('reason')}) "
                f"-- quarantining")
            _quarantine_sidecar(sc_path, snap, verdict_info,
                                "dub sidecar rejected during consolidation")
            return None
        local = _sidecar_as_srt(sc_path, workdir, data=text.encode("utf-8"))
    except (OSError, subprocess.SubprocessError) as exc:   # _Unclassifiable is an OSError
        log(f"dub sidecar {sc_path} left in place: {exc}")
        return None
    if local is None or not _finish_dub(local):
        log(f"dub sidecar {sc_path} left in place: could not be converted to SRT "
            f"or has no dialogue")
        return None
    return local, snap.identity


def _dub_srt(workdir: Path, sidecars: list[Path], embedded: list[_SubCandidate],
             consumed: dict[Path, Identity]) -> Path | None:
    """The cleaned working SRT for the dub track, else None: the first usable `.en.dub.srt`
    (recorded in `consumed`), otherwise the best-scoring dub-like embedded candidate."""
    for sc_path in sidecars:
        got = _dub_from_sidecar(sc_path, workdir)
        if got is not None:
            consumed[sc_path] = got[1]
            return got[0]
    best = max(embedded, key=lambda c: c.score, default=None)
    return best.path if best is not None and _finish_dub(best.path) else None


def _verify_added_dub(out: Path, src_info: dict) -> str | None:
    """Re-probe `out` of a dub add; the first mismatch with the source plus one new subtitle,
    else None: video, audio and the existing subtitles unchanged in count and default/forced
    flags, the CONSOLIDATED_SUBS tag kept, and the new "English (dub)" subtitle last
    (English, neither default nor forced)."""
    info = probe(out)
    if not info:
        return "probe of the remuxed file failed"

    def of(i: dict, kind: str) -> list[dict]:
        return [s for s in i.get("streams", []) if s.get("codec_type") == kind]

    def flags(ss: list[dict]) -> list[tuple[bool, bool]]:
        return [(bool((s.get("disposition") or {}).get("default")),
                 bool((s.get("disposition") or {}).get("forced"))) for s in ss]

    for kind in ("video", "audio"):
        if flags(of(info, kind)) != flags(of(src_info, kind)):
            return f"{kind} streams or their flags changed"
    out_subs, src_subs = of(info, "subtitle"), of(src_info, "subtitle")
    if len(out_subs) != len(src_subs) + 1 or flags(out_subs[:-1]) != flags(src_subs):
        return "the existing subtitles changed"
    new = out_subs[-1]
    if (not is_english_tag((new.get("tags") or {}).get("language"))
            or (new.get("tags") or {}).get("title") != ENG_DUB_TRACK_NAME
            or flags([new]) != [(False, False)]):
        return "the new subtitle is not an English dub track that is neither default nor forced"
    if already_processed(src_info) and not already_processed(info):
        return "the CONSOLIDATED_SUBS tag was lost"
    return None


def _add_late_dub(path: Path, path_str: str, dry_run: bool, info: dict | None,
                  prev: dict | None, video_identity: Identity | None) -> dict | None:
    """An ALREADY-PROCESSED anime file that keeps both Japanese and English audio and has no dub
    subtitle yet takes a `.en.dub.srt` as a new "English (dub)" track: a plain mkvmerge copy
    of the file plus that track, nothing else reprocessed. Verified before the swap, state
    updated, the consumed sidecar removed. Returns the result, or None when there is nothing
    to do (no dub sidecar, not dual anime, a dub track already there, the sidecar unusable) so the
    late-sidecar sweep goes on. No playback gate: like every swap here it is an atomic
    `os.replace`, and a session playing the old file keeps its open inode."""
    sidecars = find_dub_sidecars(path)
    if not sidecars or path.suffix.lower() != ".mkv":
        return None
    if info is None:
        info = probe(path)
    if not info:
        return None
    if not (is_dual_audio([s for s in info.get("streams", []) if s.get("codec_type") == "audio"])
            and is_anime(path)):
        return None
    if has_dub_subtitle(info):
        log(f"dub sidecar {sidecars[0]} left in place: {path.name} already has a dub subtitle")
        return None
    if dry_run:
        return {"path": path_str, "status": "SKIP",
                "detail": f"dry-run: would add dub subtitle ({sidecars[0].name})"}
    workdir = Path(tempfile.mkdtemp(prefix=f"consolsub_{os.getpid()}_", dir=path.parent))
    out = path.with_suffix(path.suffix + f".consol.{os.getpid()}.tmp.mkv")
    try:
        got = None
        for sc_path in sidecars:
            got = _dub_from_sidecar(sc_path, workdir)
            if got is not None:
                break
        if got is None:
            return None
        dub_srt, sc_ident = got
        out.unlink(missing_ok=True)
        ok, detail = add_dub_subtitle(path, out, dub_srt)
        if not ok:
            return {"path": path_str, "status": "FAIL", "detail": f"dub remux: {detail}"}
        problem = _verify_added_dub(out, info)
        if problem:
            return {"path": path_str, "status": "FAIL", "detail": f"dub verify: {problem}"}
        if file_identity(path) != video_identity:
            log(f"{path.name} changed while its dub subtitle was added; original kept")
            return {"path": path_str, "status": "SKIP",
                    "detail": "dub: the video changed, original kept"}
        early_result, _tag_written = _expose_tag_normalize(
            out, path, path, path_str, tag=False, normalize=False)
        if early_result is not None:
            return early_result
        if file_identity(sc_path) == sc_ident:
            sc_path.unlink(missing_ok=True)
        else:
            log(f"WARNING: dub sidecar {sc_path} changed since it was read; left in place")
        size, mtime = file_key(path)
        update_state_entry(STATE_FILE, str(path.resolve()),
                           {**(prev or {"status": "skip-tag"}),
                            "size": size, "mtime": mtime, "v": PIPELINE_VERSION})
        return {"path": path_str, "status": "FIXED",
                "detail": f"dub subtitle added from {sc_path.name}"}
    finally:
        out.unlink(missing_ok=True)
        shutil.rmtree(workdir, ignore_errors=True)


# Generated -> real subtitle upgrade. A file consolidated with only a Whisper transcript
# ("English (generated)") promises a real subtitle will replace it. Runs under the per-file
# lock BEFORE the state-cache / tag-present skips. A sidecar is used only once proven the
# right episode's English subtitle: English text whose cues match the generated track in
# >= 2 of 3 bounded windows (`judge_sidecar_against_windows`).
UPGRADE_REJECTED_KEY = "upgrade_rejected"
# A sidecar is read ONCE; anything bigger is refused, never judged on a prefix.
UPGRADE_MAX_SIDECAR_BYTES = 8 * 1024 * 1024

class _Snapshot(NamedTuple):
    """One sidecar, read once: what was verified is exactly these bytes."""
    identity: Identity  # fstat on the same handle, unchanged across the read
    digest: str          # sha256 of `data`
    data: bytes


class _Verified(NamedTuple):
    srt: bytes           # the verified working SRT bytes
    sdh: bool            # ranking only
    identity: Identity
    digest: str          # sha256 of the sidecar as read (before conversion)


class _Upgrade(NamedTuple):
    # The full pass muxes these exact bytes, then deletes the one it muxed if still exactly
    # what was read, and nothing else.
    sidecars: dict[Path, _Verified]
    # Re-checked before the remux and the swap; any change aborts.
    video_identity: Identity


_KEEPABLE_SUB_CODECS = TEXT_CODECS | IMAGE_CODECS


def _upgrade_kept_tracks(sub_streams: list[dict]) -> list[dict]:
    """Every forced English subtitle track (text or image) and every English dub track; an
    upgrade carries each over."""
    return [s for s in sub_streams
            if is_english_tag((s.get("tags") or {}).get("language"))
            and (is_forced_track(s) or is_dub_track(s))]


def _read_snapshot(p: Path) -> _Snapshot | None:
    """Read `p` once (fstat and bytes from the same handle). None if over
    `UPGRADE_MAX_SIDECAR_BYTES` (never verify, mux or quarantine a prefix) or not read
    completely and stably. Raises OSError if unreadable (`_snapshot_sidecar` maps that to
    None)."""
    limit = UPGRADE_MAX_SIDECAR_BYTES
    with p.open("rb") as fh:
        before = os.fstat(fh.fileno())
        if before.st_size > limit:
            return None
        data = fh.read(limit + 1)  # may have grown since fstat
        after = os.fstat(fh.fileno())
    if len(data) > limit:
        return None
    # Complete and stable; a file being written or truncated is retried next scan.
    if len(data) != before.st_size:
        return None
    if identity_from_stat(before)[1:] != identity_from_stat(after)[1:]:
        return None
    return _Snapshot(identity_from_stat(before), hashlib.sha256(data).hexdigest(), data)


def _snapshot_sidecar(p: Path) -> _Snapshot | None:
    """`_read_snapshot`, with unreadable also None (retried next scan)."""
    try:
        return _read_snapshot(p)
    except OSError:
        return None


def _sidecar_fingerprint(snaps: dict[Path, _Snapshot]) -> list[list]:
    """`[name, size, sha256]` per sidecar, for matching a recorded rejection. The hash
    catches a same-size rewrite that size and mtime miss."""
    return [[p.name, s.identity[1], s.digest]
            for p, s in sorted(snaps.items(), key=lambda kv: kv[0].name)]


def _rejection_record(video_identity: Identity, fingerprint: list[list],
                      reasons: dict[str, str] | None = None) -> dict:
    """Fingerprint of each sidecar with a NON-TRANSIENT outcome (a verdict, or undecided)
    plus the video identity; a transiently failed sidecar is left out so it stays
    retryable. `reasons` ({name: why}) is quoted when the late sweep quarantines a rejected
    sidecar; a record without it is the older format and is rejected."""
    return {"video": list(video_identity), "sidecars": fingerprint,
            "reasons": dict(reasons or {})}


def _recorded_verdicts(record: object, video_identity: Identity,
                       snaps: dict[Path, "_Snapshot"]) -> dict[str, str | None] | None:
    """What a rejection `record` still says PER SIDECAR. None if it does not apply to this
    video (old format or identity differs); else `{name: reason or None}` for sidecars
    whose fingerprint still matches (reason = wrong-episode verdict, None = judged with no
    verdict). A changed or unrecorded sidecar is absent and verified again."""
    if not isinstance(record, dict) or not isinstance(record.get("reasons"), dict):
        return None
    if not identity_matches(record.get("video"), video_identity, allow_legacy=False):
        return None
    entries = record.get("sidecars")
    recorded = {e[0]: e for e in entries if isinstance(e, list) and len(e) == 3
                and isinstance(e[0], str)} if isinstance(entries, list) else {}
    reasons = record["reasons"]
    out: dict[str, str | None] = {}
    for sc, snap in snaps.items():
        if recorded.get(sc.name) == [sc.name, snap.identity[1], snap.digest]:
            why = reasons.get(sc.name)
            out[sc.name] = str(why) if why else None
    return out


def _record_upgrade_rejected(key: str, prev: dict, rejection: dict) -> None:
    """Best effort: lets the next scan skip the window reads."""
    try:
        update_state_entry(STATE_FILE, key, {**prev, UPGRADE_REJECTED_KEY: rejection})
    except Exception as exc:
        log(f"UPGRADE-SKIP: could not record rejection for {key}: "
            f"{type(exc).__name__}: {exc}")


def _generated_upgrade_check(path: Path, cur: tuple[int, int], prev: dict | None
                             ) -> tuple[_Upgrade | None, dict | None, dict | None]:
    """Should `path` (locked by the caller) have its generated English subtitle replaced by
    a real sidecar? Returns `(upgrade, info, rejected)`.

    `upgrade` is None unless: a MAIN non-generated English sidecar exists; the file is
    already consolidated (tag, or a `fixed` state entry for the current size/mtime, since
    the tag write can fail); every main English text track is "(generated)"; and a sidecar
    passes the English-text and same-episode checks. Each sidecar is read once and those
    exact bytes go to the full pass.

    `info` is the ffprobe result if this probed. `rejected` is the record to remember when
    verification judged the sidecars wrong, never for anything possibly transient.
    """
    real = [(p, sdh) for (p, forced, sdh) in find_sidecar_subs(path)
            if not forced and not is_generated_sidecar(path, p)]
    if not real:
        return None, None, None

    notes: list[str] = []
    reasons: dict[str, str] = {}  # name -> why REJECTED (a verdict, not a failure)
    judged: dict[Path, _Snapshot] = {}
    snaps: dict[Path, _Snapshot] = {}
    for sc, _sdh in real:
        snap = _snapshot_sidecar(sc)
        if snap is None:
            notes.append(f"{sc.name}: unreadable or larger than "
                         f"{UPGRADE_MAX_SIDECAR_BYTES} bytes")
            continue
        snaps[sc] = snap
    if not snaps:
        log(f"UPGRADE-SKIP: {path.name}: {'; '.join(notes)}")
        return None, None, None

    video_identity = file_identity(path)  # before the probe
    if video_identity is None:
        return None, None, None
    remembered = _recorded_verdicts((prev or {}).get(UPGRADE_REJECTED_KEY),
                                    video_identity, snaps) or {}
    if len(snaps) == len(real) and all(sc.name in remembered for sc in snaps):
        return None, None, None

    info = probe(path)
    consolidated = bool(info) and (
        already_processed(info)
        or (bool(prev) and prev.get("size") == cur[0] and prev.get("mtime") == cur[1]
            and prev.get("v") == PIPELINE_VERSION and prev.get("status") == "fixed"))
    if not info or not consolidated:
        return None, info, None
    gen_tracks = [s for s in info.get("streams", [])
                  if s.get("codec_type") == "subtitle"
                  and is_english_tag((s.get("tags") or {}).get("language"))
                  and s.get("codec_name") in TEXT_CODECS and not is_forced_track(s)
                  and not is_dub_track(s)]
    if not gen_tracks or not all(is_generated_track(s) for s in gen_tracks):
        return None, info, None
    gen = gen_tracks[0]
    subs_src = [s for s in info.get("streams", []) if s.get("codec_type") == "subtitle"]
    unkeepable = [s for s in _upgrade_kept_tracks(subs_src)
                  if s.get("codec_name") not in _KEEPABLE_SUB_CODECS]
    if unkeepable:
        # The upgrade rebuilds subtitles, so it must carry every forced and dub track over.
        log(f"UPGRADE-SKIP: {path.name}: forced or dub English subtitle track(s) it cannot keep "
            f"(codec {', '.join(str(s.get('codec_name')) for s in unkeepable)}); "
            f"keeping the generated subtitle")
        return None, info, None

    try:
        duration = float((info.get("format") or {}).get("duration"))
    except (TypeError, ValueError):
        duration = 0.0

    sdh_of = dict(real)
    verified: dict[Path, _Verified] = {}
    windows = None
    with tempfile.TemporaryDirectory(prefix="consolsub-upgrade-") as td:
        for sc, snap in snaps.items():
            if sc.name in remembered:
                judged[sc] = snap
                if remembered[sc.name]:
                    reasons[sc.name] = str(remembered[sc.name])
                continue
            decoded = decode_subtitle_text(snap.data)
            if decoded is None:
                notes.append(f"{sc.name}: not decodable as text")
                judged[sc] = snap
                continue
            try:
                srt = _sidecar_as_srt(sc, Path(td), decoded.encode("utf-8"))
                srt_bytes = srt.read_bytes() if srt is not None else None
            except (OSError, subprocess.SubprocessError) as exc:
                notes.append(f"{sc.name}: could not be converted ({type(exc).__name__})")
                continue   # transient: retried next scan
            if srt_bytes is None:
                notes.append(f"{sc.name}: could not be converted to SRT")
                continue
            sc_text = clean_srt_text(srt_bytes.decode("utf-8", errors="replace"))
            verdict = sublang.score(sc_text)["verdict"]
            if verdict != "eng":
                notes.append(f"{sc.name}: not usable English text (language check: {verdict})")
                # No reason: this test is stricter than the content gate ("la la la" or a
                # short unlisted-language file fail it), so it does not prove "foreign".
                judged[sc] = snap
                continue
            if windows is None:
                windows = generated_track_windows(path, gen, duration)
                if windows is None:
                    notes.append("could not read the generated track's cues from the file")
                    break   # transient and video-wide
                if not windows:
                    notes.append("the file's duration is missing or zero, so no verification "
                                 "windows can be placed")
                    break
            ok, why, usable, evidence = judge_sidecar_against_windows(windows, sc_text)
            if ok:
                verified[sc] = _Verified(srt_bytes, sdh_of[sc], snap.identity, snap.digest)
            else:
                notes.append(f"{sc.name}: not this file's subtitle ({why})")
                judged[sc] = snap
                if evidence >= UPGRADE_MIN_PASSING_WINDOWS:
                    # A verdict needs enough text on BOTH sides in >= 2 windows; a partial
                    # sidecar is "not enough evidence" and the late sweep must not act on it.
                    reasons[sc.name] = f"not this file's subtitle ({why})"

    if not verified:
        log(f"UPGRADE-SKIP: {path.name}: {'; '.join(notes)}")
        return None, info, (_rejection_record(video_identity, _sidecar_fingerprint(judged), reasons)
                            if judged else None)
    return _Upgrade(verified, video_identity), info, None


def _upgrade_track_flags(video_streams: list[dict], audio_streams: list[dict],
                         kept_subs: list[dict]) -> dict[int, dict[str, bool]]:
    """Explicit default AND forced flag from the source dispositions (else mkvmerge picks)."""
    flags: dict[int, dict[str, bool]] = {}
    for s in [*video_streams, *audio_streams, *kept_subs]:
        disp = s.get("disposition") or {}
        flags[s["index"]] = {"default": bool(disp.get("default")),
                             "forced": bool(disp.get("forced"))}
    return flags


def _verify_upgrade_output(out: Path, video_streams: list[dict], audio_streams: list[dict],
                           kept_subs: list[dict], flags: dict[int, dict[str, bool]]
                           ) -> str | None:
    """Re-probe `out`; the first mismatch with what an upgrade must produce, else None:
    every source video/audio stream in order with its flags; then every kept forced English
    or dub track (source codec, English tag, flags); then the new main English subtitle
    (English, default, not forced)."""
    info = probe(out)
    if not info:
        return "probe of the remuxed file failed"
    streams = info.get("streams", [])
    out_v = [s for s in streams if s.get("codec_type") == "video"]
    out_a = [s for s in streams if s.get("codec_type") == "audio"]
    out_s = [s for s in streams if s.get("codec_type") == "subtitle"]
    expect_subs = len(kept_subs) + 1
    if (len(out_v), len(out_a), len(out_s)) != (len(video_streams), len(audio_streams), expect_subs):
        return (f"expected video={len(video_streams)} audio={len(audio_streams)} "
                f"subs={expect_subs}, got video={len(out_v)} audio={len(out_a)} "
                f"subs={len(out_s)}")

    def _flags(ss: list[dict]) -> list[tuple[bool, bool]]:
        return [(bool((s.get("disposition") or {}).get("default")),
                 bool((s.get("disposition") or {}).get("forced"))) for s in ss]

    want_v = [(flags[s["index"]]["default"], flags[s["index"]]["forced"]) for s in video_streams]
    want_a = [(flags[s["index"]]["default"], flags[s["index"]]["forced"]) for s in audio_streams]
    if _flags(out_v) != want_v or _flags(out_a) != want_a:
        return (f"default/forced flag mismatch -- video expected={want_v} got={_flags(out_v)}, "
                f"audio expected={want_a} got={_flags(out_a)}")

    def _english(s: dict) -> bool:
        return is_english_tag((s.get("tags") or {}).get("language"))

    got_subs = _flags(out_s)
    for pos, src in enumerate(kept_subs):
        got = out_s[pos]
        if got.get("codec_name") != src.get("codec_name"):
            return (f"kept subtitle #{pos + 1} changed codec: "
                    f"{src.get('codec_name')} -> {got.get('codec_name')}")
        if not _english(got):
            return f"kept subtitle #{pos + 1} is not tagged English"
        want = (flags[src["index"]]["default"], flags[src["index"]]["forced"])
        if got_subs[pos] != want:
            return (f"kept subtitle #{pos + 1} flags changed: default/forced "
                    f"expected={want} got={got_subs[pos]}")
    main = out_s[-1]
    main_default, main_forced = got_subs[-1]
    if not _english(main):
        return "the new main subtitle is not tagged English"
    if not main_default or main_forced:
        return (f"the new main subtitle must be default and not forced, got "
                f"default={main_default} forced={main_forced}")
    return None


def _verify_kept_source_subs(out: Path, n_video: int, n_audio: int, src: "dict | None",
                             kept: dict[int, dict[str, bool]], dub: bool = False) -> str | None:
    """Re-probe `out` of a full pass that stream-copied an unreadable embedded forced
    English track (`kept`); the first mismatch, else None. Expects the source track (codec,
    English tag, exact flags), then the new main English subtitle (default, not forced), then
    with `dub` the "English (dub)" one (neither)."""
    if src is None:
        return "no source track to compare with"
    info = probe(out)
    if not info:
        return "probe of the remuxed file failed"
    streams = info.get("streams", [])
    out_v = [s for s in streams if s.get("codec_type") == "video"]
    out_a = [s for s in streams if s.get("codec_type") == "audio"]
    out_s = [s for s in streams if s.get("codec_type") == "subtitle"]
    n_subs = 3 if dub else 2
    if (len(out_v), len(out_a), len(out_s)) != (n_video, n_audio, n_subs):
        return (f"expected video={n_video} audio={n_audio} subs={n_subs}, got "
                f"video={len(out_v)} audio={len(out_a)} subs={len(out_s)}")

    def _flags(s: dict) -> tuple[bool, bool]:
        d = s.get("disposition") or {}
        return bool(d.get("default")), bool(d.get("forced"))

    first, main = out_s[:2]
    want = kept[src["index"]]
    if first.get("codec_name") != src.get("codec_name"):
        return (f"the kept forced track changed codec: {src.get('codec_name')} -> "
                f"{first.get('codec_name')}")
    if not is_english_tag((first.get("tags") or {}).get("language")):
        return "the kept forced track is not tagged English"
    if _flags(first) != (want["default"], want["forced"]):
        return (f"the kept forced track's flags changed: default/forced expected="
                f"{(want['default'], want['forced'])} got={_flags(first)}")
    if not is_english_tag((main.get("tags") or {}).get("language")):
        return "the new main subtitle is not tagged English"
    if _flags(main) != (True, False):
        return f"the new main subtitle must be default and not forced, got {_flags(main)}"
    if dub and (not is_english_tag((out_s[2].get("tags") or {}).get("language"))
                or _flags(out_s[2]) != (False, False)):
        return "the dub subtitle must be English, neither default nor forced"
    return None


def _upgrade_video_changed(path: Path, path_str: str, upgrade: _Upgrade, when: str,
                           out: Path | None = None) -> dict | None:
    """The SKIP result that aborts the upgrade if the video is no longer the file the gate
    probed (temp output removed, retried next scan); else None."""
    if file_identity(path) == upgrade.video_identity:
        return None
    if out is not None:
        out.unlink(missing_ok=True)
    log(f"UPGRADE-SKIP: {path.name}: the video changed {when}; keeping it as it is")
    return {"path": path_str, "status": "SKIP",
            "detail": f"upgrade: the video changed {when}, original kept"}


def _process_locked(path: Path, path_str: str, dry_run: bool = False) -> dict:
    if not path.is_file():
        return {"path": path_str, "status": "SKIP", "detail": "file gone after lock"}

    state = load_state(STATE_FILE)
    key = f"{path.resolve()}"
    cur = file_key(path)
    video_identity = file_identity(path)
    prev = state.get(key)

    # The upgrade check runs first, ahead of the two skips. If it upgrades, the late sweep
    # does NOT run (the full pass consumes the verified sidecars); otherwise the late sweep
    # runs on either skip path, with the check's rejection.
    try:
        upgrade, info, rejected = _generated_upgrade_check(path, cur, prev)
    except Exception as exc:
        log(f"UPGRADE-SKIP: {path.name}: check failed: {type(exc).__name__}: {exc}")
        upgrade, info, rejected = None, None, None
    if upgrade is not None:
        names = ", ".join(sorted(p.name for p in upgrade.sidecars))
        if dry_run:
            return {"path": path_str, "status": "SKIP",
                    "detail": f"dry-run: would UPGRADE generated -> real subtitle ({names})"}
        log(f"UPGRADE: generated -> real subtitle ({names})")
    else:
        remembered = rejected or (prev or {}).get(UPGRADE_REJECTED_KEY)
        if prev and prev.get("size") == cur[0] and prev.get("mtime") == cur[1] and prev.get("v") == PIPELINE_VERSION:
            if rejected and not dry_run:
                _record_upgrade_rejected(key, prev, rejected)
            added = _add_late_dub(path, path_str, dry_run, info, prev, video_identity)
            if added is not None:
                return added
            foreign, redundant = _late_sidecar_sweep(path, dry_run, info, remembered,
                                                     video_identity)
            return {"path": path_str, "status": "SKIP",
                    "detail": _late_sidecar_detail("state cache", foreign, redundant, dry_run)}

        if info is None:
            info = probe(path)
        if not info:
            return {"path": path_str, "status": "FAIL", "detail": "probe failed"}

        if already_processed(info):
            added = _add_late_dub(path, path_str, dry_run, info, None, video_identity)
            if added is not None:
                return added
            if not dry_run:
                entry = {"size": cur[0], "mtime": cur[1],
                         "v": PIPELINE_VERSION, "status": "skip-tag"}
                if rejected:
                    entry[UPGRADE_REJECTED_KEY] = rejected
                update_state_entry(STATE_FILE, key, entry)
            foreign, redundant = _late_sidecar_sweep(path, dry_run, info, remembered,
                                                     video_identity)
            return {"path": path_str, "status": "SKIP",
                    "detail": _late_sidecar_detail("tag present", foreign, redundant, dry_run)}
    assert info is not None

    streams = info.get("streams", [])
    sub_streams = [s for s in streams if s.get("codec_type") == "subtitle"]

    # No early return on empty sub_streams: sidecar SRTs are a valid source. An untagged
    # or "und" track is deliberately not a candidate.
    eng_streams = [s for s in sub_streams
                   if is_english_tag((s.get("tags") or {}).get("language"))]

    eng_text   = [s for s in eng_streams if s.get("codec_name") in TEXT_CODECS]
    eng_image  = [s for s in eng_streams if s.get("codec_name") in IMAGE_CODECS]

    forced_text = next((s for s in eng_text if is_forced_track(s)), None)
    main_candidates = [s for s in eng_text if s is not forced_text]

    # The only sidecars FIXED may remove, and only while this identity still holds.
    consumed: dict[Path, Identity] = {}
    accepted_forced: dict[Path, _Snapshot] = {}
    sidecars: list = []
    unclassified_forced: list[Path] = []
    if upgrade is None:
        sidecars = find_sidecar_subs(path)
        if not dry_run:
            sidecars = _gate_forced_sidecars(sidecars, accepted_forced, unclassified_forced)
        sidecar_main = [(p, sdh, is_generated_sidecar(path, p))
                        for (p, forced, sdh) in sidecars if not forced]
        sidecar_forced = next(((p) for (p, forced, sdh) in sidecars if forced), None)
    else:
        # Exactly the verified snapshots; the generated embedded track is being replaced.
        sidecar_main = [(p, v.sdh, False) for p, v in upgrade.sidecars.items()]
        sidecar_forced = None
        main_candidates = []
        eng_image = []
        forced_text = None

    # Skipped in a dry run (no extraction) and an upgrade (forced tracks are kept).
    forced_gate: _ForcedGate | None = None
    rejected_forced_idx: int | None = None
    forced_track_unreadable = False
    unreadable_forced_stream: dict | None = None
    if forced_text is not None and upgrade is None and not dry_run:
        forced_gate = _gate_embedded_forced(path, forced_text)
        if forced_gate.state == "rejected":
            rejected_forced_idx = forced_text["index"]
            forced_text = None
        elif forced_gate.state == "unreadable":
            forced_track_unreadable = True
            unreadable_forced_stream = forced_text
            forced_text = None
    unclassified_parts = []
    if unclassified_forced:
        unclassified_parts.append(f"{len(unclassified_forced)} forced sidecar(s) unreadable or not text")
    if forced_track_unreadable:
        unclassified_parts.append("embedded forced track unreadable or not text")
    unclassified_note = "; ".join(unclassified_parts)
    early_proven_foreign = (frozenset({rejected_forced_idx})
                            if rejected_forced_idx is not None else frozenset())

    audio_streams = [s for s in streams if s.get("codec_type") == "audio"]
    video_streams_all = [s for s in streams if s.get("codec_type") == "video"]
    real_video = real_video_streams(streams)

    # The cleanup path gets the raw `video_streams_all`; the FULL path drops cover art and
    # keeps one real video stream.
    video_streams = real_video
    if len(video_streams) > 1:
        default_v = next(
            (s for s in video_streams if (s.get("disposition") or {}).get("default")),
            video_streams[0],
        )
        video_streams = [default_v]
    keep_video_idx = [s["index"] for s in video_streams]
    dropped_video = max(0, len(video_streams_all) - len(video_streams))
    if upgrade is not None:
        # An upgrade changes only the English subtitle.
        keep_video_idx = [s["index"] for s in video_streams_all]
        dropped_video = 0

    # `original_language_for` (Radarr/Sonarr) detects foreign-original films without
    # trusting the Matroska `default` flag, which release groups routinely set on a DUB.
    # multi_keep: anime, per-title opt-ins and foreign-original films keep the original track
    # AND the English dub; the Korean keyword path keeps a single best track.
    upgrade_flags: dict[int, dict[str, bool]] | None = None
    kept_subs: list[dict] = []
    if upgrade is None:
        allowed_langs, primary_lang, multi_keep = get_audio_lang_pref(
            path, audio_streams, original_language=original_language_for(path))
        keep_audio_idx = select_keep_audio(audio_streams, allowed_langs, primary_lang, multi_keep)
    else:
        # An upgrade changes only the subtitle: audio, forced English and dub tracks stay
        # as they are, with their source flags (via remux's cleanup_flags).
        primary_lang = "unchanged"
        keep_audio_idx = [s["index"] for s in audio_streams]
        kept_subs = _upgrade_kept_tracks(sub_streams)
        upgrade_flags = _upgrade_track_flags(video_streams_all, audio_streams, kept_subs)
    dropped_audio = max(0, len(audio_streams) - len(keep_audio_idx))

    # An ANIME file that keeps Japanese AND English audio gets a second English subtitle
    # matching the dub: a `.en.dub.srt`, else an embedded dub/CC/SDH track, which then stays
    # out of the main ranking (unless nothing else is a candidate: see below). An upgrade keeps
    # the dub track it already has.
    dual = (upgrade is None
            and is_dual_audio([s for s in audio_streams if s["index"] in keep_audio_idx])
            and is_anime(path))
    dub_sidecars = find_dub_sidecars(path) if upgrade is None else []
    dub_idx = {s["index"] for s in main_candidates if is_dub_track(s)} if dual else set()
    if dub_sidecars and not dual:
        log(f"dub sidecar(s) {', '.join(p.name for p in dub_sidecars)} left in place: "
            f"{path.name} is not anime keeping both Japanese and English audio")

    if not main_candidates and not eng_image and not sidecar_main:
        if sidecar_forced or forced_text:
            return _cleanup_no_main_english(
                path, path_str, "FAIL",
                "only forced English" + (f" ({unclassified_note})" if unclassified_note else ""),
                sub_streams, audio_streams, video_streams_all,
                forced_text, eng_image, dry_run, early_proven_foreign,
            )
        if unclassified_note:
            # A tooling problem, not "no English candidates" (sub-fallback would place a
            # main sidecar beside the unreadable forced one).
            return _cleanup_no_main_english(
                path, path_str, "FAIL",
                f"all extracts failed ({unclassified_note}"
                + ("; 1 rejected by content gate" if rejected_forced_idx is not None else "") + ")",
                sub_streams, audio_streams, video_streams_all,
                forced_text, eng_image, dry_run, early_proven_foreign,
            )
        return _cleanup_no_main_english(
            path, path_str, "NEEDS_BAZARR",
            "no English candidates"
            + (" (rejected by content gate)" if rejected_forced_idx is not None else ""),
            sub_streams, audio_streams, video_streams_all,
            forced_text, eng_image, dry_run, early_proven_foreign,
        )

    if dry_run:
        sources = []
        if main_candidates: sources.append(f"embedded={len(main_candidates)}")
        if sidecar_main:    sources.append(f"sidecar={len(sidecar_main)}")
        if eng_image:       sources.append(f"image={len(eng_image)} (needs subliminal)")
        return {"path": path_str, "status": "SKIP",
                "detail": f"dry-run: would FIX ({', '.join(sources)})"}

    workdir = Path(tempfile.mkdtemp(prefix=f"consolsub_{os.getpid()}_", dir=path.parent))
    try:
        chosen_srt: Path | None = None

        scored: list[_SubCandidate] = []
        dub_cands: list[_SubCandidate] = []   # the dub-like embedded ones among `scored`
        # content_rejected: TEXT scored non-English despite an English tag/name.
        # extraction_failed: could not be extracted/read/converted (tooling, kept visible).
        # rejected_embedded: embedded tracks proven non-English (the cleanup drops them).
        content_rejected = 1 if rejected_forced_idx is not None else 0
        extraction_failed = len(unclassified_forced) + (1 if forced_track_unreadable else 0)
        rejected_embedded: set[int] = (
            {rejected_forced_idx} if rejected_forced_idx is not None else set())

        for s in main_candidates:
            candidate, outcome = _embedded_candidate(path, s, workdir)
            if candidate is not None:
                scored.append(candidate)
                if s["index"] in dub_idx:
                    dub_cands.append(candidate)
            elif outcome == "rejected":
                content_rejected += 1
                rejected_embedded.add(s["index"])
            else:
                extraction_failed += 1

        consumed_from: dict[Path, Path] = {}  # working file -> its sidecar (upgrade)
        for n, (sc_path, is_sdh, sc_generated) in enumerate(sidecar_main):
            sc_ident: Identity | None = None
            if upgrade is not None:
                # The verified bytes, not the file as it is now.
                try:
                    local = _workfile(workdir, f"verified{n}", ".srt")
                    local.write_bytes(upgrade.sidecars[sc_path].srt)
                except OSError:
                    continue
                consumed_from[local] = sc_path
            else:
                # ONE read: classified, the working copy and (if rejected) quarantined.
                try:
                    snap, text = _snapshot_text(sc_path)
                    is_eng, verdict_info = classify_candidate(text)
                    if not is_eng:
                        content_rejected += 1
                        log(f"content-gate REJECT sidecar {sc_path}: named English, text "
                            f"scored {verdict_info.get('verdict')} "
                            f"(reason={verdict_info.get('reason')}) -- quarantining")
                        _quarantine_sidecar(sc_path, snap, verdict_info,
                                            "sidecar candidate rejected during consolidation")
                        continue
                    local = _sidecar_as_srt(sc_path, workdir, data=text.encode("utf-8"))
                    sc_ident = snap.identity
                except _Unclassifiable as exc:
                    log(f"content-gate SKIP sidecar {sc_path}: cannot be classified ({exc}); "
                        f"left in place, not a candidate")
                    extraction_failed += 1
                    continue
                except OSError:
                    extraction_failed += 1
                    continue
                if local is None:
                    extraction_failed += 1
                    continue
            lc = count_subtitle_lines(local)
            sidecar_score = 1100 + min(lc, 2000) // 5 + 50
            scored.append(_SubCandidate(sidecar_score, local, "sidecar",
                                        sc_generated, is_sdh))
            if sc_ident is not None:
                consumed[sc_path] = sc_ident

        proven_foreign = frozenset(rejected_embedded)

        if not scored and upgrade is not None:
            return {"path": path_str, "status": "SKIP",
                    "detail": "upgrade: no verified subtitle could be used, "
                              "kept the generated subtitle"}

        fetched_rejected: dict | None = None
        fetched_coverage_pct: int | None = None
        fetch_read_failed = False
        if not scored and eng_image:
            fetched = fetch_via_subliminal(path, workdir)
            if fetched is not None:
                try:
                    fetched_ok, fetched_info = _gate_local_file(fetched)
                except OSError as exc:
                    fetch_read_failed = True
                    extraction_failed += 1
                    log(f"WARNING: cannot read the subliminal download for {path}: "
                        f"{type(exc).__name__}: {exc}")
                else:
                    # A last cue well before the end means a partial subtitle (disc 1 of 2).
                    # Skipped, never a rejection, if duration or cues are unknown.
                    short: tuple[float, float] | None = None   # (last cue start, duration)
                    if fetched_ok:
                        duration = parse_duration((info.get("format") or {}).get("duration"))
                        last_cue = _last_cue_start(fetched)
                        if duration is None or last_cue is None:
                            log(f"coverage check SKIPPED for subliminal download for {path}: "
                                f"{'duration unknown' if duration is None else 'no timed cues'}")
                        elif last_cue < SUBLIMINAL_MIN_COVERAGE * duration:
                            short = (last_cue, duration)
                    if short is not None:
                        fetched_coverage_pct = int(100 * short[0] / short[1])
                        content_rejected += 1
                        log(f"coverage REJECT subliminal download for {path}: last cue at "
                            f"{_hms(short[0])} of {_hms(short[1])} ({fetched_coverage_pct}%)")
                    elif fetched_ok:
                        lc = count_subtitle_lines(fetched)
                        # Below a sidecar (1100): less vetted.
                        scored.append(_SubCandidate(1000 + min(lc, 2000) // 5, fetched,
                                                    "subliminal", False, False))
                    else:
                        fetched_rejected = fetched_info
                        content_rejected += 1
                        log(f"content-gate REJECT subliminal download for {path}: "
                            f"text scored {fetched_info.get('verdict')} "
                            f"(reason={fetched_info.get('reason')})")

        if not scored:
            # A tooling failure stays visible as FAIL; NEEDS_BAZARR only when every
            # non-accepted candidate was content-rejected (or the fetch came back empty).
            if extraction_failed:
                why = []
                if unclassified_note:
                    why.append(unclassified_note)
                if fetch_read_failed:
                    why.append("subliminal download unreadable")
                if content_rejected:
                    why.append(f"{content_rejected} rejected by content gate")
                return _cleanup_no_main_english(
                    path, path_str, "FAIL",
                    "all extracts failed" + (f" ({'; '.join(why)})" if why else ""),
                    sub_streams, audio_streams, video_streams_all,
                    forced_text, eng_image, dry_run, proven_foreign,
                )
            if eng_image:
                return _cleanup_no_main_english(
                    path, path_str, "NEEDS_BAZARR",
                    f"only image subs (subliminal result covers only {fetched_coverage_pct}% of the film)"
                    if fetched_coverage_pct is not None else
                    "only image subs (subliminal fetch failed)"
                    if fetched_rejected is None else
                    "only image subs (subliminal result rejected by content gate)",
                    sub_streams, audio_streams, video_streams_all,
                    forced_text, eng_image, dry_run, proven_foreign,
                )
            if content_rejected:
                return _cleanup_no_main_english(
                    path, path_str, "NEEDS_BAZARR",
                    "no English candidates (rejected by content gate)",
                    sub_streams, audio_streams, video_streams_all,
                    forced_text, eng_image, dry_run, proven_foreign,
                )
            return _cleanup_no_main_english(
                path, path_str, "FAIL", "all extracts failed",
                sub_streams, audio_streams, video_streams_all,
                forced_text, eng_image, dry_run, proven_foreign,
            )

        # Dub-like candidates leave the main ranking only when another candidate can be the main;
        # a sole CC/SDH track stays the main, as before.
        mains = [c for c in scored if c not in dub_cands]
        if mains:
            scored = mains
        else:
            dub_cands = []
        scored.sort(key=lambda c: c.rank, reverse=True)
        chosen = scored[0]
        chosen_srt, chosen_generated = chosen.path, chosen.generated

        forced_srt_final = None
        if forced_gate is not None and forced_gate.state == "accepted" and forced_gate.text is not None:
            ftmp = _workfile(workdir, "forced", ".srt")
            ftmp.write_text(forced_gate.text, encoding="utf-8")
            forced_srt_final = ftmp

        # No usable embedded forced track: embed an accepted forced SIDECAR rather than let
        # the full path discard it. Only a sidecar embedded here counts as consumed. Never
        # in an upgrade (it keeps the source's own forced tracks).
        if forced_srt_final is None and accepted_forced and upgrade is None:
            # Until one converts; only one is embedded.
            for fside in sorted(accepted_forced):
                fsnap = accepted_forced[fside]
                try:
                    fsrt = _sidecar_as_srt(
                        fside, workdir, data=decode_subtitle_bytes(fsnap.data).encode("utf-8"))
                    if fsrt is None:
                        log(f"WARNING: forced sidecar {fside} could not be converted to SRT; "
                            f"left in place, trying the next one")
                        continue
                    fsrt.write_text(clean_srt_text(fsrt.read_text(encoding="utf-8", errors="replace")),
                                    encoding="utf-8")
                    forced_srt_final = fsrt
                    consumed[fside] = fsnap.identity
                    break
                except Exception as exc:
                    log(f"WARNING: forced sidecar {fside} could not be embedded: "
                        f"{type(exc).__name__}: {exc}; left in place, trying the next one")

        dub_srt = _dub_srt(workdir, dub_sidecars, dub_cands, consumed) if dual else None

        try:
            txt = chosen_srt.read_text(encoding="utf-8", errors="replace")
            cleaned = clean_srt_text(txt)
            chosen_srt.write_text(cleaned, encoding="utf-8")
        except Exception as e:
            return {"path": path_str, "status": "FAIL", "detail": f"clean failed: {e}"}

        synced_srt = _workfile(workdir, "synced", ".srt")
        sync_ok = run_ffsubsync(path, chosen_srt, synced_srt)
        final_srt = synced_srt if sync_ok else chosen_srt

        out = path.with_suffix(path.suffix + f".consol.{os.getpid()}.tmp.mkv")
        out.unlink(missing_ok=True)
        if upgrade is not None:
            # ffsubsync ran since the gate probed: remux only if unchanged.
            changed = _upgrade_video_changed(path, path_str, upgrade, "before the remux")
            if changed:
                return changed
        # An unreadable forced track is not proven foreign: unless a readable one replaces
        # it, it is stream-copied unchanged.
        kept_source_subs: dict[int, dict[str, bool]] | None = None
        if (upgrade is None and unreadable_forced_stream is not None
                and forced_srt_final is None):
            disp = unreadable_forced_stream.get("disposition") or {}
            kept_source_subs = {unreadable_forced_stream["index"]: {
                "default": bool(disp.get("default")), "forced": bool(disp.get("forced"))}}
            log(f"keeping the unreadable embedded forced track "
                f"{unreadable_forced_stream['index']} of {path} unchanged")
        extra: dict = {}
        if kept_source_subs:
            extra["kept_source_subs"] = kept_source_subs
        if dub_srt:
            extra["dub_srt"] = dub_srt
        ok, detail = remux(
            path, out, keep_video_idx, keep_audio_idx, final_srt, forced_srt_final,
            keep_subtitle_indices=[s["index"] for s in kept_subs] or None,
            cleanup_flags=upgrade_flags,
            eng_track_name=ENG_GENERATED_TRACK_NAME if chosen_generated else ENG_TRACK_NAME,
            **extra)
        if not ok:
            out.unlink(missing_ok=True)
            return {"path": path_str, "status": "FAIL", "detail": f"remux: {detail}"}
        if kept_source_subs is not None:
            problem = _verify_kept_source_subs(
                out, len(keep_video_idx), len(keep_audio_idx), unreadable_forced_stream,
                kept_source_subs, dub=dub_srt is not None)
            if problem:
                out.unlink(missing_ok=True)
                return {"path": path_str, "status": "FAIL",
                        "detail": f"forced track verify: {problem}"}
        if upgrade is not None:
            assert upgrade_flags is not None
            problem = _verify_upgrade_output(
                out, video_streams_all, audio_streams, kept_subs, upgrade_flags)
            if problem:
                out.unlink(missing_ok=True)
                return {"path": path_str, "status": "FAIL",
                        "detail": f"upgrade verify: {problem}"}

        # Refuse to overwrite a sibling .mkv ("Movie.mp4" must not destroy "Movie.mkv").
        replacement = path.with_suffix(".mkv")
        if replacement != path and replacement.exists():
            out.unlink(missing_ok=True)
            return {"path": path_str, "status": "FAIL",
                    "detail": f"collision: sibling .mkv exists for {path.name}"}
        if upgrade is not None:
            changed = _upgrade_video_changed(path, path_str, upgrade, "before the swap", out)
            if changed:
                return changed
        # An upgrade never normalizes: its audio and video must stay the verified source.
        early_result, tag_written = _expose_tag_normalize(
            out, path, replacement, path_str, normalize=upgrade is None)
        if early_result is not None:
            return early_result

        # Delete ONLY sidecars read and accepted as English (`consumed`), and only while the
        # file still has the identity it was read with (one replaced by Bazarr or
        # sub-fallback.py meanwhile stays). An upgrade deletes only the verified sidecar it
        # muxed, and only if identity AND content hash match.
        dropped_sidecars = 0
        if upgrade is None:
            for sc_path, sc_ident in consumed.items():
                if sc_path == replacement:
                    continue
                now_ident = file_identity(sc_path)
                if now_ident is None:
                    continue
                if now_ident != sc_ident:
                    log(f"WARNING: sidecar {sc_path} changed since it was read; left in place")
                    continue
                try:
                    sc_path.unlink()
                    dropped_sidecars += 1
                except OSError:
                    pass
            for sc_path, _is_forced, _is_sdh in sidecars:
                if sc_path not in consumed and sc_path.exists():
                    log(f"sidecar {sc_path} was not consumed by this run; left in place")
        else:
            muxed_from = consumed_from.get(chosen_srt)
            verified = upgrade.sidecars.get(muxed_from) if muxed_from is not None else None
            if muxed_from is not None and verified is not None and muxed_from.exists():
                # Re-read and compare sha256: a same-length rewrite with the mtime restored
                # keeps size and mtime. Unreadable or oversized counts as changed.
                now = _snapshot_sidecar(muxed_from)
                if (now is None or now.identity != verified.identity
                        or now.digest != verified.digest):
                    log(f"UPGRADE: left {muxed_from.name} in place (changed after the check)")
                else:
                    # ACCEPTED RACE: a replacement landing between the re-check and this
                    # unlink is lost. No atomic compare-and-unlink exists (renameat2
                    # RENAME_NOREPLACE gives EINVAL on this mergerfs pool); microseconds,
                    # the same window `_quarantine_sidecar` accepts.
                    try:
                        muxed_from.unlink()
                        dropped_sidecars += 1
                    except OSError:
                        pass

        new_key = f"{replacement.resolve()}"
        new_size, new_mtime = file_key(replacement)
        update_state_entry(STATE_FILE, new_key, {"size": new_size, "mtime": new_mtime,
                                                 "v": PIPELINE_VERSION,
                                                 "status": "fixed", "synced": sync_ok,
                                                 "tag_written": tag_written,
                                                 **({"upgraded": True} if upgrade else {})})
        _clear_needs_english(replacement, dry_run)

        dropped_embedded = max(0, len(sub_streams) - (
            1 + (len(kept_subs) if upgrade else (1 if forced_srt_final else 0))
            + (1 if kept_source_subs else 0) + (1 if dub_srt else 0)))
        return {"path": str(replacement), "status": "FIXED",
                "detail": (f"{'UPGRADE generated->real ' if upgrade else ''}"
                           f"synced={sync_ok} dropped_subs={dropped_embedded} "
                           f"dropped_audio={dropped_audio} "
                           f"dropped_video={dropped_video} "
                           f"dropped_sidecars={dropped_sidecars} "
                           f"audio_lang={primary_lang}"
                           f"{' dub_subtitle=1' if dub_srt else ''}")}
    finally:
        shutil.rmtree(workdir, ignore_errors=True)


def walk(root: Path):
    # Prune `.normalize-tmp`: its scratch renders were treated as media and got lock files.
    for r, dirs, files in os.walk(root):
        dirs[:] = [d for d in dirs if d != NORMALIZE_TMP_DIR]
        for f in files:
            fl = f.lower()
            if not fl.endswith((".mkv", ".mp4")):
                continue
            if fl.endswith(".tmp.mkv"):
                continue
            yield Path(r) / f


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("target", nargs="?", help="file or directory")
    ap.add_argument("--scan", help="walk directory")
    ap.add_argument("--jobs", type=int, default=1)
    ap.add_argument("--limit", type=int, default=0, help="cap files in scan mode (0 = no cap)")
    ap.add_argument("--sweep-only", metavar="DIR",
                    help="sweep orphan tmp files / workdirs under DIR and exit")
    ap.add_argument("--no-sweep", action="store_true",
                    help="skip the orphan sweep that runs at the start of --scan")
    ap.add_argument("--dry-run", action="store_true",
                    help="with --sweep-only or --scan: report what would be deleted, delete nothing")
    args = ap.parse_args()

    if args.sweep_only:
        log(f"sweep-only start: {args.sweep_only} (dry_run={args.dry_run})")
        res = sweep_orphans(Path(args.sweep_only), dry_run=args.dry_run)
        _log_sweep_result(res, dry=args.dry_run)
        return 0

    if args.scan:
        if not args.no_sweep:
            log(f"sweep before scan: {args.scan} (dry_run={args.dry_run})")
            res = sweep_orphans(Path(args.scan), dry_run=args.dry_run)
            _log_sweep_result(res, dry=args.dry_run)
        files = list(walk(Path(args.scan)))
        if args.limit:
            files = files[: args.limit]
        log(f"scan start: {len(files)} files, jobs={args.jobs}, dry_run={args.dry_run}")
        if args.jobs <= 1:
            for f in files:
                r = process_file(str(f), dry_run=args.dry_run)
                log(f"{r['status']}: {Path(r['path']).name} | {r.get('detail','')}")
        else:
            with ProcessPoolExecutor(max_workers=args.jobs) as ex:
                futs = {ex.submit(process_file, str(f), args.dry_run): f for f in files}
                for fut in as_completed(futs):
                    r = fut.result()
                    log(f"{r['status']}: {Path(r['path']).name} | {r.get('detail','')}")
        log("scan complete")
        return 0

    if args.target:
        r = process_file(args.target, dry_run=args.dry_run)
        log(f"{r['status']}: {Path(r['path']).name} | {r.get('detail','')}")
        return 0 if r["status"] in ("FIXED", "SKIP", "NEEDS_BAZARR") else 1

    ap.print_help()
    return 2


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