diff --git a/README.md b/README.md index 1a050b9..338c2bf 100644 --- a/README.md +++ b/README.md @@ -141,6 +141,55 @@ IMG_002.jpg: Louvre IMG_003.jpg: First line | Second line ``` +## MVLog worker + +`mvlog_worker.py` can be run from this machine by cron or a systemd timer to process MVLog jobs stored on the web machine. + +Default paths: + +- remote host: `192.168.0.204` +- remote web root: `/var/www/html/mvlog` +- inputs: `in-dir//` +- outputs: `out-dir/YYYYMMDD_title.mp4` +- per-job state: `in-dir//.movmaker-state.json` + +The worker: + +1. scans remote `in-dir/` +2. computes a fingerprint from input filenames, sizes, and mtimes +3. skips jobs whose fingerprint matches a completed state file and whose output exists +4. copies changed jobs locally +5. renders with `movmaker.py` +6. uploads the MP4 atomically to remote `out-dir/` +7. removes the old output if the regenerated filename changed +8. clears the web metadata cache + +Run once: + +```bash +./mvlog_worker.py +``` + +Process one job: + +```bash +./mvlog_worker.py --job romo2026 +``` + +Force regeneration: + +```bash +./mvlog_worker.py --job romo2026 --force +``` + +Pass extra movmaker options: + +```bash +./mvlog_worker.py --movmaker-arg=--preview +``` + +Recommended scheduling is a systemd timer or cron with a separate lock, e.g. every 5 minutes. + ## Usage Basic: diff --git a/mvlog_worker.py b/mvlog_worker.py new file mode 100755 index 0000000..0c4bc9d --- /dev/null +++ b/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())