212 lines
7.2 KiB
Python
212 lines
7.2 KiB
Python
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
|