feat: doco 抽帧过滤优化 - 空白帧过滤 + pHash 阈值调至 5 + 统计输出

doco/src/doco/video_split.py

@@ -39,6 +39,16 @@ DEEPSEEK_API_KEY = os.environ.get("DEEPSEEK_API_KEY", "").strip()
 SUBTITLE_CROP = "iw:ih*0.2:0:ih*0.8"
 
 
+# ========================================================================
+# 空白帧检测参数(黑底白字场景专用)
+# ========================================================================
+
+# 亮度阈值:0-255,>200 视为接近白色像素
+BLANK_FRAME_BRIGHTNESS_THRESHOLD = 200
+# 白色像素占比阈值:< 0.5% 则判定为空白帧(留气口黑画面)
+BLANK_FRAME_WHITE_PIXEL_RATIO = 0.005
+
+
 # ========================================================================
 # FFmpeg 封装
 # ========================================================================
@@ -151,6 +161,23 @@ def extract_audio(
 # ========================================================================
 
 
+def is_blank_frame(image_path: Path) -> bool:
+    """
+    判断是否为空白帧(留气口黑画面)
+
+    算法:转灰度,统计亮度 > 200(接近白)的像素数量
+         如果占比 < 0.5%,判定为空白帧
+
+    适用于:黑底白字场景(军事科技栏目)
+    """
+    img = Image.open(image_path).convert("L")
+    pixels = list(img.getdata())
+    total = len(pixels)
+    white_count = sum(1 for p in pixels if p > BLANK_FRAME_BRIGHTNESS_THRESHOLD)
+    white_ratio = white_count / total if total > 0 else 0
+    return white_ratio < BLANK_FRAME_WHITE_PIXEL_RATIO
+
+
 def compute_phash(image_path: Path) -> str:
     """计算图片的 pHash,返回 hex 字符串"""
     img = Image.open(image_path)
@@ -297,7 +324,7 @@ def split_video(
     video_path: Path,
     output_dir: Path,
     episode_id: str,
-    phash_threshold: int = 8,
+    phash_threshold: int = 5,
     fps: int = 1,
     dry_run: bool = False,
 ) -> Dict[str, any]:
@@ -308,7 +335,7 @@ def split_video(
         video_path: 输入视频路径
         output_dir: 输出目录(work/ 路径)
         episode_id: 节目 ID
-        phash_threshold: pHash 海明距离阈值,默认 8
+        phash_threshold: pHash 海明距离阈值,默认 5
         fps: 抽帧帧率,默认 1(每秒一帧)
         dry_run: True 则不调 OCR,只输出裁切帧和 keyframes.json
 
@@ -319,6 +346,7 @@ def split_video(
             "keyframes_path": Path,
             "keyframe_count": int,
             "dry_run": bool,
+            "filter_stats": dict,
         }
     """
     video_path = Path(video_path)
@@ -330,18 +358,43 @@ def split_video(
     # 创建输出目录
     output_dir.mkdir(parents=True, exist_ok=True)
     frames_dir = output_dir / "frames"
-    frames_dir.mkdir(parents=True, exist_ok=True)
+
+    # dry-run 前自动清空上次产物,避免干扰
+    if dry_run:
+        if frames_dir.exists():
+            shutil.rmtree(frames_dir)
+        frames_dir.mkdir(parents=True, exist_ok=True)
+        keyframes_json_path = output_dir / "keyframes.json"
+        if keyframes_json_path.exists():
+            keyframes_json_path.unlink()
 
     print(f"[video_split] 开始处理: {video_path.name}")
     print(f"[video_split] fps={fps}, pHash threshold={phash_threshold}, dry_run={dry_run}")
 
-    # ---- A 路:抽帧 + pHash 检测 + OCR ----
+    # ---- A 路:抽帧 + 空白帧过滤 + pHash 检测 + OCR ----
     print(f"[video_split] A路:抽帧({'裁切模式' if dry_run else '完整帧模式'})...")
     frames = extract_frames(video_path, output_dir, fps=fps, crop=SUBTITLE_CROP, dry_run=dry_run)
-    print(f"[video_split] 抽帧完成,共 {len(frames)} 帧")
+    total_extracted = len(frames)
+    print(f"[video_split] 抽帧完成,共 {total_extracted} 帧")
 
+    # 空白帧过滤
+    print("[video_split] 空白帧过滤...")
+    non_blank_frames = []
+    blank_count = 0
+    for frame_index, timestamp_ms, image_path in frames:
+        if is_blank_frame(image_path):
+            blank_count += 1
+        else:
+            non_blank_frames.append((frame_index, timestamp_ms, image_path))
+    print(f"[stats] 原始抽帧: {total_extracted} 张")
+    print(f"[stats] 空白帧过滤后: {len(non_blank_frames)} 张 (筛掉 {blank_count} 张纯黑)")
+
+    # pHash 去重
     print("[video_split] pHash 变化检测...")
-    keyframes = find_keyframes(frames, threshold=phash_threshold)
+    keyframes = find_keyframes(non_blank_frames, threshold=phash_threshold)
+    duplicate_count = len(non_blank_frames) - len(keyframes)
+    print(f"[stats] pHash 去重后: {len(keyframes)} 张 (筛掉 {duplicate_count} 张同字幕相邻)")
+    print(f"[stats] 最终关键帧: {len(keyframes)} 张")
     print(f"[video_split] 检测到 {len(keyframes)} 个关键帧")
 
     if dry_run:
@@ -362,6 +415,13 @@ def split_video(
     print(f"[video_split] 音频提取完成: {audio_path}")
 
     # ---- 输出产物 ----
+    filter_stats = {
+        "total_extracted_frames": total_extracted,
+        "blank_frames_removed": blank_count,
+        "duplicate_frames_removed": duplicate_count,
+        "final_keyframes": len(keyframes),
+    }
+
     if dry_run:
         # dry-run:不写 B 稿,只写 keyframes.json
         keyframes_data = {
@@ -369,6 +429,7 @@ def split_video(
             "fps_sampled": fps,
             "phash_threshold": phash_threshold,
             "dry_run": True,
+            "filter_stats": filter_stats,
             "crop_params": {
                 "width_ratio": 1.0,
                 "height_ratio": 0.2,
@@ -392,6 +453,7 @@ def split_video(
             "keyframe_count": len(keyframes),
             "dry_run": True,
             "b_manuscript_path": None,
+            "filter_stats": filter_stats,
         }
     else:
         # 正式:写 B 稿 + keyframes.json
@@ -405,6 +467,7 @@ def split_video(
             "fps_sampled": fps,
             "phash_threshold": phash_threshold,
             "dry_run": False,
+            "filter_stats": filter_stats,
             "crop_params": {
                 "width_ratio": 1.0,
                 "height_ratio": 0.2,
@@ -427,4 +490,5 @@ def split_video(
             "keyframes_path": str(keyframes_path),
             "keyframe_count": len(keyframes),
             "dry_run": False,
+            "filter_stats": filter_stats,
         }