Add MVLog remote worker

mvlog_worker.py

@@ -0,0 +1,229 @@
+#!/usr/bin/env python3
+"""Remote MVLog worker.
+
+Scans the web machine's in-dir, renders changed jobs locally with movmaker,
+and publishes generated MP4 files back to the web machine's out-dir.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+from datetime import datetime, timezone
+from pathlib import Path
+from shlex import quote
+
+ROOT = Path(__file__).resolve().parent
+MOVMAKER = ROOT / "movmaker.py"
+STATE_NAME = ".movmaker-state.json"
+LOCK_NAME = ".movmaker-lock"
+
+
+def now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def run(cmd: list[str], *, input_text: str | None = None, check: bool = True) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(cmd, input=input_text, text=True, capture_output=True, check=check)
+
+
+def ssh(host: str, command: str, *, check: bool = True, input_text: str | None = None) -> subprocess.CompletedProcess[str]:
+    return run(["ssh", host, command], input_text=input_text, check=check)
+
+
+def remote_q(path: str) -> str:
+    return quote(path)
+
+
+def list_jobs(host: str, remote_root: str) -> list[str]:
+    cmd = f"cd {remote_q(remote_root)} && find in-dir -mindepth 1 -maxdepth 1 -type d -printf '%f\\n' | sort"
+    out = ssh(host, cmd).stdout
+    return [line.strip() for line in out.splitlines() if line.strip()]
+
+
+def remote_file_exists(host: str, path: str) -> bool:
+    return ssh(host, f"test -f {remote_q(path)}", check=False).returncode == 0
+
+
+def acquire_lock(host: str, job_dir: str) -> bool:
+    lock = f"{job_dir}/{LOCK_NAME}"
+    cmd = f"mkdir {remote_q(lock)} 2>/dev/null && date -Iseconds > {remote_q(lock + '/started_at')}"
+    return ssh(host, cmd, check=False).returncode == 0
+
+
+def release_lock(host: str, job_dir: str) -> None:
+    ssh(host, f"rm -rf {remote_q(job_dir + '/' + LOCK_NAME)}", check=False)
+
+
+def remote_fingerprint(host: str, job_dir: str) -> str:
+    script = r'''
+import hashlib, json, os, sys
+root = sys.argv[1]
+entries = []
+for dirpath, dirnames, filenames in os.walk(root):
+    dirnames[:] = [d for d in dirnames if d != '.movmaker-lock']
+    for name in filenames:
+        if name == '.movmaker-state.json':
+            continue
+        path = os.path.join(dirpath, name)
+        rel = os.path.relpath(path, root)
+        st = os.stat(path)
+        entries.append([rel, st.st_size, st.st_mtime_ns])
+entries.sort()
+print(hashlib.sha256(json.dumps(entries, ensure_ascii=False, separators=(',', ':')).encode('utf-8')).hexdigest())
+'''
+    cmd = f"python3 -c {quote(script)} {remote_q(job_dir)}"
+    return ssh(host, cmd).stdout.strip()
+
+
+def read_state(host: str, job_dir: str) -> dict:
+    path = f"{job_dir}/{STATE_NAME}"
+    res = ssh(host, f"cat {remote_q(path)} 2>/dev/null", check=False)
+    if res.returncode != 0 or not res.stdout.strip():
+        return {}
+    try:
+        return json.loads(res.stdout)
+    except json.JSONDecodeError:
+        return {}
+
+
+def write_state(host: str, job_dir: str, state: dict) -> None:
+    path = f"{job_dir}/{STATE_NAME}"
+    data = json.dumps(state, ensure_ascii=False, indent=2) + "\n"
+    ssh(host, f"cat > {remote_q(path)}", input_text=data)
+
+
+def copy_job_from_remote(host: str, job_dir: str, local_input: Path) -> None:
+    local_input.mkdir(parents=True, exist_ok=True)
+    remote_cmd = (
+        f"tar --exclude={quote(LOCK_NAME)} --exclude={quote(STATE_NAME)} "
+        f"-C {remote_q(job_dir)} -cf - ."
+    )
+    p1 = subprocess.Popen(["ssh", host, remote_cmd], stdout=subprocess.PIPE)
+    try:
+        p2 = subprocess.run(["tar", "-xf", "-", "-C", str(local_input)], stdin=p1.stdout, check=True)
+        assert p1.stdout is not None
+        p1.stdout.close()
+        rc = p1.wait()
+        if rc != 0:
+            raise subprocess.CalledProcessError(rc, ["ssh", host, remote_cmd])
+    except Exception:
+        p1.kill()
+        raise
+
+
+def publish_output(host: str, remote_root: str, local_mp4: Path, old_output: str | None) -> str:
+    out_name = local_mp4.name
+    remote_tmp = f"{remote_root}/out-dir/.{out_name}.tmp"
+    remote_final = f"{remote_root}/out-dir/{out_name}"
+    subprocess.run(["scp", str(local_mp4), f"{host}:{remote_tmp}"], check=True)
+    cmds = [
+        f"mv {remote_q(remote_tmp)} {remote_q(remote_final)}",
+        f"rm -f {remote_q(remote_root + '/cache/videos.json')}",
+    ]
+    if old_output and old_output != out_name:
+        cmds.append(f"rm -f {remote_q(remote_root + '/out-dir/' + old_output)}")
+    ssh(host, " && ".join(cmds))
+    return out_name
+
+
+def render_job(local_input: Path, local_out: Path, extra_args: list[str]) -> Path:
+    local_out.mkdir(parents=True, exist_ok=True)
+    cmd = [sys.executable, str(MOVMAKER), str(local_input), "--out-dir", str(local_out), *extra_args]
+    subprocess.run(cmd, check=True)
+    outputs = sorted(local_out.glob("*.mp4"), key=lambda p: p.stat().st_mtime, reverse=True)
+    if not outputs:
+        raise RuntimeError("movmaker finished but no MP4 was produced")
+    return outputs[0]
+
+
+def process_job(host: str, remote_root: str, job: str, work_dir: Path, extra_args: list[str], force: bool) -> None:
+    job_dir = f"{remote_root}/in-dir/{job}"
+    if not acquire_lock(host, job_dir):
+        print(f"skip locked: {job}")
+        return
+    try:
+        fingerprint_before = remote_fingerprint(host, job_dir)
+        state = read_state(host, job_dir)
+        old_output = state.get("output") if isinstance(state.get("output"), str) else None
+        if not force and state.get("status") == "done" and state.get("fingerprint") == fingerprint_before and old_output:
+            if remote_file_exists(host, f"{remote_root}/out-dir/{old_output}"):
+                print(f"up to date: {job} -> {old_output}")
+                return
+
+        write_state(host, job_dir, {
+            "status": "processing",
+            "fingerprint": fingerprint_before,
+            "started_at": now(),
+            "output": old_output,
+        })
+
+        local_job = work_dir / job
+        if local_job.exists():
+            shutil.rmtree(local_job)
+        local_input = local_job / "input"
+        local_out = local_job / "out"
+        copy_job_from_remote(host, job_dir, local_input)
+        print(f"rendering: {job}")
+        local_mp4 = render_job(local_input, local_out, extra_args)
+
+        fingerprint_after = remote_fingerprint(host, job_dir)
+        if fingerprint_after != fingerprint_before:
+            write_state(host, job_dir, {
+                "status": "stale",
+                "fingerprint": fingerprint_after,
+                "previous_fingerprint": fingerprint_before,
+                "updated_at": now(),
+                "message": "Input changed during render; will regenerate on next run.",
+                "output": old_output,
+            })
+            print(f"stale during render, not publishing: {job}")
+            return
+
+        output = publish_output(host, remote_root, local_mp4, old_output)
+        write_state(host, job_dir, {
+            "status": "done",
+            "fingerprint": fingerprint_after,
+            "output": output,
+            "generated_at": now(),
+        })
+        print(f"published: {job} -> {output}")
+    except Exception as exc:
+        try:
+            write_state(host, job_dir, {
+                "status": "error",
+                "updated_at": now(),
+                "message": str(exc),
+            })
+        finally:
+            raise
+    finally:
+        release_lock(host, job_dir)
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Render changed MVLog input dirs from a remote web server.")
+    parser.add_argument("--host", default="192.168.0.204", help="SSH host for the web server")
+    parser.add_argument("--remote-root", default="/var/www/html/mvlog", help="Remote MVLog web root")
+    parser.add_argument("--work-dir", default=str(Path.home() / ".cache" / "mvlog-worker"), help="Local work directory")
+    parser.add_argument("--job", action="append", help="Only process this input dir name; may be repeated")
+    parser.add_argument("--force", action="store_true", help="Render even if fingerprint is unchanged")
+    parser.add_argument("--movmaker-arg", action="append", default=[], help="Extra argument passed to movmaker; may be repeated")
+    args = parser.parse_args()
+
+    work_dir = Path(args.work_dir)
+    work_dir.mkdir(parents=True, exist_ok=True)
+    jobs = args.job or list_jobs(args.host, args.remote_root)
+    for job in jobs:
+        process_job(args.host, args.remote_root, job, work_dir, args.movmaker_arg, args.force)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())