""" Path utilities for cross-session / cross-project isolation. Goal: - Avoid file overwrites across concurrent users/projects by namespacing all temp artifacts under temp/projects/{project_id}/... - Provide safe unique filename helpers. """ from __future__ import annotations import os import re import time import uuid from pathlib import Path from typing import Optional import config _SAFE_CHARS_RE = re.compile(r"[^A-Za-z0-9._-]+") def sanitize_filename(name: str) -> str: """Keep only safe filename characters and strip path separators.""" if not isinstance(name, str): return "file" name = name.replace("\\", "_").replace("/", "_").strip() name = _SAFE_CHARS_RE.sub("_", name) return name or "file" def ensure_dir(path: Path) -> Path: path.mkdir(parents=True, exist_ok=True) return path def project_root(project_id: str) -> Path: pid = sanitize_filename(project_id or "UNKNOWN") return ensure_dir(config.TEMP_DIR / "projects" / pid) def project_upload_dir(project_id: str) -> Path: return ensure_dir(project_root(project_id) / "uploads") def project_images_dir(project_id: str) -> Path: return ensure_dir(project_root(project_id) / "images") def project_videos_dir(project_id: str) -> Path: return ensure_dir(project_root(project_id) / "videos") def project_audio_dir(project_id: str) -> Path: return ensure_dir(project_root(project_id) / "audio") def project_compose_dir(project_id: str, output_name: str) -> Path: out = sanitize_filename(output_name or f"compose_{int(time.time())}") return ensure_dir(project_root(project_id) / "compose" / out) def unique_filename( prefix: str, ext: str, project_id: Optional[str] = None, scene_id: Optional[int] = None, extra: Optional[str] = None, ) -> str: """ Build a unique filename. Example: scene_1_PROJ-xxx_173..._a1b2c3.mp4 """ pfx = sanitize_filename(prefix or "file") e = (ext or "").lstrip(".") or "bin" pid = sanitize_filename(project_id) if project_id else None sid = str(int(scene_id)) if scene_id is not None else None ex = sanitize_filename(extra) if extra else None ts = str(int(time.time() * 1000)) rnd = uuid.uuid4().hex[:8] parts = [pfx] if sid: parts.append(f"s{sid}") if pid: parts.append(pid) if ex: parts.append(ex) parts.extend([ts, rnd]) return f"{'_'.join(parts)}.{e}"