File size: 1,477 Bytes
1425afc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 | 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)
|