from __future__ import annotations import asyncio import json import os import shutil import subprocess import uuid from dataclasses import dataclass from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any from urllib.parse import parse_qsl, urlsplit import httpx from app.config import settings from app.services.resource_signed_url_service import build_resource_signed_url from app.services.video_cover_service import get_ffmpeg_bin @dataclass(slots=True) class VideoMediaInfo: width: int height: int duration_seconds: float fps: float codec_name: str | None = None color_transfer: str | None = None color_primaries: str | None = None color_space: str | None = None def build_part_mp4_path(final_path: str) -> str: path = Path(final_path) return str(path.with_name(f"{path.stem}.{uuid.uuid4().hex}.part.mp4")) def safe_remove(path: str | None) -> bool: if not path: return True try: if os.path.exists(path): os.remove(path) return not os.path.exists(path) except OSError: return False def is_valid_file(path: str | None) -> bool: if not path: return False try: return os.path.isfile(path) and os.path.getsize(path) > 0 except OSError: return False def get_ffprobe_bin() -> str: ffmpeg = Path(get_ffmpeg_bin()) sibling = ffmpeg.with_name("ffprobe.exe" if ffmpeg.suffix.lower() == ".exe" else "ffprobe") if sibling.exists(): return str(sibling) found = shutil.which("ffprobe") or shutil.which("ffprobe.exe") if found: return found raise RuntimeError("未找到 ffprobe，请确保其与 FFMPEG_BIN 同目录或已加入 PATH") def _parse_fps(value: str | None) -> float: if not value: return 0.0 try: if "/" in value: left, right = value.split("/", 1) denominator = float(right) return float(left) / denominator if denominator else 0.0 return float(value) except Exception: return 0.0 def probe_video_sync(path: str, timeout_seconds: int = 30) -> VideoMediaInfo: if not is_valid_file(path): raise RuntimeError(f"视频文件不存在或为空: {path}") cmd = [ get_ffprobe_bin(), "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=width,height,codec_name,avg_frame_rate,color_transfer,color_primaries,color_space:format=duration", "-of", "json", path, ] result = subprocess.run( cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, encoding="utf-8", errors="replace", timeout=max(5, int(timeout_seconds)), shell=False, ) if result.returncode != 0: raise RuntimeError(f"ffprobe 校验失败: {(result.stderr or '').strip()[-2000:]}") try: payload = json.loads(result.stdout or "{}") stream = (payload.get("streams") or [])[0] width = int(stream.get("width") or 0) height = int(stream.get("height") or 0) duration = float((payload.get("format") or {}).get("duration") or 0) except Exception as exc: raise RuntimeError(f"ffprobe 响应解析失败: {exc}") from exc if width <= 0 or height <= 0: raise RuntimeError("ffprobe 未读取到有效视频宽高") return VideoMediaInfo( width=width, height=height, duration_seconds=duration, fps=_parse_fps(stream.get("avg_frame_rate")), codec_name=stream.get("codec_name"), color_transfer=stream.get("color_transfer"), color_primaries=stream.get("color_primaries"), color_space=stream.get("color_space"), ) async def probe_video(path: str, timeout_seconds: int = 30) -> VideoMediaInfo: return await asyncio.to_thread(probe_video_sync, path, timeout_seconds) def parse_tos_signed_url_expiry(url: str | None) -> tuple[datetime | None, datetime | None]: if not url: return None, None params = {key.lower(): value for key, value in parse_qsl(urlsplit(url).query, keep_blank_values=True)} date_text = params.get("x-tos-date") expires_text = params.get("x-tos-expires") if not date_text or not expires_text: return None, None try: signed_at = datetime.strptime(date_text, "%Y%m%dT%H%M%SZ").replace(tzinfo=timezone.utc) expires_seconds = int(expires_text) if expires_seconds <= 0: return None, None return signed_at, signed_at + timedelta(seconds=expires_seconds) except Exception: return None, None async def probe_remote_url(url: str) -> bool: timeout = httpx.Timeout( connect=max(1, int(settings.VIDEO_UPSCALE_REMOTE_URL_PROBE_CONNECT_TIMEOUT_SECONDS or 3)), read=max(1, int(settings.VIDEO_UPSCALE_REMOTE_URL_PROBE_READ_TIMEOUT_SECONDS or 5)), write=5, pool=5, ) try: async with httpx.AsyncClient(timeout=timeout, follow_redirects=True) as client: async with client.stream("GET", url, headers={"Range": "bytes=0-0"}) as response: if response.status_code not in {200, 206}: return False async for chunk in response.aiter_bytes(): return bool(chunk) return False except Exception: return False def build_source_resource_url(source_local_path: str) -> str: root = Path(settings.STORAGE_LOCAL_PATH).resolve() path = Path(source_local_path).resolve() try: relative = path.relative_to(root).as_posix() except ValueError as exc: raise RuntimeError("超分源视频不在 STORAGE_LOCAL_PATH 下，无法生成签名 URL") from exc return f"{settings.BASE_URL.rstrip('/')}/generate/videos/{relative}" def build_local_source_signed_url(source_local_path: str, expire_seconds: int) -> str: return build_resource_signed_url( build_source_resource_url(source_local_path), expire_seconds=max(600, int(expire_seconds)), ) async def download_video_to_path(url: str, final_path: str, timeout_seconds: int) -> str: if is_valid_file(final_path): try: await probe_video(final_path) return final_path except Exception: safe_remove(final_path) os.makedirs(os.path.dirname(final_path), exist_ok=True) part_path = build_part_mp4_path(final_path) timeout = httpx.Timeout(max(30, int(timeout_seconds))) try: async with httpx.AsyncClient(timeout=timeout, follow_redirects=True) as client: async with client.stream("GET", url) as response: response.raise_for_status() with open(part_path, "wb") as file_obj: async for chunk in response.aiter_bytes(chunk_size=1024 * 1024): file_obj.write(chunk) if not is_valid_file(part_path): raise RuntimeError("视频下载完成但临时文件为空") await probe_video(part_path) os.replace(part_path, final_path) return final_path except Exception: safe_remove(part_path) raise