from __future__ import annotations import json import os import subprocess from dataclasses import dataclass from pathlib import Path from urllib.parse import urlparse from fastapi import HTTPException from app.config import settings VIDEO_EXTENSIONS = {".mp4", ".mov", ".m4v", ".webm", ".avi", ".mkv"} @dataclass(slots=True) class UploadVideoAsset: url: str path: Path duration_seconds: float def _project_root() -> Path: return Path.cwd() def _abs_path(value: str | Path) -> Path: path = Path(value) if not path.is_absolute(): path = _project_root() / path return path.resolve() def upload_root() -> Path: return _abs_path(settings.UPLOAD_LOCAL_PATH) def shot_segment_root() -> Path: return _abs_path(getattr(settings, "SHOT_SEGMENT_LOCAL_PATH", "./storage/uploads/shot_segments")) def _strip_base_url(url: str) -> str: text = str(url or "").strip() if not text: return text base_url = str(getattr(settings, "BASE_URL", "") or "").strip().rstrip("/") if base_url and text.startswith(base_url + "/"): return text[len(base_url):] parsed = urlparse(text) if parsed.scheme in ("http", "https"): # 只接受本系统 BASE_URL 下的上传资源；外部 URL 不允许 ffmpeg 本地切片。 raise HTTPException(status_code=400, detail="拆镜源视频必须来自本系统上传接口，不能传外部 http/https URL") return text def _safe_relative_from_upload_url(url: str) -> str: value = _strip_base_url(url) value = value.split("?", 1)[0].split("#", 1)[0] if value.startswith("/uploads/"): rel = value.replace("/uploads/", "", 1) elif value.startswith("uploads/"): rel = value.replace("uploads/", "", 1) else: raise HTTPException(status_code=400, detail="拆镜源视频链接必须是 /uploads/ 下的上传资源") rel = rel.lstrip("/") if not rel or ".." in Path(rel).parts: raise HTTPException(status_code=400, detail="上传视频路径非法") return rel def resolve_upload_video_path(video_url: str) -> Path: rel = _safe_relative_from_upload_url(video_url) root = upload_root() path = (root / rel).resolve() try: path.relative_to(root) except ValueError as exc: raise HTTPException(status_code=400, detail="上传视频路径越界") from exc if path.suffix.lower() not in VIDEO_EXTENSIONS: raise HTTPException(status_code=400, detail="上传资源不是支持的视频格式") if not path.exists() or not path.is_file(): raise HTTPException(status_code=404, detail=f"上传视频文件不存在: {video_url}") return path def build_upload_url_from_path(path: str | Path) -> str: root = upload_root() checked_path = _abs_path(path) try: rel = checked_path.relative_to(root).as_posix() except ValueError as exc: raise HTTPException(status_code=500, detail="生成上传资源 URL 失败：路径不在 uploads 目录下") from exc return f"/uploads/{rel}" def get_ffmpeg_bin() -> str: return str(getattr(settings, "FFMPEG_BIN", "") or "ffmpeg") def get_ffprobe_bin() -> str: configured = str(getattr(settings, "FFPROBE_BIN", "") or "").strip() if configured: return configured ffmpeg_bin = get_ffmpeg_bin() if ffmpeg_bin.endswith("ffmpeg.exe"): return ffmpeg_bin[:-10] + "ffprobe.exe" if ffmpeg_bin.endswith("ffmpeg"): return ffmpeg_bin[:-6] + "ffprobe" return "ffprobe" def probe_video_duration_seconds(video_path: str | Path) -> float: path = _abs_path(video_path) timeout = int(getattr(settings, "SHOT_FFPROBE_TIMEOUT_SECONDS", 20) or 20) cmd = [ get_ffprobe_bin(), "-v", "error", "-show_entries", "format=duration", "-of", "json", str(path), ] try: completed = subprocess.run( cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, timeout=timeout, check=False, ) except Exception as exc: raise HTTPException(status_code=400, detail=f"ffprobe 获取视频时长失败: {exc}") from exc if completed.returncode != 0: raise HTTPException(status_code=400, detail=f"ffprobe 获取视频时长失败: {completed.stderr.strip()}") try: data = json.loads(completed.stdout or "{}") duration = float((data.get("format") or {}).get("duration") or 0) except Exception as exc: raise HTTPException(status_code=400, detail="ffprobe 返回的视频时长无法解析") from exc if duration <= 0: raise HTTPException(status_code=400, detail="视频时长无效") return round(duration, 3) def validate_upload_video_asset(video_url: str, frontend_duration_seconds: float | None = None) -> UploadVideoAsset: path = resolve_upload_video_path(video_url) real_duration = probe_video_duration_seconds(path) if frontend_duration_seconds is not None and frontend_duration_seconds > 0: tolerance = float(getattr(settings, "SHOT_DURATION_TOLERANCE_SECONDS", 1.0) or 1.0) # 超出误差时以后端 ffprobe 为准，不拒绝，避免前端浮点或浏览器 metadata 偏差导致创建失败。 if abs(float(frontend_duration_seconds) - real_duration) <= tolerance: real_duration = round(float(frontend_duration_seconds), 3) return UploadVideoAsset(url=_strip_base_url(video_url), path=path, duration_seconds=real_duration) def validate_split_range(*, start_second: float, end_second: float, video_duration_seconds: float) -> tuple[float, float, float]: start = round(float(start_second), 3) end = round(float(end_second), 3) if start < 0: raise HTTPException(status_code=400, detail="开始秒不能小于0") if end <= start: raise HTTPException(status_code=400, detail="结束秒必须大于开始秒") tolerance = float(getattr(settings, "SHOT_SPLIT_END_TOLERANCE_SECONDS", 0.5) or 0.5) if end > float(video_duration_seconds) + tolerance: raise HTTPException(status_code=400, detail="结束秒不能超过视频总时长") duration = round(end - start, 3) min_seconds = float(getattr(settings, "SHOT_SPLIT_MIN_SECONDS", 1) or 1) max_seconds = float(getattr(settings, "SHOT_SPLIT_MAX_SECONDS", 120) or 120) if duration < min_seconds: raise HTTPException(status_code=400, detail=f"拆镜片段不能低于 {min_seconds:g} 秒") if duration > max_seconds: raise HTTPException(status_code=400, detail=f"拆镜片段不能超过 {max_seconds:g} 秒") return start, end, duration def format_second(value: float) -> int | float: checked = float(value) if checked.is_integer(): return int(checked) return round(checked, 2) def build_time_node(start_second: float, end_second: float) -> str: return f"{format_second(start_second)}-{format_second(end_second)}秒" def ensure_shot_segment_dir(date_dir: str) -> Path: root = shot_segment_root() output_dir = (root / date_dir).resolve() try: output_dir.relative_to(root) except ValueError as exc: raise RuntimeError("拆镜输出目录越界") from exc os.makedirs(output_dir, exist_ok=True) return output_dir