[3.1-3.5] FAM-Edge 全链路 - API-Gateway/Video-Preprocessor/AI-Orchestrator/模型适配器(基类+Ollama+Gemini)/熔断器/JSON解析容错 + 配置

fam-edge/src/fam_edge/video_preprocessor/preprocessor.py

@@ -0,0 +1,255 @@
+"""
+Video-Preprocessor - 视频预处理
+
+流程:
+1. 下载视频（超时 60s）
+2. FFmpeg 等距粗抽 30 张候选帧
+3. OpenCV 帧差分析筛选 5-8 张关键帧（MSE 阈值）
+4. 压缩（长边 ≤ 1024px，JPEG 质量 80）
+
+异常兜底:
+- ffprobe 失败 -> 退化为按 60s 间隔抽帧
+- 帧差分析异常 -> 退化为等距抽 5 帧
+- OpenCV 压缩失败 -> 跳过该帧，记录 WARN
+"""
+import os
+import time
+import subprocess
+import requests
+import cv2
+import numpy as np
+from typing import List, Tuple, Optional
+
+from ..logger import setup_logger, log_task
+from ..config_loader import load_config
+
+logger = setup_logger('fam-edge.preprocessor')
+
+
+class VideoPreprocessor:
+    """视频预处理器"""
+
+    def __init__(self, task_id: int):
+        self.task_id = task_id
+        cfg = load_config()
+        video_cfg = cfg.get('video', {})
+        self.candidate_frames = video_cfg.get('candidate_frames', 30)
+        self.min_key_frames = video_cfg.get('min_key_frames', 5)
+        self.max_key_frames = video_cfg.get('max_key_frames', 8)
+        self.mse_threshold = video_cfg.get('mse_threshold', 500)
+        self.jpeg_quality = video_cfg.get('jpeg_quality', 80)
+        self.max_long_edge = video_cfg.get('max_long_edge', 1024)
+
+        timeout_cfg = cfg.get('timeout', {})
+        self.download_timeout = timeout_cfg.get('download', 60)
+
+        # 临时目录
+        self.work_dir = f"/tmp/fam_media/task_{task_id}"
+        self.video_path = os.path.join(self.work_dir, f"video_{task_id}.mp4")
+        self.frames_dir = os.path.join(self.work_dir, "frames")
+        self.keyframes_dir = os.path.join(self.work_dir, "keyframes")
+
+    def download_video(self, video_url: str) -> str:
+        """下载视频"""
+        os.makedirs(self.work_dir, exist_ok=True)
+        start = time.time()
+        log_task(logger, self.task_id, 'download', f'开始下载: {video_url}')
+
+        resp = requests.get(video_url, stream=True, timeout=self.download_timeout)
+        if resp.status_code != 200:
+            raise Exception(f"下载失败: HTTP {resp.status_code}")
+
+        with open(self.video_path, 'wb') as f:
+            for chunk in resp.iter_content(chunk_size=8192):
+                f.write(chunk)
+
+        duration_ms = int((time.time() - start) * 1000)
+        size_mb = os.path.getsize(self.video_path) / (1024 * 1024)
+        log_task(logger, self.task_id, 'download', f'下载完成: {size_mb:.1f}MB', duration_ms=duration_ms)
+        return self.video_path
+
+    def _get_video_duration(self, video_path: str) -> float:
+        """用 ffprobe 获取视频时长（秒）"""
+        try:
+            cmd = [
+                'ffprobe', '-v', 'error',
+                '-show_entries', 'format=duration',
+                '-of', 'default=noprint_wrappers=1:nokey=1',
+                video_path
+            ]
+            result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
+            if result.returncode == 0:
+                return float(result.stdout.strip())
+        except Exception as e:
+            logger.warning(f"[task_id={self.task_id}] ffprobe 失败: {e}")
+        return 0.0
+
+    def extract_candidate_frames(self, video_path: str) -> List[str]:
+        """等距粗抽候选帧"""
+        os.makedirs(self.frames_dir, exist_ok=True)
+        duration = self._get_video_duration(video_path)
+
+        if duration > 0:
+            interval = duration / self.candidate_frames
+        else:
+            # 兜底: 每 60s 抽一帧
+            interval = 60
+            logger.warning(f"[task_id={self.task_id}] ffprobe 失败，退化为 60s 间隔抽帧")
+
+        cmd = [
+            'ffmpeg', '-i', video_path,
+            '-vf', f'fps=1/{interval}',
+            '-q:v', '2',
+            os.path.join(self.frames_dir, 'frame_%04d.jpg')
+        ]
+        try:
+            subprocess.run(cmd, capture_output=True, timeout=120, check=True)
+        except subprocess.CalledProcessError as e:
+            logger.error(f"[task_id={self.task_id}] FFmpeg 抽帧失败: {e}")
+            raise
+
+        # 收集候选帧路径
+        frames = sorted([
+            os.path.join(self.frames_dir, f)
+            for f in os.listdir(self.frames_dir)
+            if f.endswith('.jpg')
+        ])
+        log_task(logger, self.task_id, 'extract', f'粗抽 {len(frames)} 张候选帧')
+        return frames
+
+    def select_key_frames(self, candidate_frames: List[str]) -> List[str]:
+        """帧差分析筛选关键帧"""
+        if len(candidate_frames) <= self.min_key_frames:
+            return candidate_frames[:self.max_key_frames]
+
+        try:
+            # 加载所有候选帧
+            images = []
+            for path in candidate_frames:
+                img = cv2.imread(path)
+                if img is not None:
+                    images.append((path, img))
+
+            if len(images) < 2:
+                return candidate_frames[:self.max_key_frames]
+
+            # 计算每帧与前一关键帧的 MSE
+            key_indices = [0]  # 首帧必选
+            last_key_img = images[0][1]
+
+            for i in range(1, len(images)):
+                mse = self._compute_mse(last_key_img, images[i][1])
+                if mse > self.mse_threshold:
+                    key_indices.append(i)
+                    last_key_img = images[i][1]
+
+            # 末帧必选
+            if key_indices[-1] != len(images) - 1:
+                key_indices.append(len(images) - 1)
+
+            # 若 < min_key_frames，从剩余中均匀补足
+            if len(key_indices) < self.min_key_frames:
+                remaining = [i for i in range(len(images)) if i not in key_indices]
+                step = max(1, len(remaining) // (self.min_key_frames - len(key_indices)))
+                for i in range(0, len(remaining), step):
+                    if len(key_indices) >= self.min_key_frames:
+                        break
+                    key_indices.append(remaining[i])
+                key_indices.sort()
+
+            # 若 > max_key_frames，按差异值降序取前 N
+            if len(key_indices) > self.max_key_frames:
+                # 计算每个关键帧与前一帧的差异
+                diffs = []
+                for idx in key_indices[1:-1]:  # 不含首末帧
+                    diff = self._compute_mse(images[idx-1][1], images[idx][1])
+                    diffs.append((idx, diff))
+                diffs.sort(key=lambda x: x[1], reverse=True)
+                # 保留首末帧 + 差异最大的
+                keep = {0, len(images)-1}
+                for idx, _ in diffs[:self.max_key_frames - 2]:
+                    keep.add(idx)
+                key_indices = sorted(keep)
+
+            key_frames = [images[i][0] for i in key_indices]
+            log_task(logger, self.task_id, 'select_keyframes', f'筛选 {len(key_frames)} 张关键帧')
+            return key_frames
+
+        except Exception as e:
+            logger.warning(f"[task_id={self.task_id}] 帧差分析异常: {e}，退化为等距抽 5 帧")
+            step = max(1, len(candidate_frames) // self.min_key_frames)
+            return candidate_frames[::step][:self.min_key_frames]
+
+    def _compute_mse(self, img1, img2) -> float:
+        """计算两帧的 MSE"""
+        # 转灰度并统一尺寸
+        h = min(img1.shape[0], img2.shape[0])
+        w = min(img1.shape[1], img2.shape[1])
+        g1 = cv2.cvtColor(img1, cv2.COLOR_BGR2GRAY)
+        g2 = cv2.cvtColor(img2, cv2.COLOR_BGR2GRAY)
+        g1 = cv2.resize(g1, (w, h))
+        g2 = cv2.resize(g2, (w, h))
+        diff = g1.astype(np.float64) - g2.astype(np.float64)
+        mse = np.mean(diff ** 2)
+        return float(mse)
+
+    def compress_frames(self, frame_paths: List[str]) -> List[str]:
+        """压缩关键帧（长边 ≤ max_long_edge，JPEG 质量 80）"""
+        os.makedirs(self.keyframes_dir, exist_ok=True)
+        compressed = []
+
+        for i, path in enumerate(frame_paths):
+            out_path = os.path.join(self.keyframes_dir, f"keyframe_{i+1:02d}.jpg")
+            try:
+                img = cv2.imread(path)
+                if img is None:
+                    logger.warning(f"[task_id={self.task_id}] 读取图片失败: {path}")
+                    continue
+
+                h, w = img.shape[:2]
+                if max(h, w) > self.max_long_edge:
+                    scale = self.max_long_edge / max(h, w)
+                    img = cv2.resize(img, (int(w * scale), int(h * scale)))
+
+                cv2.imwrite(out_path, img, [cv2.IMWRITE_JPEG_QUALITY, self.jpeg_quality])
+                compressed.append(out_path)
+
+            except Exception as e:
+                logger.warning(f"[task_id={self.task_id}] 压缩失败 {path}: {e}")
+                continue
+
+        log_task(logger, self.task_id, 'compress', f'压缩 {len(compressed)} 张关键帧')
+        return compressed
+
+    def compute_timestamps(self, video_path: str, frame_count: int,
+                            event_start_time: str) -> List[str]:
+        """计算每帧的绝对时间戳 = 视频开始时间 + 帧偏移"""
+        from datetime import datetime, timedelta
+
+        duration = self._get_video_duration(video_path)
+        if duration <= 0:
+            duration = frame_count * 60  # 兜底
+
+        interval = duration / frame_count
+        try:
+            start_dt = datetime.fromisoformat(event_start_time.replace('Z', '+00:00'))
+        except Exception:
+            start_dt = datetime.now()
+
+        timestamps = []
+        for i in range(frame_count):
+            offset = interval * i
+            ts = start_dt + timedelta(seconds=offset)
+            timestamps.append(ts.strftime('%Y-%m-%d %H:%M:%S'))
+
+        return timestamps
+
+    def cleanup(self):
+        """清理临时文件"""
+        import shutil
+        try:
+            if os.path.exists(self.work_dir):
+                shutil.rmtree(self.work_dir)
+                log_task(logger, self.task_id, 'cleanup', f'清理临时目录: {self.work_dir}')
+        except Exception as e:
+            logger.warning(f"[task_id={self.task_id}] 清理失败: {e}")