simonkoson/tps-dashboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: doco 关键帧过滤 - dHash算法 + IoU保底 + 诊断CSV + 474vs374排查

Browse Source
This commit is contained in:
simonkoson
2026-06-12 17:56:30 +08:00
parent b138b98bc4
commit a826212302
1 changed files with 251 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
doco/src/doco/video_split.py
+251 -62
View File
@@ -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,
}
}