Files
video-flow/modules/path_utils.py

94 lines
2.4 KiB
Python

"""
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}"