Startseite Erkunden Hilfe
Registrieren Anmelden
haiany
/
superpowers
Mirror von https://github.com/obra/superpowers.git
1
0
Fork 0
Dateien Issues 0 Wiki
Quellcode durchsuchen

fix(movie): finish native Windows media tools

Drew Ritter vor 2 Wochen
Ursprung
470bb9996e
Commit
5621d1d6aa
14 geänderte Dateien mit 809 neuen und 115 gelöschten Zeilen
Gesamtansicht Diff-Statistik anzeigen
  1. 12 25
      skills/proving-it-works-with-a-movie/scripts/assemble
  2. 120 0
      skills/proving-it-works-with-a-movie/scripts/browser_tools.py
  3. 22 13
      skills/proving-it-works-with-a-movie/scripts/burn-subtitles
  4. 7 4
      skills/proving-it-works-with-a-movie/scripts/check-movie
  5. 6 3
      skills/proving-it-works-with-a-movie/scripts/make-subtitles
  6. 66 47
      skills/proving-it-works-with-a-movie/scripts/narrate
  7. 229 0
      skills/proving-it-works-with-a-movie/scripts/windows_jobs.py
  8. 26 0
      tests/proving-it-works-with-a-movie/fixtures.py
  9. 3 3
      tests/proving-it-works-with-a-movie/run-tests.py
  10. 2 20
      tests/proving-it-works-with-a-movie/test_assembly.py
  11. 73 0
      tests/proving-it-works-with-a-movie/test_browser.py
  12. 80 0
      tests/proving-it-works-with-a-movie/test_narration.py
  13. 64 0
      tests/proving-it-works-with-a-movie/test_processes.py
  14. 99 0
      tests/proving-it-works-with-a-movie/test_subtitles.py

+ 12 - 25
skills/proving-it-works-with-a-movie/scripts/assemble
Datei anzeigen 

@@ -31,18 +31,12 @@ import shutil
 
 import subprocess
 
 import subprocess
 
 import sys
 
 import sys
 
 import tempfile
 
 import tempfile
 
-import urllib.parse
 
 from pathlib import Path
 
 from pathlib import Path
 
 
 
 import yaml
 
 import yaml
 
+from browser_tools import find_browser, render_card
 
 from media_paths import ffconcat_entry, stage_frames
 
 from media_paths import ffconcat_entry, stage_frames
 
 
 
-BROWSERS = [
 
-    "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome",
 
-    "/Applications/Chromium.app/Contents/MacOS/Chromium",
 
-    "chromium", "chromium-browser", "google-chrome", "google-chrome-stable",
 
-]
 
-
 
 CARD_HTML = """<!doctype html><meta charset="utf-8">
 
 CARD_HTML = """<!doctype html><meta charset="utf-8">
 
 <style>
 
 <style>
 
  html,body{{margin:0;width:{w}px;height:{h}px;background:{bg};color:#e8e6e1;
 
  html,body{{margin:0;width:{w}px;height:{h}px;background:{bg};color:#e8e6e1;
 
@@ -61,8 +55,9 @@ def die(msg):
 
     sys.exit(1)
 
     sys.exit(1)
 
 
 
 
 
-def run(cmd):
 
-    r = subprocess.run(cmd, capture_output=True, text=True)
 
+def run(cmd, *, cwd=None):
 
+    r = subprocess.run(cmd, cwd=cwd, capture_output=True, text=True,
 
+                       encoding="utf-8", errors="replace")
 
     if r.returncode != 0:
 
     if r.returncode != 0:
 
         die(f"{' '.join(map(str, cmd))}\n{r.stderr.strip()[:500]}")
 
         die(f"{' '.join(map(str, cmd))}\n{r.stderr.strip()[:500]}")
 
     return r
 
     return r
 
@@ -74,18 +69,6 @@ def dur(path):
 
     return float(r.stdout.strip())
 
     return float(r.stdout.strip())
 
 
 
 
 
-def find_browser(explicit):
 
-    for cand in ([explicit] if explicit else []) + BROWSERS:
 
-        if not cand:
 
-            continue
 
-        if os.path.sep in cand and Path(cand).exists():
 
-            return cand
 
-        found = shutil.which(cand)
 
-        if found:
 
-            return found
 
-    return None
 
-
 
-
 
 def make_card(scene, png, w, h, browser):
 
 def make_card(scene, png, w, h, browser):
 
     if not browser:
 
     if not browser:
 
         die("a `card` scene needs a browser (Chrome/Chromium) to render text; "
 
         die("a `card` scene needs a browser (Chrome/Chromium) to render text; "
 
@@ -97,12 +80,16 @@ def make_card(scene, png, w, h, browser):
 
         TITLE=scene.get("title", ""), SUB=scene.get("subtitle", ""))
 
         TITLE=scene.get("title", ""), SUB=scene.get("subtitle", ""))
 
     tmp = png.with_suffix(".html")
 
     tmp = png.with_suffix(".html")
 
     tmp.write_text(html, encoding="utf-8")
 
     tmp.write_text(html, encoding="utf-8")
 
-    run([browser, "--headless=new", "--disable-gpu", "--hide-scrollbars",
 
-         f"--screenshot={png}", f"--window-size={w},{h}",
 
-         "--force-device-scale-factor=1", "file://" + urllib.parse.quote(str(tmp))])
 
+    try:
 
+        render_card(tmp, png, browser=browser, width=w, height=h)
 
+    finally:
 
+        tmp.unlink(missing_ok=True)
 
 
 
 
 
 def main():
 
 def main():
 
+    for stream in (sys.stdout, sys.stderr):
 
+        if hasattr(stream, "reconfigure"):
 
+            stream.reconfigure(errors="backslashreplace")
 
     ap = argparse.ArgumentParser()
 
     ap = argparse.ArgumentParser()
 
     ap.add_argument("scenes", type=Path)
 
     ap.add_argument("scenes", type=Path)
 
     ap.add_argument("out", type=Path)
 
     ap.add_argument("out", type=Path)
 
@@ -115,7 +102,7 @@ def main():
 
         if not shutil.which(tool):
 
         if not shutil.which(tool):
 
             die(f"{tool} not on PATH")
 
             die(f"{tool} not on PATH")
 
 
 
-    doc = yaml.safe_load(args.scenes.read_text(encoding="utf-8"))
 
+    doc = yaml.safe_load(args.scenes.read_text(encoding="utf-8-sig"))
 
     base = args.scenes.parent
 
     base = args.scenes.parent
 
     res = doc.get("resolution", {}) or {}
 
     res = doc.get("resolution", {}) or {}
 
     W, H = int(res.get("width", 1920)), int(res.get("height", 1080))
 
     W, H = int(res.get("width", 1920)), int(res.get("height", 1080))

+ 120 - 0
skills/proving-it-works-with-a-movie/scripts/browser_tools.py
Datei anzeigen 

@@ -0,0 +1,120 @@
 
+"""Owned headless browser discovery and bounded card screenshots."""
 
+
 
+from __future__ import annotations
 
+
 
+import os
 
+import shutil
 
+import signal
 
+import subprocess
 
+import sys
 
+import tempfile
 
+import time
 
+from pathlib import Path
 
+
 
+from windows_jobs import WindowsJob
 
+
 
+
 
+UNIX_BROWSERS = [
 
+    "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome",
 
+    "/Applications/Chromium.app/Contents/MacOS/Chromium",
 
+    "chromium", "chromium-browser", "google-chrome", "google-chrome-stable",
 
+]
 
+
 
+
 
+def _windows_browsers() -> list[Path]:
 
+    locations = []
 
+    for variable in ("LOCALAPPDATA", "PROGRAMFILES", "PROGRAMFILES(X86)", "PROGRAMW6432"):
 
+        value = os.environ.get(variable)
 
+        if value:
 
+            root = Path(value)
 
+            locations.extend([
 
+                root / "Google/Chrome/Application/chrome.exe",
 
+                root / "Microsoft/Edge/Application/msedge.exe",
 
+            ])
 
+    return locations
 
+
 
+
 
+def _resolve(candidate: str | Path) -> str | None:
 
+    path = Path(candidate).expanduser()
 
+    if path.is_file() and (sys.platform == "win32" or os.access(path, os.X_OK)):
 
+        return str(path.resolve())
 
+    return shutil.which(str(candidate))
 
+
 
+
 
+def find_browser(explicit: str | None) -> str | None:
 
+    """Return a usable Chrome-family executable, honoring explicit values."""
 
+    if explicit:
 
+        found = _resolve(explicit)
 
+        if found:
 
+            return found
 
+        raise FileNotFoundError(f"explicit browser is not usable: {explicit}")
 
+    candidates: list[str | Path] = (
 
+        _windows_browsers() + ["chrome.exe", "msedge.exe"]
 
+        if sys.platform == "win32" else UNIX_BROWSERS
 
+    )
 
+    for candidate in candidates:
 
+        found = _resolve(candidate)
 
+        if found:
 
+            return found
 
+    return None
 
+
 
+
 
+def render_card(html: Path, png: Path, *, browser: str, width: int,
 
+                height: int, timeout: float = 20) -> None:
 
+    """Render one local HTML page and release every process it launched."""
 
+    html = Path(html).resolve()
 
+    png = Path(png).resolve()
 
+    if not html.is_file():
 
+        raise FileNotFoundError(f"card HTML does not exist: {html}")
 
+    png.parent.mkdir(parents=True, exist_ok=True)
 
+    png.unlink(missing_ok=True)
 
+    with tempfile.TemporaryDirectory(prefix="movie-browser-") as profile:
 
+        profile_path = Path(profile)
 
+        log = profile_path / "browser.log"
 
+        job = None
 
+        process = None
 
+        try:
 
+            argv = [
 
+                str(Path(browser).resolve()) if Path(browser).is_file() else browser,
 
+                "--headless=new", "--disable-gpu", "--hide-scrollbars",
 
+                "--no-first-run", "--no-default-browser-check",
 
+                f"--user-data-dir={profile_path}", f"--screenshot={png}",
 
+                f"--window-size={width},{height}", "--force-device-scale-factor=1",
 
+                html.as_uri(),
 
+            ]
 
+            with log.open("wb") as output:
 
+                if sys.platform == "win32":
 
+                    job = WindowsJob()
 
+                    pid = job.spawn(argv, profile_path, log)
 
+                else:
 
+                    process = subprocess.Popen(argv, cwd=profile_path,
 
+                        stdin=subprocess.DEVNULL, stdout=output, stderr=subprocess.STDOUT,
 
+                        start_new_session=True)
 
+                deadline = time.monotonic() + timeout
 
+                while time.monotonic() < deadline:
 
+                    # A fresh profile may keep background services alive after
 
+                    # taking the screenshot. A complete PNG is the render result.
 
+                    if png.is_file():
 
+                        data = png.read_bytes()
 
+                        if data.startswith(b"\x89PNG\r\n\x1a\n") and data.endswith(b"IEND\xaeB`\x82"):
 
+                            return
 
+                    try:
 
+                        remaining = min(0.05, max(0, deadline - time.monotonic()))
 
+                        code = job.wait(pid, remaining) if job else process.wait(timeout=remaining)
 
+                    except (TimeoutError, subprocess.TimeoutExpired):
 
+                        continue
 
+                    if code != 0 or not png.is_file() or png.stat().st_size == 0:
 
+                        detail = log.read_text(encoding="utf-8", errors="replace")[-1000:]
 
+                        raise RuntimeError(f"Browser exited with status {code} without a complete PNG: {detail}")
 
+                    # Read the file on the next iteration after a successful exit.
 
+                    time.sleep(0.01)
 
+                raise TimeoutError(f"Browser exceeded {timeout:g}s")
 
+        finally:
 
+            if job is not None:
 
+                job.close()
 
+            if process is not None:
 
+                try:
 
+                    os.killpg(process.pid, signal.SIGKILL)
 
+                except ProcessLookupError:
 
+                    pass
 
+                process.wait(timeout=5)

+ 22 - 13
skills/proving-it-works-with-a-movie/scripts/burn-subtitles
Datei anzeigen 

@@ -20,18 +20,19 @@ import argparse
 
 import shutil
 
 import shutil
 
 import subprocess
 
 import subprocess
 
 import sys
 
 import sys
 
+import tempfile
 
 from pathlib import Path
 
 from pathlib import Path
 
 
 
 
 
 def has_libass():
 
 def has_libass():
 
     out = subprocess.run(["ffmpeg", "-hide_banner", "-filters"],
 
     out = subprocess.run(["ffmpeg", "-hide_banner", "-filters"],
 
-                         capture_output=True, text=True)
 
+                         capture_output=True, text=True, encoding="utf-8", errors="replace")
 
     return any(line.split()[1:2] == ["subtitles"]
 
     return any(line.split()[1:2] == ["subtitles"]
 
                for line in out.stdout.splitlines() if line.strip())
 
                for line in out.stdout.splitlines() if line.strip())
 
 
 
 
 
-def run(cmd):
 
-    r = subprocess.run(cmd, capture_output=True, text=True)
 
+def run(cmd, *, cwd=None):
 
+    r = subprocess.run(cmd, cwd=cwd, capture_output=True, text=True, encoding="utf-8", errors="replace")
 
     if r.returncode != 0:
 
     if r.returncode != 0:
 
         print(" ".join(map(str, cmd)), file=sys.stderr)
 
         print(" ".join(map(str, cmd)), file=sys.stderr)
 
         print(r.stderr.strip()[:600], file=sys.stderr)
 
         print(r.stderr.strip()[:600], file=sys.stderr)
 
@@ -39,6 +40,9 @@ def run(cmd):
 
 
 
 
 
 def main():
 
 def main():
 
+    for stream in (sys.stdout, sys.stderr):
 
+        if hasattr(stream, "reconfigure"):
 
+            stream.reconfigure(errors="backslashreplace")
 
     ap = argparse.ArgumentParser()
 
     ap = argparse.ArgumentParser()
 
     ap.add_argument("movie", type=Path)
 
     ap.add_argument("movie", type=Path)
 
     ap.add_argument("subs", type=Path)
 
     ap.add_argument("subs", type=Path)
 
@@ -56,7 +60,10 @@ def main():
 
         if not f.exists():
 
         if not f.exists():
 
             sys.exit(f"no such file: {f}")
 
             sys.exit(f"no such file: {f}")
 
 
 
-    if not args.soft and has_libass():
 
+    args.movie, args.subs, args.out = (path.resolve() for path in
 
+                                       (args.movie, args.subs, args.out))
 
+    libass = has_libass() if not args.soft else False
 
+    if not args.soft and libass:
 
         # ffmpeg 8 dropped positional filter options, so name it explicitly:
 
         # ffmpeg 8 dropped positional filter options, so name it explicitly:
 
         # `subtitles=movie.srt` parses on 5.x and fails on 8.x, but
 
         # `subtitles=movie.srt` parses on 5.x and fails on 8.x, but
 
         # `subtitles=filename=movie.srt` works on both
 
         # `subtitles=filename=movie.srt` works on both
 
@@ -67,14 +74,16 @@ def main():
 
                  f"BorderStyle=3,Outline=1,Shadow=0,MarginV={args.margin},"
 
                  f"BorderStyle=3,Outline=1,Shadow=0,MarginV={args.margin},"
 
                  f"PrimaryColour=&H00FFFFFF&,OutlineColour=&HB0101014&,"
 
                  f"PrimaryColour=&H00FFFFFF&,OutlineColour=&HB0101014&,"
 
                  f"BackColour=&HB0101014&")
 
                  f"BackColour=&HB0101014&")
 
-        # run from the subtitle's directory: the filter treats ':' and '\' in
 
-        # paths as its own syntax, and quoting around that is a losing game
 
-        ok = run(["ffmpeg", "-nostdin", "-y", "-v", "error",
 
-                  "-i", str(args.movie.resolve()),
 
-                  "-vf", f"subtitles=filename={args.subs.name}:"
 
-                         f"force_style='{style}'",
 
-                  "-c:a", "copy", "-c:v", "libx264", "-preset", "medium",
 
-                  "-pix_fmt", "yuv420p", str(args.out.resolve())])
 
+        # Only the safe basename enters filter syntax; media paths stay absolute.
 
+        with tempfile.TemporaryDirectory(prefix="movie-subtitles-") as temporary:
 
+            directory = Path(temporary)
 
+            shutil.copyfile(args.subs, directory / "captions.srt")
 
+            ok = run(["ffmpeg", "-nostdin", "-y", "-v", "error",
 
+                      "-i", str(args.movie),
 
+                      "-vf", "subtitles=filename=captions.srt:"
 
+                             f"force_style='{style}'",
 
+                      "-c:a", "copy", "-c:v", "libx264", "-preset", "medium",
 
+                      "-pix_fmt", "yuv420p", str(args.out)], cwd=directory)
 
         if ok:
 
         if ok:
 
             print(f"burned into the picture -> {args.out}")
 
             print(f"burned into the picture -> {args.out}")
 
             return 0
 
             return 0
 
@@ -87,7 +96,7 @@ def main():
 
     if not ok:
 
     if not ok:
 
         return 1
 
         return 1
 
     print(f"embedded a soft subtitle track -> {args.out}")
 
     print(f"embedded a soft subtitle track -> {args.out}")
 
-    if not args.soft:
 
+    if not args.soft and not libass:
 
         print("NOTE: this ffmpeg has no libass, so the subtitles are a track a "
 
         print("NOTE: this ffmpeg has no libass, so the subtitles are a track a "
 
               "player must choose to show, not pixels. Anything that autoplays "
 
               "player must choose to show, not pixels. Anything that autoplays "
 
               "without subtitle UI (Slack, PR previews) will show none. Install "
 
               "without subtitle UI (Slack, PR previews) will show none. Install "

+ 7 - 4
skills/proving-it-works-with-a-movie/scripts/check-movie
Datei anzeigen 

@@ -64,7 +64,7 @@ def sample_picture(movie, workdir):
 
     out = subprocess.run(
 
     out = subprocess.run(
 
         ["ffmpeg", "-nostdin", "-v", "error", "-i", str(movie),
 
         ["ffmpeg", "-nostdin", "-v", "error", "-i", str(movie),
 
          "-vf", f"fps=1,scale={THUMB_W}:-1", "-f", "image2", str(frames / "s%05d.png")],
 
          "-vf", f"fps=1,scale={THUMB_W}:-1", "-f", "image2", str(frames / "s%05d.png")],
 
-        capture_output=True, text=True)
 
+        capture_output=True, text=True, encoding="utf-8", errors="replace")
 
     if out.returncode != 0:
 
     if out.returncode != 0:
 
         die(f"frame sampling failed: {out.stderr.strip()[:200]}")
 
         die(f"frame sampling failed: {out.stderr.strip()[:200]}")
 
     paths = sorted(frames.glob("s*.png"))
 
     paths = sorted(frames.glob("s*.png"))
 
@@ -122,6 +122,9 @@ def contact_sheet(paths, out_path, count=12):
 
 
 
 
 
 def main():
 
 def main():
 
+    for stream in (sys.stdout, sys.stderr):
 
+        if hasattr(stream, "reconfigure"):
 
+            stream.reconfigure(errors="backslashreplace")
 
     ap = argparse.ArgumentParser()
 
     ap = argparse.ArgumentParser()
 
     ap.add_argument("movie", type=Path)
 
     ap.add_argument("movie", type=Path)
 
     ap.add_argument("--out", type=Path, default=None)
 
     ap.add_argument("--out", type=Path, default=None)
 
@@ -146,7 +149,7 @@ def main():
 
     meta = subprocess.run(
 
     meta = subprocess.run(
 
         ["ffprobe", "-v", "error", "-print_format", "json",
 
         ["ffprobe", "-v", "error", "-print_format", "json",
 
          "-show_format", "-show_streams", str(args.movie)],
 
          "-show_format", "-show_streams", str(args.movie)],
 
-        capture_output=True, text=True)
 
+        capture_output=True, text=True, encoding="utf-8", errors="replace")
 
     if meta.returncode != 0:
 
     if meta.returncode != 0:
 
         die(f"ffprobe failed: {meta.stderr.strip()[:200]}")
 
         die(f"ffprobe failed: {meta.stderr.strip()[:200]}")
 
     info = json.loads(meta.stdout)
 
     info = json.loads(meta.stdout)
 
@@ -186,7 +189,7 @@ def main():
 
         embedded = any(s["codec_type"] == "subtitle" for s in info["streams"])
 
         embedded = any(s["codec_type"] == "subtitle" for s in info["streams"])
 
         if srt.exists():
 
         if srt.exists():
 
             last = 0.0
 
             last = 0.0
 
-            for line in srt.read_text(errors="replace").splitlines():
 
+            for line in srt.read_text(encoding="utf-8-sig", errors="replace").splitlines():
 
                 if "-->" in line:
 
                 if "-->" in line:
 
                     end = line.split("-->")[1].strip().split()[0]
 
                     end = line.split("-->")[1].strip().split()[0]
 
                     hh, mm, rest = end.split(":")
 
                     hh, mm, rest = end.split(":")
 
@@ -246,7 +249,7 @@ def main():
 
         (workdir / "check.json").write_text(json.dumps(
 
         (workdir / "check.json").write_text(json.dumps(
 
             {"duration": duration, "change_seconds": changes,
 
             {"duration": duration, "change_seconds": changes,
 
              "talk_seconds": talking, "failures": failures,
 
              "talk_seconds": talking, "failures": failures,
 
-             "warnings": warnings}, indent=2))
 
+             "warnings": warnings}, indent=2), encoding="utf-8")
 
 
 
     if failures:
 
     if failures:
 
         print("\nNOT SHIPPABLE. Fix, regenerate, re-run.")
 
         print("\nNOT SHIPPABLE. Fix, regenerate, re-run.")

+ 6 - 3
skills/proving-it-works-with-a-movie/scripts/make-subtitles
Datei anzeigen 

@@ -71,6 +71,9 @@ def ts(seconds):
 
 
 
 
 
 def main():
 
 def main():
 
+    for stream in (sys.stdout, sys.stderr):
 
+        if hasattr(stream, "reconfigure"):
 
+            stream.reconfigure(errors="backslashreplace")
 
     ap = argparse.ArgumentParser()
 
     ap = argparse.ArgumentParser()
 
     ap.add_argument("manifest", type=Path)
 
     ap.add_argument("manifest", type=Path)
 
     ap.add_argument("out", type=Path)
 
     ap.add_argument("out", type=Path)
 
@@ -84,11 +87,11 @@ def main():
 
     ap.add_argument("--max-secs", type=float, default=MAX_SECS)
 
     ap.add_argument("--max-secs", type=float, default=MAX_SECS)
 
     args = ap.parse_args()
 
     args = ap.parse_args()
 
 
 
-    manifest = json.loads(args.manifest.read_text())
 
+    manifest = json.loads(args.manifest.read_text(encoding="utf-8-sig"))
 
     overrides = {}
 
     overrides = {}
 
     if args.offsets_json:
 
     if args.offsets_json:
 
         overrides.update({k: float(v) for k, v in
 
         overrides.update({k: float(v) for k, v in
 
-                          json.loads(args.offsets_json.read_text()).items()})
 
+                          json.loads(args.offsets_json.read_text(encoding="utf-8-sig")).items()})
 
     for spec in args.offsets:
 
     for spec in args.offsets:
 
         k, _, v = spec.partition("=")
 
         k, _, v = spec.partition("=")
 
         overrides[k] = float(v)
 
         overrides[k] = float(v)
 
@@ -116,7 +119,7 @@ def main():
 
         if b <= a:
 
         if b <= a:
 
             b = a + MIN_SECS
 
             b = a + MIN_SECS
 
         lines += [str(i), f"{ts(a)} --> {ts(b)}", text, ""]
 
         lines += [str(i), f"{ts(a)} --> {ts(b)}", text, ""]
 
-    args.out.write_text("\n".join(lines))
 
+    args.out.write_text("\n".join(lines), encoding="utf-8")
 
     print(f"{len(cues)} cues, ends at {ts(cues[-1][1])} -> {args.out}")
 
     print(f"{len(cues)} cues, ends at {ts(cues[-1][1])} -> {args.out}")
 
     return 0
 
     return 0

+ 66 - 47
skills/proving-it-works-with-a-movie/scripts/narrate
Datei anzeigen 

@@ -28,6 +28,7 @@ import os
 
 import re
 
 import re
 
 import subprocess
 
 import subprocess
 
 import sys
 
 import sys
 
+import tempfile
 
 import urllib.request
 
 import urllib.request
 
 import wave
 
 import wave
 
 from pathlib import Path
 
 from pathlib import Path
 
@@ -67,7 +68,9 @@ import sys
 
 from faster_whisper import WhisperModel
 
 from faster_whisper import WhisperModel
 
 m = WhisperModel(sys.argv[2], device="cpu", compute_type="int8")
 
 m = WhisperModel(sys.argv[2], device="cpu", compute_type="int8")
 
 segs, _ = m.transcribe(sys.argv[1])
 
 segs, _ = m.transcribe(sys.argv[1])
 
-print(" ".join(s.text.strip() for s in segs))
 
+from pathlib import Path
 
+import json
 
+Path(sys.argv[3]).write_text(json.dumps({"text": " ".join(s.text.strip() for s in segs)}), encoding="utf-8")
 
 """
 
 """
 
 
 
 
 
@@ -75,13 +78,26 @@ def transcribe_local(wav, model="base.en"):
 
     """Transcribe with a local ASR, in its own uv env so narrate stays light.
 
     """Transcribe with a local ASR, in its own uv env so narrate stays light.
 
     Returns None when faster-whisper isn't available."""
 
     Returns None when faster-whisper isn't available."""
 
     try:
 
     try:
 
-        out = subprocess.run(
 
-            ["uv", "run", "--quiet", "--with", "faster-whisper", "python3",
 
-             "-c", ASR_SNIPPET, str(wav), model],
 
-            capture_output=True, text=True, timeout=900)
 
-    except Exception:  # noqa: BLE001 - no uv, no network: gate simply unavailable
 
+        with tempfile.TemporaryDirectory(prefix="movie-asr-") as temporary:
 
+            result_path = Path(temporary) / "transcript.json"
 
+            out = subprocess.run(
 
+                ["uv", "run", "--no-config", "--no-project", "--isolated",
 
+                 "--python", sys.executable, "--with", "faster-whisper", "python",
 
+                 "-c", ASR_SNIPPET, str(wav.resolve()), model, str(result_path)],
 
+                cwd=temporary, capture_output=True, text=True,
 
+                encoding="utf-8", errors="replace", timeout=900)
 
+            for diagnostic in (out.stdout, out.stderr):
 
+                if diagnostic.strip():
 
+                    print(f"local ASR: {diagnostic.strip()[-600:]}", file=sys.stderr)
 
+            if out.returncode != 0:
 
+                print(f"local ASR exited with status {out.returncode}", file=sys.stderr)
 
+                return None
 
+            data = json.loads(result_path.read_text(encoding="utf-8"))
 
+            text = data.get("text") if isinstance(data, dict) else None
 
+            return text.strip() if isinstance(text, str) and text.strip() else None
 
+    except (OSError, ValueError, subprocess.SubprocessError) as error:
 
+        print(f"local ASR unavailable: {error}", file=sys.stderr)
 
         return None
 
         return None
 
-    return out.stdout.strip() if out.returncode == 0 and out.stdout.strip() else None
 
 
 
 
 
 def structural_drift(text, heard):
 
 def structural_drift(text, heard):
 
@@ -153,14 +169,17 @@ def say_piper(text, out_wav, voice):
 
 def duration(path):
 
 def duration(path):
 
     out = subprocess.run(
 
     out = subprocess.run(
 
         ["ffprobe", "-v", "error", "-show_entries", "format=duration",
 
         ["ffprobe", "-v", "error", "-show_entries", "format=duration",
 
-         "-of", "csv=p=0", str(path)], capture_output=True, text=True)
 
+         "-of", "csv=p=0", str(path)], capture_output=True, text=True, encoding="utf-8", errors="replace")
 
     return round(float(out.stdout.strip()), 3)
 
     return round(float(out.stdout.strip()), 3)
 
 
 
 
 
 def main():
 
 def main():
 
+    for stream in (sys.stdout, sys.stderr):
 
+        if hasattr(stream, "reconfigure"):
 
+            stream.reconfigure(errors="backslashreplace")
 
     if len(sys.argv) == 4 and sys.argv[1] == "--drift-check":
 
     if len(sys.argv) == 4 and sys.argv[1] == "--drift-check":
 
-        script = Path(sys.argv[2]).read_text()
 
-        heard = Path(sys.argv[3]).read_text()
 
+        script = Path(sys.argv[2]).read_text(encoding="utf-8-sig")
 
+        heard = Path(sys.argv[3]).read_text(encoding="utf-8-sig")
 
         delta, worst = structural_drift(script, heard)
 
         delta, worst = structural_drift(script, heard)
 
         bad = delta > 0.15 or worst >= 4
 
         bad = delta > 0.15 or worst >= 4
 
         print(f"length change {delta:.0%}, worst run {worst} -> "
 
         print(f"length change {delta:.0%}, worst run {worst} -> "
 
@@ -181,7 +200,7 @@ def main():
 
     ap.add_argument("--asr-model", default="base.en")
 
     ap.add_argument("--asr-model", default="base.en")
 
     args = ap.parse_args()
 
     args = ap.parse_args()
 
 
 
-    doc = yaml.safe_load(args.scenes.read_text())
 
+    doc = yaml.safe_load(args.scenes.read_text(encoding="utf-8-sig"))
 
     scenes = [s for s in doc.get("scenes", []) if (s.get("narration") or "").strip()]
 
     scenes = [s for s in doc.get("scenes", []) if (s.get("narration") or "").strip()]
 
     if not scenes:
 
     if not scenes:
 
         die("no scenes with narration")
 
         die("no scenes with narration")
 
@@ -207,7 +226,7 @@ def main():
 
     if prior_path.exists():
 
     if prior_path.exists():
 
         try:
 
         try:
 
             prior = {e["id"]: e.get("text", "") for e in
 
             prior = {e["id"]: e.get("text", "") for e in
 
-                     json.loads(prior_path.read_text())}
 
+                     json.loads(prior_path.read_text(encoding="utf-8-sig"))}
 
         except Exception:  # noqa: BLE001 - a corrupt manifest just means no cache
 
         except Exception:  # noqa: BLE001 - a corrupt manifest just means no cache
 
             prior = {}
 
             prior = {}
 
     manifest, failures = [], []
 
     manifest, failures = [], []
 
@@ -216,13 +235,14 @@ def main():
 
         sid = sc["id"]
 
         sid = sc["id"]
 
         text = " ".join((sc["narration"] or "").split())
 
         text = " ".join((sc["narration"] or "").split())
 
         wav = args.outdir / f"{sid}.wav"
 
         wav = args.outdir / f"{sid}.wav"
 
-        if wav.exists() and not args.force and prior.get(sid) == text:
 
+        cached = wav.exists() and not args.force and prior.get(sid) == text
 
+        if cached:
 
             print(f"{sid}: cached")
 
             print(f"{sid}: cached")
 
         elif wav.exists() and not args.force and sid in prior:
 
         elif wav.exists() and not args.force and sid in prior:
 
             print(f"{sid}: text changed since this clip was rendered - redoing")
 
             print(f"{sid}: text changed since this clip was rendered - redoing")
 
-            args.force = True
 
-        else:
 
-            for attempt in (1, 2):
 
+        for attempt in ((1,) if cached else (1, 2)):
 
+            claimed = None
 
+            if not cached:
 
                 if engine == "openai":
 
                 if engine == "openai":
 
                     claimed = say_openai(key, text, wav, args.voice)
 
                     claimed = say_openai(key, text, wav, args.voice)
 
                 elif engine == "openai-chat":
 
                 elif engine == "openai-chat":
 
@@ -230,42 +250,41 @@ def main():
 
                 else:
 
                 else:
 
                     claimed = say_piper(text, wav, args.voice)
 
                     claimed = say_piper(text, wav, args.voice)
 
 
 
-                # a chat model reports what it said: hold it to that exactly,
 
-                # because "Sure, here it is:" is the failure it introduces
 
-                if claimed is not None:
 
-                    want, got = norm(text), norm(claimed)
 
-                    drift = abs(len(want) - len(got)) + sum(
 
-                        1 for a, b in zip(want, got) if a != b)
 
-                    if drift > max(2, len(want) // 25):
 
-                        print(f"{sid}: engine ad-libbed (attempt {attempt}, "
 
-                              f"drift {drift})")
 
-                        continue
 
-
 
-                # every engine: listen back. An ASR mangles unusual names, so
 
-                # only missing or invented CONTENT counts as a failure here.
 
-                if verify:
 
-                    heard = transcribe_local(wav, args.asr_model)
 
-                    if heard is None:
 
-                        print(f"{sid}: ok (no local ASR available - gate skipped)")
 
-                        break
 
-                    delta, worst = structural_drift(text, heard)
 
-                    if delta > 0.15 or worst >= 4:
 
-                        print(f"{sid}: what came out does not match the script "
 
-                              f"(attempt {attempt}: {delta:.0%} length change, "
 
-                              f"{worst} words in a row wrong)")
 
-                        print(f"       heard: {heard[:120]}")
 
-                        continue
 
-                    print(f"{sid}: ok (verified by ear: {delta:.0%} length "
 
-                          f"change, worst run {worst})")
 
+            # Preserve the engine transcript gate and the ASR drift thresholds.
 
+            if claimed is not None:
 
+                want, got = norm(text), norm(claimed)
 
+                drift = abs(len(want) - len(got)) + sum(
 
+                    1 for a, b in zip(want, got) if a != b)
 
+                if drift > max(2, len(want) // 25):
 
+                    print(f"{sid}: engine ad-libbed (attempt {attempt}, drift {drift})")
 
+                    continue
 
+            if verify:
 
+                heard = transcribe_local(wav, args.asr_model)
 
+                if heard is None:
 
+                    if args.verify == "on":
 
+                        print(f"{sid}: required verification unavailable", file=sys.stderr)
 
+                        failures.append(sid)
 
+                    else:
 
+                        print(f"{sid}: verification unavailable (no local ASR)")
 
                     break
 
                     break
 
-                print(f"{sid}: ok")
 
+                delta, worst = structural_drift(text, heard)
 
+                if delta > 0.15 or worst >= 4:
 
+                    print(f"{sid}: what came out does not match the script "
 
+                          f"(attempt {attempt}: {delta:.0%} length change, "
 
+                          f"{worst} words in a row wrong)")
 
+                    print(f"       heard: {heard[:120]}")
 
+                    continue
 
+                print(f"{sid}: ok (verified by ear: {delta:.0%} length "
 
+                      f"change, worst run {worst})")
 
                 break
 
                 break
 
-            else:
 
-                failures.append(sid)
 
+            print(f"{sid}: ok")
 
+            break
 
+        else:
 
+            failures.append(sid)
 
         manifest.append({"id": sid, "text": text, "wav": wav.name,
 
         manifest.append({"id": sid, "text": text, "wav": wav.name,
 
                          "duration": duration(wav)})
 
                          "duration": duration(wav)})
 
 
 
-    (args.outdir / "manifest.json").write_text(json.dumps(manifest, indent=2))
 
+    (args.outdir / "manifest.json").write_text(json.dumps(manifest, indent=2), encoding="utf-8")
 
     total = sum(m["duration"] for m in manifest)
 
     total = sum(m["duration"] for m in manifest)
 
     print(f"\n{len(manifest)} clips, {total:.1f}s total -> {args.outdir}/manifest.json")
 
     print(f"\n{len(manifest)} clips, {total:.1f}s total -> {args.outdir}/manifest.json")
 
     if engine == "piper":
 
     if engine == "piper":

+ 229 - 0
skills/proving-it-works-with-a-movie/scripts/windows_jobs.py
Datei anzeigen 

@@ -0,0 +1,229 @@
 
+"""Owned Windows process trees for movie capture, extracted from the native probe."""
 
+
 
+import os
 
+import subprocess
 
+import sys
 
+import time
 
+from pathlib import Path
 
+
 
+
 
+class WindowsJob:
 
+    """Suspended roots enter an unnamed, non-inheritable job before resuming."""
 
+
 
+    def __init__(self):
 
+        if sys.platform != "win32":
 
+            raise RuntimeError("WindowsJob requires native Windows")
 
+        import ctypes
 
+        from ctypes import wintypes as W
 
+        self.ctypes = ctypes
 
+        U64, SIZE = ctypes.c_ulonglong, ctypes.c_size_t
 
+
 
+        class STARTUPINFOW(ctypes.Structure):
 
+            _fields_ = [("cb", W.DWORD), ("lpReserved", W.LPWSTR),
 
+                        ("lpDesktop", W.LPWSTR), ("lpTitle", W.LPWSTR),
 
+                        ("dwX", W.DWORD), ("dwY", W.DWORD), ("dwXSize", W.DWORD),
 
+                        ("dwYSize", W.DWORD), ("dwXCountChars", W.DWORD),
 
+                        ("dwYCountChars", W.DWORD), ("dwFillAttribute", W.DWORD),
 
+                        ("dwFlags", W.DWORD), ("wShowWindow", W.WORD),
 
+                        ("cbReserved2", W.WORD), ("lpReserved2", ctypes.POINTER(W.BYTE)),
 
+                        ("hStdInput", W.HANDLE), ("hStdOutput", W.HANDLE), ("hStdError", W.HANDLE)]
 
+
 
+        class PROCESS_INFORMATION(ctypes.Structure):
 
+            _fields_ = [("hProcess", W.HANDLE), ("hThread", W.HANDLE),
 
+                        ("dwProcessId", W.DWORD), ("dwThreadId", W.DWORD)]
 
+
 
+        class BASIC_LIMIT(ctypes.Structure):
 
+            _fields_ = [("PerProcessUserTimeLimit", ctypes.c_longlong),
 
+                        ("PerJobUserTimeLimit", ctypes.c_longlong), ("LimitFlags", W.DWORD),
 
+                        ("MinimumWorkingSetSize", SIZE), ("MaximumWorkingSetSize", SIZE),
 
+                        ("ActiveProcessLimit", W.DWORD), ("Affinity", SIZE),
 
+                        ("PriorityClass", W.DWORD), ("SchedulingClass", W.DWORD)]
 
+
 
+        class IO_COUNTERS(ctypes.Structure):
 
+            _fields_ = [(name, U64) for name in ("ReadOperationCount", "WriteOperationCount",
 
+                        "OtherOperationCount", "ReadTransferCount", "WriteTransferCount", "OtherTransferCount")]
 
+
 
+        class EXTENDED_LIMIT(ctypes.Structure):
 
+            _fields_ = [("BasicLimitInformation", BASIC_LIMIT), ("IoInfo", IO_COUNTERS),
 
+                        ("ProcessMemoryLimit", SIZE), ("JobMemoryLimit", SIZE),
 
+                        ("PeakProcessMemoryUsed", SIZE), ("PeakJobMemoryUsed", SIZE)]
 
+
 
+        self.sizes = {"STARTUPINFOW": ctypes.sizeof(STARTUPINFOW),
 
+                      "PROCESS_INFORMATION": ctypes.sizeof(PROCESS_INFORMATION),
 
+                      "BASIC_LIMIT": ctypes.sizeof(BASIC_LIMIT),
 
+                      "IO_COUNTERS": ctypes.sizeof(IO_COUNTERS),
 
+                      "EXTENDED_LIMIT": ctypes.sizeof(EXTENDED_LIMIT)}
 
+        self.SI, self.PI = STARTUPINFOW, PROCESS_INFORMATION
 
+        self.k = ctypes.WinDLL("kernel32", use_last_error=True)
 
+        signatures = {
 
+            "CreateJobObjectW": ([ctypes.c_void_p, W.LPCWSTR], W.HANDLE),
 
+            "SetInformationJobObject": ([W.HANDLE, ctypes.c_int, ctypes.c_void_p, W.DWORD], W.BOOL),
 
+            "QueryInformationJobObject": ([W.HANDLE, ctypes.c_int, ctypes.c_void_p, W.DWORD, ctypes.POINTER(W.DWORD)], W.BOOL),
 
+            "CreateProcessW": ([W.LPCWSTR, W.LPWSTR, ctypes.c_void_p, ctypes.c_void_p, W.BOOL,
 
+                                W.DWORD, ctypes.c_void_p, W.LPCWSTR, ctypes.POINTER(STARTUPINFOW),
 
+                                ctypes.POINTER(PROCESS_INFORMATION)], W.BOOL),
 
+            "AssignProcessToJobObject": ([W.HANDLE, W.HANDLE], W.BOOL),
 
+            "ResumeThread": ([W.HANDLE], W.DWORD),
 
+            "TerminateProcess": ([W.HANDLE, W.UINT], W.BOOL),
 
+            "TerminateJobObject": ([W.HANDLE, W.UINT], W.BOOL),
 
+            "CloseHandle": ([W.HANDLE], W.BOOL),
 
+            "WaitForSingleObject": ([W.HANDLE, W.DWORD], W.DWORD),
 
+            "GetExitCodeProcess": ([W.HANDLE, ctypes.POINTER(W.DWORD)], W.BOOL),
 
+            "OpenProcess": ([W.DWORD, W.BOOL, W.DWORD], W.HANDLE),
 
+            "GetProcessTimes": ([W.HANDLE] + [ctypes.POINTER(W.FILETIME)] * 4, W.BOOL),
 
+            "IsProcessInJob": ([W.HANDLE, W.HANDLE, ctypes.POINTER(W.BOOL)], W.BOOL),
 
+        }
 
+        for name, (arguments, returns) in signatures.items():
 
+            function = getattr(self.k, name)
 
+            function.argtypes, function.restype = arguments, returns
 
+        self.handle = self.k.CreateJobObjectW(None, None)
 
+        if not self.handle:
 
+            raise ctypes.WinError(ctypes.get_last_error())
 
+        limits = EXTENDED_LIMIT()
 
+        limits.BasicLimitInformation.LimitFlags = 0x2000  # KILL_ON_JOB_CLOSE; no breakaway
 
+        if not self.k.SetInformationJobObject(self.handle, 9, ctypes.byref(limits), ctypes.sizeof(limits)):
 
+            error = ctypes.get_last_error()
 
+            self.k.CloseHandle(self.handle)
 
+            raise ctypes.WinError(error)
 
+        self.roots = {}
 
+
 
+    def spawn(self, argv: list[str], directory: Path, log: Path,
 
+              env: dict[str, str] | None = None) -> int:
 
+        import msvcrt
 
+        ctypes = self.ctypes
 
+        if not self.handle:
 
+            raise RuntimeError("Job is closed")
 
+
 
+        if not argv or not Path(argv[0]).is_absolute():
 
+            raise ValueError("An absolute executable path is required")
 
+        environment = dict(os.environ) if env is None else env
 
+        block = ctypes.create_unicode_buffer("\0".join(f"{k}={v}" for k, v in sorted(environment.items())) + "\0\0")
 
+        si, pi = self.SI(), self.PI()
 
+        si.cb, si.dwFlags = ctypes.sizeof(si), 0x100  # STARTF_USESTDHANDLES
 
+        with open(os.devnull, "rb") as stdin, log.open("ab", buffering=0) as output:
 
+            handles = [msvcrt.get_osfhandle(f.fileno()) for f in (stdin, output)]
 
+            for handle in handles:
 
+                os.set_handle_inheritable(handle, True)
 
+            si.hStdInput, si.hStdOutput, si.hStdError = handles[0], handles[1], handles[1]
 
+            try:
 
+                created = self.k.CreateProcessW(argv[0], ctypes.create_unicode_buffer(subprocess.list2cmdline(argv)),
 
+                    None, None, True, 0x404, block, str(directory), ctypes.byref(si), ctypes.byref(pi))
 
+            finally:
 
+                for handle in handles:
 
+                    os.set_handle_inheritable(handle, False)
 
+        if not created:
 
+            raise ctypes.WinError(ctypes.get_last_error())
 
+        try:
 
+            if not self.k.AssignProcessToJobObject(self.handle, pi.hProcess):
 
+                raise ctypes.WinError(ctypes.get_last_error())
 
+            if self.k.ResumeThread(pi.hThread) == 0xFFFFFFFF:
 
+                raise ctypes.WinError(ctypes.get_last_error())
 
+        except BaseException:
 
+            self.k.TerminateProcess(pi.hProcess, 1)
 
+            self.k.WaitForSingleObject(pi.hProcess, 5000)
 
+            self.k.CloseHandle(pi.hProcess)
 
+            raise
 
+        finally:
 
+            self.k.CloseHandle(pi.hThread)
 
+        self.roots[pi.dwProcessId] = pi.hProcess
 
+        return pi.dwProcessId
 
+
 
+    def wait(self, pid: int, timeout: float) -> int:
 
+        handle = self.roots[pid]
 
+        result = self.k.WaitForSingleObject(handle, max(0, int(timeout * 1000)))
 
+        if result == 258:
 
+            raise TimeoutError(f"Owned process {pid} exceeded {timeout:g}s")
 
+        if result != 0:
 
+            raise self.ctypes.WinError(self.ctypes.get_last_error())
 
+        from ctypes import wintypes as W
 
+        code = W.DWORD()
 
+        if not self.k.GetExitCodeProcess(handle, self.ctypes.byref(code)):
 
+            raise self.ctypes.WinError(self.ctypes.get_last_error())
 
+        return int(code.value)
 
+
 
+    def pids(self):
 
+        ctypes = self.ctypes
 
+        from ctypes import wintypes as W
 
+
 
+        class PROCESS_LIST(ctypes.Structure):
 
+            _fields_ = [("assigned", W.DWORD), ("count", W.DWORD), ("pids", ctypes.c_size_t * 1024)]
 
+
 
+        info = PROCESS_LIST()
 
+        if not self.k.QueryInformationJobObject(self.handle, 3, ctypes.byref(info), ctypes.sizeof(info), None):
 
+            raise ctypes.WinError(ctypes.get_last_error())
 
+        return list(info.pids[:info.count])
 
+
 
+    def process_time(self, handle):
 
+        ctypes = self.ctypes
 
+        from ctypes import wintypes as W
 
+
 
+        times = [W.FILETIME() for _ in range(4)]
 
+        if not self.k.GetProcessTimes(handle, *(ctypes.byref(value) for value in times)):
 
+            raise ctypes.WinError(ctypes.get_last_error())
 
+        return (times[0].dwHighDateTime << 32) | times[0].dwLowDateTime
 
+
 
+    def open_process(self, pid, creation=None, terminate=False):
 
+        ctypes = self.ctypes
 
+        handle = self.k.OpenProcess(0x100000 | 0x1000 | int(terminate), False, pid)
 
+        if not handle:
 
+            raise ctypes.WinError(ctypes.get_last_error())
 
+        if creation is not None and self.process_time(handle) != creation:
 
+            self.k.CloseHandle(handle)
 
+            raise RuntimeError("Process creation time changed; refusing stale PID")
 
+        return handle
 
+
 
+    def snapshot(self):
 
+        ctypes = self.ctypes
 
+        from ctypes import wintypes as W
 
+
 
+        processes = []
 
+        try:
 
+            for pid in self.pids():
 
+                try:
 
+                    handle = self.open_process(pid)
 
+                except OSError as error:
 
+                    if error.winerror == 87:  # Exited between enumeration and open.
 
+                        continue
 
+                    raise
 
+                owned = W.BOOL()
 
+                if not self.k.IsProcessInJob(handle, self.handle, ctypes.byref(owned)) or not owned.value:
 
+                    self.k.CloseHandle(handle)
 
+                    raise RuntimeError("Process is no longer a member of the owned job")
 
+                processes.append({"pid": pid, "handle": handle, "creation": self.process_time(handle)})
 
+            return processes
 
+        except BaseException:
 
+            for process in processes:
 
+                self.k.CloseHandle(process["handle"])
 
+            raise
 
+
 
+    def close(self) -> None:
 
+        ctypes = self.ctypes
 
+        if not self.handle:
 
+            return
 
+        processes = []
 
+        try:
 
+            # The job's active PID list can empty before terminated processes
 
+            # finish releasing files. Retain identities and wait for exit too.
 
+            processes = self.snapshot()
 
+            if not self.k.TerminateJobObject(self.handle, 1):
 
+                raise ctypes.WinError(ctypes.get_last_error())
 
+            deadline = time.monotonic() + 5
 
+            for process in processes:
 
+                remaining = max(0, int((deadline - time.monotonic()) * 1000))
 
+                status = self.k.WaitForSingleObject(process["handle"], remaining)
 
+                if status == 258:
 
+                    raise TimeoutError("Owned process did not finish termination")
 
+                if status != 0:
 
+                    raise ctypes.WinError(ctypes.get_last_error())
 
+            while self.pids() and time.monotonic() < deadline:
 
+                time.sleep(0.05)
 
+            if self.pids():
 
+                raise TimeoutError("Owned processes remain after job termination")
 
+        finally:
 
+            self.k.CloseHandle(self.handle)
 
+            self.handle = None
 
+            for process in processes:
 
+                self.k.CloseHandle(process["handle"])
 
+            for handle in self.roots.values():
 
+                self.k.CloseHandle(handle)
 
+            self.roots.clear()

+ 26 - 0
tests/proving-it-works-with-a-movie/fixtures.py
Datei anzeigen 

@@ -1,8 +1,12 @@
 
 """Portable fixtures for the imported movie regression suites."""
 
 """Portable fixtures for the imported movie regression suites."""
 
 
 
 import json
 
 import json
 
+import importlib.util
 
+import importlib.machinery
 
 import shutil
 
 import shutil
 
 import subprocess
 
 import subprocess
 
+import types
 
+import sys
 
 from pathlib import Path
 
 from pathlib import Path
 
 
 
 
 
@@ -44,6 +48,28 @@ def run_tool(
 
     )
 
     )
 
 
 
 
 
+def load_script(name: str) -> types.ModuleType:
 
+    """Load an extensionless movie tool as a test module."""
 
+    script = (
 
+        Path(__file__).resolve().parents[2]
 
+        / "skills/proving-it-works-with-a-movie/scripts"
 
+        / name
 
+    )
 
+    if not script.exists():
 
+        script = script.with_suffix(".py")
 
+    sys.path.insert(0, str(script.parent))
 
+    loader = importlib.machinery.SourceFileLoader(f"movie_tool_{name}", str(script))
 
+    spec = importlib.util.spec_from_loader(loader.name, loader)
 
+    if spec is None or spec.loader is None:
 
+        raise ImportError(f"cannot load movie tool {name!r}")
 
+    module = importlib.util.module_from_spec(spec)
 
+    try:
 
+        spec.loader.exec_module(module)
 
+    finally:
 
+        sys.path.pop(0)
 
+    return module
 
+
 
+
 
 def duration(path: Path) -> float:
 
 def duration(path: Path) -> float:
 
     """Measure a media file's container duration with ffprobe."""
 
     """Measure a media file's container duration with ffprobe."""
 
     ffprobe = shutil.which("ffprobe")
 
     ffprobe = shutil.which("ffprobe")

+ 3 - 3
tests/proving-it-works-with-a-movie/run-tests.py
Datei anzeigen 

@@ -13,14 +13,14 @@ from pathlib import Path
 
 
 
 IMPLEMENTED_SUITES = {
 
 IMPLEMENTED_SUITES = {
 
     "assembly": "test_assembly.py",
 
     "assembly": "test_assembly.py",
 
+    "browser": "test_browser.py",
 
     "checker": "test_checker.py",
 
     "checker": "test_checker.py",
 
     "narration": "test_narration.py",
 
     "narration": "test_narration.py",
 
     "paths": "test_paths.py",
 
     "paths": "test_paths.py",
 
+    "processes": "test_processes.py",
 
+    "subtitles": "test_subtitles.py",
 
 }
 
 }
 
 RESERVED_SUITES = {
 
 RESERVED_SUITES = {
 
-    "browser",
 
-    "subtitles",
 
-    "processes",
 
     "shells",
 
     "shells",
 
     "terminal",
 
     "terminal",
 
     "routes",
 
     "routes",

+ 2 - 20
tests/proving-it-works-with-a-movie/test_assembly.py
Datei anzeigen 

@@ -129,21 +129,7 @@ def decoded_frequency(path: Path, *, cwd: Path) -> float:
 
 
 
 
 
 def available_browser() -> str | None:
 
 def available_browser() -> str | None:
 
-    candidates = [
 
-        "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome",
 
-        "/Applications/Chromium.app/Contents/MacOS/Chromium",
 
-        "chromium",
 
-        "chromium-browser",
 
-        "google-chrome",
 
-        "google-chrome-stable",
 
-    ]
 
-    for candidate in candidates:
 
-        if Path(candidate).is_file():
 
-            return candidate
 
-        found = shutil.which(candidate)
 
-        if found:
 
-            return found
 
-    return None
 
+    return fixtures.load_script("browser_tools").find_browser(os.environ.get("MOVIE_BROWSER"))
 
 
 
 
 
 class AssemblyRegression(unittest.TestCase):
 
 class AssemblyRegression(unittest.TestCase):
 
@@ -328,10 +314,6 @@ scenes:
 
             self.assertTrue(subtitled.is_file())
 
             self.assertTrue(subtitled.is_file())
 
 
 
     def test_card_uses_longer_narration_duration(self):
 
     def test_card_uses_longer_narration_duration(self):
 
-        if sys.platform == "win32":
 
-            self.skipTest(
 
-                "native Windows card completion is pending Task5 browser discovery/file-URI work"
 
-            )
 
         missing = fixtures.missing_executables("uv", "ffmpeg", "ffprobe")
 
         missing = fixtures.missing_executables("uv", "ffmpeg", "ffprobe")
 
         if missing:
 
         if missing:
 
             self.skipTest(f"required executable(s) not on PATH: {', '.join(missing)}")
 
             self.skipTest(f"required executable(s) not on PATH: {', '.join(missing)}")
 
@@ -346,7 +328,7 @@ scenes:
 
             make_tone(narration / "card.wav", 0.8, 550, cwd=root)
 
             make_tone(narration / "card.wav", 0.8, 550, cwd=root)
 
             scenes = root / "scenes.yaml"
 
             scenes = root / "scenes.yaml"
 
             scenes.write_bytes(
 
             scenes.write_bytes(
 
-                b"resolution: { width: 160, height: 90 }\r\n"
 
+                b"\xef\xbb\xbfresolution: { width: 160, height: 90 }\r\n"
 
                 b"fps: 10\r\n"
 
                 b"fps: 10\r\n"
 
                 b"scenes:\r\n"
 
                 b"scenes:\r\n"
 
                 b"  - id: card\r\n"
 
                 b"  - id: card\r\n"

+ 73 - 0
tests/proving-it-works-with-a-movie/test_browser.py
Datei anzeigen 

@@ -0,0 +1,73 @@
 
+import os
 
+import subprocess
 
+import sys
 
+import tempfile
 
+import time
 
+import unittest
 
+from pathlib import Path
 
+from unittest.mock import patch
 
+
 
+from PIL import Image
 
+import fixtures
 
+
 
+
 
+class BrowserToolsRegression(unittest.TestCase):
 
+    def test_explicit_unusable_browser_fails_authoritatively(self):
 
+        module = fixtures.load_script("browser_tools")
 
+        with self.assertRaises(FileNotFoundError):
 
+            module.find_browser("this-browser-does-not-exist")
 
+        if os.name != "nt":
 
+            with tempfile.NamedTemporaryFile() as file:
 
+                with self.assertRaises(FileNotFoundError):
 
+                    module.find_browser(file.name)
 
+
 
+    def test_windows_chrome_edge_discovery(self):
 
+        module = fixtures.load_script("browser_tools")
 
+        with tempfile.TemporaryDirectory() as directory:
 
+            root = Path(directory)
 
+            for relative in ("Google/Chrome/Application/chrome.exe", "Microsoft/Edge/Application/msedge.exe"):
 
+                browser = root / relative
 
+                browser.parent.mkdir(parents=True)
 
+                browser.touch()
 
+                with patch.object(module.sys, "platform", "win32"), patch.dict(os.environ, {"LOCALAPPDATA": str(root)}, clear=True), patch.object(module.shutil, "which", return_value=None):
 
+                    self.assertEqual(module.find_browser(None), str(browser.resolve()))
 
+                browser.unlink()
 
+            with patch.object(module.sys, "platform", "win32"), patch.dict(os.environ, {}, clear=True), patch.object(module.shutil, "which", side_effect=lambda name: "C:/Edge/msedge.exe" if name == "msedge.exe" else None):
 
+                self.assertEqual(module.find_browser(None), "C:/Edge/msedge.exe")
 
+
 
+    def browser(self, module):
 
+        browser = module.find_browser(os.environ.get("MOVIE_BROWSER"))
 
+        if not browser:
 
+            self.skipTest("Chrome/Edge is required")
 
+        return browser
 
+
 
+    def test_render_card_special_path_is_a_real_png(self):
 
+        module = fixtures.load_script("browser_tools")
 
+        browser = self.browser(module)
 
+        with tempfile.TemporaryDirectory() as directory:
 
+            work = Path(directory) / "special path O'Brien λ # %"
 
+            work.mkdir()
 
+            html, png = work / "card page.html", work / "card.png"
 
+            html.write_text("<meta charset='utf-8'><style>body{margin:0;background:rgb(255,0,0)}</style><p>λ</p>", encoding="utf-8")
 
+            module.render_card(html, png, browser=browser, width=640, height=360)
 
+            with Image.open(png) as image:
 
+                self.assertEqual(image.size, (640, 360))
 
+                self.assertEqual(image.convert("RGB").getpixel((500, 200)), (255, 0, 0))
 
+
 
+    def test_render_card_timeout_preserves_unrelated_process(self):
 
+        module = fixtures.load_script("browser_tools")
 
+        browser = self.browser(module)
 
+        sentinel = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"])
 
+        try:
 
+            with tempfile.TemporaryDirectory() as directory:
 
+                work = Path(directory)
 
+                html = work / "card.html"
 
+                html.write_text("<p>timeout</p>", encoding="utf-8")
 
+                start = time.monotonic()
 
+                with self.assertRaises(TimeoutError):
 
+                    module.render_card(html, work / "card.png", browser=browser, width=640, height=360, timeout=0)
 
+                self.assertLess(time.monotonic() - start, 10)
 
+                self.assertIsNone(sentinel.poll())
 
+        finally:
 
+            sentinel.terminate()
 
+            sentinel.wait(timeout=10)

+ 80 - 0
tests/proving-it-works-with-a-movie/test_narration.py
Datei anzeigen 

@@ -1,6 +1,7 @@
 
 import tempfile
 
 import tempfile
 
 import unittest
 
 import unittest
 
 from pathlib import Path
 
 from pathlib import Path
 
+from unittest.mock import patch
 
 
 
 import fixtures
 
 import fixtures
 
 
 
@@ -56,6 +57,85 @@ class NarrationDriftRegression(unittest.TestCase):
 
     def test_empty_clip_fails(self):
 
     def test_empty_clip_fails(self):
 
         self.drift(1, "you")
 
         self.drift(1, "you")
 
 
 
+    def test_cached_audio_requires_requested_verification(self):
 
+        import json
 
+        import sys
 
+        module = fixtures.load_script("narrate")
 
+        with tempfile.TemporaryDirectory() as tmp:
 
+            work = Path(tmp)
 
+            output = work / "voice"
 
+            output.mkdir()
 
+            (output / "clip.wav").write_bytes(b"cached audio fixture")
 
+            (output / "manifest.json").write_text(json.dumps([
 
+                {"id": "clip", "text": "Read this sentence.", "wav": "clip.wav",
 
+                 "duration": 1.0}
 
+            ]), encoding="utf-8")
 
+            scenes = work / "scenes.yaml"
 
+            scenes.write_text(json.dumps({"scenes": [
 
+                {"id": "clip", "narration": "Read this sentence."}
 
+            ]}), encoding="utf-8")
 
+            argv = ["narrate", str(scenes), str(output),
 
+                    "--engine", "piper", "--verify", "on"]
 
+            with patch.object(sys, "argv", argv), \
 
+                 patch.object(module, "openai_key", return_value=None), \
 
+                 patch.object(module, "duration", return_value=1.0), \
 
+                 patch.object(module, "transcribe_local", return_value=None):
 
+                self.assertNotEqual(module.main(), 0)
 
+
 
+class TranscriptionProtocolRegression(unittest.TestCase):
 
+    def test_owned_json_is_used_instead_of_library_stdout(self):
 
+        import json
 
+        import subprocess
 
+        import sys
 
+        import io
 
+        from contextlib import redirect_stderr
 
+        module = fixtures.load_script("narrate")
 
+        def child(argv, **kwargs):
 
+            self.assertIn("--isolated", argv)
 
+            self.assertIn("--no-project", argv)
 
+            self.assertIn("--no-config", argv)
 
+            self.assertEqual(argv[argv.index("--python") + 1], sys.executable)
 
+            self.assertNotEqual(Path(kwargs["cwd"]), Path.cwd())
 
+            Path(argv[-1]).write_text(json.dumps({"text": "Correct λ transcript"}), encoding="utf-8")
 
+            return subprocess.CompletedProcess(argv, 0, "native library warning", "diagnostic")
 
+        diagnostics = io.StringIO()
 
+        with patch.object(module.subprocess, "run", side_effect=child), redirect_stderr(diagnostics):
 
+            self.assertEqual(module.transcribe_local(Path("clip.wav")), "Correct λ transcript")
 
+        self.assertIn("native library warning", diagnostics.getvalue())
 
+        self.assertIn("diagnostic", diagnostics.getvalue())
 
+
 
+    def test_failed_absent_and_malformed_child_results_are_unavailable(self):
 
+        import subprocess
 
+        module = fixtures.load_script("narrate")
 
+        for payload, code in ((None, 0), ("garbage", 0), ('{"text": 7}', 0), ('{"text": ""}', 0), ('{"text": "words"}', 1)):
 
+            with self.subTest(payload=payload, code=code):
 
+                def child(argv, **kwargs):
 
+                    if payload is not None and "--isolated" in argv:
 
+                        Path(argv[-1]).write_text(payload, encoding="utf-8")
 
+                    return subprocess.CompletedProcess(argv, code, "misleading stdout", "error")
 
+                with patch.object(module.subprocess, "run", side_effect=child):
 
+                    self.assertIsNone(module.transcribe_local(Path("clip.wav")))
 
+
 
+    def test_fresh_and_off_then_on_clips_require_asr(self):
 
+        import json
 
+        import sys
 
+        module = fixtures.load_script("narrate")
 
+        for cached in (False, True):
 
+            with self.subTest(cached=cached), tempfile.TemporaryDirectory() as directory:
 
+                root = Path(directory)
 
+                scenes = root / "scenes.yaml"
 
+                scenes.write_text(json.dumps({"scenes": [{"id": "clip", "narration": "Read this sentence."}]}), encoding="utf-8-sig")
 
+                output = root / "voice"
 
+                def synthesize(text, wav, voice):
 
+                    wav.write_bytes(b"branch policy fixture")
 
+                argv = ["narrate", str(scenes), str(output), "--engine", "piper", "--verify"]
 
+                with patch.object(module, "openai_key", return_value=None), patch.object(module, "say_piper", side_effect=synthesize), patch.object(module, "duration", return_value=1.0), patch.object(module, "transcribe_local", return_value=None):
 
+                    if cached:
 
+                        with patch.object(sys, "argv", [*argv, "off"]):
 
+                            self.assertEqual(module.main(), 0)
 
+                    with patch.object(sys, "argv", [*argv, "on"]):
 
+                        self.assertNotEqual(module.main(), 0)
 
+
 
 
 
 if __name__ == "__main__":
 
 if __name__ == "__main__":
 
     unittest.main()
 
     unittest.main()

+ 64 - 0
tests/proving-it-works-with-a-movie/test_processes.py
Datei anzeigen 

@@ -0,0 +1,64 @@
 
+import os
 
+import tempfile
 
+import unittest
 
+from pathlib import Path
 
+
 
+import fixtures
 
+
 
+
 
+@unittest.skipUnless(os.name == "nt", "Windows Job ownership is native Windows only")
 
+class WindowsJobRegression(unittest.TestCase):
 
+    def test_job_wait_and_close_own_child_process(self):
 
+        module = fixtures.load_script("windows_jobs")
 
+        with tempfile.TemporaryDirectory() as directory:
 
+            root = Path(directory)
 
+            log = root / "child.log"
 
+            job = module.WindowsJob()
 
+            try:
 
+                pid = job.spawn([str(Path(os.environ["ComSpec"])), "/c", "exit 7"], root, log)
 
+                self.assertEqual(job.wait(pid, 10), 7)
 
+            finally:
 
+                job.close()
 
+
 
+@unittest.skipUnless(os.name == "nt", "Windows Job ownership is native Windows only")
 
+class WindowsJobCleanupRegression(unittest.TestCase):
 
+    def test_timeout_closes_descendants_without_touching_sentinel(self):
 
+        import subprocess
 
+        import sys
 
+        import time
 
+        module = fixtures.load_script("windows_jobs")
 
+        sentinel = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"])
 
+        try:
 
+            with tempfile.TemporaryDirectory() as directory:
 
+                root = Path(directory)
 
+                child = "import time; time.sleep(60)"
 
+                code = f"import subprocess,sys,time; subprocess.Popen([sys.executable, '-c', {child!r}]); time.sleep(60)"
 
+                job = module.WindowsJob()
 
+                snapshot = []
 
+                try:
 
+                    pid = job.spawn([sys.executable, "-c", code], root, root / "log")
 
+                    with self.assertRaises(TimeoutError):
 
+                        job.wait(pid, 0.05)
 
+                    deadline = time.monotonic() + 5
 
+                    while len(job.pids()) < 2 and time.monotonic() < deadline:
 
+                        time.sleep(0.05)
 
+                    snapshot = job.snapshot()
 
+                    self.assertGreaterEqual(len(snapshot), 2)
 
+                    start = time.monotonic()
 
+                    job.close()
 
+                    job.close()
 
+                    self.assertLess(time.monotonic() - start, 10)
 
+                    for process in snapshot:
 
+                        self.assertEqual(job.k.WaitForSingleObject(process["handle"], 0), 0)
 
+                    self.assertIsNone(sentinel.poll())
 
+                finally:
 
+                    job.close()
 
+                    for process in snapshot:
 
+                        job.k.CloseHandle(process["handle"])
 
+        finally:
 
+            sentinel.terminate()
 
+            sentinel.wait(timeout=10)
 
+
 
+
 
+if __name__ == "__main__":
 
+    unittest.main()

+ 99 - 0
tests/proving-it-works-with-a-movie/test_subtitles.py
Datei anzeigen 

@@ -0,0 +1,99 @@
 
+import tempfile
 
+import sys
 
+import io
 
+from contextlib import redirect_stderr, redirect_stdout
 
+import unittest
 
+from pathlib import Path
 
+from unittest.mock import patch
 
+
 
+import fixtures
 
+
 
+
 
+class SubtitlePathRegression(unittest.TestCase):
 
+    def test_hard_burn_runs_from_safe_directory_with_absolute_media(self):
 
+        module = fixtures.load_script("burn-subtitles")
 
+        with tempfile.TemporaryDirectory() as directory:
 
+            root = Path(directory) / "movie O'Brien λ"
 
+            root.mkdir()
 
+            movie, subs, output = root / "in.mp4", root / "nested" / "captions.srt", root / "out.mp4"
 
+            subs.parent.mkdir()
 
+            movie.write_bytes(b"movie")
 
+            subs.write_bytes("\ufeff1\r\n00:00:00,000 --> 00:00:01,000\r\nλ\r\n".encode("utf-8"))
 
+            calls = []
 
+
 
+            def fake_run(cmd, *, cwd=None):
 
+                calls.append((cmd, cwd))
 
+                if cwd is not None:
 
+                    self.assertEqual((cwd / "captions.srt").read_bytes(), subs.read_bytes())
 
+                return True
 
+
 
+            with patch.object(sys, "argv", ["burn-subtitles", str(movie), str(subs), str(output)]), patch.object(module, "has_libass", return_value=True), patch.object(module, "run", side_effect=fake_run):
 
+                self.assertEqual(module.main(), 0)
 
+            command, cwd = calls[0]
 
+            self.assertEqual(cwd.name.startswith("movie-subtitles-"), True)
 
+            self.assertIn(str(movie.resolve()), command)
 
+            self.assertIn(str(output.resolve()), command)
 
+            self.assertTrue(any(value.startswith("subtitles=filename=captions.srt") for value in command))
 
+            self.assertFalse(cwd.exists())
 
+
 
+    def test_burn_failure_is_reported_separately_from_missing_libass(self):
 
+        module = fixtures.load_script("burn-subtitles")
 
+        with tempfile.TemporaryDirectory() as directory:
 
+            root = Path(directory)
 
+            movie, subs, output = root / "in.mp4", root / "captions.srt", root / "out.mp4"
 
+            movie.write_bytes(b"movie")
 
+            subs.write_text("1\n00:00:00,000 --> 00:00:01,000\ncaption\n", encoding="utf-8")
 
+            with patch.object(sys, "argv", ["burn-subtitles", str(movie), str(subs), str(output)]), patch.object(module, "has_libass", return_value=True), patch.object(module, "run", return_value=False):
 
+                self.assertEqual(module.main(), 1)
 
+
 
+class SubtitleIntegrationRegression(unittest.TestCase):
 
+    def test_bom_manifest_and_offsets_write_utf8_under_legacy_console(self):
 
+        import json
 
+        import os
 
+        with tempfile.TemporaryDirectory() as directory:
 
+            root = Path(directory)
 
+            manifest, offsets, out = root / "manifest.json", root / "offsets.json", root / "λ.srt"
 
+            manifest.write_bytes(('\ufeff' + json.dumps([{"id": "clip", "text": "Unicode λ café", "duration": 1}], ensure_ascii=False) + '\r\n').encode('utf-8'))
 
+            offsets.write_text('{"clip": 2}', encoding="utf-8-sig")
 
+            env = dict(os.environ, PYTHONIOENCODING="cp1252", PYTHONUTF8="0")
 
+            result = fixtures.run_tool("make-subtitles", [str(manifest), str(out), "--offsets-json", str(offsets)], cwd=root, env=env)
 
+            self.assertEqual(result.returncode, 0, fixtures.output_text(result))
 
+            text = out.read_text(encoding="utf-8")
 
+            self.assertIn("Unicode λ café", text)
 
+            self.assertIn("00:00:02,000 --> 00:00:03,000", text)
 
+            self.assertFalse(out.read_bytes().startswith(b'\xef\xbb\xbf'))
 
+
 
+    def test_hard_subtitles_are_pixels_in_nested_special_path(self):
 
+        import subprocess
 
+        module = fixtures.load_script("burn-subtitles")
 
+        if not module.has_libass():
 
+            self.skipTest("libass FFmpeg is required for hard subtitle pixels")
 
+        with tempfile.TemporaryDirectory() as directory:
 
+            root = Path(directory) / "O'Brien λ # %"
 
+            root.mkdir()
 
+            movie, subs, output = root / "in.mp4", root / "nested" / "O'Brien.srt", root / "out.mp4"
 
+            subs.parent.mkdir()
 
+            subs.write_bytes("\ufeff1\r\n00:00:00,000 --> 00:00:01,000\r\nVisible caption\r\n".encode("utf-8"))
 
+            fixtures._run_ffmpeg(["-f", "lavfi", "-i", "color=c=black:s=640x360:d=1", "-c:v", "libx264", str(movie)], cwd=root)
 
+            result = fixtures.run_tool("burn-subtitles", [str(movie), str(subs), str(output)], cwd=Path(directory))
 
+            self.assertEqual(result.returncode, 0, fixtures.output_text(result))
 
+            self.assertIn("burned into the picture", fixtures.output_text(result))
 
+            frame = subprocess.run(["ffmpeg", "-v", "error", "-ss", "0.5", "-i", str(output), "-frames:v", "1", "-pix_fmt", "gray", "-f", "rawvideo", "-"], capture_output=True, check=True).stdout
 
+            self.assertGreater(sum(value > 180 for value in frame), 50)
 
+
 
+    def test_burn_failure_fallback_does_not_claim_missing_libass(self):
 
+        module = fixtures.load_script("burn-subtitles")
 
+        for libass in (True, False):
 
+            with self.subTest(libass=libass), tempfile.TemporaryDirectory() as directory:
 
+                root = Path(directory)
 
+                movie, subs = root / "in.mp4", root / "in.srt"
 
+                movie.touch(); subs.touch()
 
+                stdout, stderr = io.StringIO(), io.StringIO()
 
+                with patch.object(sys, "argv", ["burn-subtitles", str(movie), str(subs), str(root / "out.mp4")]), patch.object(module, "has_libass", return_value=libass), patch.object(module, "run", side_effect=[False, True] if libass else [True]), redirect_stdout(stdout), redirect_stderr(stderr):
 
+                    self.assertEqual(module.main(), 0)
 
+                self.assertEqual("no libass" in stdout.getvalue(), not libass)
 
+                self.assertEqual("burn failed" in stderr.getvalue(), libass)
 
+
 
+
 
+if __name__ == "__main__":
 
+    unittest.main()