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 logging
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"
PUSH_SENDER = ROOT / "send_push.js"
PUSH_CONFIG = Path.home() / ".config" / "mvlog" / "push.json"
STATE_NAME = ".movmaker-state.json"
LOCK_NAME = ".movmaker-lock"
ENABLED_NAME = ".movmaker-enabled"
PREVIEW_NAME = ".movmaker-preview"
HIDDEN_NAME = ".mvlog-hidden"
NOTIFIED_CACHE = "cache/push_notifications.json"


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', '.movmaker-preview', '.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_remote_json(host: str, path: str) -> dict:
    res = ssh(host, f"cat {remote_q(path)} 2>/dev/null", check=False)
    if res.returncode != 0 or not res.stdout.strip():
        return {}
    try:
        data = json.loads(res.stdout)
        return data if isinstance(data, dict) else {}
    except json.JSONDecodeError:
        return {}


def write_remote_json(host: str, path: str, data: dict) -> None:
    tmp_path = f"{path}.tmp.$$"
    payload = json.dumps(data, ensure_ascii=False, indent=2) + "\n"
    cmd = (
        f"mkdir -p {remote_q(str(Path(path).parent))} && "
        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=payload)


def read_state(host: str, job_dir: str) -> dict:
    return read_remote_json(host, f"{job_dir}/{STATE_NAME}")


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 send_push_notification(host: str, remote_root: str, output: str) -> bool:
    if not PUSH_SENDER.exists() or not PUSH_CONFIG.exists():
        return False
    res = ssh(host, f"cat {remote_q(remote_root + '/cache/push_subscriptions.json')} 2>/dev/null", check=False)
    if res.returncode != 0 or not res.stdout.strip():
        return False
    title = output.rsplit('.', 1)[0].replace('_', ' ').title()
    payload = json.dumps({
        "title": "New MVLog video",
        "body": title,
        "url": "https://bubulescu.org/mvlog/",
        "tag": f"mvlog-{output}",
    }, ensure_ascii=False)
    proc = subprocess.run(["node", str(PUSH_SENDER), str(PUSH_CONFIG), payload], input=res.stdout, text=True, capture_output=True, check=False)
    if proc.stdout.strip():
        print(proc.stdout.strip())
    if proc.stderr.strip():
        print(proc.stderr.strip(), file=sys.stderr)
    return proc.returncode == 0


def maybe_send_first_show_notification(host: str, remote_root: str, job_dir: str, job: str, state: dict | None = None) -> None:
    if remote_file_exists(host, f"{job_dir}/{HIDDEN_NAME}"):
        return
    state = state if state is not None else read_state(host, job_dir)
    output = state.get("output") if isinstance(state.get("output"), str) else ""
    if not output or not remote_file_exists(host, f"{remote_root}/out-dir/{output}"):
        return
    cache_path = f"{remote_root}/{NOTIFIED_CACHE}"
    cache = read_remote_json(host, cache_path)
    sent = cache.get("sent") if isinstance(cache.get("sent"), dict) else {}
    if output in sent:
        return
    if send_push_notification(host, remote_root, output):
        sent[output] = {"job": job, "sent_at": now()}
        cache["sent"] = sent
        write_remote_json(host, cache_path, cache)
        print(f"notification sent: {job} -> {output}")


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}"
    state = read_state(host, job_dir)
    maybe_send_first_show_notification(host, remote_root, job_dir, job, state)
    full_requested = remote_file_exists(host, f"{job_dir}/{ENABLED_NAME}")
    preview_requested = remote_file_exists(host, f"{job_dir}/{PREVIEW_NAME}")
    if not full_requested and not preview_requested:
        print(f"disabled: {job}")
        return
    preview_mode = preview_requested and not full_requested
    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
        old_preview_output = state.get("preview_output") if isinstance(state.get("preview_output"), str) else None
        if preview_mode:
            if not force and state.get("preview_fingerprint") == fingerprint_before and old_preview_output:
                if remote_file_exists(host, f"{remote_root}/out-dir/{old_preview_output}"):
                    print(f"preview up to date: {job} -> {old_preview_output}")
                    return
        elif 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

        processing_state = dict(state)
        processing_state.update({
            "status": "processing",
            "mode": "preview" if preview_mode else "full",
            "started_at": now(),
        })
        if preview_mode:
            processing_state["preview_fingerprint"] = fingerprint_before
        else:
            processing_state["fingerprint"] = fingerprint_before
            processing_state["output"] = old_output
        write_state(host, job_dir, processing_state)

        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)
        render_args = [*extra_args, "--preview"] if preview_mode else extra_args
        print(f"rendering {'preview: ' if preview_mode else ''}{job}")
        local_mp4 = render_job(local_input, local_out, render_args)
        if preview_mode:
            preview_mp4 = local_mp4.with_name(f"{local_mp4.stem}_preview{local_mp4.suffix}")
            local_mp4.rename(preview_mp4)
            local_mp4 = preview_mp4

        fingerprint_after = remote_fingerprint(host, job_dir)
        if fingerprint_after != fingerprint_before:
            stale_state = dict(state)
            stale_state.update({
                "status": "stale",
                "updated_at": now(),
                "message": "Input changed during render; will regenerate on next run.",
            })
            if preview_mode:
                stale_state["preview_fingerprint"] = fingerprint_after
                stale_state["previous_preview_fingerprint"] = fingerprint_before
            else:
                stale_state["fingerprint"] = fingerprint_after
                stale_state["previous_fingerprint"] = fingerprint_before
                stale_state["output"] = old_output
            write_state(host, job_dir, stale_state)
            print(f"stale during render, not publishing: {job}")
            return

        if preview_mode:
            preview_output = publish_output(host, remote_root, local_mp4, old_preview_output)
            new_state = dict(state)
            new_state.update({
                "status": "done",
                "mode": "preview",
                "preview_fingerprint": fingerprint_after,
                "preview_output": preview_output,
                "preview_generated_at": now(),
            })
            write_state(host, job_dir, new_state)
            print(f"published preview: {job} -> {preview_output}")
            return

        output = publish_output(host, remote_root, local_mp4, old_output)
        new_state = dict(state)
        new_state.update({
            "status": "done",
            "mode": "full",
            "fingerprint": fingerprint_after,
            "output": output,
            "generated_at": now(),
        })
        maybe_send_first_show_notification(host, remote_root, job_dir, job, new_state)
        write_state(host, job_dir, new_state)
        print(f"published: {job} -> {output}")
        return
    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__":
    logging.info('Worker finished.')
    raise SystemExit(main())