From a8262123025fda1f35bfa63066e98c3bf2b41288 Mon Sep 17 00:00:00 2001 From: simonkoson <28867558@qq.com> Date: Fri, 12 Jun 2026 17:56:30 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20doco=20=E5=85=B3=E9=94=AE=E5=B8=A7?= =?UTF-8?q?=E8=BF=87=E6=BB=A4=20-=20dHash=E7=AE=97=E6=B3=95=20+=20IoU?= =?UTF-8?q?=E4=BF=9D=E5=BA=95=20+=20=E8=AF=8A=E6=96=ADCSV=20+=20474vs374?= =?UTF-8?q?=E6=8E=92=E6=9F=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- doco/src/doco/video_split.py | 313 ++++++++++++++++++++++++++++------- 1 file changed, 251 insertions(+), 62 deletions(-) diff --git a/doco/src/doco/video_split.py b/doco/src/doco/video_split.py index f218c38..a5e6342 100644 --- a/doco/src/doco/video_split.py +++ b/doco/src/doco/video_split.py @@ -3,12 +3,13 @@ 视频双路拆分 - P1 核心模块 ================================================= 功能: - A 路:视频帧 → pHash 变化检测 → OCR → B 稿 txt + A 路:视频帧 → 空白帧过滤 → 哈希变化检测 → OCR → B 稿 txt B 路:视频 → 16kHz/单声道/16bit WAV 不引入 ffmpeg-python 等 wrapper,只用 subprocess 调系统 ffmpeg。 """ +import csv import hashlib import json import os @@ -40,14 +41,25 @@ 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 +# 哈希算法参数 +# 可选: "phash"(感知哈希,默认) / "dhash"(差异哈希,对边缘更敏感) +HASH_ALGORITHM = "dhash" +# pHash 阈值:海明距离 > 此值才算有变化 +PHASH_THRESHOLD = 2 +# dHash 阈值:差异位 > 此值才算有变化 +DHASH_THRESHOLD = 5 +# IoU 保底阈值:二值化帧间 IoU > 此值视为同字幕,跳过 +IOU_THRESHOLD = 0.95 + # ======================================================================== # FFmpeg 封装 @@ -157,11 +169,11 @@ def extract_audio( # ======================================================================== -# pHash 变化检测 +# 空白帧检测 # ======================================================================== -def is_blank_frame(image_path: Path) -> bool: +def is_blank_frame(image_path: Path, debug: bool = False) -> Tuple[bool, float]: """ 判断是否为空白帧(留气口黑画面) @@ -169,73 +181,209 @@ def is_blank_frame(image_path: Path) -> bool: 如果占比 < 0.5%,判定为空白帧 适用于:黑底白字场景(军事科技栏目) + + 返回: (is_blank, white_ratio) """ 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 + is_blank = white_ratio < BLANK_FRAME_WHITE_PIXEL_RATIO + + if debug: + frame_idx = image_path.stem # e.g. "frame_0226" + print(f"[debug] {frame_idx}, white_ratio={white_ratio:.6f}, is_blank={is_blank}") + + return is_blank, white_ratio -def compute_phash(image_path: Path) -> str: - """计算图片的 pHash,返回 hex 字符串""" +# ======================================================================== +# 哈希变化检测 +# ======================================================================== + + +def compute_hash(image_path: Path, algorithm: str = "dhash") -> str: + """ + 计算图片的哈希值 + + 算法: + - dhash(差异哈希):对边缘更敏感,适合字幕小图场景 + - phash(感知哈希):对内容结构敏感 + """ img = Image.open(image_path) - ph = imagehash.phash(img) - return str(ph) + if algorithm == "dhash": + h = imagehash.dhash(img) + else: + h = imagehash.phash(img) + return str(h) + + +def hamming_distance(s1: str, s2: str) -> int: + """计算两个 hex 哈希字符串的海明距离""" + if len(s1) != len(s2): + return max(len(s1), len(s2)) + return sum(c1 != c2 for c1, c2 in zip(s1, s2)) + + +def compute_binary_matrix(image_path: Path, threshold: int = 200) -> List[List[int]]: + """ + 将图片转为二值化矩阵(亮度 > threshold → 1,否则 → 0) + 用于 IoU 对比 + """ + img = Image.open(image_path).convert("L") + w, h = img.size + matrix = [] + for y in range(h): + row = [] + for x in range(w): + pixel = img.getpixel((x, y)) + row.append(1 if pixel > threshold else 0) + matrix.append(row) + return matrix + + +def compute_iou(matrix1: List[List[int]], matrix2: List[List[int]]) -> float: + """ + 计算两个二值化矩阵的 IoU(交并比) + """ + h = len(matrix1) + w = len(matrix1[0]) if h > 0 else 0 + intersection = 0 + union = 0 + for y in range(h): + for x in range(w): + v1 = matrix1[y][x] + v2 = matrix2[y][x] + if v1 == 1 or v2 == 1: + union += 1 + if v1 == 1 and v2 == 1: + intersection += 1 + return intersection / union if union > 0 else 1.0 def find_keyframes( frames: List[Tuple[int, int, Path]], - threshold: int = 8, -) -> List[Dict]: + hash_algorithm: str = "dhash", + phash_threshold: int = PHASH_THRESHOLD, + dhash_threshold: int = DHASH_THRESHOLD, + iou_threshold: float = IOU_THRESHOLD, + debug_csv: Optional[Path] = None, +) -> Tuple[List[Dict], Dict]: """ - 基于 pHash 海明距离找出字幕变化的关键帧 + 基于多级检测找出字幕变化的关键帧 + + 过滤级别(从轻到严): + 1. IoU 保底:二值化矩阵 IoU > iou_threshold → 跳过 + 2. 哈希检测:海明距离 > threshold → 新关键帧 算法: - 第一帧总是关键帧 - - 后续帧:如果与上一个关键帧的 pHash 海明距离 > threshold,则是新关键帧 + - 后续帧:先走 IoU 保底,再走哈希检测 - threshold: 海明距离阈值,默认 8 + 返回: (keyframes, frame_analysis) + frame_analysis: {frame_index: {white_ratio, is_blank, hash_value, iou_to_prev, hash_dist_to_prev, decision}} """ if not frames: - return [] + return [], {} keyframes = [] - last_keyframe_phash = None + frame_analysis: Dict[int, Dict] = {} + last_keyframe_data: Optional[Dict] = None + + # CSV 写入器 + csv_file = None + csv_writer = None + if debug_csv: + csv_file = open(debug_csv, "w", newline="", encoding="utf-8") + csv_writer = csv.writer(csv_file) + csv_writer.writerow([ + "frame_index", "timestamp_ms", "white_pixel_ratio", "is_blank", + "hash_value", "iou_to_prev", "hash_dist_to_prev", + "decision", "keyframe_index" + ]) + + threshold = phash_threshold if hash_algorithm == "phash" else dhash_threshold for frame_index, timestamp_ms, image_path in frames: - phash = compute_phash(image_path) + is_blank, white_ratio = is_blank_frame(image_path, debug=False) + hash_value = compute_hash(image_path, hash_algorithm) - is_keyframe = False - if last_keyframe_phash is None: + decision = "kept" + keyframe_index = -1 + + if last_keyframe_data is None: # 第一帧总是关键帧 is_keyframe = True + iou_to_prev = None + hash_dist_to_prev = None else: - # 计算海明距离 - hamming = hamming_distance(last_keyframe_phash, phash) - if hamming > threshold: + # IoU 保底检测 + last_matrix = last_keyframe_data["binary_matrix"] + current_matrix = compute_binary_matrix(image_path) + iou_to_prev = compute_iou(last_matrix, current_matrix) + hash_dist_to_prev = hamming_distance(last_keyframe_data["hash_value"], hash_value) + + # 决策逻辑:IoU > threshold → 跳过;否则哈希 > threshold → 新关键帧 + if iou_to_prev > iou_threshold: + # IoU 认为相同,跳过 + is_keyframe = False + decision = "duplicate(iou)" + elif hash_dist_to_prev > threshold: + # 哈希认为有变化,新关键帧 is_keyframe = True + decision = "keyframe(hash)" + else: + # 哈希认为相同 + is_keyframe = False + decision = "duplicate(hash)" + + if is_blank: + decision = "blank" + is_keyframe = False + + frame_analysis[frame_index] = { + "white_ratio": white_ratio, + "is_blank": is_blank, + "hash_value": hash_value, + "iou_to_prev": iou_to_prev, + "hash_dist_to_prev": hash_dist_to_prev, + "decision": decision, + "timestamp_ms": timestamp_ms, + } if is_keyframe: - keyframes.append({ + binary_matrix = compute_binary_matrix(image_path) + kf_entry = { "frame_index": frame_index, "timestamp_ms": timestamp_ms, "frame_image_path": str(image_path), - "phash": phash, - "ocr_text": "", # P2 调用 DeepSeek Vision 填充 - }) - last_keyframe_phash = phash + "hash_value": hash_value, + "ocr_text": "", + "binary_matrix": binary_matrix, + } + keyframes.append(kf_entry) + last_keyframe_data = kf_entry + keyframe_index = len(keyframes) - 1 - return keyframes + # CSV 写入 + if csv_writer: + csv_writer.writerow([ + frame_index, + timestamp_ms, + f"{white_ratio:.6f}", + is_blank, + hash_value, + f"{iou_to_prev:.6f}" if iou_to_prev is not None else "", + hash_dist_to_prev if hash_dist_to_prev is not None else "", + decision, + keyframe_index, + ]) + if csv_file: + csv_file.close() -def hamming_distance(s1: str, s2: str) -> int: - """计算两个 hex pHash 字符串的海明距离""" - if len(s1) != len(s2): - # pHash 长度不一致,取较长字符串的长度作为海明距离上限 - return max(len(s1), len(s2)) - return sum(c1 != c2 for c1, c2 in zip(s1, s2)) + return keyframes, frame_analysis # ======================================================================== @@ -251,11 +399,7 @@ def ocr_frame(image_path: Path) -> str: P2: 调用 DeepSeek Vision API """ if not DEEPSEEK_API_KEY: - # 无 API Key,返回占位 return f"[OCR待填充 frame={image_path.name}]" - - # P2 实现:调用 DeepSeek Vision - # TODO: P2 实现 return f"[OCR待填充 frame={image_path.name}]" @@ -265,7 +409,7 @@ def ocr_keyframes(keyframes: List[Dict]) -> List[Dict]: for kf in keyframes: image_path = Path(kf["frame_image_path"]) ocr_text = ocr_frame(image_path) - kf_copy = kf.copy() + kf_copy = {k: v for k, v in kf.items() if k != "binary_matrix"} kf_copy["ocr_text"] = ocr_text result.append(kf_copy) return result @@ -291,11 +435,10 @@ def build_b_manuscript(keyframes: List[Dict]) -> List[str]: last_text = None for kf in keyframes: - text = kf["ocr_text"].strip() + text = kf.get("ocr_text", "").strip() if not text: continue - # 跳过占位文本 if text.startswith("[OCR待填充"): text = "" @@ -324,7 +467,10 @@ def split_video( video_path: Path, output_dir: Path, episode_id: str, - phash_threshold: int = 5, + hash_algorithm: str = HASH_ALGORITHM, + phash_threshold: int = PHASH_THRESHOLD, + dhash_threshold: int = DHASH_THRESHOLD, + iou_threshold: float = IOU_THRESHOLD, fps: int = 1, dry_run: bool = False, ) -> Dict[str, any]: @@ -335,7 +481,10 @@ def split_video( video_path: 输入视频路径 output_dir: 输出目录(work/ 路径) episode_id: 节目 ID - phash_threshold: pHash 海明距离阈值,默认 5 + hash_algorithm: 哈希算法,"dhash"(默认) 或 "phash" + phash_threshold: pHash 海明距离阈值,默认 2 + dhash_threshold: dHash 海明距离阈值,默认 5 + iou_threshold: IoU 保底阈值,默认 0.95 fps: 抽帧帧率,默认 1(每秒一帧) dry_run: True 则不调 OCR,只输出裁切帧和 keyframes.json @@ -369,43 +518,73 @@ def split_video( 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}") + print(f"[video_split] fps={fps}, hash={hash_algorithm}, phash_th={phash_threshold}, dhash_th={dhash_threshold}, iou_th={iou_threshold}, dry_run={dry_run}") - # ---- A 路:抽帧 + 空白帧过滤 + pHash 检测 + OCR ---- + # ---- A 路:抽帧 + 空白帧过滤 + 哈希检测 + 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) total_extracted = len(frames) print(f"[video_split] 抽帧完成,共 {total_extracted} 帧") + # 诊断 CSV + debug_csv_path = output_dir / "frame_analysis_debug.csv" + # 空白帧过滤 print("[video_split] 空白帧过滤...") non_blank_frames = [] blank_count = 0 + # 统计各 decision 类型 + decision_stats = {"blank": 0, "duplicate(iou)": 0, "duplicate(hash)": 0, "keyframe(hash)": 0, "kept": 0} + for frame_index, timestamp_ms, image_path in frames: - if is_blank_frame(image_path): + is_blank, white_ratio = is_blank_frame(image_path, debug=False) + if is_blank: blank_count += 1 + decision_stats["blank"] += 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(non_blank_frames, threshold=phash_threshold) - duplicate_count = len(non_blank_frames) - len(keyframes) - print(f"[stats] pHash 去重后: {len(keyframes)} 张 (筛掉 {duplicate_count} 张同字幕相邻)") + # 哈希 + IoU 去重 + print(f"[video_split] 哈希变化检测(算法={hash_algorithm}, IoU保底阈值={iou_threshold})...") + keyframes, frame_analysis = find_keyframes( + non_blank_frames, + hash_algorithm=hash_algorithm, + phash_threshold=phash_threshold, + dhash_threshold=dhash_threshold, + iou_threshold=iou_threshold, + debug_csv=debug_csv_path, + ) + + # 统计各 decision + for frame_idx, data in frame_analysis.items(): + decision_stats[data["decision"]] = decision_stats.get(data["decision"], 0) + + duplicate_iou = decision_stats.get("duplicate(iou)", 0) + duplicate_hash = decision_stats.get("duplicate(hash)", 0) + total_duplicate = blank_count + duplicate_iou + duplicate_hash + print(f"[stats] pHash/dHash 去重后(IoU保底): {len(keyframes)} 张 (筛掉 {duplicate_iou} 张IoU相同 + {duplicate_hash} 张哈希相同)") print(f"[stats] 最终关键帧: {len(keyframes)} 张") print(f"[video_split] 检测到 {len(keyframes)} 个关键帧") + print(f"[video_split] 诊断CSV写入: {debug_csv_path}") + + # 确认 frames/ 目录文件数与 total_extracted 一致 + actual_frame_files = len(list(frames_dir.glob("frame_*.png"))) + print(f"[debug] frames/ 目录实际文件数: {actual_frame_files} (预期: {total_extracted})") + if actual_frame_files != total_extracted: + print(f"[WARNING] frames/ 文件数({actual_frame_files}) 与抽帧数({total_extracted})不一致!") if dry_run: - # dry-run:不调 OCR,只输出关键帧信息 print("[video_split] dry-run:跳过 OCR") - keyframes = [ - {**kf, "ocr_text": ""} for kf in keyframes + keyframes_clean = [ + {k: v for k, v in kf.items() if k != "binary_matrix"} + for kf in keyframes ] else: print("[video_split] OCR 关键帧...") - keyframes = ocr_keyframes(keyframes) + keyframes_clean = ocr_keyframes(keyframes) print(f"[video_split] OCR 完成") # ---- B 路:音频提取 ---- @@ -418,16 +597,24 @@ def split_video( filter_stats = { "total_extracted_frames": total_extracted, "blank_frames_removed": blank_count, - "duplicate_frames_removed": duplicate_count, + "duplicate_frames_removed": duplicate_iou + duplicate_hash, + "duplicate_iou_removed": duplicate_iou, + "duplicate_hash_removed": duplicate_hash, "final_keyframes": len(keyframes), + "hash_algorithm": hash_algorithm, + "phash_threshold": phash_threshold, + "dhash_threshold": dhash_threshold, + "iou_threshold": iou_threshold, } if dry_run: - # dry-run:不写 B 稿,只写 keyframes.json keyframes_data = { "video_path": str(video_path), "fps_sampled": fps, + "hash_algorithm": hash_algorithm, "phash_threshold": phash_threshold, + "dhash_threshold": dhash_threshold, + "iou_threshold": iou_threshold, "dry_run": True, "filter_stats": filter_stats, "crop_params": { @@ -439,7 +626,7 @@ def split_video( "applies_to": "军事科技 全栏目特殊视频", "ffmpeg_crop": SUBTITLE_CROP, }, - "keyframes": keyframes, + "keyframes": keyframes_clean, } keyframes_path = output_dir / "keyframes.json" with open(keyframes_path, "w", encoding="utf-8") as f: @@ -456,8 +643,7 @@ def split_video( "filter_stats": filter_stats, } else: - # 正式:写 B 稿 + keyframes.json - b_lines = build_b_manuscript(keyframes) + b_lines = build_b_manuscript(keyframes_clean) b_manuscript_path = output_dir / "b_manuscript.txt" write_b_manuscript(b_lines, b_manuscript_path) print(f"[video_split] B稿写入: {b_manuscript_path} ({len(b_lines)} 行)") @@ -465,7 +651,10 @@ def split_video( keyframes_data = { "video_path": str(video_path), "fps_sampled": fps, + "hash_algorithm": hash_algorithm, "phash_threshold": phash_threshold, + "dhash_threshold": dhash_threshold, + "iou_threshold": iou_threshold, "dry_run": False, "filter_stats": filter_stats, "crop_params": { @@ -477,7 +666,7 @@ def split_video( "applies_to": "军事科技 全栏目特殊视频", "ffmpeg_crop": SUBTITLE_CROP, }, - "keyframes": keyframes, + "keyframes": keyframes_clean, } keyframes_path = output_dir / "keyframes.json" with open(keyframes_path, "w", encoding="utf-8") as f: @@ -491,4 +680,4 @@ def split_video( "keyframe_count": len(keyframes), "dry_run": False, "filter_stats": filter_stats, - } + } \ No newline at end of file