| 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 |
| path.write_text(json.dumps(serializable, indent=2, default=str), encoding="utf-8") |
|
|
|
|
| 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) |
|
|