feat(cache): 优化歌词缓存同步逻辑，追加写入避免全量读取

- 修改 sync_inserted_lyrics_cache 函数，新增 seen_keys 参数用于跳过已存在歌词缓存 - 实现追加写入方式，同步新增歌词缓存时不再全量读取，提升性能 - 在主逻辑中初始化并传递 lyrics_cache_seen_keys，避免重复写入缓存 - 更新测试用例，增加追加写入场景覆盖，防止误读缓存文件读取操作 - 修正缓存文件路径统一使用 lyrics_cache_path，增强代码一致性

feat(cache): 优化歌词缓存同步逻辑，追加写入避免全量读取
- 修改 sync_inserted_lyrics_cache 函数，新增 seen_keys 参数用于跳过已存在歌词缓存 - 实现追加写入方式，同步新增歌词缓存时不再全量读取，提升性能 - 在主逻辑中初始化并传递 lyrics_cache_seen_keys，避免重复写入缓存 - 更新测试用例，增加追加写入场景覆盖，防止误读缓存文件读取操作 - 修正缓存文件路径统一使用 lyrics_cache_path，增强代码一致性
沈秋雨
Commit 3f2cfa61 ... 3f2cfa61e7ee2ead10834d37d6dc455bcfb38f87 authored 2026-07-06 22:52:38 +0800 by 沈秋雨
Showing 2 changed files with 64 additions and 15 deletions
import_hk_songs.py
test_dedup.py
--- a/import_hk_songs.py
View file @3f2cfa6
+++ b/import_hk_songs.py
View file @3f2cfa6
@@ -752,13 +752,16 @@ def _read_existing_lyrics_cache(cache_path: Path) -> dict[tuple[str, str], dict]
    return cached_by_key
-def sync_inserted_lyrics_cache(cache_path: Path, inserted_rows: list[tuple[tuple, dict]]) -> int:
+def sync_inserted_lyrics_cache(
-    """把本次已成功插入目标库的歌词同步到 L2 启动缓存。"""
+    cache_path: Path,
+    inserted_rows: list[tuple[tuple, dict]],
+    seen_keys: set[tuple[str, str]] | None = None,
+) -> int:
+    """把本次已成功插入目标库的歌词追加同步到 L2 启动缓存。"""
    if not inserted_rows:
        return 0
-    cached_by_key = _read_existing_lyrics_cache(cache_path)
+    new_items = []
-    added = 0
    for tuple_row, source_row in inserted_rows:
        url = tuple_row[OSS_LYRIC_INDEX]
        lyrics = source_row.get('lyrics_txt_content')
@@ -769,9 +772,9 @@ def sync_inserted_lyrics_cache(cache_path: Path, inserted_rows: list[tuple[tuple
        record_id = str(tuple_row[0])
        key = (record_id, str(url))
-        if key not in cached_by_key:
+        if seen_keys is not None and key in seen_keys:
-            added += 1
+            continue
-        cached_by_key[key] = {
+        item = {
            'id': record_id,
            'url': str(url),
            'lyrics': str(lyrics),
@@ -780,10 +783,18 @@ def sync_inserted_lyrics_cache(cache_path: Path, inserted_rows: list[tuple[tuple
            'lyricist': tuple_row[2],
            'composer': tuple_row[3],
        }
+        new_items.append(item)
+        if seen_keys is not None:
+            seen_keys.add(key)
-    if added:
+    if not new_items:
-        _write_existing_lyrics_cache(cache_path, cached_by_key)
+        return 0
-    return added
+    cache_path.parent.mkdir(parents=True, exist_ok=True)
+    with cache_path.open('a', encoding='utf-8') as f:
+        for item in new_items:
+            f.write(json.dumps(item, ensure_ascii=False) + '\n')
+    return len(new_items)
 def load_existing_l2_candidates(
@@ -1627,6 +1638,8 @@ def main():
    dedup_report = None
    review_report = None
    source_record_counts: dict[str, int] = {}
+    lyrics_cache_path = CACHE_DIR / f'{TARGET_TABLE_NAME}_existing_lyrics.jsonl'
+    lyrics_cache_seen_keys: set[tuple[str, str]] = set()
    if not args.skip_dedup:
        logger.info("正在构建 L1 元数据索引...")
@@ -1660,14 +1673,14 @@ def main():
                with target_conn.cursor(pymysql.cursors.DictCursor) as cursor:
                    cursor.execute(lyric_sql)
                    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, lyrics_cache_path)
-                existing_records, cache_stats = load_existing_l2_candidates(existing_lyric_rows, cache_path)
+                lyrics_cache_seen_keys = set(_read_existing_lyrics_cache(lyrics_cache_path).keys())
                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}"
+                    f"缓存文件={lyrics_cache_path}"
                )
            except Exception as e:
                logger.error(f"加载已有歌词失败: {e}")
@@ -1776,8 +1789,9 @@ def main():
                        with stats_lock:
                            stats['inserted'] += batch_inserted
                        cache_added = sync_inserted_lyrics_cache(
-                            CACHE_DIR / f'{TARGET_TABLE_NAME}_existing_lyrics.jsonl',
+                            lyrics_cache_path,
                            batch_cache_rows,
+                            lyrics_cache_seen_keys,
                        )
                        if cache_added:
                            logger.info(f"目标库歌词缓存同步新增: {cache_added} 条")
@@ -1795,8 +1809,9 @@ def main():
                                    stats['inserted'] += 1
                                if row_i < len(batch_cache_rows):
                                    sync_inserted_lyrics_cache(
-                                        CACHE_DIR / f'{TARGET_TABLE_NAME}_existing_lyrics.jsonl',
+                                        lyrics_cache_path,
                                        [batch_cache_rows[row_i]],
+                                        lyrics_cache_seen_keys,
                                    )
                        except Exception as e2:
                            with stats_lock:
--- a/test_dedup.py
View file @3f2cfa6
+++ b/test_dedup.py
View file @3f2cfa6
@@ -23,6 +23,7 @@ from tqdm import tqdm
 # 确保项目根目录在 sys.path
 sys.path.insert(0, str(Path(__file__).resolve().parent))
+import import_hk_songs as import_script
 from import_hk_songs import (
    AGGREGATE_SQL,
    REPORT_DIR,
@@ -301,6 +302,39 @@ class TestExistingLyricsCache:
        assert cache_items[1]['url'] == 'https://oss.example.test/200.txt'
        assert cache_items[1]['lyrics'] == '新歌词正文'
+    def test_sync_inserted_lyrics_cache_appends_without_reading_full_cache(self, tmp_path, monkeypatch):
+        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] = 201
+        tuple_row[1] = '新歌'
+        tuple_row[2] = '新词'
+        tuple_row[3] = '新曲'
+        tuple_row[8] = 'https://oss.example.test/201.txt'
+        tuple_row[33] = '新歌手'
+        def fail_if_full_cache_read(path):
+            raise AssertionError('should not read full cache during append sync')
+        monkeypatch.setattr(import_script, '_read_existing_lyrics_cache', fail_if_full_cache_read)
+        added = sync_inserted_lyrics_cache(
+            cache_path,
+            [(tuple(tuple_row), {'lyrics_txt_content': '新增歌词'})],
+        )
+        cache_lines = cache_path.read_text(encoding='utf-8').splitlines()
+        assert added == 1
+        assert len(cache_lines) == 2
+        assert json.loads(cache_lines[-1])['id'] == '201'
 # ====================================================================
 # L1 元数据去重 check_l1