"""Template-based remote jobs. Remote jobs are intentionally explicit templates instead of free-form shell input. This keeps the UI safer and makes future destructive operations easier to wrap in confirmations/dry-runs. """ from __future__ import annotations import logging import shlex from dataclasses import dataclass from typing import Mapping from media_library_viewer_api.clients.ssh import CommandResult, RemoteSSHClient logger = logging.getLogger(__name__) @dataclass(frozen=True) class JobTemplate: """Description and command template for one remote job.""" name: str description: str command_template: str destructive: bool = False def render(self, values: Mapping[str, str]) -> str: """Render the command with shell-quoted template values. This is what keeps paths with spaces safe when inserted into job commands. """ safe_values = {key: shlex.quote(value) for key, value in values.items()} return self.command_template.format(**safe_values) # Phase 1 jobs are intentionally conservative. Add your own templates here later. JOB_TEMPLATES: dict[str, JobTemplate] = { "disk_usage": JobTemplate( name="Disk usage for selected path", description="Runs du -sh on the selected remote path.", command_template="du -sh {path}", ), "ffprobe": JobTemplate( name="ffprobe JSON", description="Prints raw ffprobe stream/format metadata.", command_template="ffprobe -v error -show_format -show_streams -print_format json {path}", ), "dry_run_find_empty_dirs": JobTemplate( name="Find empty directories dry-run", description="Lists empty directories under the selected path. Does not delete anything.", command_template="find {path} -type d -empty -print", ), } def run_job(ssh: RemoteSSHClient, job_key: str, path: str, timeout: int = 600) -> CommandResult: """Render and execute a configured job template for a selected remote path.""" template = JOB_TEMPLATES[job_key] command = template.render({"path": path}) logger.info("Executing job template key=%s path=%s timeout=%s", job_key, path, timeout) return ssh.run(command, timeout=timeout)