Skip to content

沈秋雨 / import_hk_songs

Go to a project
Toggle navigation
Toggle navigation pinning
Commit 93b03ea9 93b03ea91557e1bebbe201d31f48271e2d68830e by 沈秋雨
Options

feat(dedup): 优化歌词去重逻辑并引入作者字段增量合并 

- 新增 L2 候选加载缓存机制，避免重复下载歌词内容
- L2 去重返回 top-K 召回候选信息，丰富判定依据
- 分类函数新增跳过导入的规则，如空歌词且作者不详、原声类歌曲名等
- 实现作者字段（作词、作曲）增量合并功能，避免重复覆盖
- 对合并冲突添加异常处理并记录日志，确保稳定执行
- 引入后台线程异步写入数据库，提升批量处理性能
- 优化 L1 索引更新，避免去重误差，提高去重准确率
- 扩展统计字段，新增作者合并和跳过导入计数
- 前端增加决策过滤选项，支持根据决策筛选展示记录
- 修改界面显示逻辑，支持删除已审核样本，完善用户操作
- 调整页面显示候选数目为 Top5，优化视觉体验
- 参数新增 L2 精确哈希自动合并的最小歌词长度配置
- 添加数据库表结构自动迁移，新增 recalled_candidates 字段支持新功能
- 增强程序中断时数据库写入的安全关闭与日志提示
1 parent 25cb261f
Inline Side-by-side
Showing 5 changed files with 844 additions and 130 deletions
......@@ -6,8 +6,10 @@ hk_music_record 聚合数据导入脚本
import argparse
import csv
import json
import logging
import os
import queue
import re
import sys
import time
......@@ -40,8 +42,10 @@ load_dotenv()
OUTPUT_DIR = Path(__file__).resolve().parent / 'output'
LOG_DIR = OUTPUT_DIR / 'logs'
REPORT_DIR = OUTPUT_DIR / 'reports'
CACHE_DIR = OUTPUT_DIR / 'cache'
LOG_DIR.mkdir(parents=True, exist_ok=True)
REPORT_DIR.mkdir(parents=True, exist_ok=True)
CACHE_DIR.mkdir(parents=True, exist_ok=True)
# 日志配置
logging.basicConfig(
......@@ -245,7 +249,8 @@ INSERT INTO {table} (
lyric_archive_element_id, melody_archive_element_id, audio_fingerprint,
record_count,
dedup_action, dedup_decision, dedup_confidence, matched_song_id, l1_matched_id,
merge_authors, dedup_reason, biz_review_status, staging_status, imported_song_id
merge_authors, dedup_reason, biz_review_status, staging_status, imported_song_id,
recalled_candidates
) VALUES (
%s,%s,
%s,%s,%s,%s,%s,%s,
......@@ -261,7 +266,8 @@ INSERT INTO {table} (
%s,%s,%s,
%s,%s,%s,%s,%s,
%s,
%s,%s,%s,%s,%s
%s,%s,%s,%s,%s,
%s
)
ON DUPLICATE KEY UPDATE
dedup_action = VALUES(dedup_action),
......@@ -274,6 +280,7 @@ ON DUPLICATE KEY UPDATE
biz_review_status = VALUES(biz_review_status),
staging_status = VALUES(staging_status),
imported_song_id = VALUES(imported_song_id),
recalled_candidates = VALUES(recalled_candidates),
error_message = NULL
"""
......@@ -566,11 +573,18 @@ def build_staging_tuple(tuple_row: tuple, action: dict, import_batch_id: str, re
biz_review_status = 'not_required'
staging_status = 'skipped'
imported_song_id = None
elif dedup_action == 'skip':
biz_review_status = 'not_required'
staging_status = 'skipped'
imported_song_id = None
else:
biz_review_status = 'pending'
staging_status = 'staged'
imported_song_id = None
recalled_candidates = action.get('recalled_candidates')
recalled_json = json.dumps(recalled_candidates, ensure_ascii=False) if recalled_candidates else None
return (
ID_GENERATOR.next_id(),
import_batch_id,
......@@ -586,6 +600,7 @@ def build_staging_tuple(tuple_row: tuple, action: dict, import_batch_id: str, re
biz_review_status,
staging_status,
imported_song_id,
recalled_json,
)
......@@ -605,6 +620,8 @@ def process_row_with_oss(args_tuple):
_T2S = opencc.OpenCC('t2s')
_METADATA_PUNCT = set(' \t\n\r，。！？；：、"“”‘’·…—～！￥（）【】《》〈〉「」『』﹏,.:;!?()[]{}<>|/\\_-')
_INSTRUMENTAL_LYRIC_RE = re.compile(r'(?:纯音乐|純音樂|无歌词|無歌詞|没有歌词|沒有歌詞|instrumental)', re.IGNORECASE)
_ORIGINAL_SOUND_RE = re.compile(r'^@\S+的原声$')
_UNKNOWN_AUTHORS = {'不详', '未知', '无', '佚名', 'unknown', 'none', 'n/a', ''}
def _normalize_meta(text: str | None) -> str:
......@@ -616,6 +633,26 @@ def _normalize_meta(text: str | None) -> str:
return ''.join(c for c in text if c not in _METADATA_PUNCT)
def mark_rows_in_l1_index(rows: list[dict], l1_index: dict[tuple[str, str, str], int | str]) -> None:
"""把已接受处理的行预先写入内存 L1 索引，供后续批次去重使用。"""
for row in rows:
key = (
_normalize_meta(row.get('name')),
_normalize_meta(row.get('lyricist')),
_normalize_meta(row.get('composer')),
)
if key[0] and key not in l1_index:
l1_index[key] = row['source_id']
def stop_db_writer(write_queue: queue.Queue, writer_thread: threading.Thread) -> None:
"""等待已入队 DB 写入完成，再停止后台写入线程。"""
write_queue.join()
if writer_thread.is_alive():
write_queue.put(None)
writer_thread.join()
def is_instrumental_lyrics(text: str | None) -> bool:
"""识别纯音乐/无歌词提示文本，这类内容不参与 L2 歌词去重。"""
if not text:
......@@ -678,6 +715,86 @@ def load_source_record_counts(source_conn, source_song_ids: set[str]) -> dict[st
return counts
def _decode_lyric_content(content: bytes) -> str:
try:
return content.decode('utf-8')
except UnicodeDecodeError:
return content.decode('utf-8', errors='replace')
def load_existing_l2_candidates(
rows: list[dict],
cache_path: Path,
downloader=download_file,
) -> tuple[list[LyricRecord], dict[str, int]]:
"""从目标库 lyrics_url 行构建 L2 候选；本地缓存已下载歌词，避免重启全量拉 OSS。"""
cached_by_key: dict[tuple[str, str], dict] = {}
if cache_path.exists():
with cache_path.open('r', encoding='utf-8') as f:
for line in f:
line = line.strip()
if not line:
continue
try:
item = json.loads(line)
except json.JSONDecodeError:
continue
record_id = str(item.get('id') or '')
url = str(item.get('url') or '')
lyrics = item.get('lyrics')
if record_id and url and isinstance(lyrics, str) and lyrics.strip():
cached_by_key[(record_id, url)] = item
records: list[LyricRecord] = []
next_cache: dict[tuple[str, str], dict] = {}
stats = {'cached': 0, 'downloaded': 0, 'failed': 0}
for row in rows:
url = row.get('lyrics_url')
if not url or not str(url).startswith(('http://', 'https://')):
continue
record_id = str(row['id'])
key = (record_id, str(url))
item = cached_by_key.get(key)
if item:
stats['cached'] += 1
else:
content, _ = downloader(str(url), timeout=10)
if not content:
stats['failed'] += 1
continue
item = {
'id': record_id,
'url': str(url),
'lyrics': _decode_lyric_content(content),
'name': row.get('name'),
'singer': row.get('singer'),
'lyricist': row.get('lyricist'),
'composer': row.get('composer'),
}
stats['downloaded'] += 1
next_cache[key] = item
records.append(LyricRecord(
record_id=record_id,
lyrics=item['lyrics'],
title=row.get('name') or item.get('name'),
artist=row.get('singer') or item.get('singer'),
lyricist=row.get('lyricist') or item.get('lyricist'),
composer=row.get('composer') or item.get('composer'),
))
cache_path.parent.mkdir(parents=True, exist_ok=True)
tmp_path = cache_path.with_suffix(cache_path.suffix + '.tmp')
with tmp_path.open('w', encoding='utf-8') as f:
for item in next_cache.values():
f.write(json.dumps(item, ensure_ascii=False) + '\n')
tmp_path.replace(cache_path)
return records, stats
def check_l1(row: dict, l1_index: dict) -> tuple[bool, int | None]:
"""L1 元数据去重，返回 (is_duplicate, matched_id)"""
key = (
......@@ -760,13 +877,16 @@ def check_l2(
row: dict,
checker: DuplicateChecker,
candidates: list[LyricRecord] | L2CandidateIndex,
) -> tuple[str, float, str | None, str]:
"""L2 歌词内容去重，返回 (decision, confidence, matched_id, reason)"""
) -> tuple[str, float, str | None, str, list[dict]]:
"""L2 歌词内容去重，返回 (decision, confidence, matched_id, reason, recalled_candidates)。
recalled_candidates: top-K 召回候选信息列表，即使最终判定为 new 也返回。
"""
lyrics_text = row.get('lyrics_txt_content')
if not lyrics_text or not lyrics_text.strip():
return 'new', 1.0, None, '无歌词内容，跳过 L2'
return 'new', 1.0, None, '无歌词内容，跳过 L2', []
if is_instrumental_lyrics(lyrics_text):
return 'new', 1.0, None, '歌词内容包含纯音乐/无歌词提示，跳过 L2'
return 'new', 1.0, None, '歌词内容包含纯音乐/无歌词提示，跳过 L2', []
record_id = row.get('source_id', row.get('id'))
record = LyricRecord(
......@@ -778,18 +898,37 @@ def check_l2(
composer=row.get('composer'),
)
if isinstance(candidates, L2CandidateIndex):
candidates = candidates.recall(record)
if not candidates:
return 'new', 1.0, None, '无候选集'
recalled_records = candidates.recall(record)
else:
recalled_records = candidates
if not recalled_records:
return 'new', 1.0, None, '无候选集', []
result = checker.check_record_against_candidates(record, recalled_records, max_candidates=5)
# 构建召回候选信息（保留 top 10，即使判定为 new）
recalled_candidates: list[dict] = []
candidate_record_map: dict[str, LyricRecord] = {r.record_id: r for r in recalled_records}
for cm in result.candidates[:10]:
cr = candidate_record_map.get(cm.record_id)
recalled_candidates.append({
'id': cm.record_id,
'name': cr.title if cr else None,
'lyricist': cr.lyricist if cr else None,
'composer': cr.composer if cr else None,
'decision': cm.decision.value,
'confidence': round(cm.confidence, 4),
'jaccard': round(cm.jaccard, 4),
'line_coverage': round(cm.line_coverage, 4),
})
result = checker.check_record_against_candidates(record, candidates, max_candidates=5)
# new 记录不返回 matched_id，避免前端误认为有命中
if result.decision == DuplicateDecision.NEW:
matched_id = None
else:
matched_id = result.candidates[0].record_id if result.candidates else None
return result.decision.value, result.confidence, matched_id, result.reason
return result.decision.value, result.confidence, matched_id, result.reason, recalled_candidates
def _candidate_by_id(
......@@ -817,6 +956,61 @@ def _writers_differ(row: dict, candidate: LyricRecord | None) -> bool:
)
_AUTHOR_SEP_RE = re.compile(r'[,，/;；、]')
_AUTHOR_MERGE_SQL = 'UPDATE `{table}` SET {{col}} = %s WHERE id = %s'.format(table=TARGET_TABLE_NAME)
def _merge_author_field(existing: str | None, new: str | None) -> str:
"""增量合并作者字段：保留已有作者，追加新作者（去重）。"""
if not new:
return existing or ''
if not existing:
return new
def _split_authors(text: str) -> list[str]:
return [a.strip() for a in _AUTHOR_SEP_RE.split(text) if a.strip()]
existing_list = _split_authors(existing)
new_list = _split_authors(new)
existing_norm = {_normalize_meta(a) for a in existing_list}
result = list(existing_list)
for author in new_list:
if _normalize_meta(author) not in existing_norm:
result.append(author)
existing_norm.add(_normalize_meta(author))
merged = '、'.join(result)
return merged if merged != existing else ''
def execute_author_merges(cursor, author_merge_tasks: list[dict]) -> int:
"""执行作者字段增量合并 UPDATE，返回成功合并的记录数。"""
merged_count = 0
for task in author_merge_tasks:
try:
if task['lyricist']:
cursor.execute(
_AUTHOR_MERGE_SQL.format(col='lyricist'),
(task['lyricist'], task['matched_id']),
)
if task['composer']:
cursor.execute(
_AUTHOR_MERGE_SQL.format(col='composer'),
(task['composer'], task['matched_id']),
)
merged_count += 1
except Exception as e:
logger.error(
f"作者字段合并失败 matched_id={task['matched_id']}: {e}"
)
return merged_count
def _is_short_exact_l2_review(decision: str, reason: str) -> bool:
return decision == 'review' and '规范化后的原文哈希一致' in reason and '有效歌词过短' in reason
def classify_dedup_action(
row: dict,
l1_index: dict,
......@@ -830,8 +1024,48 @@ def classify_dedup_action(
当判定为 duplicate 时，根据源库关联录音数量决定合并方向：
录音少的合并到录音多的。
"""
l1_matched, l1_matched_id = check_l1(row, l1_index)
# 规则1：歌词为空 + 词曲作者不详 → 不导
lyrics_text = row.get('lyrics_txt_content') or ''
no_lyrics = not lyrics_text or not lyrics_text.strip()
lyricist_raw = (row.get('lyricist') or '').strip()
composer_raw = (row.get('composer') or '').strip()
if no_lyrics and lyricist_raw.lower() in _UNKNOWN_AUTHORS and composer_raw.lower() in _UNKNOWN_AUTHORS:
return {
'action': 'skip',
'decision': 'skip',
'confidence': 1.0,
'matched_id': None,
'reason': '歌词为空且词曲作者不详，不导入',
'l1_matched': False,
'l1_matched_id': None,
'merge_authors': False,
'matched_candidate': None,
'merge_direction': None,
'new_record_count': row.get('record_count'),
'existing_record_count': None,
'recalled_candidates': [],
}
# 规则2：歌名形如 "@XXX的原声" → 不导
song_name = (row.get('name') or '').strip()
if _ORIGINAL_SOUND_RE.match(song_name):
return {
'action': 'skip',
'decision': 'skip',
'confidence': 1.0,
'matched_id': None,
'reason': f'歌名「{song_name}」为原声类内容，不导入',
'l1_matched': False,
'l1_matched_id': None,
'merge_authors': False,
'matched_candidate': None,
'merge_direction': None,
'new_record_count': row.get('record_count'),
'existing_record_count': None,
'recalled_candidates': [],
}
l1_matched, l1_matched_id = check_l1(row, l1_index)
l2_candidates_for_check: list[LyricRecord] | L2CandidateIndex = candidates
if isinstance(candidates, L2CandidateIndex):
record = LyricRecord(
......@@ -849,7 +1083,57 @@ def classify_dedup_action(
recalled.append(l1_candidate)
l2_candidates_for_check = recalled
decision, confidence, matched_id, reason = check_l2(row, checker, l2_candidates_for_check)
decision, confidence, matched_id, reason, recalled_candidates = check_l2(row, checker, l2_candidates_for_check)
if _is_short_exact_l2_review(decision, reason):
if not l1_matched:
return {
'action': 'new',
'decision': 'new',
'confidence': 1.0,
'matched_id': None,
'reason': f'{reason}；L1 元数据未命中，按新歌处理',
'l1_matched': l1_matched,
'l1_matched_id': l1_matched_id,
'merge_authors': False,
'matched_candidate': None,
'merge_direction': None,
'new_record_count': row.get('record_count'),
'existing_record_count': None,
'recalled_candidates': recalled_candidates,
}
matched_id = str(l1_matched_id)
matched_candidate = _candidate_by_id(l2_candidates_for_check, matched_id)
new_count = (row.get('record_count') or 0) if source_record_counts is not None else None
existing_count = None
if source_record_counts is not None and l1_matched_id is not None:
try:
existing_sid = _target_id_to_source_sid.get(int(l1_matched_id))
if existing_sid:
existing_count = source_record_counts.get(existing_sid, 0)
except (ValueError, TypeError):
pass
merge_direction = 'new_into_existing'
if new_count is not None and existing_count is not None and new_count > existing_count:
merge_direction = 'existing_into_new'
return {
'action': 'merge',
'decision': 'duplicate',
'confidence': 1.0,
'matched_id': matched_id,
'reason': f'{reason}；L1 元数据命中，按 L1 判定重复',
'l1_matched': l1_matched,
'l1_matched_id': l1_matched_id,
'merge_authors': _writers_differ(row, matched_candidate),
'matched_candidate': matched_candidate,
'merge_direction': merge_direction,
'new_record_count': new_count,
'existing_record_count': existing_count,
'recalled_candidates': recalled_candidates,
}
if decision == 'duplicate':
matched_candidate = _candidate_by_id(l2_candidates_for_check, matched_id)
......@@ -866,6 +1150,27 @@ def classify_dedup_action(
except (ValueError, TypeError):
pass
# 规则3：歌词相似但歌名与词曲作者均不一致 → 洗盗蹭嫌疑，强制 review
if matched_candidate:
_name_match = _normalize_meta(row.get('name')) == _normalize_meta(matched_candidate.title)
_authors_match = not _writers_differ(row, matched_candidate)
if not _name_match and not _authors_match:
return {
'action': 'review',
'decision': 'review',
'confidence': confidence,
'matched_id': matched_id,
'reason': f'歌词相似但歌名与词曲作者均不一致，有洗盗蹭嫌疑（{reason}）',
'l1_matched': l1_matched,
'l1_matched_id': l1_matched_id,
'merge_authors': True,
'matched_candidate': matched_candidate,
'merge_direction': None,
'new_record_count': new_count,
'existing_record_count': existing_count,
'recalled_candidates': recalled_candidates,
}
# 歌词质量兜底：现有记录无歌词但新记录有歌词 → 强制人工复核
existing_has_lyrics = matched_candidate and bool(matched_candidate.lyrics and matched_candidate.lyrics.strip())
new_has_lyrics = bool(lyrics_text and lyrics_text.strip())
......@@ -883,6 +1188,7 @@ def classify_dedup_action(
'merge_direction': 'existing_into_new',
'new_record_count': new_count,
'existing_record_count': existing_count,
'recalled_candidates': recalled_candidates,
}
merge_direction = 'new_into_existing'
......@@ -903,6 +1209,7 @@ def classify_dedup_action(
'merge_direction': merge_direction,
'new_record_count': new_count,
'existing_record_count': existing_count,
'recalled_candidates': recalled_candidates,
}
# 空歌词 / 歌词获取失败：L2 无法比对，必须人工复核，不能自动入库
......@@ -921,6 +1228,7 @@ def classify_dedup_action(
'merge_direction': None,
'new_record_count': row.get('record_count'),
'existing_record_count': None,
'recalled_candidates': recalled_candidates,
}
if decision == 'review' or (l1_matched and reason == '无候选集'):
......@@ -939,6 +1247,7 @@ def classify_dedup_action(
'merge_direction': None,
'new_record_count': row.get('record_count'),
'existing_record_count': None,
'recalled_candidates': recalled_candidates,
}
return {
......@@ -954,6 +1263,7 @@ def classify_dedup_action(
'merge_direction': None,
'new_record_count': row.get('record_count'),
'existing_record_count': None,
'recalled_candidates': recalled_candidates,
}
......@@ -1096,6 +1406,8 @@ def main():
help='启动时从目标库下载已有歌词文本构建 L2 候选集（默认关闭）')
parser.add_argument('--dedup-threshold', type=float, default=0.78,
help='L2 去重 Jaccard 阈值（默认 0.78）')
parser.add_argument('--dedup-min-primary-chars', type=int, default=40,
help='L2 精确哈希自动合并所需的最小规范化原文歌词长度（默认 40，过短转人工复核）')
parser.add_argument('--l2-recall', choices=['topk', 'full'], default='topk',
help='L2 候选召回模式：topk=倒排索引召回，full=全量候选精排（默认 topk）')
parser.add_argument('--l2-recall-top-k', type=int, default=200,
......@@ -1112,6 +1424,7 @@ def main():
logger.info(f"参数: limit={args.limit}, offset={args.offset}, skip_oss={args.skip_oss}, "
f"batch_size={args.batch_size}, workers={args.workers}, "
f"skip_dedup={args.skip_dedup}, dedup_threshold={args.dedup_threshold}, "
f"dedup_min_primary_chars={args.dedup_min_primary_chars}, "
f"l2_recall={args.l2_recall}, l2_recall_top_k={args.l2_recall_top_k}, "
f"dedup_only={args.dedup_only}, review_result_csv={args.review_result_csv}, "
f"dry_run={args.dry_run}")
......@@ -1137,6 +1450,25 @@ def main():
staging_insert_sql = STAGING_INSERT_SQL_TEMPLATE.format(table=TARGET_TABLE_NAME_TMP)
import_batch_id = datetime.now().strftime("%Y%m%d_%H%M%S")
# 自动迁移：确保暂存表存在 recalled_candidates 列
try:
with target_conn.cursor() as cursor:
cursor.execute(
f"SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS "
f"WHERE TABLE_SCHEMA = %s AND TABLE_NAME = %s AND COLUMN_NAME = 'recalled_candidates'",
(TARGET_DB_CONFIG['database'], TARGET_TABLE_NAME_TMP),
)
if not cursor.fetchone():
cursor.execute(
f"ALTER TABLE `{TARGET_TABLE_NAME_TMP}` "
f"ADD COLUMN `recalled_candidates` text DEFAULT NULL COMMENT 'L2 召回候选 JSON' "
f"AFTER `imported_song_id`"
)
target_conn.commit()
logger.info(f"暂存表 {TARGET_TABLE_NAME_TMP} 新增 recalled_candidates 列")
except Exception as e:
logger.warning(f"暂存表 recalled_candidates 列迁移跳过: {e}")
# ===== 加载已处理的 source_song_id 集合（增量去重）=====
# 包括：1) 已写入目标表的 2) 已在暂存表中标记过 dedup_action 的（review/merge/new）
imported_ids: set[str] = set()
......@@ -1244,6 +1576,7 @@ def main():
logger.info(f"初始化 L2 歌词去重检查器 (阈值={args.dedup_threshold})")
l2_checker = DuplicateChecker(
duplicate_jaccard_threshold=args.dedup_threshold,
exact_duplicate_min_primary_chars=args.dedup_min_primary_chars,
)
l2_candidates = L2CandidateIndex(
l2_checker,
......@@ -1259,24 +1592,16 @@ def main():
lyric_sql = f"SELECT id, name, singer, lyricist, composer, lyrics_url FROM {TARGET_TABLE_NAME} WHERE deleted = '0' AND lyrics_url IS NOT NULL"
with target_conn.cursor(pymysql.cursors.DictCursor) as cursor:
cursor.execute(lyric_sql)
for r in cursor.fetchall():
url = r.get('lyrics_url')
if url and url.startswith(('http://', 'https://')):
content, _ = download_file(url, timeout=10)
if content:
try:
text = content.decode('utf-8')
except UnicodeDecodeError:
text = content.decode('utf-8', errors='replace')
l2_candidates.add(LyricRecord(
record_id=str(r['id']),
lyrics=text,
title=r.get('name'),
artist=r.get('singer'),
lyricist=r.get('lyricist'),
composer=r.get('composer'),
))
logger.info(f"L2 候选集加载完成: {len(l2_candidates)} 条")
existing_lyric_rows = cursor.fetchall()
cache_path = CACHE_DIR / f'{TARGET_TABLE_NAME}_existing_lyrics.jsonl'
existing_records, cache_stats = load_existing_l2_candidates(existing_lyric_rows, cache_path)
for record in existing_records:
l2_candidates.add(record)
logger.info(
f"L2 候选集加载完成: {len(l2_candidates)} 条 | "
f"缓存命中={cache_stats['cached']} 下载={cache_stats['downloaded']} 失败={cache_stats['failed']} | "
f"缓存文件={cache_path}"
)
except Exception as e:
logger.error(f"加载已有歌词失败: {e}")
......@@ -1336,7 +1661,8 @@ def main():
dedup_report.close()
logger.info(
f"去重审核清单完成: {review_report.filepath} | "
f"new={decision_counts['new']} review={decision_counts['review']} merge={decision_counts['merge']}"
f"new={decision_counts['new']} review={decision_counts['review']} "
f"merge={decision_counts['merge']} skip={decision_counts['skip']}"
)
return
......@@ -1345,6 +1671,7 @@ def main():
stats = {
'inserted': 0, 'errors': 0,
'l1_dup': 0, 'l2_dup': 0, 'l2_review': 0, 'l2_new': 0,
'author_merged': 0, 'skipped': 0,
}
start_time = time.time()
......@@ -1355,16 +1682,84 @@ def main():
pbar_db = tqdm(total=total, desc="DB 写入", unit="条", position=1, leave=True,
bar_format='{l_bar}{bar}| {n_fmt}/{total_fmt} [{elapsed}<{remaining}, {rate_fmt}]')
# ===== 后台 DB 写入线程：与下一批次的 OSS 处理并行 =====
db_write_queue: queue.Queue = queue.Queue(maxsize=2)
db_write_errors: list[str] = []
def _background_db_writer():
"""后台线程：顺序写入 DB，与主线程 OSS 处理并行。"""
while True:
item = db_write_queue.get()
if item is None:
db_write_queue.task_done()
break
batch_data, staging_data, batch_rows, batch_size_actual, batch_start, author_merge_tasks = item
try:
with target_conn.cursor() as cursor:
if batch_data:
cursor.executemany(insert_sql, batch_data)
if staging_data:
cursor.executemany(staging_insert_sql, staging_data)
if author_merge_tasks:
merged_cnt = execute_author_merges(cursor, author_merge_tasks)
with stats_lock:
stats['author_merged'] += merged_cnt
target_conn.commit()
batch_inserted = len(batch_data)
with stats_lock:
stats['inserted'] += batch_inserted
except Exception as e:
logger.error(f"批量写入失败 (offset {batch_start}): {e}")
target_conn.rollback()
db_write_errors.append(f"batch@{batch_start}: {e}")
# 降级逐条写入
for row_i, tuple_row in enumerate(batch_data):
try:
with target_conn.cursor() as cursor:
cursor.execute(insert_sql, tuple_row)
target_conn.commit()
with stats_lock:
stats['inserted'] += 1
except Exception as e2:
with stats_lock:
stats['errors'] += 1
logger.error(f"单条写入失败 (batch offset {batch_start + row_i}): {e2}")
# 降级模式下也尝试执行作者合并
if author_merge_tasks:
try:
with target_conn.cursor() as cursor:
merged_cnt = execute_author_merges(cursor, author_merge_tasks)
target_conn.commit()
with stats_lock:
stats['author_merged'] += merged_cnt
except Exception as e_merge:
logger.error(f"降级作者合并失败 (offset {batch_start}): {e_merge}")
pbar_db.update(batch_size_actual)
with stats_lock:
pbar_db.set_postfix(
ins=stats['inserted'],
err=stats['errors'],
l1=stats['l1_dup'],
l2=stats['l2_dup'],
rev=stats['l2_review'],
auth=stats['author_merged'],
skip=stats['skipped'],
refresh=False
)
db_write_queue.task_done()
db_writer_thread = threading.Thread(target=_background_db_writer, daemon=True)
db_writer_thread.start()
for batch_start in range(0, total, args.batch_size):
batch_rows = rows[batch_start:batch_start + args.batch_size]
batch_size_actual = len(batch_rows)
# ===== 步骤1：去重分类（纯内存，先于 OSS）=====
# L1/L2 去重只依赖源库查询结果（name/lyricist/composer/lyrics_txt_content），
# 不需要任何 URL 下载，因此可以在 OSS 之前完成，避免对 merge/review 记录做无效上传。
# ===== 步骤1：去重分类（纯内存）=====
staging_data = []
new_rows: list = []
row_actions: dict = {}
author_merge_tasks: list[dict] = []
if not args.skip_dedup and not args.review_result_csv:
for row in batch_rows:
......@@ -1393,6 +1788,29 @@ def main():
if review_report:
review_report.write(row, action)
# 收集作者增量合并任务
if action.get('merge_authors') and action.get('matched_id'):
matched_cand = action.get('matched_candidate')
merged_lyricist = _merge_author_field(
getattr(matched_cand, 'lyricist', None) if matched_cand else None,
row.get('lyricist'),
)
merged_composer = _merge_author_field(
getattr(matched_cand, 'composer', None) if matched_cand else None,
row.get('composer'),
)
if merged_lyricist or merged_composer:
author_merge_tasks.append({
'matched_id': int(action['matched_id']),
'lyricist': merged_lyricist,
'composer': merged_composer,
'source_id': record_id,
})
logger.debug(
f"作者合并任务: matched_id={action['matched_id']}, "
f"lyricist→{merged_lyricist!r}, composer→{merged_composer!r}"
)
elif action['action'] == 'review':
with stats_lock:
stats['l2_review'] += 1
......@@ -1402,6 +1820,13 @@ def main():
if review_report:
review_report.write(row, action)
elif action['action'] == 'skip':
with stats_lock:
stats['skipped'] += 1
dedup_report.write(record_id, record_name, 'PRE', 'skip',
confidence='1.0000',
reason=reason)
else: # new
with stats_lock:
stats['l2_new'] += 1
......@@ -1413,7 +1838,6 @@ def main():
review_report.write(row, action)
new_rows.append(row)
# 加入 L2 候选集供后续批次比对
lyrics_text = row.get('lyrics_txt_content')
if lyrics_text and lyrics_text.strip():
l2_candidates.add(LyricRecord(
......@@ -1425,7 +1849,6 @@ def main():
composer=row.get('composer'),
))
else:
# skip_dedup 或 review_result_csv 模式：全部视为 new
new_rows = list(batch_rows)
# ===== 步骤2：OSS 并发处理（只对 new 记录）=====
......@@ -1445,15 +1868,13 @@ def main():
results_map[idx] = tuple_row
pbar_oss.update(1)
# merge/review 记录跳过 OSS，直接推进进度条
pbar_oss.update(batch_size_actual - len(new_rows))
# 按顺序组装 batch_data（写入 hk_songs 的 new 记录）
# 按顺序组装 batch_data
batch_data = [results_map[i] for i in range(len(new_rows)) if i in results_map]
# ===== 步骤3：构建 staging tuples =====
if not args.skip_dedup and not args.review_result_csv:
# new 记录：使用 OSS 处理后的 tuple_row
for i, row in enumerate(new_rows):
if i not in results_map:
continue
......@@ -1465,10 +1886,9 @@ def main():
staging_data.append(build_staging_tuple(results_map[i], action, import_batch_id,
record_count=row.get('record_count')))
# merge/review 记录：透传源 URL，跳过 OSS
for row in batch_rows:
action = row_actions.get(row['source_id'])
if action and action['action'] in ('merge', 'review'):
if action and action['action'] in ('merge', 'review', 'skip'):
tuple_row = build_row_tuple(row, None, skip_oss=True)
staging_data.append(build_staging_tuple(tuple_row, action, import_batch_id,
record_count=row.get('record_count')))
......@@ -1477,56 +1897,13 @@ def main():
pbar_db.update(batch_size_actual)
continue
# ===== 步骤4：顺序写入数据库（单连接，保证不出错）=====
try:
with target_conn.cursor() as cursor:
if batch_data:
cursor.executemany(insert_sql, batch_data)
if staging_data:
cursor.executemany(staging_insert_sql, staging_data)
target_conn.commit()
batch_inserted = len(batch_data)
with stats_lock:
stats['inserted'] += batch_inserted
# 更新 L1 索引（防止批次内重复）
# ===== 步骤4：交给后台线程写入 DB（主线程继续下一批次 OSS）=====
if not args.skip_dedup:
for i, row in enumerate(batch_rows):
key = (
_normalize_meta(row.get('name')),
_normalize_meta(row.get('lyricist')),
_normalize_meta(row.get('composer')),
)
if key[0] and key not in l1_index:
l1_index[key] = row['source_id']
except Exception as e:
logger.error(f"批量写入失败 (offset {batch_start}): {e}")
target_conn.rollback()
# 降级逐条写入，保证其他行不受影响
for row_i, tuple_row in enumerate(batch_data):
try:
with target_conn.cursor() as cursor:
cursor.execute(insert_sql, tuple_row)
target_conn.commit()
with stats_lock:
stats['inserted'] += 1
except Exception as e2:
with stats_lock:
stats['errors'] += 1
logger.error(f"单条写入失败 (batch offset {batch_start + row_i}): {e2}")
mark_rows_in_l1_index(batch_rows, l1_index)
db_write_queue.put((batch_data, staging_data, batch_rows, batch_size_actual, batch_start, author_merge_tasks))
pbar_db.update(batch_size_actual)
# 更新进度条描述（附加统计）
with stats_lock:
pbar_db.set_postfix(
ins=stats['inserted'],
err=stats['errors'],
l1=stats['l1_dup'],
l2=stats['l2_dup'],
rev=stats['l2_review'],
refresh=False
)
# 所有批次已入队；等待后台 DB 写入收尾后停止线程
stop_db_writer(db_write_queue, db_writer_thread)
pbar_oss.close()
pbar_db.close()
......@@ -1540,7 +1917,8 @@ def main():
if not args.skip_dedup:
logger.info(f"去重统计: L1跳过={stats['l1_dup']} | "
f"L2合并命中={stats['l2_dup']} | L2待审={stats['l2_review']} | "
f"L2新词={stats['l2_new']}")
f"L2新词={stats['l2_new']} | 作者合并={stats['author_merged']} | "
f"跳过={stats['skipped']}")
if dedup_report:
logger.info(f"去重报告: {dedup_report.filepath}")
if review_report:
......@@ -1548,6 +1926,17 @@ def main():
logger.info(f"耗时: {elapsed_total:.1f}s | 平均速度: {total/elapsed_total:.1f} 条/秒")
logger.info("=" * 60)
except KeyboardInterrupt:
logger.warning("收到中断信号，正在等待已入队 DB 写入完成...")
if 'db_write_queue' in locals() and 'db_writer_thread' in locals():
stop_db_writer(db_write_queue, db_writer_thread)
if 'pbar_oss' in locals():
pbar_oss.close()
if 'pbar_db' in locals():
pbar_db.close()
logger.warning("已停止导入；未入队的 OSS 结果不会写入 DB，可安全重跑继续。")
raise
finally:
# 显式关闭 OSS session，避免 __del__ 中的 Bad file descriptor / ConnectionPool 警告
if bucket is not None:
......
......@@ -460,6 +460,13 @@
<select id="runSelect"></select>
<label id="topKLabel" for="topKSelect">topK</label>
<select id="topKSelect"></select>
<label for="decisionFilter">决策</label>
<select id="decisionFilter">
<option value="">全部</option>
<option value="new">new</option>
<option value="merge">merge</option>
<option value="review">review</option>
</select>
<input id="searchInput" type="search" placeholder="搜索歌名 / ID">
<label for="pageSizeSelect">每页</label>
<select id="pageSizeSelect">
......@@ -520,6 +527,7 @@
pageSize: 50,
mode: 'retrieval',
topK: '',
decisionFilter: '',
queryId: '',
candidateId: '',
lyricsCache: new Map()
......@@ -529,6 +537,7 @@
runSelect: document.getElementById('runSelect'),
topKSelect: document.getElementById('topKSelect'),
topKLabel: document.getElementById('topKLabel'),
decisionFilter: document.getElementById('decisionFilter'),
searchInput: document.getElementById('searchInput'),
pageSizeSelect: document.getElementById('pageSizeSelect'),
exportBtn: document.getElementById('exportBtn'),
......@@ -576,7 +585,7 @@
}
function rowDecision(row) {
return row.decision || row.candidate_decision || '';
return row.candidate_decision || row.decision || row.action || '';
}
function isConflict(row) {
......@@ -706,7 +715,8 @@
function selectedCandidate(rows) {
if (!rows.length) return null;
if (!state.candidateId || !rows.some(row => row.candidate_id === state.candidateId)) {
state.candidateId = rows[0].candidate_id || '';
const firstReal = rows.find(row => row.candidate_id);
state.candidateId = firstReal ? firstReal.candidate_id : (rows[0].candidate_id || '');
}
return rows.find(row => row.candidate_id === state.candidateId) || rows[0];
}
......@@ -726,6 +736,7 @@
const conflict = isConflict(row);
const l1 = l1Match(row);
const extraClass = `${conflict ? ' conflict' : ''}${l1 ? ' l1hint' : ''}`;
const l2Decision = row.candidate_decision || '';
return `<div class="candidate ${esc(decision)}${active}${extraClass}" data-candidate-id="${esc(row.candidate_id)}">
<div class="candidate-head">
<span class="rank">${esc(row.rank || '')}</span>
......@@ -737,6 +748,7 @@
<div class="query-meta">词 ${esc(row.candidate_lyricist || '-')} · 曲 ${esc(row.candidate_composer || '-')}</div>
<div style="margin-top:7px;display:flex;gap:5px;flex-wrap:wrap">
${l1 ? '<span class="pill l1">L1 元数据命中</span>' : '<span class="pill">L1 未命中</span>'}
${l2Decision ? `<span class="pill ${esc(l2Decision)}">L2 ${esc(l2Decision)}</span>` : ''}
${conflict ? '<span class="pill conflict">冲突</span>' : ''}
</div>
<div class="score-row">
......@@ -764,6 +776,7 @@
approved_import: '确认不重复（已入库）',
rejected_duplicate: '确认重复',
unsure: '待确认',
deleted: '已删除',
};
async function renderDetail() {
......@@ -824,7 +837,7 @@
<div class="section-head">
<h2 class="section-title">人工标注</h2>
${query.review_status && query.review_status !== 'pending' && query.review_status !== 'not_required'
? `<span class="pill ${query.review_status === 'approved_import' ? 'new' : query.review_status === 'rejected_duplicate' ? 'duplicate' : ''}">已审核</span>`
? `<span class="pill ${query.review_status === 'approved_import' ? 'new' : query.review_status === 'rejected_duplicate' ? 'duplicate' : query.review_status === 'deleted' ? 'duplicate' : ''}">${query.review_status === 'deleted' ? '已删除' : '已审核'}</span>`
: ''}
</div>
<div class="section-body">
......@@ -846,14 +859,15 @@
</button>`).join('')}
<input id="reviewNote" value="${esc(selectedReview.note || '')}" placeholder="人工备注">
<button id="saveReviewBtn" class="secondary">保存</button>
<button id="deleteReviewBtn" class="secondary" style="color:#c0392b;border-color:#c0392b">删除</button>
</div>
`}
</div>
</div>
<div class="section">
<div class="section-head"><h2 class="section-title">Top10 候选</h2></div>
<div class="section-body"><div class="candidates">${rows.map(candidateCard).join('') || '<div class="empty">无召回候选</div>'}</div></div>
<div class="section-head"><h2 class="section-title">Top5 候选</h2></div>
<div class="section-body"><div class="candidates">${rows.filter(r => r.candidate_id).map(candidateCard).join('') || '<div class="empty">无召回候选</div>'}</div></div>
</div>
<div class="section">
......@@ -900,6 +914,21 @@
renderDetail();
});
}
const deleteBtn = document.getElementById('deleteReviewBtn');
if (deleteBtn) {
deleteBtn.addEventListener('click', async () => {
if (!confirm('确认删除此样本？删除后将不再导入。')) return;
deleteBtn.disabled = true;
try {
await updateStagingReview(query.query_source_id, 'deleted',
document.getElementById('reviewNote')?.value || '');
await loadQueryRows(query.query_source_id);
} catch (e) {
alert(`删除失败: ${e.message}`);
}
renderDetail();
});
}
const undoBtn = document.getElementById('undoReviewBtn');
if (undoBtn) {
undoBtn.addEventListener('click', async () => {
......@@ -1024,6 +1053,7 @@
path: dataPath(),
top_k: state.topK,
mode: state.mode,
decision: state.decisionFilter,
q: els.searchInput.value.trim(),
page: String(state.page),
page_size: String(state.pageSize)
......@@ -1100,6 +1130,14 @@
await loadGroups();
renderAll();
});
els.decisionFilter.addEventListener('change', async () => {
state.decisionFilter = els.decisionFilter.value;
state.page = 1;
state.queryId = '';
state.candidateId = '';
await loadGroups();
renderAll();
});
els.reloadBtn.addEventListener('click', loadRuns);
els.exportBtn.addEventListener('click', exportAnnotations);
els.importReviewedBtn.addEventListener('click', importReviewed);
......
......@@ -91,6 +91,7 @@ class DuplicateChecker:
metadata_duplicate_line_coverage_threshold: float = 0.65,
metadata_duplicate_min_primary_lines: int = 8,
auto_duplicate_min_primary_lines: int = 4,
exact_duplicate_min_primary_chars: int = 40,
) -> None:
self.duplicate_jaccard_threshold = duplicate_jaccard_threshold
self.duplicate_line_coverage_threshold = duplicate_line_coverage_threshold
......@@ -111,6 +112,7 @@ class DuplicateChecker:
self.metadata_duplicate_line_coverage_threshold = metadata_duplicate_line_coverage_threshold
self.metadata_duplicate_min_primary_lines = metadata_duplicate_min_primary_lines
self.auto_duplicate_min_primary_lines = auto_duplicate_min_primary_lines
self.exact_duplicate_min_primary_chars = exact_duplicate_min_primary_chars
def check_record_against_candidates(
self,
......@@ -209,6 +211,14 @@ class DuplicateChecker:
decision = DuplicateDecision.REVIEW
confidence = 0.95
reason = "原文哈希一致，但疑似整段翻译结构拆分置信度较低，需要人工复核"
elif not _has_enough_primary_text_chars(
query.normalized,
candidate.normalized,
min_chars=self.exact_duplicate_min_primary_chars,
):
decision = DuplicateDecision.REVIEW
confidence = 0.95
reason = "规范化后的原文哈希一致，但有效歌词过短，需要人工复核"
elif _is_generic_primary_lyrics(query.normalized) or _is_generic_primary_lyrics(candidate.normalized):
decision = DuplicateDecision.REVIEW
confidence = 0.95
......@@ -510,6 +520,21 @@ def _has_enough_primary_lyrics(
return len(meaningful) >= min_lines
def _has_enough_primary_text_chars(
left: NormalizedLyrics,
right: NormalizedLyrics,
*,
min_chars: int,
) -> bool:
left_chars = _normalized_primary_text_length(left)
right_chars = _normalized_primary_text_length(right)
return min(left_chars, right_chars) >= min_chars
def _normalized_primary_text_length(normalized: NormalizedLyrics) -> int:
return len(_normalize_meta("".join(normalized.primary_lines)))
def _is_metadata_like_line(line: str) -> bool:
normalized = _normalize_meta(line)
metadata_prefixes = (
......
......@@ -164,7 +164,7 @@ def _staging_dashboard_row(row: dict[str, object]) -> dict[str, str]:
}
def _staging_groups_response(mode: str, term: str, page: int, page_size: int) -> dict[str, object]:
def _staging_groups_response(mode: str, term: str, page: int, page_size: int, decision: str = '') -> dict[str, object]:
where_clauses = []
params: list[object] = []
if term:
......@@ -174,6 +174,10 @@ def _staging_groups_response(mode: str, term: str, page: int, page_size: int) ->
# hits 模式下只展示有命中的记录（merge/review）
if mode == "hits":
where_clauses.append("dedup_action IN ('merge', 'review')")
# 决策筛选：new / merge / review
if decision and decision in ('new', 'merge', 'review'):
where_clauses.append("dedup_action = %s")
params.append(decision)
where = "WHERE " + " AND ".join(where_clauses) if where_clauses else ""
# 同一首歌可能因多次导入批次产生多行，按 source_song_id 去重取最新一行
count_sql = f"SELECT COUNT(DISTINCT source_song_id) AS n FROM {TARGET_TABLE_NAME_TMP} {where}"
......@@ -216,6 +220,7 @@ def _staging_groups_response(mode: str, term: str, page: int, page_size: int) ->
"has_hit": row.get("dedup_action") in {"merge", "review"},
"has_conflict": False,
"has_l1_hint": bool(row.get("l1_matched_id")),
"decision": row.get("dedup_action") or "",
}
for row in rows
]
......@@ -248,6 +253,9 @@ def _staging_query_rows(query_id: str) -> dict[str, object]:
)
matched_row = cursor.fetchone()
dashboard_row = _staging_dashboard_row(row)
has_main_match = bool(matched_song_id)
# 主行 rank：仅 merge/review 有主匹配时显示 "1"，new 样本置空以便召回候选从 1 开始编号
dashboard_row["rank"] = "1" if has_main_match else ""
if matched_row:
dashboard_row["candidate_name"] = str(matched_row.get("name") or "")
dashboard_row["candidate_lyricist"] = str(matched_row.get("lyricist") or "")
......@@ -256,13 +264,98 @@ def _staging_query_rows(query_id: str) -> dict[str, object]:
if cand_lyrics and not cand_lyrics.startswith(("http://", "https://", "/")):
cand_lyrics = f"_raw:{cand_lyrics}"
dashboard_row["candidate_lyrics_path"] = cand_lyrics
return {"rows": [dashboard_row]}
rows = [dashboard_row]
# 解析 recalled_candidates JSON，为每个召回候选生成一张卡片
recalled_json = row.get("recalled_candidates")
if recalled_json:
try:
recalled_list = json.loads(recalled_json) if isinstance(recalled_json, str) else recalled_json
main_candidate_id = str(matched_song_id or "")
# 批量查找召回候选的歌词 URL（目标表 + 暂存表 fallback）
recalled_ids_str = [
str(c.get("id") or "")
for c in recalled_list
if c.get("id") and str(c.get("id")) != main_candidate_id
]
# 转 int 避免 bigint/string 类型不匹配导致 IN 查询返空
recalled_ids_int: list[int] = []
for rid in recalled_ids_str:
try:
recalled_ids_int.append(int(rid))
except (ValueError, TypeError):
pass
lyrics_map: dict[str, dict] = {}
if recalled_ids_int:
placeholders = ",".join(["%s"] * len(recalled_ids_int))
with _target_conn() as tconn, tconn.cursor() as tcursor:
# 目标表查找
tcursor.execute(
f"SELECT id, name, lyricist, composer, lyrics_url"
f" FROM {TARGET_TABLE_NAME} WHERE id IN ({placeholders})",
recalled_ids_int,
)
for mrow in tcursor.fetchall():
lyrics_map[str(mrow.get("id"))] = mrow
# 暂存表 fallback：目标表找不到的 ID 从暂存表补查
missing_ids = [rid for rid in recalled_ids_int if str(rid) not in lyrics_map]
if missing_ids:
ph2 = ",".join(["%s"] * len(missing_ids))
tcursor.execute(
f"SELECT source_song_id, name, lyricist, composer, lyrics_url"
f" FROM {TARGET_TABLE_NAME_TMP}"
f" WHERE source_song_id IN ({ph2})"
f" ORDER BY staging_id DESC",
[str(mid) for mid in missing_ids],
)
for mrow in tcursor.fetchall():
sid = str(mrow.get("source_song_id"))
if sid not in lyrics_map:
lyrics_map[sid] = mrow
rank = 2 if has_main_match else 1
for cand in recalled_list:
cand_id = str(cand.get("id") or "")
# 跳过已经是主匹配候选的（避免重复显示）
if cand_id and cand_id == main_candidate_id:
continue
cand_row = dict(dashboard_row)
cand_row["rank"] = str(rank)
cand_row["candidate_id"] = cand_id
# 优先用目标表/暂存表查到的元数据，回退到 JSON 中的信息
target_info = lyrics_map.get(cand_id)
if target_info:
cand_row["candidate_name"] = str(target_info.get("name") or cand.get("name") or "")
cand_row["candidate_lyricist"] = str(target_info.get("lyricist") or cand.get("lyricist") or "")
cand_row["candidate_composer"] = str(target_info.get("composer") or cand.get("composer") or "")
cand_lyrics = str(target_info.get("lyrics_url") or "")
if cand_lyrics and not cand_lyrics.startswith(("http://", "https://", "/")):
cand_lyrics = f"_raw:{cand_lyrics}"
cand_row["candidate_lyrics_path"] = cand_lyrics
else:
cand_row["candidate_name"] = str(cand.get("name") or "")
cand_row["candidate_lyricist"] = str(cand.get("lyricist") or "")
cand_row["candidate_composer"] = str(cand.get("composer") or "")
cand_row["candidate_lyrics_path"] = ""
cand_row["candidate_decision"] = str(cand.get("decision") or "new")
cand_row["candidate_confidence"] = str(cand.get("confidence") or "")
cand_row["candidate_jaccard"] = str(cand.get("jaccard") or "")
cand_row["candidate_line_coverage"] = str(cand.get("line_coverage") or "")
cand_row["candidate_reason"] = ""
rows.append(cand_row)
rank += 1
except (json.JSONDecodeError, TypeError):
pass
return {"rows": rows}
def _update_staging_review(source_song_id: str, decision: str, note: str, reviewer: str = "dashboard") -> dict[str, object]:
"""更新暂存表的人工审核状态。
decision: 'approved_import' | 'rejected_duplicate' | 'unsure' | 'pending' (撤销)
decision: 'approved_import' | 'rejected_duplicate' | 'unsure' | 'pending' (撤销) | 'deleted'
"""
if decision == "pending":
# 撤销审核
......@@ -275,6 +368,18 @@ def _update_staging_review(source_song_id: str, decision: str, note: str, review
WHERE source_song_id = %s AND dedup_action = 'review'
"""
params = [note, source_song_id]
elif decision == "deleted":
# 删除：不限定 dedup_action，对 review/skip 等各种 action 均可执行
sql = f"""
UPDATE {TARGET_TABLE_NAME_TMP}
SET staging_status = 'deleted',
biz_review_status = 'deleted',
reviewed_by = %s,
reviewed_at = NOW(),
biz_review_note = NULLIF(%s, '')
WHERE source_song_id = %s
"""
params = [reviewer, note, source_song_id]
else:
sql = f"""
UPDATE {TARGET_TABLE_NAME_TMP}
......@@ -502,6 +607,7 @@ def _group_index(path: Path, top_k: str) -> list[dict[str, object]]:
"has_hit": False,
"has_conflict": False,
"has_l1_hint": False,
"decision": _row_decision(row),
}
groups_by_id[query_id] = group
order += 1
......@@ -528,11 +634,14 @@ def _group_index(path: Path, top_k: str) -> list[dict[str, object]]:
return groups
def _groups_response(path: Path, top_k: str, mode: str, term: str, page: int, page_size: int) -> dict[str, object]:
def _groups_response(path: Path, top_k: str, mode: str, term: str, page: int, page_size: int, decision: str = '') -> dict[str, object]:
groups = _group_index(path, top_k)
# hits 模式下只展示有命中的记录（merge/review），排除纯 new
if mode == "hits":
groups = [g for g in groups if g.get("has_hit")]
# 决策筛选
if decision and decision in ('new', 'merge', 'review', 'duplicate'):
groups = [g for g in groups if g.get("decision") == decision or (decision == 'merge' and g.get("decision") == 'merge')]
filtered = [group for group in groups if _matches_term(group, term)]
start = max(0, (page - 1) * page_size)
end = start + page_size
......@@ -638,13 +747,14 @@ class Handler(BaseHTTPRequestHandler):
top_k = params.get("top_k", [""])[0]
mode = params.get("mode", ["retrieval"])[0]
term = _norm(params.get("q", [""])[0])
decision = _norm(params.get("decision", [""])[0])
page = max(1, int(params.get("page", ["1"])[0]))
page_size = min(200, max(10, int(params.get("page_size", ["50"])[0])))
if raw_path == STAGING_DB_PATH:
_json_response(self, _staging_groups_response(mode, term, page, page_size))
_json_response(self, _staging_groups_response(mode, term, page, page_size, decision))
return
path = _safe_path(raw_path)
_json_response(self, _groups_response(path, top_k, mode, term, page, page_size))
_json_response(self, _groups_response(path, top_k, mode, term, page, page_size, decision))
except Exception as exc: # noqa: BLE001 - local diagnostic API
_error(self, str(exc), status=404)
return
......@@ -694,7 +804,7 @@ class Handler(BaseHTTPRequestHandler):
reviewer = str(payload.get("reviewer", "dashboard")).strip() or "dashboard"
if not source_id:
raise ValueError("source_id is required")
if decision not in ("approved_import", "rejected_duplicate", "unsure", "pending"):
if decision not in ("approved_import", "rejected_duplicate", "unsure", "pending", "deleted"):
raise ValueError(f"invalid decision: {decision}")
result = _update_staging_review(source_id, decision, note, reviewer)
_json_response(self, result)
......
......@@ -5,9 +5,12 @@
"""
import csv
import json
import os
import queue
import sys
import tempfile
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
......@@ -28,6 +31,8 @@ from import_hk_songs import (
TARGET_TABLE_NAME,
OSS_CONFIG,
_is_lrc_format,
mark_rows_in_l1_index,
stop_db_writer,
_normalize_meta,
build_row_tuple,
classify_dedup_action,
......@@ -39,6 +44,7 @@ from import_hk_songs import (
DedupReport,
L2CandidateIndex,
load_approved_import_ids,
load_existing_l2_candidates,
)
from lyric_dedup import DuplicateChecker, DuplicateDecision, LyricRecord
from lyric_dedup.checker import CandidateMatch
......@@ -117,6 +123,97 @@ class TestNormalizeMeta:
assert _normalize_meta('你好！？世界') == '你好世界'
class TestPendingL1Index:
"""测试异步 DB 写入时的内存 L1 索引维护"""
def test_marks_rows_before_background_db_writer_finishes(self):
l1_index = {}
rows = [
{'source_id': 101, 'name': '晴天', 'lyricist': '周杰伦', 'composer': '周杰伦'},
{'source_id': 102, 'name': '', 'lyricist': '无标题', 'composer': '匿名'},
]
mark_rows_in_l1_index(rows, l1_index)
key = (_normalize_meta('晴天'), _normalize_meta('周杰伦'), _normalize_meta('周杰伦'))
assert l1_index[key] == 101
assert len(l1_index) == 1
class TestDbWriterShutdown:
"""测试后台 DB 写入线程的优雅停止"""
def test_stop_db_writer_drains_queued_items_before_stopping(self):
write_queue = queue.Queue()
processed = []
def writer():
while True:
item = write_queue.get()
if item is None:
write_queue.task_done()
break
processed.append(item)
write_queue.task_done()
thread = threading.Thread(target=writer)
thread.start()
write_queue.put('batch-1')
write_queue.put('batch-2')
stop_db_writer(write_queue, thread)
assert processed == ['batch-1', 'batch-2']
assert not thread.is_alive()
class TestExistingLyricsCache:
"""测试目标库已有歌词加载缓存"""
def test_load_existing_l2_candidates_reuses_cached_lyrics(self, tmp_path):
cache_path = tmp_path / 'lyrics_cache.jsonl'
cached = {
'id': '100',
'url': 'https://oss.example.test/100.txt',
'lyrics': '缓存里的歌词\n第二行',
'name': '旧歌',
'singer': '旧歌手',
'lyricist': '旧词',
'composer': '旧曲',
}
cache_path.write_text(json.dumps(cached, ensure_ascii=False) + '\n', encoding='utf-8')
rows = [
{
'id': 100,
'lyrics_url': 'https://oss.example.test/100.txt',
'name': '旧歌',
'singer': '旧歌手',
'lyricist': '旧词',
'composer': '旧曲',
},
{
'id': 101,
'lyrics_url': 'https://oss.example.test/101.txt',
'name': '新歌',
'singer': '新歌手',
'lyricist': '新词',
'composer': '新曲',
},
]
downloaded_urls = []
def downloader(url, timeout=10):
downloaded_urls.append(url)
return '下载的新歌词\n第二行'.encode('utf-8'), 'text/plain'
records, stats = load_existing_l2_candidates(rows, cache_path, downloader=downloader)
assert downloaded_urls == ['https://oss.example.test/101.txt']
assert [r.record_id for r in records] == ['100', '101']
assert records[0].lyrics == '缓存里的歌词\n第二行'
assert stats == {'cached': 1, 'downloaded': 1, 'failed': 0}
# ====================================================================
# L1 元数据去重 check_l1
# ====================================================================
......@@ -192,13 +289,13 @@ class TestCheckL2:
def test_empty_lyrics(self):
"""空歌词应返回 new"""
row = {'id': 1, 'lyrics_txt_content': '', 'name': 'test', 'singer': 'test'}
decision, confidence, matched_id, reason = check_l2(row, self.checker, [])
decision, confidence, matched_id, reason, _ = check_l2(row, self.checker, [])
assert decision == 'new'
assert '无歌词内容' in reason
def test_none_lyrics(self):
row = {'id': 1, 'lyrics_txt_content': None, 'name': 'test', 'singer': 'test'}
decision, _, _, _ = check_l2(row, self.checker, [])
decision, _, _, _, _ = check_l2(row, self.checker, [])
assert decision == 'new'
def test_no_candidates(self):
......@@ -209,7 +306,7 @@ class TestCheckL2:
'name': '你是我的眼',
'singer': '萧煌奇',
}
decision, confidence, matched_id, reason = check_l2(row, self.checker, [])
decision, confidence, matched_id, reason, _ = check_l2(row, self.checker, [])
assert decision == 'new'
assert '无候选集' in reason
......@@ -227,7 +324,7 @@ class TestCheckL2:
'singer': '测试歌手',
}
decision, confidence, matched_id, reason = check_l2(row, self.checker, [candidate])
decision, confidence, matched_id, reason, _ = check_l2(row, self.checker, [candidate])
assert decision == 'new'
assert matched_id is None
......@@ -247,7 +344,7 @@ class TestCheckL2:
'singer': '测试歌手',
}
decision, confidence, matched_id, reason = check_l2(row, self.checker, [candidate])
decision, confidence, matched_id, reason, _ = check_l2(row, self.checker, [candidate])
assert decision == 'new'
assert matched_id is None
......@@ -267,7 +364,7 @@ class TestCheckL2:
'singer': '测试歌手',
}
decision, confidence, matched_id, reason = check_l2(row, self.checker, [candidate])
decision, confidence, matched_id, reason, _ = check_l2(row, self.checker, [candidate])
assert decision == 'new'
assert matched_id is None
......@@ -292,13 +389,30 @@ class TestCheckL2:
'name': '你是我的眼',
'singer': '萧煌奇',
}
decision, confidence, matched_id, reason = check_l2(
decision, confidence, matched_id, reason, _ = check_l2(
row, self.checker, [candidate]
)
assert decision == 'duplicate'
assert confidence >= 0.9
assert matched_id == 'existing_1'
def test_short_exact_hash_match_is_review_not_duplicate(self):
"""规范化后歌词过短时，即使哈希完全一致也不能自动合并。"""
short_lyrics = '[00:00.730]春风吹过我的心'
candidate = LyricRecord(record_id='existing_short', lyrics=short_lyrics, title='旧歌')
row = {
'id': 22,
'lyrics_txt_content': '[00:01.120]春风吹过我的心',
'name': '另一首歌',
'singer': '不同歌手',
}
decision, _, matched_id, reason, _ = check_l2(row, self.checker, [candidate])
assert decision == 'review'
assert matched_id == 'existing_short'
assert '过短' in reason
def test_similar_lyrics_high_overlap(self):
"""高度相似的歌词应判为 duplicate 或 review"""
base_lyrics = '\n'.join([
......@@ -335,7 +449,7 @@ class TestCheckL2:
'name': '如果有一天(另一版)',
'singer': '未知',
}
decision, confidence, matched_id, reason = check_l2(
decision, confidence, matched_id, reason, _ = check_l2(
row, self.checker, [candidate]
)
# 高相似度 -> 应该不是 new（duplicate 或 review）
......@@ -362,7 +476,7 @@ class TestCheckL2:
'name': '夜晚',
'singer': '未知',
}
decision, confidence, matched_id, reason = check_l2(
decision, confidence, matched_id, reason, _ = check_l2(
row, self.checker, [candidate]
)
assert decision == 'new'
......@@ -392,7 +506,7 @@ class TestCheckL2:
'name': '片段歌曲',
'singer': '未知',
}
decision, _, _, _ = check_l2(row, self.checker, [candidate])
decision, _, _, _, _ = check_l2(row, self.checker, [candidate])
assert decision == 'review'
def test_lrc_with_timestamps_still_detects(self):
......@@ -424,7 +538,7 @@ class TestCheckL2:
'name': '小苹果(LRC版)',
'singer': '筷子兄弟',
}
decision, confidence, matched_id, reason = check_l2(
decision, confidence, matched_id, reason, _ = check_l2(
row, self.checker, [candidate]
)
assert decision == 'duplicate', f"Expected duplicate, got {decision}: {reason}"
......@@ -453,7 +567,7 @@ class TestCheckL2:
'name': '小小鸟',
'singer': '赵传',
}
decision, confidence, matched_id, _ = check_l2(
decision, confidence, matched_id, _, _ = check_l2(
row, self.checker, [candidate_different, candidate_exact]
)
assert decision == 'duplicate'
......@@ -477,7 +591,7 @@ class TestCheckL2:
# 极低阈值 -> 更容易判为 duplicate
strict_checker = DuplicateChecker(duplicate_jaccard_threshold=0.3)
decision_strict, _, _, _ = check_l2(row, strict_checker, [candidate])
decision_strict, _, _, _, _ = check_l2(row, strict_checker, [candidate])
assert decision_strict == 'duplicate'
def test_strong_review_same_writers_promotes_to_duplicate(self):
......@@ -527,7 +641,7 @@ class TestCheckL2:
'composer': '王琪',
}
decision, _, matched_id, reason = check_l2(row, self.checker, [candidate])
decision, _, matched_id, reason, _ = check_l2(row, self.checker, [candidate])
assert decision == 'duplicate', reason
assert matched_id == 'c1'
......@@ -580,7 +694,7 @@ class TestCheckL2:
'composer': '陈粒',
}
decision, _, matched_id, reason = check_l2(row, self.checker, [candidate])
decision, _, matched_id, reason, _ = check_l2(row, self.checker, [candidate])
assert decision == 'review', reason
assert matched_id == 'c1'
......@@ -603,7 +717,7 @@ class TestCheckL2:
'composer': '安智英/Vanilla Man',
}
decision, _, _, reason = check_l2(row, self.checker, [candidate])
decision, _, _, reason, _ = check_l2(row, self.checker, [candidate])
assert decision != 'duplicate', reason
......@@ -643,7 +757,7 @@ class TestCheckL2:
'composer': '共同作者',
}
decision, _, matched_id, reason = check_l2(row, self.checker, [candidate])
decision, _, matched_id, reason, _ = check_l2(row, self.checker, [candidate])
assert decision != 'duplicate', reason
......@@ -661,7 +775,7 @@ class TestCheckL2:
'singer': '测试歌手',
}
decision, _, matched_id, reason = check_l2(row, self.checker, [candidate])
decision, _, matched_id, reason, _ = check_l2(row, self.checker, [candidate])
assert decision != 'duplicate', reason
......@@ -693,7 +807,7 @@ class TestCheckL2:
'composer': '新曲作者',
}
decision, _, matched_id, reason = check_l2(row, self.checker, [candidate])
decision, _, matched_id, reason, _ = check_l2(row, self.checker, [candidate])
assert decision == 'duplicate', reason
assert matched_id == 'c1'
......@@ -841,7 +955,7 @@ class TestL2CandidateIndex:
])
index.add(LyricRecord(record_id='existing', lyrics=lyrics, title='你是我的眼'))
decision, confidence, matched_id, reason = check_l2(
decision, confidence, matched_id, reason, _ = check_l2(
{
'source_id': 999,
'lyrics_txt_content': lyrics,
......@@ -979,7 +1093,7 @@ class TestDedupIntegration:
assert is_dup is False
# L2 应命中
decision, confidence, matched_id, reason = check_l2(
decision, confidence, matched_id, reason, _ = check_l2(
row, self.checker, self.l2_candidates
)
assert decision == 'duplicate'
......@@ -1009,7 +1123,7 @@ class TestDedupIntegration:
assert is_dup is False
# L2 也不命中
decision, _, _, _ = check_l2(row, self.checker, self.l2_candidates)
decision, _, _, _, _ = check_l2(row, self.checker, self.l2_candidates)
assert decision == 'new'
def test_l1_match_with_different_lyrics_is_not_skipped(self):
......@@ -1056,6 +1170,44 @@ class TestDedupIntegration:
assert action['merge_authors'] is True
assert action['matched_id'] == '100'
def test_short_l2_exact_match_without_l1_is_new(self):
"""L2 过短 exact-hash 命中但 L1 未命中时，直接作为新歌。"""
candidate = LyricRecord(record_id='102', lyrics='春风吹过我的心', title='旧歌')
row = {
'id': 205,
'name': '另一首歌',
'lyricist': '不同词作者',
'composer': '不同曲作者',
'lyrics_txt_content': '[00:00.730]春风吹过我的心',
'singer': '不同歌手',
}
action = classify_dedup_action(row, self.l1_index, self.checker, [candidate])
assert action['action'] == 'new'
assert action['decision'] == 'new'
assert action['matched_id'] is None
assert action['l1_matched'] is False
def test_short_l2_exact_match_with_l1_is_merge(self):
"""L2 过短 exact-hash 命中且 L1 命中时，按 L1 结果自动合并。"""
candidate = LyricRecord(record_id='100', lyrics='春风吹过我的心', title='晴天')
row = {
'id': 206,
'name': '晴天',
'lyricist': '周杰伦',
'composer': '周杰伦',
'lyrics_txt_content': '[00:00.730]春风吹过我的心',
'singer': '周杰伦',
}
action = classify_dedup_action(row, self.l1_index, self.checker, [candidate])
assert action['action'] == 'merge'
assert action['decision'] == 'duplicate'
assert action['matched_id'] == '100'
assert action['l1_matched'] is True
def test_full_pipeline_simulation(self):
"""模拟完整 L1 -> L2 -> 写入 的管线流程"""
batch = [
......@@ -1100,7 +1252,7 @@ class TestDedupIntegration:
continue
# L2
decision, confidence, l2_matched_id, reason = check_l2(
decision, confidence, l2_matched_id, reason, _ = check_l2(
row, self.checker, self.l2_candidates
)
if decision == 'duplicate':
......@@ -1138,7 +1290,7 @@ class TestEdgeCases:
checker = DuplicateChecker()
candidate = LyricRecord(record_id='c1', lyrics='爱', title='短歌')
row = {'id': 1, 'lyrics_txt_content': '爱', 'name': '短歌', 'singer': 'test'}
decision, _, _, _ = check_l2(row, checker, [candidate])
decision, _, _, _, _ = check_l2(row, checker, [candidate])
# 极短歌词可能是 duplicate（哈希匹配）或 new，但不应崩溃
assert decision in ('duplicate', 'review', 'new')
......@@ -1146,7 +1298,7 @@ class TestEdgeCases:
"""纯空白歌词"""
checker = DuplicateChecker()
row = {'id': 1, 'lyrics_txt_content': ' \n\n ', 'name': 'test', 'singer': 'test'}
decision, _, _, reason = check_l2(row, checker, [])
decision, _, _, reason, _ = check_l2(row, checker, [])
assert decision == 'new'
def test_l1_index_with_duplicate_names(self):
......@@ -1171,7 +1323,7 @@ class TestEdgeCases:
candidate = LyricRecord(record_id='c1', lyrics=plain, title='我爱你中国')
row = {'id': 2, 'lyrics_txt_content': with_ts, 'name': '我爱你中国', 'singer': '汪峰'}
decision, _, _, _ = check_l2(row, checker, [candidate])
decision, _, _, _, _ = check_l2(row, checker, [candidate])
# normalization 应该去掉时间戳后识别为相同歌词
assert decision == 'duplicate'
......@@ -1780,7 +1932,7 @@ def _run_l2_retrieval_benchmark(
'name': record.title,
'singer': record.artist,
}
decision, confidence, matched_id, reason = check_l2(row, checker, index)
decision, confidence, matched_id, reason, _ = check_l2(row, checker, index)
decision_counts[decision] += 1
if show_progress:
source_iterable.set_postfix(
......