From b138b98bc49b0f2f9348f856d9435b16f5c3220c Mon Sep 17 00:00:00 2001 From: simonkoson <28867558@qq.com> Date: Fri, 12 Jun 2026 17:38:12 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20doco=20=E6=8A=BD=E5=B8=A7=E8=BF=87?= =?UTF-8?q?=E6=BB=A4=E4=BC=98=E5=8C=96=20-=20=E7=A9=BA=E7=99=BD=E5=B8=A7?= =?UTF-8?q?=E8=BF=87=E6=BB=A4=20+=20pHash=20=E9=98=88=E5=80=BC=E8=B0=83?= =?UTF-8?q?=E8=87=B3=205=20+=20=E7=BB=9F=E8=AE=A1=E8=BE=93=E5=87=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- doco/src/doco/video_split.py | 76 +++++++++++++++++++++++++++++++++--- 1 file changed, 70 insertions(+), 6 deletions(-) diff --git a/doco/src/doco/video_split.py b/doco/src/doco/video_split.py index 8325e10..f218c38 100644 --- a/doco/src/doco/video_split.py +++ b/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, }