universal-video-transcode/scripts/library_run.py

#!/usr/bin/env python3
"""Library re-encode driver: same look, SMALLER file, playable on iPad + Android.

Policy (see references/hevc-conversion-policy.md):
  * classify each source by `bpp` and pick the cheapest tier that keeps the quality:
      lean  (bpp < 0.025) -> `-c:v copy`            video bit-identical, nothing to gain
      mid   (0.025-0.040) -> hevc_nvenc cq 30       ~6.4x realtime
      fat   (> 0.040)     -> libx265 crf 24         ~3x realtime, 9-16% more efficient
  * `-maxrate` = 1.0x the source's **video** bitrate (never the file's total) and a
    post-encode assertion that the output video bitrate did not exceed it
  * quality pinned to a measured number: per-frame SSIM of a sampled window vs the source
  * audio: copy when already AAC, else AAC at the SOURCE's own bitrate (never a fixed 256k)
  * `-fps_mode cfr` + `-pix_fmt yuv420p` + `-tag:v hvc1` on every re-encode

Per file, in order:
  1. copy the source OFF the network share to LOCAL staging (never encode over SMB/WebDAV)
  2. encode locally (tier above)
  3. structural gate + full decode pass + SSIM-vs-source + size assertion, all on the LOCAL output
  4. copy the verified output to the share as <name>.new.mp4
  5. read the share copy back end-to-end and compare SHA-256 with the local file
  6. ONLY then rename it over the original, and re-probe the result on the share
A file that fails at any step leaves the original untouched and is retried next run.
"""
import argparse
import csv
import ctypes
import hashlib
import json
import os
import re
import shutil
import subprocess
import sys
import time

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import transcode as T  # noqa: E402

FFMPEG = T.FFMPEG
FFPROBE = T.FFPROBE
ROOT = r"Z:\Class\樊登读书\樊登读书会（每周更新）"
WORK = r"C:\AI\transcode-run"
STAGE = os.path.join(WORK, "stage")
LEDGER = os.path.join(WORK, "ledger.jsonl")
LOCKF = os.path.join(WORK, "run.lock")
VIDEO_EXT = {".mkv", ".mp4", ".mov", ".avi", ".webm", ".flv", ".ts", ".m4v", ".wmv",
             ".mpg", ".mpeg", ".m2ts", ".3gp", ".ogv", ".vob", ".rmvb", ".rm", ".divx"}
DECODE_CHECK = True

# ---- policy (all numbers measured; see references/hevc-conversion-policy.md §10) ----------
CQ_NVENC = 31            # measured: min SSIM 0.989-0.994 at 43-74% of the source video bitrate
CRF_X265 = 26            # measured: min SSIM 0.988-0.993 at 34-67% -- 10-22% smaller than NVENC
                         # at matched SSIM. crf24 = the conservative option (min 0.990-0.994, 43-84%)
NVENC_PRESET = "p6"
X265_PRESET = "medium"
X265_POOLS = 10          # E-cores only (affinity 0x3FF0000)
E_CORE_MASK = 0x3FF0000
SSIM_TARGET = 0.985      # per-frame floor vs the source window (480x270 comparison)
SSIM_WINDOW = 60         # seconds sampled for the SSIM gate
BAD_FRAME_SSIM = 0.85    # a frame this far below the source is 'bad' for the share test
MAX_BAD_SHARE = 0.01     # ... and more than this share of them means a systematic defect, not fades
FLAT_LUMA_STD = 3.0      # a source frame flatter than this: SSIM on it means nothing


def _flat_frames(raw_path, n, w=480, h=270, thresh=FLAT_LUMA_STD):
    """Indices of near-flat source frames (fades, black, plain titles) in a raw YUV window.

    SSIM collapses on such frames - two nearly black frames score ~0.25 while their neighbours
    score 0.998 - so a minimum taken over every frame condemns perfectly good encodes (measured on
    this library: the file that failed had 5 of 1500 frames below 0.85, all isolated, at a 0.98
    mean, while the genuine decoder-defect files had 44 of 1499).
    """
    import numpy as np
    fsz = w * h * 3 // 2
    flat = set()
    with open(raw_path, "rb") as f:
        for i in range(n):
            luma = f.read(w * h)
            if len(luma) < w * h:
                break
            if float(np.frombuffer(luma, dtype=np.uint8).std()) < thresh:
                flat.add(i)
            f.seek(fsz - w * h, 1)
    return flat


SIZE_CEIL = 1.0          # output video bitrate must not exceed the source's
QUALITY_STEPS = 2        # on a quality-gate failure, retry with the quality raised this many
                         # steps (cq -2 / crf -2) before giving up; the retry only fires on a
                         # file the cheaper setting could not carry
SSIM_MEAN_MIN = 0.98     # the window-mean bar for a normal encode
SSIM_MEAN_MIN_REDO = 0.975   # ... and for a RE-DO of a file that already plays everywhere
                         # (HEVC hvc1 + AAC). Owner decision 2026-10-08: measured on this library,
                         # 11 of a pass's attempts died on a mean of 0.9655-0.9795 while files with
                         # WORSE frame minima (0.33-0.90) passed - the bar was splitting a
                         # continuum, not separating defects (0718 马斯克原理 failed with 0.00% of
                         # frames below 0.85 and min 0.887). A re-do's only job is to reclaim the
                         # old settings' bloat, so it must not be stranded with the bigger file for
                         # ever; 0.975 still rejects the two genuine stragglers (0.9655, 0.9739).


def log(*a):
    print(time.strftime("%H:%M:%S") + " " + " ".join(str(x) for x in a), flush=True)


def disp(p):
    """Path for logs/ledger: relative to the library when possible.

    A single-file run can point anywhere (measured: os.path.relpath raises ValueError when the file
    is on D: and the library on Z:), so fall back to the absolute path.
    """
    try:
        return os.path.relpath(p, ROOT)
    except ValueError:
        return p


def probe(p, tries=3):
    """T.probe with retries: this share times a probe out (measured: 120 s for a 30 MB read), and
    one hiccup must not fail an item that is otherwise fine. The per-item try/except in the main
    loop would have turned a hiccup into a FAIL, leaving the file unconverted."""
    last = None
    for a in range(1, tries + 1):
        try:
            v = T.probe(p)
            if v:
                return v
            last = "no streams"
        except Exception as e:
            last = e
        log("   ! probe attempt %d/%d failed (%s) - retrying" % (a, tries, last))
        time.sleep(5 * a)
    raise IOError("could not probe %s after %d attempts: %s" % (p, tries, last))


def run(cmd, timeout=86400, **kw):
    return subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8",
                          errors="replace", timeout=timeout, **kw)


STALL_S = 900   # an encode with no output growth for this long is wedged, not slow


def run_watched(cmd, watch, stall_s=STALL_S, timeout=86400):
    """run() with a STALL WATCHDOG on the file the command writes.

    Measured: a hevc_nvenc encode sat FROZEN for 1.5 hours with its output unchanged - the user had
    updated the NVIDIA driver while it was running, which wedges in-flight GPU work without any
    error. A multi-day unattended run must recover from that by itself: no output growth for
    `stall_s` -> kill the command and report a failure the normal retry path can pick up.

    NB the child's stderr goes to a FILE, never a pipe. ffmpeg with `-stats` writes a line every
    0.5 s, which fills a 64 KB pipe within minutes; a blocked ffmpeg stops writing its output, so
    the watchdog then kills a perfectly healthy encode (measured: ~270 kills in one night, almost no
    work done, while the encoder sat idle in write() on stderr).
    """
    errf = watch + ".err"
    err = ""
    with open(errf, "wb") as ferr:
        p = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=ferr)
        last, last_t, t0 = -1, time.time(), time.time()
        while p.poll() is None:
            time.sleep(30)
            if time.time() - t0 > timeout:
                p.kill()
                p.wait()
                err = "timeout after %d s" % timeout
                break
            try:
                sz = os.path.getsize(watch) if os.path.exists(watch) else 0
            except OSError:
                sz = last
            if sz != last:
                last, last_t = sz, time.time()
            elif time.time() - last_t > stall_s:
                p.kill()
                p.wait()
                log("   !! STALL: no output growth for %d s - killed the wedged encode" % stall_s)
                err = "killed: wedged (no growth for %ds)" % stall_s
                break
    try:
        with open(errf, "rb") as f:
            data = f.read()
        tail = data[-4000:].decode("utf-8", "replace")
        err = (tail + "\n" + err).strip()
        os.remove(errf)
    except OSError:
        pass
    rc = p.returncode if p.returncode is not None else -9
    return subprocess.CompletedProcess(cmd, rc, "", err)


# ---------------------------------------------------------------- integrity

def _sha_local(path, label="local"):
    h = hashlib.sha256()
    n = 0
    with open(path, "rb") as f:
        while True:
            b = f.read(8 << 20)
            if not b:
                break
            h.update(b)
            n += len(b)
    return h.hexdigest(), n


def _sha_remote(path, label="remote", tries=3):
    """Hash a file over the network. getsize() lies on RaiDrive/Dokan, so the byte
    count here is what was actually read, never a metadata figure."""
    last = None
    for attempt in range(1, tries + 1):
        try:
            h = hashlib.sha256()
            n = 0
            with open(path, "rb") as f:
                while True:
                    b = f.read(8 << 20)
                    if not b:
                        break
                    h.update(b)
                    n += len(b)
            return h.hexdigest(), n
        except OSError as e:
            last = e
            log("   ! %s read attempt %d failed: %s" % (label, attempt, e))
            time.sleep(5 * attempt)
    raise last


def copy_remote_to_local(remote, local, expect_size=None, tries=5):
    """Network -> local. Retried; a short read is a failure, never a silent pass.

    Measured on this host: RaiDrive/Dokan throws `[Errno 9] Bad file descriptor` on a 1 GB read
    when three jobs hammer the share at once (the user sees RaiDrive's own upload-failure toasts at
    the same time). Smaller chunks, more attempts and a longer backoff ride those hiccups out
    instead of losing the file - and each failed attempt costs minutes on a 1 GB source, so the
    point is to stop retrying as fast as we can.
    """
    for attempt in range(1, tries + 1):
        try:
            n = 0
            with open(remote, "rb") as fi, open(local, "wb") as fo:
                while True:
                    b = fi.read(4 << 20)
                    if not b:
                        break
                    fo.write(b)
                    n += len(b)
                fo.flush()
                os.fsync(fo.fileno())
            if expect_size and n != expect_size:
                raise IOError("short read: got %d, container says %d" % (n, expect_size))
            # A read of the RIGHT LENGTH can still be wrong on this share: RaiDrive reports stale
            # sizes, and a listing read here stopped halfway with no error at all (it looked like a
            # truncated file and was not). A corrupt staged source would be encoded into a corrupt
            # output, so pay for one more pass and compare hashes of what we wrote against a fresh
            # read of the share copy.
            if n > (64 << 20):
                a = _sha_local(local, "staged source")
                b = _sha_remote(remote, "share source")
                if a != b:
                    raise IOError("staged source hash mismatch (local %s.. != share %s..)"
                                  % (a[:12], b[:12]))
            return n
        except OSError as e:
            log("   ! source copy attempt %d/%d failed: %s" % (attempt, tries, e))
            time.sleep(15 * attempt)
    raise IOError("could not copy source off the share after %d attempts" % tries)


def copy_local_to_remote(local, remote_tmp, tries=4):
    """Local -> network temp name. Chunked; never writes the final name.

    Measured failure mode on this host: RaiDrive/Dokan aborts a ~500 MB write with
    "the operation didn't complete within specific time limit" (and leaves a partial `.new.mp4`),
    which the user sees as an upload-failed toast. Three jobs reading/writing the share at once is
    what provokes it. So: smaller chunks, a periodic flush so nothing huge sits in a buffer, more
    attempts, and a longer backoff. A partial file is always deleted before retrying (never resumed
    from `os.path.getsize()` on the share - RaiDrive reports stale sizes, so an offset taken from it
    would silently corrupt the upload).
    """
    for attempt in range(1, tries + 1):
        try:
            written = 0
            with open(local, "rb") as fi, open(remote_tmp, "wb") as fo:
                while True:
                    b = fi.read(4 << 20)
                    if not b:
                        break
                    fo.write(b)
                    written += len(b)
                    if written % (32 << 20) < (4 << 20):
                        fo.flush()
                        os.fsync(fo.fileno())
                fo.flush()
                os.fsync(fo.fileno())
            return True
        except OSError as e:
            log("   ! upload attempt %d/%d failed after %.0f MB: %s" % (
                attempt, tries, written / 1e6, e))
            try:
                os.remove(remote_tmp)
            except OSError:
                pass
            time.sleep(20 * attempt)
    return False


def probe_ok(path, info, vcodec="hevc"):
    """Structural gate on any path (local or share)."""
    return T.verify(info, path, vcodec, "copy" if info["acodec"] in ("aac", "mp4a", "") else "aac",
                    check_decode=False)


# ---------------------------------------------------------------- encode

def src_bitrate(info):
    if info["dur"] > 0 and info["size"] > 0:
        return info["size"] * 8 / info["dur"]
    return 0


def match_audio_kbps(path, floor=96, ceil=256, fallback=192):
    """AAC at the SOURCE's own bitrate, not a fixed 256k.

    Measured on this library: the weekly series ships mp3 at 128 kbps, so a fixed
    -b:a 256k added ~128 kbps for no audible gain - about +10% file size on a
    1.3 Mbps file, which by itself cancels the video savings. AAC is more efficient
    than mp3, so matching the bitrate is already a quality upgrade, never a loss.
    """
    r = run([FFPROBE, "-v", "error", "-select_streams", "a:0",
             "-show_entries", "stream=bit_rate", "-of", "csv=p=0", path], timeout=120)
    try:
        b = int((r.stdout or "").strip().splitlines()[0].split(",")[0])
    except Exception:
        return fallback
    return max(floor, min(ceil, round(b / 1000)))


def src_video_bps(path, info):
    """The SOURCE's video-stream bitrate. Never the file's total: mixing them turned a 109%
    (grow) into a false 86% (shrink) in one analysis, because the total carries the audio."""
    r = run([FFPROBE, "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=bit_rate",
             "-of", "csv=p=0", path], timeout=180)
    try:
        v = int((r.stdout or "").strip().splitlines()[0].split(",")[0])
        if v > 0:
            return v
    except Exception:
        pass
    tot = info["size"] * 8 / info["dur"] if info.get("dur") else 0
    return max(0, int(tot - (info.get("abps") or 0)))


def src_fps(info):
    """transcode.probe() exposes rfps/afps, not `fps`."""
    return info.get("rfps") or info.get("afps") or 0


def bpp_of(info, vbps):
    """bits per pixel per frame: how much fat the source still carries."""
    d = (info.get("w") or 0) * (info.get("h") or 0) * src_fps(info)
    return (vbps / d) if d else 0.0


def tier_for_bpp(b):
    if b < 0.025:
        return "copy"
    return "nvenc" if b <= 0.040 else "x265"


def out_video_bps(path, dur):
    r = run([FFPROBE, "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=bit_rate",
             "-of", "csv=p=0", path], timeout=180)
    try:
        v = int((r.stdout or "").strip().splitlines()[0].split(",")[0])
        if v > 0:
            return v
    except Exception:
        pass
    return int(os.path.getsize(path) * 8 / dur) if dur else 0


def ssim_vs_source(src_local, out_local, info, workdir, tag, dur=SSIM_WINDOW):
    """Per-frame SSIM of a sampled window: the OUTPUT against the SOURCE.

    The comparison is done at 480x270 (a wrong/scaled picture is obvious at that size and the
    decode stays cheap). NB the stats file must be a BARE name inside workdir - a Windows path
    inside `-lavfi` has its backslashes eaten by the filter parser, which silently yields zero
    frames and 0.0000 SSIM.

    Both sides are decoded with `-fps_mode passthrough` so the frame counts are the REAL ones. If
    they differ by more than 1% the source's timeline is irregular (measured on this library: a
    source with a 1.7 s hole - 1459 frames where 1500 belong - while the CFR encode fills it with
    duplicated frames), and a frame-by-frame comparison is *meaningless*: after the hole every
    frame pairs with the wrong one and SSIM collapses to 0.25 while the encode is perfectly good.
    In that case the picture gate is skipped (reported, not silently) and the caller's other
    assertions - structure, full decode, frame count not lower than the source, size ceiling -
    carry the verdict.
    """
    start = max(0.0, (info["dur"] / 2) - dur / 2)
    cmp_scale = "480:270"
    # The raw frame files MUST be per-run: the tiers run side by side (by design - copy is I/O,
    # x265 is CPU, nvenc is the GPU ASIC) and they share this stage directory, so fixed names let
    # two concurrent gates overwrite each other's decodes. Measured: a comparison came out with
    # 100% of frames below 0.85 (a convincing-looking 'systematic defect') that was nothing but two
    # runs writing ssim_ref.yuv at the same time.
    ref = os.path.join(workdir, "ssim_%d_%s_ref.yuv" % (os.getpid(), tag))
    out = os.path.join(workdir, "ssim_%d_%s_out.yuv" % (os.getpid(), tag))
    counts = []
    # Decode WITHOUT an input seek, and cut the window with the trim filter on absolute PTS.
    #
    # Two traps live here, both measured on real files in this library:
    #   1. An input seek snaps to a keyframe, and it snaps to a DIFFERENT keyframe for the source
    #      and for the encode (two 1500-frame decodes whose start times differed by ~10 s), so the
    #      frames met the wrong partners and a good encode scored mean 0.45.
    #   2. Worse, on an H.264 source with open GOPs an input seek decodes the frames after the
    #      landing point differently from a sequential decode: the SAME file compared with itself,
    #      once seeked and once decoded in order, scored min 0.25 / mean 0.82. A seek-based
    #      reference therefore condemns perfectly good encodes.
    # Sequential decoding costs a pass over the file up to the window at 480x270 - small next to the
    # encode it is judging, and it is the only decode that is guaranteed to match what the encoder
    # itself saw.
    vf = ("trim=start=%.3f:end=%.3f,setpts=PTS-STARTPTS,scale=%s:flags=bilinear,format=yuv420p"
          % (start, start + dur, cmp_scale))
    for path, dstp in ((src_local, ref), (out_local, out)):
        r = run_watched([FFMPEG, "-y", "-v", "error", "-i", path,
                         "-map", "0:v:0", "-an", "-vf", vf,
                         "-fps_mode", "passthrough", "-f", "rawvideo", dstp], dstp, timeout=7200)
        if r.returncode != 0 or not os.path.exists(dstp) or not os.path.getsize(dstp):
            return None, None, 0, None, {}
        counts.append(os.path.getsize(dstp) // (480 * 270 * 3 // 2))
    n_src, n_out = counts
    if n_src and abs(n_out - n_src) / float(n_src) > 0.01:
        for f in (ref, out):
            try:
                os.remove(f)
            except OSError:
                pass
        return None, None, 0, ("source timeline irregular: %d frames vs %d in the sampled window "
                               "(%.1f%% difference) - the CFR encode fills the source's holes, so a "
                               "frame-by-frame comparison is not applicable" % (n_src, n_out,
                                                                               100.0 * (n_out - n_src) / n_src)), {}
    subprocess.run([FFMPEG, "-y", "-v", "error", "-f", "rawvideo", "-pix_fmt", "yuv420p", "-s", cmp_scale,
                    "-i", ref, "-f", "rawvideo", "-pix_fmt", "yuv420p", "-s", cmp_scale, "-i", out,
                    "-lavfi", "ssim=stats_file=ssim_%s.log" % tag, "-f", "null", "-"],
                   capture_output=True, cwd=workdir)
    vals = []
    p = os.path.join(workdir, "ssim_%s.log" % tag)
    if os.path.exists(p):
        with open(p, encoding="utf-8", errors="replace") as fh:
            for line in fh:
                m = re.search(r"All:([0-9.]+)", line)
                if m:
                    vals.append(float(m.group(1)))
    n = len(vals)
    # SSIM is meaningless on a nearly flat frame: two almost-black frames of a fade scored 0.25
    # while their neighbours scored 0.998, so a MINIMUM over every frame condemns good encodes.
    # Measure the source frames' flatness and drop them from the verdict (reported, not hidden).
    flat = set()
    try:
        flat = _flat_frames(ref, n)
    except Exception as e:  # numpy missing / unreadable raw: fall back to using every frame
        log("   ! flat-frame scan skipped: %s" % e)
    if flat and len(flat) < n * 0.5:
        vals = [v for i, v in enumerate(vals) if i not in flat]
        log("   %d near-flat source frame(s) excluded from the picture verdict (SSIM is "
            "meaningless on them)" % len(flat))
    n = len(vals)
    lo = min(vals) if vals else None
    mean = (sum(vals) / len(vals)) if vals else None
    bad = sum(1 for v in vals if v < BAD_FRAME_SSIM)
    stats = {"below": bad, "flat": len(flat), "total": n}
    note = None
    # If the direct pairing looks bad, try to EXPLAIN the mismatch as a timeline offset before
    # condemning the file: a source whose timeline drifts (or had a hole earlier) makes the encode
    # lag it by a fixed number of frames, and a wrong-picture defect can NOT be fixed by any single
    # shift. Requiring a good score at some offset is therefore a sound discriminator.
    if n and (lo is None or lo < SSIM_TARGET or mean < 0.985):
        best = None
        nf = min(n, 1200)
        for off in range(-100, 101, 10):
            if off == 0:
                continue
            mn, mm, k = _paired_ssim(ref, out, off, nf, workdir, "off%d" % abs(off))
            if mn is not None and (best is None or mn > best[0]):
                best = (mn, mm, k, off)
        if best and best[0] >= 0.90 and best[1] >= 0.98:
            note = ("timeline offset of %+d frames between source and encode - pictures verified at "
                    "that alignment (min SSIM %.4f mean %.4f)" % (best[3], best[0], best[1]))
            lo, mean, n = best[0], best[1], best[2]
    for f in (ref, out):
        try:
            os.remove(f)
        except OSError:
            pass
    return lo, mean, n, note, stats


def _paired_ssim(ref_raw, out_raw, off, nf, workdir, tag):
    """SSIM of the first `nf` frames of out_raw against ref_raw starting `off` frames in."""
    fs = 480 * 270 * 3 // 2
    with open(ref_raw, "rb") as f:
        f.seek(max(0, off) * fs)
        a = f.read(nf * fs)
    pa = os.path.join(workdir, "pair_a_%s.yuv" % tag)
    pb = os.path.join(workdir, "pair_b_%s.yuv" % tag)
    with open(pa, "wb") as f:
        f.write(a)
    with open(out_raw, "rb") as f:
        b = f.read(nf * fs)
    with open(pb, "wb") as f:
        f.write(b)
    vals = []
    log = os.path.join(workdir, "pair_%s.log" % tag)
    subprocess.run([FFMPEG, "-y", "-v", "error", "-f", "rawvideo", "-pix_fmt", "yuv420p", "-s", "480:270",
                    "-i", pa, "-f", "rawvideo", "-pix_fmt", "yuv420p", "-s", "480:270", "-i", pb,
                    "-lavfi", "ssim=stats_file=%s" % os.path.basename(log), "-f", "null", "-"],
                   capture_output=True, cwd=workdir)
    if os.path.exists(log):
        with open(log, encoding="utf-8", errors="replace") as fh:
            for line in fh:
                m = re.search(r"All:([0-9.]+)", line)
                if m:
                    vals.append(float(m.group(1)))
    for f in (pa, pb):
        try:
            os.remove(f)
        except OSError:
            pass
    return (min(vals), sum(vals) / len(vals), len(vals)) if vals else (None, None, 0)


def pin_to_e_cores():
    """x265 tiers: keep the P-cores free for the desktop (measured 3.51x vs 4.37x realtime)."""
    try:
        ctypes.windll.kernel32.SetProcessAffinityMask(
            ctypes.windll.kernel32.GetCurrentProcess(), ctypes.c_size_t(E_CORE_MASK))
        return True
    except Exception:
        return False


def encode(src_local, dst_local, info, args, tier, vbps, q_off=0):
    cmd = [FFMPEG, "-y", "-hide_banner", "-loglevel", "warning", "-stats",
           "-i", src_local, "-map", "0:v:0", "-map", "0:a:0?", "-map_metadata", "0"]
    if tier == "copy":
        # Nothing to win: the video is copied bit-for-bit, which is the strongest possible
        # answer to "same picture quality" and the smallest file that still plays everywhere.
        cmd += ["-c:v", "copy"]
    elif tier == "nvenc":
        cmd += ["-c:v", "hevc_nvenc", "-preset", args.nvenc_preset, "-tune", "hq", "-rc", "vbr",
                "-cq", str(max(1, args.cq - q_off)), "-b:v", "0",
                # efficiency knobs verified accepted by this build (multipass 2pass-full and
                # weighted_pred are REJECTED here)
                "-spatial-aq", "1", "-temporal-aq", "1", "-rc-lookahead", "32",
                "-b_ref_mode", "middle", "-bf", "4", "-g", "240"]
    else:
        cmd += ["-c:v", "libx265", "-preset", args.x265_preset, "-crf", str(max(1, args.crf - q_off)),
                "-x265-params", "pools=%d" % args.pools]
    if tier != "copy" and vbps > 0:
        # Hard ceiling at the source's own video bitrate: the output physically cannot grow.
        cmd += ["-maxrate", str(int(vbps * SIZE_CEIL)), "-bufsize", str(int(vbps * SIZE_CEIL * 2.5))]
    if tier != "copy":
        cmd += ["-profile:v", "main", "-pix_fmt", "yuv420p", "-tag:v", "hvc1", "-fps_mode", "cfr"]
    if info["acodec"] in ("aac", "mp4a", ""):
        cmd += ["-c:a", "copy"]
    else:
        # AAC at the SOURCE's own bitrate: a fixed 256k added ~128 kbps for no audible gain on
        # this library's 128k mp3 sources, which by itself cancelled the video saving.
        abr = args.audio or ("%dk" % match_audio_kbps(src_local))
        cmd += ["-c:a", "aac", "-b:a", abr, "-ac", "2"]
    cmd += ["-movflags", "+faststart", dst_local]
    t0 = time.time()
    r = run_watched(cmd, dst_local)
    return r, time.time() - t0



def decode_check(path):
    """Full decode pass. The only check that proves the payload is not truncated/corrupt."""
    r = run([FFMPEG, "-v", "error", "-i", path, "-f", "null", "-"], timeout=14400)
    return r.returncode == 0, (r.stderr or "").strip()[:300]


# ---------------------------------------------------------------- per file

def dst_path_for(src):
    """In-place: the SAME name for an existing .mp4/.MP4 (the share is case-sensitive),
    <stem>.mp4 for anything else."""
    if os.path.splitext(src)[1].lower() == ".mp4":
        return src
    return os.path.splitext(src)[0] + ".mp4"


def already_hevc(src, info):
    """Self-describing resume test: the share copy is already HEVC hvc1 and matches."""
    if info["vcodec"] != "hevc" or info["tag"] != "hvc1":
        return False
    return True


def process(src, args, dest=None, origin=None):
    info = probe(src)
    if not info:
        return "FAIL unreadable", None
    if info["dur"] < 1.0:
        return "FAIL stub (%.1fs) - needs a re-download, not a transcode" % info["dur"], None
    live = dest or dst_path_for(src)
    redo_playable = False    # True once we know this run is re-doing a file that ALREADY plays

    if info["vcodec"] == "hevc" and info["tag"] == "hvc1" and info["pix"] in ("yuv420p",) \
            and not T.unusable_timeline(info, src)[0]:
        # Already converted. Only worth touching if the ORIGINAL is available and the current
        # file is BIGGER than it: that is precisely the old settings' damage (a fixed cq27 above
        # the source's own quality plus a fixed 256k audio). A file that is already smaller than
        # its original is left alone, which makes the whole run naturally idempotent.
        if not (origin and os.path.exists(origin)):
            return "SKIP already HEVC hvc1", None
        o_size = os.path.getsize(origin)
        if info["size"] <= o_size * 0.99:
            return "SKIP already HEVC hvc1 (%.0f MB <= original %.0f MB)" % (
                info["size"] / 1e6, o_size / 1e6), None
        log("   re-doing from the original: live %.0f MB > original %.0f MB" % (
            info["size"] / 1e6, o_size / 1e6))
        # A re-do replaces a file that already plays on iPad + Android; its only purpose is to
        # reclaim the old settings' bloat, so it earns the relaxed mean bar (owner decision,
        # SSIM_MEAN_MIN_REDO) - but only when the live file really is playable end to end
        # (HEVC hvc1 video, which the branch above guarantees, plus AAC audio).
        redo_playable = info["acodec"] in ("aac", "mp4a")
        if not redo_playable:
            log("   re-do of a file whose audio is %s (not playable) - the normal bar applies"
                % (info["acodec"] or "none"))
        src, info = origin, probe(origin)
        if not info:
            return "FAIL original unreadable", None

    # ---- policy: classify the source, then work only on the requested tier (if any) ----------
    vbps = src_video_bps(src, info)
    bpp = bpp_of(info, vbps)
    tier = tier_for_bpp(bpp)
    if args.tier and args.tier != tier:
        return "SKIP classified %s (run wants %s)" % (tier, args.tier), None
    expect_vc = "hevc" if tier != "copy" else (info["vcodec"] or "h264")
    log("   tier=%s bpp=%.4f src_video=%.0f kbps %sx%s@%.0f %s" % (
        tier, bpp, vbps / 1000, info["w"], info["h"], src_fps(info), os.path.basename(src)[:46]))

    os.makedirs(STAGE, exist_ok=True)
    tag = hashlib.sha1(src.encode("utf-8")).hexdigest()[:12]
    loc_src = os.path.join(STAGE, tag + "_src" + os.path.splitext(src)[1].lower())
    loc_out = os.path.join(STAGE, tag + "_out.mp4")
    remote_tmp = live + ".new.mp4"
    for p in (loc_src, loc_out):
        if os.path.exists(p):
            os.remove(p)
    origin_used = bool(origin and os.path.abspath(src) == os.path.abspath(origin))
    try:
        if os.path.exists(remote_tmp):
            os.remove(remote_tmp)

        # 1. fetch the source to local disk
        n_in = copy_remote_to_local(src, loc_src, expect_size=info["size"] or None)
        # 2. encode (with a bounded quality ladder: if the sampled window cannot clear the SSIM
        #    bar at the tier's default setting, raise the quality and try once more)
        attempt, q_off, el = 0, 0, 0.0
        while True:
            attempt += 1
            r, el = encode(loc_src, loc_out, info, args, tier, vbps, q_off)
            if r.returncode != 0 or not os.path.exists(loc_out):
                return "FAIL ffmpeg rc=%s %s" % (r.returncode, (r.stderr or "").strip()[-300:]), None
            # 3. verify the LOCAL output (structural + full decode)
            ok, msg = T.verify(info, loc_out, expect_vc, "aac", check_decode=False)
            if not ok:
                return "FAIL verify: %s" % msg, None
            if DECODE_CHECK and not args.no_decode_check:
                ok, msg = decode_check(loc_out)
                if not ok:
                    return "FAIL decode pass: %s" % msg, None
            # 3b. the two policy gates that make "same look, never bigger" a fact and not a hope:
            #     a per-frame SSIM of a sampled window against the source, and a hard assertion
            #     that the output video bitrate did not exceed the source's.
            ssim_note, growth = "copy (video bit-identical)", 100.0
            if tier != "copy":
                lo, mean, nfr, irreg, sstat = ssim_vs_source(loc_src, loc_out, info, STAGE, tag)
                obps = out_video_bps(loc_out, info["dur"])
                growth = 100.0 * obps / vbps if vbps else 0.0
                if irreg:
                    # Either the source's timeline has holes (frame pairing is meaningless) or the
                    # mismatch was explained by a fixed timeline offset. Both are reported, not
                    # silently swallowed, and the structural/decode/size gates still apply below.
                    ssim_note = irreg
                    log("   NOTE: %s" % ssim_note)
                    if vbps and obps > vbps * SIZE_CEIL:
                        return "FAIL size: output %.0f kbps > source %.0f kbps (%.1f%%)" % (
                            obps / 1000, vbps / 1000, growth), None
                    break
                if not nfr:
                    return "FAIL ssim: no frames compared", None
                ssim_note = "ssim min %.4f mean %.4f (%d frames)%s" % (
                    lo, mean, nfr, "" if not q_off else " at quality+%d" % q_off)
                # The verdict is a DEFECT SHAPE, not a single frame. A systematic mismatch drops the
                # mean and puts a real share of frames below the floor; a handful of scattered low
                # frames at a healthy mean is what fades, black frames and hard cuts do to SSIM.
                # Measured on this library: good encodes 0 of 1500 frames below 0.95; the file that
                # triggered this rule 5 of 1500 below 0.85 (0.33%) at 0.98 mean; the genuine
                # decoder-defect files 44 of 1499 (2.9%) with the mean falling too.
                # The MEAN bar is 0.975 instead of 0.98 when this is a re-do of a file that already
                # plays (SSIM_MEAN_MIN_REDO) - the frame-share test below is unchanged and is what
                # actually catches a defect.
                bad_share = (sstat.get("below", 0) / float(nfr)) if nfr else 0.0
                mean_min = SSIM_MEAN_MIN_REDO if redo_playable else SSIM_MEAN_MIN
                systematic = (mean is not None and mean < mean_min) or bad_share > MAX_BAD_SHARE
                if systematic:
                    if q_off < QUALITY_STEPS and (attempt == 1):
                        q_off = QUALITY_STEPS
                        log("   quality gate: mean %.4f < bar %.3f, %d of %d frames below %.2f "
                            "(%.2f%%) - retrying at quality+%d" % (
                                mean, mean_min, sstat.get("below", 0), nfr,
                                BAD_FRAME_SSIM, 100.0 * bad_share, q_off))
                        continue
                    return ("FAIL quality: %s - systematic: mean %.4f < bar %.3f, %.2f%% of frames "
                            "below %.2f%s"
                            % (ssim_note, mean, mean_min, 100.0 * bad_share, BAD_FRAME_SSIM,
                               " (re-do of an already playable file)" if redo_playable else "")), None
                if lo is not None and lo < SSIM_TARGET:
                    log("   %d scattered frame(s) below %.2f (%.2f%% of the window) with a healthy "
                        "mean - treated as fades/black, not a defect"
                        % (sstat.get("below", 0), BAD_FRAME_SSIM, 100.0 * bad_share))
                if vbps and obps > vbps * SIZE_CEIL:
                    return "FAIL size: output %.0f kbps > source %.0f kbps (%.1f%%)" % (
                        obps / 1000, vbps / 1000, growth), None
                log("   %s  %.1f%% of source video" % (ssim_note, growth))
            break
        # 4. upload under a temp name
        if args.keep_local:
            return "ok", {"sha256": "", "bytes": os.path.getsize(loc_out), "src_bytes": n_in,
                          "src_mb": round(info["size"] / 1e6, 1),
                          "out_mb": round(os.path.getsize(loc_out) / 1e6, 1),
                          "ratio": round(100.0 * os.path.getsize(loc_out) / max(info["size"], 1), 1),
                          "tier": tier, "bpp": round(bpp, 4), "video_pct": round(growth, 1),
                          "ssim": ssim_note + " [LOCAL ONLY - not uploaded]", "encode_s": round(el, 1)}
        # 4-5. upload under a temp name, then read the share copy back and compare.
        # The read-back is not ceremony: RaiDrive can return from a write with NO error while the
        # remote file is a fraction of the local one (measured: 12.5 MB of 403 MB), and its own
        # "operation didn't complete within specific time limit" toast arrives afterwards. So a
        # mismatch is retried rather than reported straight away.
        h_loc, n_loc = _sha_local(loc_out)
        last = ""
        for up in range(1, 3):
            if not copy_local_to_remote(loc_out, remote_tmp):
                last = "FAIL upload"
                time.sleep(10 * up)
                continue
            h_rem, n_rem = _sha_remote(remote_tmp)
            if (h_loc, n_loc) == (h_rem, n_rem):
                break
            last = "FAIL read-back mismatch local=%s/%d remote=%s/%d" % (
                h_loc[:12], n_loc, h_rem[:12], n_rem)
            log("   ! %s (upload %d) - retrying" % (last, up))
            try:
                os.remove(remote_tmp)
            except OSError:
                pass
            time.sleep(10 * up)
        else:
            return last, None
        # 6. atomic replace, then confirm on the share
        os.replace(remote_tmp, live)
        # NEVER delete the source when it came from the snapshot origin: that is the archive of
        # the originals, and removing it would destroy the only clean copy of the library.
        if not origin_used and not getattr(args, "keep_source", False) \
                and os.path.abspath(live) != os.path.abspath(src):
            os.remove(src)
        back = probe(live)
        tag_needed = (tier != "copy")
        if not back or back["vcodec"] != expect_vc \
                or (tag_needed and back["tag"] != "hvc1") \
                or abs(back["dur"] - info["dur"]) > max(1.0, info["dur"] * 0.01):
            return "FAIL read-back probe: %s" % (back and (back["vcodec"], back["tag"], back["dur"]),), None
        ok, msg = probe_ok(live, info, expect_vc)
        if not ok:
            return "FAIL post-replace gate on share: %s" % msg, None
        # Metadata can lie on this share. RaiDrive/Dokan serves CACHED attributes, and a rename did
        # not actually land for one file while every metadata probe insisted it had: the run
        # reported "ok 1085 MB -> 478 MB" and the live file was still the H.264 original afterwards,
        # with nothing anywhere reporting an error. So read the installed file back and compare its
        # hash with what we encoded - the only check that cannot be served from a cache.
        try:
            h_back, n_back = _sha_remote(live)
        except OSError as e:
            return "FAIL install verify: cannot read the installed file back: %s" % e, None
        if (h_back, n_back) != (h_loc, n_loc):
            return ("FAIL install verify: the live file is NOT the encode we made "
                    "(live %s/%d bytes vs ours %s/%d) - the replace did not land"
                    % (h_back[:12], n_back, h_loc[:12], n_loc)), None
        return "ok", {"sha256": h_loc, "bytes": n_loc, "src_bytes": n_in,
                      "src_mb": round(info["size"] / 1e6, 1),
                      "out_mb": round(n_loc / 1e6, 1),
                      "ratio": round(100.0 * n_loc / max(info["size"], 1), 1),
                      "tier": tier, "bpp": round(bpp, 4), "video_pct": round(growth, 1),
                      "ssim": ssim_note, "encode_s": round(el, 1)}
    finally:
        for p in (loc_src, loc_out):
            try:
                if os.path.exists(p):
                    os.remove(p)
            except OSError:
                pass


def collect(root):
    out = []
    for dp, _, fn in os.walk(root):
        for f in fn:
            if os.path.splitext(f)[1].lower() in VIDEO_EXT and ".part." not in f \
                    and "_src" not in f and ".new." not in f:
                out.append(os.path.join(dp, f))
    return sorted(out)


def main():
    global DECODE_CHECK
    ap = argparse.ArgumentParser()
    ap.add_argument("--cq", type=int, default=CQ_NVENC, help="nvenc tier quality (measured default 30)")
    ap.add_argument("--crf", type=int, default=CRF_X265, help="x265 tier quality (measured default 24)")
    ap.add_argument("--x265-preset", default=X265_PRESET)
    ap.add_argument("--pools", type=int, default=X265_POOLS, help="x265 threads (match the E-core mask)")
    ap.add_argument("--tier", choices=["copy", "nvenc", "x265"],
                    help="only touch files classified into this tier (run tiers separately)")
    ap.add_argument("--origin-root", default=None,
                    help="tree holding the PRE-CONVERSION originals (e.g. a snapshot); a file that "
                         "is already HEVC and BIGGER than its original is re-done from that "
                         "original, and the original is never modified or deleted")
    ap.add_argument("--audio", default=None,
                    help="audio bitrate; default = the SOURCE's own bitrate (never a fixed 256k)")
    ap.add_argument("--nvenc-preset", default=NVENC_PRESET)
    ap.add_argument("--limit", type=int, default=0)
    ap.add_argument("--only", default=None, help="substring filter")
    ap.add_argument("--path", default=None,
                    help="convert this ONE file instead of the library tree (its tier is still "
                         "chosen by the same bpp rule); pair with --dest to install it elsewhere")
    ap.add_argument("--dest", default=None,
                    help="install the result at this exact path (default: next to the source)")
    ap.add_argument("--keep-source", action="store_true",
                    help="never delete the source file after a verified install")
    ap.add_argument("--reverse", action="store_true")
    ap.add_argument("--dry-run", action="store_true")
    ap.add_argument("--no-decode-check", action="store_true")
    ap.add_argument("--overwrite", action="store_true", help="re-do files already HEVC")
    ap.add_argument("--keep-local", action="store_true",
                    help="pilot mode: encode + verify + gate locally and report the numbers, then "
                         "THROW THE OUTPUT AWAY - nothing is uploaded and nothing is replaced")
    args = ap.parse_args()
    DECODE_CHECK = not args.no_decode_check

    os.makedirs(STAGE, exist_ok=True)
    if args.path:
        files = [args.path]
        if not os.path.exists(args.path):
            log("!!! no such file: %s" % args.path)
            return
    else:
        files = collect(ROOT)
    if args.only:
        files = [f for f in files if args.only in f]
    if args.reverse:
        files = list(reversed(files))
    if args.limit:
        files = files[:args.limit]
    if args.tier == "x265":
        # x265 is CPU: pin it to the E-cores so the P-cores stay free (children inherit the mask).
        log("=== E-core pinning: %s" % ("on" if pin_to_e_cores() else "FAILED (running unpinned)"))
    # One lock PER TIER, not one global lock: the copy tier uses no encoder, the x265 tier is CPU
    # and the nvenc tier is the GPU ASIC, so they are safe to run side by side whereas two runs of
    # the same tier would race on dst and starve a shared unit.
    lockf = os.path.join(WORK, "run-%s.lock" % (args.tier or ("single" if args.path else "all")))
    log("=== %d file(s)  tier=%s cq=%d crf=%d audio=%s decode_check=%s lock=%s" %
        (len(files), args.tier or "auto", args.cq, args.crf, args.audio or "source-matched",
         DECODE_CHECK, os.path.basename(lockf)))

    if args.dry_run:
        plan = {}
        for f in files:
            info = probe(f)
            if not info:
                log("PLAN %s -> UNREADABLE" % disp(f))
                continue
            vbps = src_video_bps(f, info)
            b = bpp_of(info, vbps)
            t = tier_for_bpp(b)
            plan[t] = plan.get(t, 0) + 1
            log("PLAN %-5s bpp=%.4f %sx%s@%.0f %6.0f kbps %7.1f MB %s" % (
                t, b, info["w"], info["h"], src_fps(info), vbps / 1000, info["size"] / 1e6,
                disp(f)[:60]))
        log("=== plan: %s" % plan)
        return

    with T.Lock(lockf):
        stats = {"ok": 0, "skip": 0, "fail": 0}
        ledger = open(LEDGER, "a", encoding="utf-8")
        for i, f in enumerate(files, 1):
            free = shutil.disk_usage(STAGE).free / 2 ** 30
            if free < 8:
                log("!!! ABORT: only %.1f GB free on C: - stopping" % free)
                break
            log("--- %d/%d %s" % (i, len(files), disp(f)))
            t0 = time.time()
            origin = None
            if args.origin_root:
                try:
                    cand = os.path.join(args.origin_root, os.path.relpath(f, ROOT))
                except ValueError:
                    cand = None
                origin = cand if cand and os.path.exists(cand) else None
            try:
                status, extra = process(f, args, dest=args.dest or dst_path_for(f), origin=origin)
            except Exception as e:
                status, extra = "FAIL %s: %s" % (type(e).__name__, e), None
            el = time.time() - t0
            if status == "ok":
                stats["ok"] += 1
                log("   ok %.0f MB -> %.0f MB (%.0f%%) video %.1f%% enc %.0fs tot %.0fs | %s | %s" % (
                    extra["src_mb"], extra["out_mb"], extra["ratio"], extra["video_pct"],
                    extra["encode_s"], el, extra["tier"], extra["ssim"]))
            elif status.startswith("SKIP"):
                stats["skip"] += 1
                log("   %s" % status)
            else:
                stats["fail"] += 1
                log("   !! %s" % status)
            ledger.write(json.dumps({"t": time.strftime("%Y-%m-%dT%H:%M:%S"),
                                     "path": disp(f), "status": status,
                                     "secs": round(el, 1), "extra": extra},
                                    ensure_ascii=False) + "\n")
            ledger.flush()
            if i % 10 == 0:
                log("   [progress] ok=%d skip=%d fail=%d" % (stats["ok"], stats["skip"], stats["fail"]))
        ledger.close()
        log("=== done: ok=%d skipped=%d failed=%d" % (stats["ok"], stats["skip"], stats["fail"]))
        if stats["fail"]:
            sys.exit(1)


if __name__ == "__main__":
    main()