diff --git a/apps/worker/video_processing/dedup.py b/apps/worker/video_processing/dedup.py index 817f9034b..cd5011f8b 100755 --- a/apps/worker/video_processing/dedup.py +++ b/apps/worker/video_processing/dedup.py @@ -1051,110 +1051,6 @@ class VideoDeduplicator: logger.info("check_batch_duplicate no match (batch=%s): best_fusion=%.3f", batch_id, best_score) return None - -# ── 文案 & 结构维度查重(Issue #P2-后端3) ──────────────────────── - - -def _normalize_text(text: str) -> str: - """文本标准化:去空白、转小写、去标点。""" - if not text: - return "" - # 去空白字符 - text = re.sub(r"\s+", "", text) - # 转小写 - text = text.lower() - # 去标点(只保留中文、字母、数字) - text = re.sub(r"[^\w\u4e00-\u9fff]", "", text) - return text - - -def compute_text_similarity(text1: str, text2: str) -> float: - """计算两段文本的相似度(0~1)。 - - 使用字符级 Jaccard 相似度:交集 / 并集。 - 适合短文本(配音脚本)的相似度比对。 - - Args: - text1: 第一段文本 - text2: 第二段文本 - - Returns: - 0~1 之间的相似度 - """ - t1 = _normalize_text(text1) - t2 = _normalize_text(text2) - - if not t1 and not t2: - return 1.0 # 都为空,视为完全相同 - if not t1 or not t2: - return 0.0 # 一个为空,完全不同 - - # 字符级 Jaccard - set1 = set(t1) - set2 = set(t2) - intersection = set1 & set2 - union = set1 | set2 - - if not union: - return 0.0 - - return len(intersection) / len(union) - - -def compute_structure_similarity(clips1: list[dict], clips2: list[dict]) -> float: - """计算两个视频的结构相似度(0~1)。 - - 结构维度包括: - 1. 片段数差异(数量越接近越相似) - 2. 片段类型序列(相同位置的片段类型是否一致) - 3. 时长分布(各片段时长占比是否相似) - - Args: - clips1: 第一个视频的片段列表,每项包含 {clip_type, duration} - clips2: 第二个视频的片段列表 - - Returns: - 0~1 之间的相似度 - """ - if not clips1 and not clips2: - return 1.0 - if not clips1 or not clips2: - return 0.0 - - # 1. 片段数相似度(数量差异越大越低) - n1, n2 = len(clips1), len(clips2) - count_sim = min(n1, n2) / max(n1, n2) - - # 2. 类型序列相似度(逐位比较,相同位置类型是否一致) - min_len = min(n1, n2) - type_matches = sum(1 for i in range(min_len) if clips1[i].get("clip_type") == clips2[i].get("clip_type")) - type_sim = type_matches / min_len if min_len > 0 else 0.0 - - # 3. 时长分布相似度(归一化后比较分布) - total1 = sum(c.get("duration", 0) for c in clips1) - total2 = sum(c.get("duration", 0) for c in clips2) - - if total1 > 0 and total2 > 0: - # 归一化为占比 - dist1 = [c.get("duration", 0) / total1 for c in clips1] - dist2 = [c.get("duration", 0) / total2 for c in clips2] - - # 比较前 min_len 个片段的占比差异(L1 距离转相似度) - l1_dist = sum(abs(dist1[i] - dist2[i]) for i in range(min_len)) - # 加上多出的片段占比 - if n1 > n2: - l1_dist += sum(dist1[i] for i in range(n2, n1)) - elif n2 > n1: - l1_dist += sum(dist2[i] for i in range(n1, n2)) - - # L1 距离范围 [0, 2],转为相似度 [0, 1] - duration_sim = 1.0 - (l1_dist / 2.0) - else: - duration_sim = 0.0 - - # 三维度加权:数量 0.3 + 类型 0.4 + 时长 0.3 - return count_sim * 0.3 + type_sim * 0.4 + duration_sim * 0.3 - def compute_duplicate_rate( self, fingerprint: VideoFingerprint, @@ -1361,6 +1257,110 @@ def compute_structure_similarity(clips1: list[dict], clips2: list[dict]) -> floa } +# ── 文案 & 结构维度查重(Issue #P2-后端3) ──────────────────────── + + +def _normalize_text(text: str) -> str: + """文本标准化:去空白、转小写、去标点。""" + if not text: + return "" + # 去空白字符 + text = re.sub(r"\s+", "", text) + # 转小写 + text = text.lower() + # 去标点(只保留中文、字母、数字) + text = re.sub(r"[^\w\u4e00-\u9fff]", "", text) + return text + + +def compute_text_similarity(text1: str, text2: str) -> float: + """计算两段文本的相似度(0~1)。 + + 使用字符级 Jaccard 相似度:交集 / 并集。 + 适合短文本(配音脚本)的相似度比对。 + + Args: + text1: 第一段文本 + text2: 第二段文本 + + Returns: + 0~1 之间的相似度 + """ + t1 = _normalize_text(text1) + t2 = _normalize_text(text2) + + if not t1 and not t2: + return 1.0 # 都为空,视为完全相同 + if not t1 or not t2: + return 0.0 # 一个为空,完全不同 + + # 字符级 Jaccard + set1 = set(t1) + set2 = set(t2) + intersection = set1 & set2 + union = set1 | set2 + + if not union: + return 0.0 + + return len(intersection) / len(union) + + +def compute_structure_similarity(clips1: list[dict], clips2: list[dict]) -> float: + """计算两个视频的结构相似度(0~1)。 + + 结构维度包括: + 1. 片段数差异(数量越接近越相似) + 2. 片段类型序列(相同位置的片段类型是否一致) + 3. 时长分布(各片段时长占比是否相似) + + Args: + clips1: 第一个视频的片段列表,每项包含 {clip_type, duration} + clips2: 第二个视频的片段列表 + + Returns: + 0~1 之间的相似度 + """ + if not clips1 and not clips2: + return 1.0 + if not clips1 or not clips2: + return 0.0 + + # 1. 片段数相似度(数量差异越大越低) + n1, n2 = len(clips1), len(clips2) + count_sim = min(n1, n2) / max(n1, n2) + + # 2. 类型序列相似度(逐位比较,相同位置类型是否一致) + min_len = min(n1, n2) + type_matches = sum(1 for i in range(min_len) if clips1[i].get("clip_type") == clips2[i].get("clip_type")) + type_sim = type_matches / min_len if min_len > 0 else 0.0 + + # 3. 时长分布相似度(归一化后比较分布) + total1 = sum(c.get("duration", 0) for c in clips1) + total2 = sum(c.get("duration", 0) for c in clips2) + + if total1 > 0 and total2 > 0: + # 归一化为占比 + dist1 = [c.get("duration", 0) / total1 for c in clips1] + dist2 = [c.get("duration", 0) / total2 for c in clips2] + + # 比较前 min_len 个片段的占比差异(L1 距离转相似度) + l1_dist = sum(abs(dist1[i] - dist2[i]) for i in range(min_len)) + # 加上多出的片段占比 + if n1 > n2: + l1_dist += sum(dist1[i] for i in range(n2, n1)) + elif n2 > n1: + l1_dist += sum(dist2[i] for i in range(n1, n2)) + + # L1 距离范围 [0, 2],转为相似度 [0, 1] + duration_sim = 1.0 - (l1_dist / 2.0) + else: + duration_sim = 0.0 + + # 三维度加权:数量 0.3 + 类型 0.4 + 时长 0.3 + return count_sim * 0.3 + type_sim * 0.4 + duration_sim * 0.3 + + def _save_fingerprint_chunks( fingerprint: VideoFingerprint, video_id: str,