from __future__ import annotations import os import re import uuid from dataclasses import dataclass from datetime import datetime from pathlib import Path from typing import Iterable from urllib.parse import urlsplit from app.config import settings from app.enums.upload_resource import UploadResourceModuleEnum, UploadResourceTypeEnum COMMON_IMAGE_RE = re.compile(r"^images/(?P\d{4})/(?P\d{2})/(?P\d{2})/video_img_(?P[^_]+)_(?P\d{8})_(?P\d{6})_(?P[0-9a-fA-F]{8})\.(?P[^/]+)$") COMMON_VIDEO_RE = re.compile(r"^videos/(?P\d{4})/(?P\d{2})/(?P\d{2})/video_ref_(?P[^_]+)_(?P\d{8})_(?P\d{6})_(?P[0-9a-fA-F]{8})\.(?P[^/]+)$") COMMON_AUDIO_RE = re.compile(r"^audios/(?P\d{4})/(?P\d{2})/(?P\d{2})/audio_ref_(?P[^_]+)_(?P\d{8})_(?P\d{6})_(?P[0-9a-fA-F]{8})\.(?P[^/]+)$") MODULE_RE = re.compile(r"^(?Phot_opening_replicate|shot_replicate)/(?Pimages|videos)/(?P\d{4})/(?P\d{2})/(?P\d{2})/(?Pvideo_img|video_ref)_(?P[^_]+)_(?P\d{8})_(?P\d{6})_(?P[0-9a-fA-F]{8})\.(?P[^/]+)$") SHOT_SEGMENT_RE = re.compile(r"^shot_segments/(?P\d{4})/(?P\d{2})/(?P\d{2})/(?P[^/]+)\.mp4$") LEGACY_GEN_RE = re.compile(r"^(?Pimages|videos)/gen_(?P[^_]+)_(?P[0-9a-zA-Z]+)\.(?P[^/]+)$") IMAGE_EXTS = {".jpg", ".jpeg", ".png", ".webp", ".gif", ".bmp", ".svg"} VIDEO_EXTS = {".mp4", ".mov", ".m4v", ".webm", ".avi", ".mkv"} AUDIO_EXTS = {".mp3", ".wav", ".m4a", ".aac", ".flac"} @dataclass(slots=True) class ParsedUploadPath: storage_path: str resource_url: str file_name: str file_ext: str file_size_bytes: int module: str resource_type: str user_id: str | None created_at: datetime | None source_model: str | None = None source_id: str | None = None skip_reason: str | None = None def upload_root() -> Path: return Path(settings.UPLOAD_LOCAL_PATH).resolve() def to_abs_path(path: str | os.PathLike[str]) -> Path: p = Path(path) return p if p.is_absolute() else Path.cwd() / p def normalize_storage_path(path: str | os.PathLike[str]) -> str: return str(to_abs_path(path).resolve()) def upload_url_to_storage_path(url: str | None) -> str | None: if not url: return None value = str(url).strip() if not value: return None value = urlsplit(value).path or value if not value.startswith("/uploads/"): return None rel = value[len("/uploads/"):].lstrip("/") if not rel or ".." in Path(rel).parts: return None return str((upload_root() / rel).resolve()) def storage_path_to_upload_url(path: str | os.PathLike[str]) -> str: abs_path = to_abs_path(path).resolve() rel = abs_path.relative_to(upload_root()).as_posix() return f"/uploads/{rel}" def _parse_created_at(parts: dict[str, str], fallback: datetime | None = None) -> datetime | None: ymd = parts.get("ymd") hms = parts.get("hms") try: if ymd and hms: return datetime.strptime(f"{ymd}_{hms}", "%Y%m%d_%H%M%S") return datetime(int(parts["year"]), int(parts["month"]), int(parts["day"])) except Exception: return fallback def _file_size(path: Path) -> int: try: return int(path.stat().st_size) except OSError: return 0 def _base(path: Path, module: str, resource_type: str, user_id: str | None, created_at: datetime | None) -> ParsedUploadPath: return ParsedUploadPath( storage_path=normalize_storage_path(path), resource_url=storage_path_to_upload_url(path), file_name=path.name, file_ext=path.suffix.lower().lstrip("."), file_size_bytes=_file_size(path), module=module, resource_type=resource_type, user_id=user_id, created_at=created_at, ) def parse_upload_path(path: str | os.PathLike[str], *, include_legacy: bool = False) -> ParsedUploadPath | None: abs_path = to_abs_path(path).resolve() if not abs_path.is_file(): return None try: rel = abs_path.relative_to(upload_root()).as_posix() except ValueError: return None ignored_prefixes = ("home_materials/",) if rel in {"site_logo.png"} or rel.startswith(ignored_prefixes) or rel.startswith("pdf_"): return ParsedUploadPath( storage_path=normalize_storage_path(abs_path), resource_url="", file_name=abs_path.name, file_ext=abs_path.suffix.lower().lstrip("."), file_size_bytes=_file_size(abs_path), module=UploadResourceModuleEnum.COMMON.value, resource_type="unknown", user_id=None, created_at=None, skip_reason="ignored_path", ) for pattern, rtype in ( (COMMON_IMAGE_RE, UploadResourceTypeEnum.IMAGE.value), (COMMON_VIDEO_RE, UploadResourceTypeEnum.VIDEO.value), (COMMON_AUDIO_RE, UploadResourceTypeEnum.AUDIO.value), ): m = pattern.match(rel) if m: d = m.groupdict() return _base(abs_path, UploadResourceModuleEnum.COMMON.value, rtype, d.get("user_id"), _parse_created_at(d)) m = MODULE_RE.match(rel) if m: d = m.groupdict() kind = d.get("kind") rtype = UploadResourceTypeEnum.IMAGE.value if kind == "images" else UploadResourceTypeEnum.VIDEO.value return _base(abs_path, d["module"], rtype, d.get("user_id"), _parse_created_at(d)) m = SHOT_SEGMENT_RE.match(rel) if m: d = m.groupdict() parsed = _base(abs_path, UploadResourceModuleEnum.SHOT_REPLICATE.value, UploadResourceTypeEnum.SHOT_SEGMENT.value, None, _parse_created_at(d)) parsed.source_model = "ShotReplicateSegment" parsed.source_id = d["segment_id"] return parsed if include_legacy: m = LEGACY_GEN_RE.match(rel) if m: d = m.groupdict() ext = abs_path.suffix.lower() if ext in IMAGE_EXTS: rtype = UploadResourceTypeEnum.IMAGE.value elif ext in VIDEO_EXTS: rtype = UploadResourceTypeEnum.VIDEO.value else: return None try: created = datetime.fromtimestamp(abs_path.stat().st_mtime) except OSError: created = None return _base(abs_path, UploadResourceModuleEnum.COMMON.value, rtype, d.get("user_id"), created) return ParsedUploadPath( storage_path=normalize_storage_path(abs_path), resource_url="", file_name=abs_path.name, file_ext=abs_path.suffix.lower().lstrip("."), file_size_bytes=_file_size(abs_path), module=UploadResourceModuleEnum.COMMON.value, resource_type="unknown", user_id=None, created_at=None, skip_reason="unmatched_path", ) def iter_files(root: str | os.PathLike[str]) -> Iterable[Path]: base = to_abs_path(root).resolve() if not base.exists(): return [] return (p for p in base.rglob("*") if p.is_file()) def build_upload_destination(*, module: str, resource_type: str, user_id: str, original_filename: str | None, gen_type: str = "video") -> tuple[Path, str, str]: now = datetime.now() date_dir = now.strftime("%Y/%m/%d") ext = Path(original_filename or "").suffix.lower() if not ext: ext = ".mp4" if resource_type == UploadResourceTypeEnum.VIDEO.value else ".mp3" if resource_type == UploadResourceTypeEnum.AUDIO.value else ".png" timestamp = now.strftime("%Y%m%d_%H%M%S") suffix = uuid.uuid4().hex[:8] if resource_type == UploadResourceTypeEnum.AUDIO.value: filename = f"audio_ref_{user_id}_{timestamp}_{suffix}{ext}" rel_dir = Path("audios") / date_dir elif resource_type == UploadResourceTypeEnum.VIDEO.value: filename = f"video_ref_{user_id}_{timestamp}_{suffix}{ext}" rel_dir = Path("videos") / date_dir if module == UploadResourceModuleEnum.COMMON.value else Path(module) / "videos" / date_dir else: filename = f"{gen_type}_img_{user_id}_{timestamp}_{suffix}{ext}" rel_dir = Path("images") / date_dir if module == UploadResourceModuleEnum.COMMON.value else Path(module) / "images" / date_dir dir_path = upload_root() / rel_dir file_path = dir_path / filename url = f"/uploads/{(rel_dir / filename).as_posix()}" return file_path, url, filename