perf(etl_to_crawler): 优化录音数据批量查询与写入性能

- 将批量查询录音数从1000调整为100，适配具体需求
- 新增按录音id批量查询录音函数，避免按歌曲id查询全部平台数据
- 修改runner流程中以录音id批量预取录音数据，减少重复查询
- 增加spider数据批量预取，避免逐条调用造成性能瓶颈
- 优化写入yinyan_song_records时采用批量更新语句，提高更新效率
- 新增测试覆盖fetch_platform_records_by_record_ids函数的查询准确性
- 调整测试用例以适配新增的批量查询函数及写入逻辑

@@ -48,5 +48,5 @@ PLATFORM_KUGOU = '2'
 PLATFORM_NETEASE = '4'
 PLATFORMS = [PLATFORM_QQ, PLATFORM_KUGOU, PLATFORM_NETEASE]
 
-BATCH_SIZE = 1000
+BATCH_SIZE = 100
 BACKFILL_BATCH_SIZE = int(os.environ.get('BACKFILL_BATCH_SIZE', '5000'))

@@ -54,6 +54,26 @@ ORDER BY sar.song_id,
          mr.id ASC
 """
 
+_PLATFORM_BY_RECORD_IDS_QUERY = """
+SELECT
+    sar.song_id AS source_song_id,
+    mr.id AS record_id,
+    mr.platform,
+    mr.platform_unique_key,
+    mr.platform_mid,
+    mr.album_audio_id,
+    sar.is_main_version,
+    mr.is_high,
+    mr.pub_time
+FROM hk_song_and_record sar
+JOIN hk_music_record mr ON mr.id = sar.record_id
+WHERE mr.id IN ({placeholders})
+  AND mr.platform IN ('1','2','4')
+  AND mr.platform_unique_key IS NOT NULL
+  AND mr.platform_unique_key != ''
+  AND mr.deleted = 0
+"""
+
 # 同 _PLATFORM_QUERY，但不做 per-platform 去重，保留同平台全部录音供 singer fallback 遍历
 _ALL_PLATFORM_RECORDS_QUERY = _PLATFORM_QUERY
 
@@ -123,6 +143,27 @@ def fetch_all_platform_records(source_conn: pymysql.Connection, song_ids: list[i
     return result
 
 
+def fetch_platform_records_by_record_ids(source_conn: pymysql.Connection, record_ids: list[int]) -> list[dict]:
+    """按状态表指定的录音 id 批量查询录音，不再按 song_id 拉取全平台候选。"""
+    if not record_ids:
+        return []
+    unique_record_ids = list(dict.fromkeys(int(record_id) for record_id in record_ids))
+    placeholders = ','.join(['%s'] * len(unique_record_ids))
+    query = _PLATFORM_BY_RECORD_IDS_QUERY.format(placeholders=placeholders)
+    with source_conn.cursor() as cur:
+        cur.execute(query, unique_record_ids)
+        rows = cur.fetchall()
+
+    seen_record_ids: set = set()
+    result = []
+    for row in rows:
+        rid = int(row['record_id'])
+        if rid not in seen_record_ids:
+            seen_record_ids.add(rid)
+            result.append(row)
+    return result
+
+
 def fetch_record_platforms(source_conn: pymysql.Connection, record_ids: list[int]) -> dict[int, str]:
     """按录音 id 查询平台代码，用于回填 yinyan_song_records.platform。"""
     if not record_ids:

@@ -10,6 +10,7 @@ from .reader import (
     fetch_hk_songs_by_source_ids,
     fetch_platform_records,
     fetch_all_platform_records,
+    fetch_platform_records_by_record_ids,
     fetch_record_platforms,
     select_primary_record,
 )
@@ -608,14 +609,40 @@ def run(
                 break
 
             song_ids = [int(r['song_id']) for r in pending_records]
+            record_ids = [int(r['record_id']) for r in pending_records]
             hk_by_song = fetch_hk_songs_by_source_ids(hk_conn, song_ids)
-            all_platform_records = fetch_all_platform_records(src_conn, song_ids)
+            platform_records = fetch_platform_records_by_record_ids(src_conn, record_ids)
             pr_by_state_key: dict[tuple[int, int, str], dict] = {}
-            for pr in all_platform_records:
+            for pr in platform_records:
                 if pr['platform'] in platforms:
                     key = (int(pr['source_song_id']), int(pr['record_id']), pr['platform'])
                     pr_by_state_key[key] = pr
 
+            # ── 批量预取 spider 数据（整批一次 IN 查询，不逐首调用）──────────
+            qq_mids = list({pr['platform_unique_key'] for pr in platform_records if pr['platform'] == PLATFORM_QQ})
+            kugou_ids = list({int(pr['platform_unique_key']) for pr in platform_records if pr['platform'] == PLATFORM_KUGOU})
+            netease_ids = list({int(pr['platform_unique_key']) for pr in platform_records if pr['platform'] == PLATFORM_NETEASE})
+
+            qq_songs_map = fetch_qq_songs(spider_conn, qq_mids) if qq_mids else {}
+            kugou_songs_map = fetch_kugou_songs(spider_conn, kugou_ids) if kugou_ids else {}
+            netease_songs_map = fetch_netease_songs(spider_conn, netease_ids) if netease_ids else {}
+
+            qq_db_song_ids = [v['id'] for v in qq_songs_map.values()]
+            qq_singers_map = fetch_qq_singers(spider_conn, qq_db_song_ids) if qq_db_song_ids else {}
+            kugou_singers_map = fetch_kugou_singers(spider_conn, kugou_ids) if kugou_ids else {}
+            netease_singers_map = fetch_netease_singers(spider_conn, netease_ids) if netease_ids else {}
+
+            songs_maps = {
+                PLATFORM_QQ: qq_songs_map,
+                PLATFORM_KUGOU: kugou_songs_map,
+                PLATFORM_NETEASE: netease_songs_map,
+            }
+            singers_maps = {
+                PLATFORM_QQ: qq_singers_map,
+                PLATFORM_KUGOU: kugou_singers_map,
+                PLATFORM_NETEASE: netease_singers_map,
+            }
+
             with pg_conn.cursor() as pg_cur:
                 pushed_count = 0
                 for pending in pending_records:

@@ -80,19 +80,22 @@ def upsert_yinyan_song_records(cur, records: list[dict]) -> None:
     """
     if not records:
         return
-    cur.executemany(
-        """
-        UPDATE yinyan_song_records
-        SET platform = %s,
-            platform_song_id = %s,
-            record_id = %s,
+    values_sql = ', '.join(['(%s::bigint, %s::bigint, %s::varchar, %s::bigint)'] * len(records))
+    params = []
+    for r in records:
+        params.extend([r['song_id'], r['record_id'], r['platform'], r['platform_song_id']])
+    cur.execute(
+        f"""
+        UPDATE yinyan_song_records AS ysr
+        SET platform = v.platform,
+            platform_song_id = v.platform_song_id,
+            record_id = v.record_id,
             is_yinyan_push = TRUE
-        WHERE song_id = %s AND is_yinyan_push = FALSE
+        FROM (VALUES {values_sql}) AS v(song_id, record_id, platform, platform_song_id)
+        WHERE ysr.song_id = v.song_id
+          AND ysr.is_yinyan_push = FALSE
         """,
-        [
-            (r['platform'], r['platform_song_id'], r['record_id'], r['song_id'])
-            for r in records
-        ],
+        tuple(params),
     )
 
 

 from unittest.mock import MagicMock
 
 from etl_to_crawler.reader import (
+    fetch_platform_records_by_record_ids,
     fetch_platform_records,
     iter_hk_songs_batches,
     select_primary_record,
@@ -59,6 +60,32 @@ def test_fetch_platform_records_keeps_one_record_per_song_and_platform():
     assert {row['record_id'] for row in result} == {100, 101, 102}
 
 
+def test_fetch_platform_records_by_record_ids_queries_exact_records_once():
+    conn = MagicMock()
+    conn.cursor.return_value = _make_cursor([
+        {
+            'source_song_id': 10,
+            'record_id': 200,
+            'platform': '2',
+            'platform_unique_key': '200',
+            'platform_mid': 'kg-hash',
+            'album_audio_id': None,
+            'is_main_version': 1,
+            'is_high': 0,
+            'pub_time': '2021-01-01',
+        },
+    ])
+
+    result = fetch_platform_records_by_record_ids(conn, [200, 200])
+
+    cur = conn.cursor.return_value
+    sql, params = cur.execute.call_args[0]
+    assert 'WHERE mr.id IN (%s)' in sql
+    assert params == [200]
+    assert result[0]['record_id'] == 200
+    assert result[0]['platform'] == '2'
+
+
 def test_select_primary_record_prefers_main_version_then_is_high():
     rows = [
         {

@@ -82,7 +82,8 @@ def test_run_imports_only_pending_yinyan_platform_record(monkeypatch):
         },
     ]
     monkeypatch.setattr(runner, 'fetch_platform_records', lambda conn, song_ids: platform_records)
-    monkeypatch.setattr(runner, 'fetch_all_platform_records', lambda conn, song_ids: platform_records)
+    monkeypatch.setattr(runner, 'fetch_all_platform_records', MagicMock())
+    monkeypatch.setattr(runner, 'fetch_platform_records_by_record_ids', lambda conn, record_ids: platform_records)
     monkeypatch.setattr(runner, '_PROCESSORS', processors)
     monkeypatch.setattr(runner, 'upsert_yinyan_song_records', yinyan_writer)
 
@@ -90,6 +91,7 @@ def test_run_imports_only_pending_yinyan_platform_record(monkeypatch):
 
     processors['1'].assert_not_called()
     processors['2'].assert_called_once()
+    runner.fetch_all_platform_records.assert_not_called()
     yinyan_writer.assert_called_once_with(pg_conn.cur, [{
         'song_id': 10,
         'record_id': 200,

@@ -52,14 +52,17 @@ def test_upsert_yinyan_song_records_marks_existing_relation_as_pushed():
         {'song_id': 11, 'record_id': 101, 'platform': '2', 'platform_song_id': 1001},
     ])
 
-    sql, rows = cur.executemany.call_args[0]
+    sql, params = cur.execute.call_args[0]
     assert 'UPDATE yinyan_song_records' in sql
-    assert 'platform = %s' in sql
-    assert 'platform_song_id = %s' in sql
-    assert 'record_id = %s' in sql
+    assert 'UPDATE yinyan_song_records AS ysr' in sql
+    assert 'SET platform = v.platform' in sql
+    assert 'platform_song_id = v.platform_song_id' in sql
+    assert 'record_id = v.record_id' in sql
     assert 'is_yinyan_push = TRUE' in sql
-    assert 'WHERE song_id = %s AND is_yinyan_push = FALSE' in sql
-    assert rows == [('1', 1000, 100, 10), ('2', 1001, 101, 11)]
+    assert 'FROM (VALUES (%s::bigint, %s::bigint, %s::varchar, %s::bigint), (%s::bigint, %s::bigint, %s::varchar, %s::bigint))' in sql
+    assert 'ysr.song_id = v.song_id' in sql
+    assert 'ysr.is_yinyan_push = FALSE' in sql
+    assert params == (10, 100, '1', 1000, 11, 101, '2', 1001)
 
 
 def test_insert_yinyan_song_records_initializes_unpushed_rows():