"""视频获取链 (V3-M2A). 从 play_url 下载视频(带平台下载头)→ imageio-ffmpeg 压到 ~4MB 低清 → base64 data URL,供 GeminiVideoClient 投喂(OpenRouter image_url)。 真实下载/压缩只在 M7 live smoke 跑;单测全 mock。 原片不做本地持久化;长期复盘资产走 OSS,这里仅下载到内存并压缩后投喂模型。 """ from __future__ import annotations import base64 import os import subprocess import tempfile import time from typing import Any import httpx import imageio_ffmpeg from content_agent.integrations import timeout_config # platform_profiles 里写的是 "iOS UA"/"PC UA" 占位,这里映射成真实串 + Referer。 _PLATFORM_DOWNLOAD_HEADERS = { "douyin": { "User-Agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.0 Mobile/15E148", "Referer": "https://www.douyin.com/", }, "shipinhao": { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120 Safari/537.36", "Referer": "https://channels.weixin.qq.com/", }, } # 已拍板压缩档:360p / 1fps / 低清,实测 ~4MB(memory/video-multimodal-analysis)。 _FFMPEG_ARGS = ["-vf", "scale=360:-2,fps=1", "-crf", "33", "-c:a", "aac", "-b:a", "32k", "-ac", "1"] MAX_INLINE_BYTES = 500 * 1024 * 1024 # 本地 inline data URL 护栏;实际上游上限由 OpenRouter/Gemini 决定 DOWNLOAD_TIMEOUT_SECONDS = timeout_config.total_timeout("video_download") # 10min(原 30min) COMPRESS_TIMEOUT_SECONDS = 600.0 # ffmpeg 压缩 10min(原 20min;正常 ~8s,subprocess 硬超时真生效) class VideoFetchError(RuntimeError): """下载/压缩/超限失败,由 GeminiVideoClient 捕获转 fail。""" def __init__(self, message: str, *, metrics: dict[str, Any] | None = None) -> None: super().__init__(message) self.metrics = metrics or {} def _download_headers(platform: str, override: dict[str, str] | None) -> dict[str, str]: if override is not None: return override return _PLATFORM_DOWNLOAD_HEADERS.get(platform, {}) def _compress( source: bytes | str, ffmpeg_exe: str, *, timeout_seconds: float = COMPRESS_TIMEOUT_SECONDS, input_path: bool = False, ) -> bytes: # 超时保护:坏视频会让 ffmpeg 卡死,进而挂住一个判定并发线程(实测正常压缩 ~8s)。 cmd = [ffmpeg_exe, "-i", str(source) if input_path else "pipe:0", *_FFMPEG_ARGS, "-f", "mp4", "-movflags", "frag_keyframe+empty_moov", "pipe:1"] try: proc = subprocess.run( cmd, input=None if input_path else source, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=timeout_seconds, ) except subprocess.TimeoutExpired as exc: raise VideoFetchError("ffmpeg compression timeout") from exc if proc.returncode != 0 or not proc.stdout: raise VideoFetchError("ffmpeg compression failed") return proc.stdout def fetch_and_compress( play_url: str, platform: str, *, headers: dict[str, str] | None = None, http_client: Any | None = None, ffmpeg_exe: str | None = None, timeout_seconds: float = DOWNLOAD_TIMEOUT_SECONDS, compress_timeout_seconds: float = COMPRESS_TIMEOUT_SECONDS, clock: Any = time.monotonic, ) -> str: data_url, _ = fetch_and_compress_with_metrics( play_url, platform, headers=headers, http_client=http_client, ffmpeg_exe=ffmpeg_exe, timeout_seconds=timeout_seconds, compress_timeout_seconds=compress_timeout_seconds, clock=clock, ) return data_url def fetch_and_compress_with_metrics( play_url: str, platform: str, *, headers: dict[str, str] | None = None, http_client: Any | None = None, ffmpeg_exe: str | None = None, timeout_seconds: float = DOWNLOAD_TIMEOUT_SECONDS, compress_timeout_seconds: float = COMPRESS_TIMEOUT_SECONDS, clock: Any = time.monotonic, ) -> tuple[str, dict[str, Any]]: metrics: dict[str, Any] = { "download_timeout_seconds": timeout_seconds, "ffmpeg_timeout_seconds": compress_timeout_seconds, } total_started_at = clock() if not play_url: raise VideoFetchError("missing play_url", metrics=metrics) client = http_client or httpx try: download_started_at = clock() video_path = _download_to_tempfile( client, play_url, platform, headers=headers, timeout_seconds=timeout_seconds, started_at=total_started_at, clock=clock, ) metrics["download_duration_ms"] = _elapsed_ms(download_started_at, clock()) except httpx.HTTPError as exc: metrics["total_duration_ms"] = _elapsed_ms(total_started_at, clock()) raise VideoFetchError(f"download failed: {type(exc).__name__}", metrics=metrics) from exc except VideoFetchError as exc: metrics.update(exc.metrics) metrics["total_duration_ms"] = _elapsed_ms(total_started_at, clock()) raise VideoFetchError(str(exc), metrics=metrics) from exc try: downloaded_bytes = os.path.getsize(video_path) metrics["downloaded_bytes"] = downloaded_bytes if downloaded_bytes <= 0: metrics["total_duration_ms"] = _elapsed_ms(total_started_at, clock()) raise VideoFetchError("empty download", metrics=metrics) ffmpeg_started_at = clock() compressed = _compress( video_path, ffmpeg_exe or imageio_ffmpeg.get_ffmpeg_exe(), timeout_seconds=compress_timeout_seconds, input_path=True, ) metrics["ffmpeg_duration_ms"] = _elapsed_ms(ffmpeg_started_at, clock()) metrics["compressed_bytes"] = len(compressed) except VideoFetchError as exc: metrics.update(exc.metrics) metrics["total_duration_ms"] = _elapsed_ms(total_started_at, clock()) raise VideoFetchError(str(exc), metrics=metrics) from exc finally: try: os.unlink(video_path) except OSError: pass if len(compressed) > MAX_INLINE_BYTES: metrics["total_duration_ms"] = _elapsed_ms(total_started_at, clock()) raise VideoFetchError(f"compressed video oversize: {len(compressed)} bytes", metrics=metrics) metrics["total_duration_ms"] = _elapsed_ms(total_started_at, clock()) return f"data:video/mp4;base64,{base64.b64encode(compressed).decode('ascii')}", metrics def _download_to_tempfile( client: Any, play_url: str, platform: str, *, headers: dict[str, str] | None, timeout_seconds: float, started_at: float, clock: Any, ) -> str: download_headers = _download_headers(platform, headers) tmp = tempfile.NamedTemporaryFile(prefix="content-agent-video-", suffix=".mp4", delete=False) tmp_path = tmp.name tmp.close() ok = False if hasattr(client, "stream"): try: with client.stream( "GET", play_url, headers=download_headers, follow_redirects=True, timeout=timeout_config.as_httpx_timeout( timeout_seconds, read=timeout_config.read_timeout("video_download") ), ) as response: response.raise_for_status() with open(tmp_path, "wb") as file: for chunk in response.iter_bytes(): if clock() - started_at > timeout_seconds: raise VideoFetchError("download timeout") file.write(chunk) ok = True return tmp_path finally: if not ok: _unlink_quietly(tmp_path) try: response = client.get( play_url, headers=download_headers, follow_redirects=True, timeout=timeout_config.as_httpx_timeout( timeout_seconds, read=timeout_config.read_timeout("video_download") ), ) response.raise_for_status() if clock() - started_at > timeout_seconds: raise VideoFetchError("download timeout") with open(tmp_path, "wb") as file: file.write(response.content) ok = True return tmp_path finally: if not ok: _unlink_quietly(tmp_path) def _unlink_quietly(path: str) -> None: try: os.unlink(path) except OSError: pass def _elapsed_ms(start: float, end: float) -> int: return max(0, int((end - start) * 1000))