""" Video-Preprocessor - 视频预处理 流程: 1. 下载视频（超时 60s） 2. 根据视频时长自适应计算候选帧数，FFmpeg 等距粗抽 3. 根据视频时长自适应计算关键帧数，OpenCV 帧差分析筛选（MSE 阈值） 4. 压缩（长边 ≤ 1024px，JPEG 质量 80） 自适应规则: - 候选帧: max(candidate_min, duration_min * candidate_per_minute), 上限 candidate_max - 关键帧: max(min_key_frames, duration / key_frame_interval_sec), 上限 max_key_frames_cap 例: 30分钟视频 → 候选60张 → 关键帧12张（每2.5分钟1张） 例: 3分钟视频 → 候选30张 → 关键帧8张（保底） 异常兜底: - ffprobe 失败 -> 退化为按 60s 间隔抽帧 - 帧差分析异常 -> 退化为等距抽 min_key_frames 帧 - 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_per_minute = video_cfg.get('candidate_per_minute', 2) self.candidate_min = video_cfg.get('candidate_min', 30) self.candidate_max = video_cfg.get('candidate_max', 120) self.key_frame_interval_sec = video_cfg.get('key_frame_interval_sec', 150) self.min_key_frames = video_cfg.get('min_key_frames', 5) self.max_key_frames_floor = video_cfg.get('max_key_frames_floor', 8) self.max_key_frames_cap = video_cfg.get('max_key_frames_cap', 30) 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) # 视频时长（秒），在 extract_candidate_frames 中填充 self.video_duration = 0.0 # 临时目录 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 save_upload(self, file_storage) -> str: """保存推送模式上传的视频文件（multipart），替代 download_video""" os.makedirs(self.work_dir, exist_ok=True) start = time.time() file_storage.save(self.video_path) duration_ms = int((time.time() - start) * 1000) size_mb = os.path.getsize(self.video_path) / (1024 * 1024) log_task(logger, self.task_id, 'upload', 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]: """等距粗抽候选帧（数量随视频时长自适应，使用快速 seek）""" os.makedirs(self.frames_dir, exist_ok=True) duration = self._get_video_duration(video_path) self.video_duration = duration if duration > 0: duration_min = duration / 60 # 自适应候选帧数：每分钟 candidate_per_minute 张，保底 candidate_min，上限 candidate_max candidate_count = min( max(self.candidate_min, int(duration_min * self.candidate_per_minute)), self.candidate_max ) interval = duration / candidate_count else: # 兜底: 每 60s 抽一帧 interval = 60 candidate_count = 0 logger.warning(f"[task_id={self.task_id}] ffprobe 失败，退化为 60s 间隔抽帧") # 快速 seek 逐帧提取（比 fps 滤镜快 6-8 倍，ARM CPU 上尤甚） timestamps = [i * interval for i in range(candidate_count)] if candidate_count > 0 else [] if not timestamps: # 兜底: 未知时长，用 ffprobe 不可用时按 60s 间隔 timestamps = [i * 60 for i in range(30)] for i, ts in enumerate(timestamps): output_path = os.path.join(self.frames_dir, f'frame_{i+1:04d}.jpg') cmd = [ 'ffmpeg', '-ss', f'{ts:.1f}', '-i', video_path, '-frames:v', '1', '-q:v', '2', output_path ] try: subprocess.run(cmd, capture_output=True, timeout=30, check=True) except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as e: logger.warning(f"[task_id={self.task_id}] seek 到 {ts:.1f}s 失败: {e}") # 收集候选帧路径 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'视频时长 {duration:.0f}s, 快速 seek 粗抽 {len(frames)} 张候选帧 (目标 {candidate_count})') return frames def _compute_adaptive_key_frame_counts(self) -> Tuple[int, int]: """根据视频时长自适应计算关键帧下限和上限""" if self.video_duration > 0: # 每隔 key_frame_interval_sec 秒 1 张关键帧 adaptive = int(self.video_duration / self.key_frame_interval_sec) max_kf = min(max(self.max_key_frames_floor, adaptive), self.max_key_frames_cap) else: max_kf = self.max_key_frames_floor min_kf = max(self.min_key_frames, max_kf // 2) return min_kf, max_kf def select_key_frames(self, candidate_frames: List[str]) -> List[str]: """帧差分析筛选关键帧（数量随视频时长自适应）""" min_kf, max_kf = self._compute_adaptive_key_frame_counts() log_task(logger, self.task_id, 'select_keyframes', f'自适应关键帧: min={min_kf}, max={max_kf} (视频时长 {self.video_duration:.0f}s)') if len(candidate_frames) <= min_kf: return candidate_frames[:max_kf] 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[:max_kf] # 计算每帧与前一关键帧的 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_kf，从剩余中均匀补足 if len(key_indices) < min_kf: remaining = [i for i in range(len(images)) if i not in key_indices] step = max(1, len(remaining) // (min_kf - len(key_indices))) for i in range(0, len(remaining), step): if len(key_indices) >= min_kf: break key_indices.append(remaining[i]) key_indices.sort() # 若 > max_kf，按差异值降序取前 N if len(key_indices) > max_kf: # 计算每个关键帧与前一帧的差异 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[:max_kf - 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}，退化为等距抽 {min_kf} 帧") step = max(1, len(candidate_frames) // min_kf) return candidate_frames[::step][:min_kf] 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 from datetime import timedelta, timezone # 统一北京时区: 视频均为北京时间录制，Edge 机器是 UTC， # fallback 不能用本地 datetime.now() try: start_dt = datetime.fromisoformat(event_start_time.replace('Z', '+00:00')) if start_dt.tzinfo is not None: start_dt = start_dt.astimezone(timezone(timedelta(hours=8))).replace(tzinfo=None) except Exception: start_dt = datetime.now(timezone(timedelta(hours=8))).replace(tzinfo=None) 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}")