movmaker/mvlog_worker.py

#!/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"
ENABLED_NAME = ".movmaker-enabled"


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 in {'.movmaker-state.json', '.movmaker-enabled', '.mvlog-hidden'}:
            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}"
    tmp_path = f"{path}.tmp.$$"
    data = json.dumps(state, ensure_ascii=False, indent=2) + "\n"
    cmd = (
        f"cat > {remote_q(tmp_path)} && "
        f"chmod 664 {remote_q(tmp_path)} && "
        f"mv {remote_q(tmp_path)} {remote_q(path)}"
    )
    ssh(host, cmd, 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 remote_file_exists(host, f"{job_dir}/{ENABLED_NAME}"):
        print(f"disabled: {job}")
        return
    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())