from __future__ import annotations import json import re import shutil import tempfile import time import uuid from dataclasses import asdict, is_dataclass from pathlib import Path from typing import Any def new_id(prefix: str = "job") -> str: return f"{prefix}_{uuid.uuid4().hex[:12]}" def safe_filename(name: str) -> str: cleaned = re.sub(r"[^A-Za-z0-9._-]+", "_", name).strip("._") return cleaned or "render.mp4" def write_json(path: Path, data: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) serializable = asdict(data) if is_dataclass(data) else data temporary = path.with_name(f".{path.name}.{uuid.uuid4().hex}.tmp") try: temporary.write_text(json.dumps(serializable, indent=2, default=str), encoding="utf-8") temporary.replace(path) finally: temporary.unlink(missing_ok=True) def read_json(path: Path, default: Any) -> Any: if not path.exists(): return default return json.loads(path.read_text(encoding="utf-8")) def now() -> float: return time.time() def temp_workdir(root: Path, prefix: str) -> tempfile.TemporaryDirectory[str]: root.mkdir(parents=True, exist_ok=True) return tempfile.TemporaryDirectory(prefix=f"{prefix}_", dir=str(root)) def cleanup_directory(path: Path, keep: set[Path] | None = None) -> None: keep = {p.resolve() for p in (keep or set())} if not path.exists(): return for child in path.iterdir(): if child.resolve() in keep: continue if child.is_dir(): shutil.rmtree(child, ignore_errors=True) else: child.unlink(missing_ok=True)