Commit 65615e7b 65615e7ba1bf368e72f2ef89e207a4fec7914fb6 by 沈秋雨

refactor(lyrics-cache): 重构歌词缓存读取写入逻辑并同步新增歌词缓存

- 抽取歌词缓存读取和写入为独立函数 _read_existing_lyrics_cache 与 _write_existing_lyrics_cache
- load_existing_l2_candidates 中改用新缓存读写函数,增加缓存加载日志和进度输出
- 处理下载异常时保证已下载缓存的歌词及时持久化,避免丢失数据
- 新增 sync_inserted_lyrics_cache 函数实现新增歌词缓存同步写入功能
- 后台写入线程写入数据库后调用歌词缓存同步,记录新增条数日志
- 修改批量写入队列数据结构,传递缓存同步相关数据
- 优化去重统计日志显示格式,新增 format_dedup_stats 函数统一格式化
- L2审核界面新增 skip 筛选选项,调整相关前端和后端代码支持该状态
- 移除 L2 审核界面中保存按钮,避免误操作手动保存
- 补充单元测试覆盖缓存同步、缓存读取异常处理和去重统计日志格式输出
1 parent 93b03ea9
......@@ -722,14 +722,19 @@ def _decode_lyric_content(content: bytes) -> str:
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。"""
def _write_existing_lyrics_cache(cache_path: Path, cache_items: dict[tuple[str, str], dict]) -> None:
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 cache_items.values():
f.write(json.dumps(item, ensure_ascii=False) + '\n')
tmp_path.replace(cache_path)
def _read_existing_lyrics_cache(cache_path: Path) -> dict[tuple[str, str], dict]:
cached_by_key: dict[tuple[str, str], dict] = {}
if cache_path.exists():
if not cache_path.exists():
return cached_by_key
with cache_path.open('r', encoding='utf-8') as f:
for line in f:
line = line.strip()
......@@ -744,15 +749,67 @@ def load_existing_l2_candidates(
lyrics = item.get('lyrics')
if record_id and url and isinstance(lyrics, str) and lyrics.strip():
cached_by_key[(record_id, url)] = item
return cached_by_key
def sync_inserted_lyrics_cache(cache_path: Path, inserted_rows: list[tuple[tuple, dict]]) -> int:
"""把本次已成功插入目标库的歌词同步到 L2 启动缓存。"""
if not inserted_rows:
return 0
cached_by_key = _read_existing_lyrics_cache(cache_path)
added = 0
for tuple_row, source_row in inserted_rows:
url = tuple_row[OSS_LYRIC_INDEX]
lyrics = source_row.get('lyrics_txt_content')
if not url or not str(url).startswith(('http://', 'https://')):
continue
if not lyrics or not str(lyrics).strip():
continue
record_id = str(tuple_row[0])
key = (record_id, str(url))
if key not in cached_by_key:
added += 1
cached_by_key[key] = {
'id': record_id,
'url': str(url),
'lyrics': str(lyrics),
'name': tuple_row[1],
'singer': tuple_row[33],
'lyricist': tuple_row[2],
'composer': tuple_row[3],
}
if added:
_write_existing_lyrics_cache(cache_path, cached_by_key)
return added
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 = _read_existing_lyrics_cache(cache_path)
records: list[LyricRecord] = []
next_cache: dict[tuple[str, str], dict] = {}
stats = {'cached': 0, 'downloaded': 0, 'failed': 0}
for row in rows:
rows_with_url = [
row for row in rows
if row.get('lyrics_url') and str(row.get('lyrics_url')).startswith(('http://', 'https://'))
]
logger.info(
f"目标库歌词缓存加载开始: rows={len(rows)}, 有效URL={len(rows_with_url)}, "
f"已有缓存={len(cached_by_key)}, 缓存文件={cache_path}"
)
try:
for idx, row in enumerate(rows_with_url, start=1):
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))
......@@ -785,12 +842,13 @@ def load_existing_l2_candidates(
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)
if idx % 50 == 0 or idx == len(rows_with_url):
logger.info(
f"目标库歌词缓存进度: {idx}/{len(rows_with_url)} | "
f"命中={stats['cached']} 下载={stats['downloaded']} 失败={stats['failed']}"
)
finally:
_write_existing_lyrics_cache(cache_path, next_cache)
return records, stats
......@@ -807,6 +865,15 @@ def check_l1(row: dict, l1_index: dict) -> tuple[bool, int | None]:
return False, None
def format_dedup_stats(stats: dict) -> str:
return (
f"L1命中={stats['l1_dup']} | "
f"L2合并命中={stats['l2_dup']} | L2待审={stats['l2_review']} | "
f"L2新词={stats['l2_new']} | 作者合并={stats['author_merged']} | "
f"最终跳过={stats['skipped']}"
)
class L2CandidateIndex:
"""L2 歌词候选召回索引,最终判定仍交给 DuplicateChecker。"""
......@@ -1693,7 +1760,7 @@ def main():
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
batch_data, staging_data, batch_cache_rows, batch_size_actual, batch_start, author_merge_tasks = item
try:
with target_conn.cursor() as cursor:
if batch_data:
......@@ -1708,6 +1775,12 @@ def main():
batch_inserted = len(batch_data)
with stats_lock:
stats['inserted'] += batch_inserted
cache_added = sync_inserted_lyrics_cache(
CACHE_DIR / f'{TARGET_TABLE_NAME}_existing_lyrics.jsonl',
batch_cache_rows,
)
if cache_added:
logger.info(f"目标库歌词缓存同步新增: {cache_added} 条")
except Exception as e:
logger.error(f"批量写入失败 (offset {batch_start}): {e}")
target_conn.rollback()
......@@ -1720,6 +1793,11 @@ def main():
target_conn.commit()
with stats_lock:
stats['inserted'] += 1
if row_i < len(batch_cache_rows):
sync_inserted_lyrics_cache(
CACHE_DIR / f'{TARGET_TABLE_NAME}_existing_lyrics.jsonl',
[batch_cache_rows[row_i]],
)
except Exception as e2:
with stats_lock:
stats['errors'] += 1
......@@ -1739,7 +1817,7 @@ def main():
pbar_db.set_postfix(
ins=stats['inserted'],
err=stats['errors'],
l1=stats['l1_dup'],
l1_hit=stats['l1_dup'],
l2=stats['l2_dup'],
rev=stats['l2_review'],
auth=stats['author_merged'],
......@@ -1872,6 +1950,11 @@ def main():
# 按顺序组装 batch_data
batch_data = [results_map[i] for i in range(len(new_rows)) if i in results_map]
batch_cache_rows = [
(results_map[i], new_rows[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:
......@@ -1900,7 +1983,7 @@ def main():
# ===== 步骤4:交给后台线程写入 DB(主线程继续下一批次 OSS)=====
if not args.skip_dedup:
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))
db_write_queue.put((batch_data, staging_data, batch_cache_rows, batch_size_actual, batch_start, author_merge_tasks))
# 所有批次已入队;等待后台 DB 写入收尾后停止线程
stop_db_writer(db_write_queue, db_writer_thread)
......@@ -1915,10 +1998,7 @@ def main():
logger.info("导入完成!")
logger.info(f"总计: {total} | 插入: {stats['inserted']} | 错误: {stats['errors']}")
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']} | 作者合并={stats['author_merged']} | "
f"跳过={stats['skipped']}")
logger.info(f"去重统计: {format_dedup_stats(stats)}")
if dedup_report:
logger.info(f"去重报告: {dedup_report.filepath}")
if review_report:
......
......@@ -466,6 +466,7 @@
<option value="new">new</option>
<option value="merge">merge</option>
<option value="review">review</option>
<option value="skip">skip</option>
</select>
<input id="searchInput" type="search" placeholder="搜索歌名 / ID">
<label for="pageSizeSelect">每页</label>
......@@ -858,7 +859,6 @@
${value === 'duplicate' ? '确认重复' : value === 'not_duplicate' ? '确认不重复' : '待确认'}
</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>
`}
......@@ -906,14 +906,6 @@
renderDetail();
});
}
const saveBtn = document.getElementById('saveReviewBtn');
if (saveBtn) {
saveBtn.addEventListener('click', () => {
if (!candidate) return;
setReview(candidate, { note: document.getElementById('reviewNote')?.value || '' });
renderDetail();
});
}
const deleteBtn = document.getElementById('deleteReviewBtn');
if (deleteBtn) {
deleteBtn.addEventListener('click', async () => {
......
......@@ -174,8 +174,8 @@ def _staging_groups_response(mode: str, term: str, page: int, page_size: int, de
# 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'):
# 决策筛选:new / merge / review / skip
if decision and decision in ('new', 'merge', 'review', 'skip'):
where_clauses.append("dedup_action = %s")
params.append(decision)
where = "WHERE " + " AND ".join(where_clauses) if where_clauses else ""
......@@ -640,7 +640,7 @@ def _groups_response(path: Path, top_k: str, mode: str, term: str, page: int, pa
if mode == "hits":
groups = [g for g in groups if g.get("has_hit")]
# 决策筛选
if decision and decision in ('new', 'merge', 'review', 'duplicate'):
if decision and decision in ('new', 'merge', 'review', 'duplicate', 'skip'):
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)
......
......@@ -43,6 +43,8 @@ from import_hk_songs import (
process_lyrics,
DedupReport,
L2CandidateIndex,
format_dedup_stats,
sync_inserted_lyrics_cache,
load_approved_import_ids,
load_existing_l2_candidates,
)
......@@ -140,6 +142,26 @@ class TestPendingL1Index:
assert len(l1_index) == 1
class TestDedupStatsLog:
"""测试去重统计日志文案"""
def test_l1_counter_is_labeled_as_match_not_skip(self):
stats = {
'l1_dup': 2,
'l2_dup': 2,
'l2_review': 23,
'l2_new': 1725,
'author_merged': 1,
'skipped': 0,
}
message = format_dedup_stats(stats)
assert 'L1命中=2' in message
assert 'L1跳过' not in message
assert '最终跳过=0' in message
class TestDbWriterShutdown:
"""测试后台 DB 写入线程的优雅停止"""
......@@ -213,6 +235,72 @@ class TestExistingLyricsCache:
assert records[0].lyrics == '缓存里的歌词\n第二行'
assert stats == {'cached': 1, 'downloaded': 1, 'failed': 0}
def test_load_existing_l2_candidates_persists_successes_before_later_download_exception(self, tmp_path):
cache_path = tmp_path / 'lyrics_cache.jsonl'
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': '曲',
},
]
def downloader(url, timeout=10):
if url.endswith('/101.txt'):
raise RuntimeError('network interrupted')
return '已下载歌词'.encode('utf-8'), 'text/plain'
with pytest.raises(RuntimeError, match='network interrupted'):
load_existing_l2_candidates(rows, cache_path, downloader=downloader)
cache_lines = cache_path.read_text(encoding='utf-8').splitlines()
assert len(cache_lines) == 1
cached = json.loads(cache_lines[0])
assert cached['id'] == '100'
assert cached['lyrics'] == '已下载歌词'
def test_sync_inserted_lyrics_cache_adds_committed_inserted_rows(self, tmp_path):
cache_path = tmp_path / 'lyrics_cache.jsonl'
cache_path.write_text(
json.dumps({
'id': '100',
'url': 'https://oss.example.test/100.txt',
'lyrics': '已有歌词',
}, ensure_ascii=False) + '\n',
encoding='utf-8',
)
tuple_row = [None] * 45
tuple_row[0] = 200
tuple_row[1] = '新歌'
tuple_row[2] = '新词'
tuple_row[3] = '新曲'
tuple_row[8] = 'https://oss.example.test/200.txt'
tuple_row[33] = '新歌手'
source_row = {'lyrics_txt_content': '新歌词正文'}
added = sync_inserted_lyrics_cache(cache_path, [(tuple(tuple_row), source_row)])
cache_items = [
json.loads(line)
for line in cache_path.read_text(encoding='utf-8').splitlines()
]
assert added == 1
assert len(cache_items) == 2
assert cache_items[1]['id'] == '200'
assert cache_items[1]['url'] == 'https://oss.example.test/200.txt'
assert cache_items[1]['lyrics'] == '新歌词正文'
# ====================================================================
# L1 元数据去重 check_l1
......