399 lines
No EOL
16 KiB
Python
Executable file
399 lines
No EOL
16 KiB
Python
Executable file
#!/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"
|
|
INTRO_LOGO = ROOT / "img" / "moto_travel.png"
|
|
STATE_NAME = ".movmaker-state.json"
|
|
LOCK_NAME = ".movmaker-lock"
|
|
ENABLED_NAME = ".movmaker-enabled"
|
|
PREVIEW_NAME = ".movmaker-preview"
|
|
HIDDEN_NAME = ".mvlog-hidden"
|
|
INTRO_LOGO_ALPHA = "0.5"
|
|
MVLOG_WORKER_RECOMMENDED_INTERVAL_MINUTES = 5
|
|
|
|
|
|
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 | None, path: str) -> bool:
|
|
if host is None:
|
|
return Path(path).is_file()
|
|
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 | None, path: str) -> dict:
|
|
if host is None:
|
|
try:
|
|
data = json.loads(Path(path).read_text(encoding="utf-8"))
|
|
return data if isinstance(data, dict) else {}
|
|
except Exception:
|
|
return {}
|
|
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 | None, job_dir: str) -> dict:
|
|
if host is None:
|
|
path = Path(job_dir) / STATE_NAME
|
|
if not path.exists():
|
|
return {}
|
|
try:
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
except Exception:
|
|
return {}
|
|
return read_remote_json(host, f"{job_dir}/{STATE_NAME}")
|
|
|
|
|
|
def write_state(host: str | None, job_dir: str, state: dict) -> None:
|
|
if host is None:
|
|
path = Path(job_dir) / STATE_NAME
|
|
path.write_text(json.dumps(state, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
path.chmod(0o664)
|
|
return
|
|
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 upload_audio_files(host: str | None, job_dir: str, local_input: Path) -> list[str]:
|
|
"""Upload audio files found inside local_input to the remote job_dir/in-dir if they are missing.
|
|
|
|
Returns a list of uploaded relative paths.
|
|
"""
|
|
AUDIO_EXTS = {".mp3", ".wav", ".m4a", ".aac", ".flac", ".ogg", ".opus", ".wma", ".aiff", ".aif"}
|
|
uploaded: list[str] = []
|
|
for path in sorted(local_input.rglob("*")):
|
|
if not path.is_file():
|
|
continue
|
|
if path.suffix.lower() not in AUDIO_EXTS:
|
|
continue
|
|
rel = path.relative_to(local_input).as_posix()
|
|
remote_path = f"{job_dir}/{rel}"
|
|
if remote_file_exists(host, remote_path):
|
|
print(f"audio exists on server, skipping: {rel}")
|
|
continue
|
|
parent = str(Path(remote_path).parent)
|
|
if host is None:
|
|
Path(parent).mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(str(path), remote_path)
|
|
else:
|
|
# Ensure parent directory exists on the remote host
|
|
ssh(host, f"mkdir -p {remote_q(parent)}")
|
|
remote_tmp = f"{remote_path}.tmp.$$"
|
|
# Upload to a temporary file then move into place for (some) atomicity
|
|
subprocess.run(["scp", str(path), f"{host}:{remote_tmp}"], check=True)
|
|
ssh(host, f"mv {remote_q(remote_tmp)} {remote_q(remote_path)}")
|
|
uploaded.append(rel)
|
|
print(f"uploaded audio: {rel}")
|
|
return uploaded
|
|
|
|
|
|
def render_job(local_input: Path, local_out: Path, extra_args: list[str]) -> Path:
|
|
local_out.mkdir(parents=True, exist_ok=True)
|
|
intro_logo = INTRO_LOGO if INTRO_LOGO.exists() else None
|
|
intro_args = []
|
|
if intro_logo is not None:
|
|
intro_args = [
|
|
"--intro-logo", str(intro_logo),
|
|
"--intro-logo-alpha", INTRO_LOGO_ALPHA,
|
|
]
|
|
cmd = [sys.executable, str(MOVMAKER), str(local_input), "--out-dir", str(local_out), *intro_args, *extra_args]
|
|
log_dir = local_input / ".logs"
|
|
log_dir.mkdir(parents=True, exist_ok=True)
|
|
log_path = log_dir / f"mvlog_worker_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log"
|
|
try:
|
|
result = subprocess.run(cmd, capture_output=True, text=True, check=True)
|
|
except subprocess.CalledProcessError as exc:
|
|
log_path.write_text(
|
|
"Command failed:\n" + " ".join(cmd) + "\n\nSTDOUT:\n" + (exc.stdout or "") + "\n\nSTDERR:\n" + (exc.stderr or ""),
|
|
encoding="utf-8",
|
|
)
|
|
raise RuntimeError(f"movmaker failed; see {log_path}") from exc
|
|
else:
|
|
if result.stdout or result.stderr:
|
|
log_path.write_text(
|
|
"Command:\n" + " ".join(cmd) + "\n\nSTDOUT:\n" + (result.stdout or "") + "\n\nSTDERR:\n" + (result.stderr or ""),
|
|
encoding="utf-8",
|
|
)
|
|
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)
|
|
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)
|
|
# Upload any audio files present in the local input directory to the remote in-dir if missing.
|
|
try:
|
|
upload_audio_files(host, job_dir, local_input)
|
|
except Exception as exc: # Upload failure should not prevent publishing the preview
|
|
print(f"warning: audio upload failed: {exc}")
|
|
# Recompute fingerprint now that audio files may have been uploaded.
|
|
final_fingerprint = remote_fingerprint(host, job_dir)
|
|
new_state = dict(state)
|
|
new_state.update({
|
|
"status": "done",
|
|
"mode": "preview",
|
|
"preview_fingerprint": final_fingerprint,
|
|
"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)
|
|
try:
|
|
upload_audio_files(host, job_dir, local_input)
|
|
except Exception as exc:
|
|
print(f"warning: audio upload failed: {exc}")
|
|
final_fingerprint = remote_fingerprint(host, job_dir)
|
|
new_state = dict(state)
|
|
new_state.update({
|
|
"status": "done",
|
|
"mode": "full",
|
|
"fingerprint": final_fingerprint,
|
|
"output": output,
|
|
"generated_at": now(),
|
|
})
|
|
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("--local", action="store_true", help="Treat remote paths as local; skip SSH")
|
|
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(None if args.local else args.host, args.remote_root)
|
|
for job in jobs:
|
|
process_job(None if args.local else 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()) |