fix: make write_smil strictly additive, add broken-SMIL scanner

scripts/find_broken_smils.py

@@ -0,0 +1,224 @@
+#!/usr/bin/env python3
+"""Find (and optionally repair) SMIL manifests damaged by the archive transcriber.
+
+Incident 2026-07-13: older versions of archive_transcriber.py created a SMIL
+from scratch when the original was missing, or regenerated it after an XML
+parse error. Such SMILs reference only ONE video variant (the best rendition,
+usually _1080p) instead of all transcoder renditions, carry no NAME, and
+inject malformed CODECS ("avc1,mp4a") into the Wowza playlist — breaking
+adaptive playback.
+
+Fingerprint of a transcriber-created SMIL:
+  - exactly one <video> node in <switch>
+  - that node has <param name="videoCodecId"/audioCodecId"> children
+    (the transcoder's own generator does not write these params)
+
+This script needs only the Python 3 standard library. Run it on a host with
+access to the storage (e.g. prod12):
+
+    # 1. Scan only (writes report + list of SMILs to regenerate; touches nothing)
+    python3 find_broken_smils.py /mnt/vod/srv/storage/transcoded --report smil_report.jsonl
+
+    # 2. Restore originals from the oldest .bak where one exists (dry-run first)
+    python3 find_broken_smils.py /mnt/vod/srv/storage/transcoded --restore-bak
+    python3 find_broken_smils.py /mnt/vod/srv/storage/transcoded --restore-bak --apply
+
+SMILs that are broken and have NO .bak never had an original (the transcoder
+pipeline never wrote one) — they are listed in --regen-list output and must be
+regenerated by the transcoder-side SMIL generator, not by this script.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import shutil
+import subprocess
+import sys
+import xml.etree.ElementTree as ET
+from pathlib import Path
+from typing import Iterator, Optional
+
+
+def iter_paths(root: Path, patterns: list[str]) -> Iterator[Path]:
+    """Yield files under root matching any of the glob patterns (uses find when available)."""
+    try:
+        find_args = ["find", str(root), "-type", "f", "("]
+        for i, pat in enumerate(patterns):
+            if i:
+                find_args.append("-o")
+            find_args += ["-name", pat]
+        find_args.append(")")
+        proc = subprocess.Popen(find_args, stdout=subprocess.PIPE, text=True)
+        assert proc.stdout is not None
+        for line in proc.stdout:
+            yield Path(line.rstrip("\n"))
+        proc.wait()
+    except FileNotFoundError:
+        for pat in patterns:
+            yield from root.rglob(pat)
+
+
+def iter_targets(root: Path) -> Iterator[tuple[str, Path]]:
+    """Yield ("smil", path) for primary SMILs and ("missing", expected_smil_path)
+    for videos that have a 1080p variant but no SMIL at all (e.g. the generator
+    had no write permission when the video was transcoded)."""
+    for p in iter_paths(root, ["*.smil", "*_1080p.mp4"]):
+        if p.name.endswith("_audio.smil"):
+            continue
+        if p.name.endswith(".smil"):
+            yield "smil", p
+        else:
+            expected = p.with_name(p.name.replace("_1080p.mp4", ".smil"))
+            if not expected.exists():
+                yield "missing", expected
+
+
+def has_transcriber_fingerprint(videos: list[ET.Element]) -> bool:
+    """True if any <video> node carries the codec <param> children only our transcriber wrote."""
+    return any(
+        param.get("name") in ("videoCodecId", "audioCodecId") for v in videos for param in v.findall("param")
+    )
+
+
+def classify(smil_path: Path) -> dict:
+    """Classify a single SMIL file. Returns a report record."""
+    record: dict = {"smil": str(smil_path), "verdict": None}
+
+    # Legacy runs (pre-timestamp) wrote a static "<name>.smil.bak"; it is the
+    # oldest backup where present, so list it first.
+    legacy_bak = smil_path.parent / (smil_path.name + ".bak")
+    baks = ([legacy_bak] if legacy_bak.exists() else []) + sorted(smil_path.parent.glob(smil_path.name + ".bak.*"))
+    record["baks"] = [str(b) for b in baks]
+
+    try:
+        tree = ET.parse(smil_path)
+    except (ET.ParseError, OSError) as exc:
+        record["verdict"] = "UNPARSEABLE"
+        record["error"] = str(exc)
+        return record
+
+    switch = tree.getroot().find("body/switch")
+    if switch is None:
+        record["verdict"] = "NO_SWITCH"
+        return record
+
+    videos = switch.findall("video")
+    textstreams = switch.findall("textstream")
+    record["n_videos"] = len(videos)
+    record["n_textstreams"] = len(textstreams)
+    record["textstream_srcs"] = [ts.get("src", "") for ts in textstreams]
+
+    has_codec_params = has_transcriber_fingerprint(videos)
+    record["has_codec_params"] = has_codec_params
+
+    missing_subs = [src for src in record["textstream_srcs"] if src and not (smil_path.parent / src).exists()]
+    if missing_subs:
+        record["missing_subtitle_files"] = missing_subs
+
+    if len(videos) == 1 and has_codec_params:
+        record["verdict"] = "BROKEN_TRANSCRIBER_CREATED" if not baks else "BROKEN_HAS_BACKUP"
+    elif len(videos) >= 2:
+        record["verdict"] = "OK_WITH_SUBS" if textstreams else "OK_UNTOUCHED"
+    else:
+        # single video without our fingerprint: possibly a legit single-rendition
+        # asset — flag for manual review rather than auto-repair
+        record["verdict"] = "SINGLE_VIDEO_REVIEW"
+
+    return record
+
+
+def oldest_valid_bak(baks: list[str]) -> Optional[Path]:
+    """Return the oldest backup that parses and contains >=1 video node without our fingerprint.
+
+    `baks` is already ordered oldest-first (legacy static .bak, then timestamped).
+    """
+    for bak in baks:
+        p = Path(bak)
+        try:
+            tree = ET.parse(p)
+        except (ET.ParseError, OSError):
+            continue
+        switch = tree.getroot().find("body/switch")
+        if switch is None:
+            continue
+        videos = switch.findall("video")
+        if videos and not has_transcriber_fingerprint(videos):
+            return p
+    return None
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    parser.add_argument("root", type=Path, help="Archive root, e.g. /mnt/vod/srv/storage/transcoded")
+    parser.add_argument("--report", type=Path, default=Path("smil_report.jsonl"), help="JSONL report output")
+    parser.add_argument(
+        "--regen-list",
+        type=Path,
+        default=Path("smils_to_regenerate.txt"),
+        help="Output list of broken SMILs with no usable backup (regenerate with the transcoder-side generator)",
+    )
+    parser.add_argument("--restore-bak", action="store_true", help="Restore broken SMILs from their oldest valid .bak")
+    parser.add_argument("--apply", action="store_true", help="Actually write changes (default is dry-run)")
+    args = parser.parse_args()
+
+    counts: dict = {}
+    broken_with_bak: list[dict] = []
+    regen_needed: list[str] = []
+    scanned = 0
+
+    with open(args.report, "w", encoding="utf-8") as report:
+        for kind, smil_path in iter_targets(args.root):
+            rec: dict
+            if kind == "missing":
+                rec = {"smil": str(smil_path), "verdict": "MISSING_SMIL", "baks": []}
+            else:
+                rec = classify(smil_path)
+            scanned += 1
+            counts[rec["verdict"]] = counts.get(rec["verdict"], 0) + 1
+            if rec["verdict"] != "OK_UNTOUCHED":
+                report.write(json.dumps(rec, ensure_ascii=False) + "\n")
+            if rec["verdict"] == "BROKEN_HAS_BACKUP":
+                broken_with_bak.append(rec)
+            elif rec["verdict"] in ("BROKEN_TRANSCRIBER_CREATED", "UNPARSEABLE", "NO_SWITCH", "MISSING_SMIL"):
+                regen_needed.append(rec["smil"])
+            if scanned % 10000 == 0:
+                print(f"scanned {scanned}...", file=sys.stderr)
+
+    args.regen_list.write_text("\n".join(regen_needed) + ("\n" if regen_needed else ""))
+
+    print(f"\nScanned {scanned} SMIL files. Verdicts:")
+    for verdict, n in sorted(counts.items()):
+        print(f"  {n:8d}  {verdict}")
+    print(f"\nReport: {args.report}")
+    print(f"Regeneration list ({len(regen_needed)} files, no usable backup): {args.regen_list}")
+
+    if args.restore_bak:
+        restored = skipped = failed = 0
+        for rec in broken_with_bak:
+            bak = oldest_valid_bak(rec["baks"])
+            if bak is None:
+                regen_needed.append(rec["smil"])
+                skipped += 1
+                continue
+            if args.apply:
+                try:
+                    shutil.copy2(bak, rec["smil"])
+                except OSError as exc:
+                    print(f"FAILED to restore {rec['smil']} <- {bak.name}: {exc}", file=sys.stderr)
+                    failed += 1
+                    continue
+            print(f"{'RESTORED' if args.apply else 'WOULD RESTORE'} {rec['smil']} <- {bak.name}")
+            restored += 1
+        mode = "" if args.apply else " (dry-run, use --apply)"
+        print(
+            f"\nRestore{mode}: {restored} restored, {failed} failed, "
+            f"{skipped} had no valid backup (added to regen list)"
+        )
+        args.regen_list.write_text("\n".join(regen_needed) + ("\n" if regen_needed else ""))
+
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())