feat(records2): 添加完整录音元数据的关联查询与初始化逻辑

- 优化长任务连接刷新和分页批处理机制

@@ -95,6 +95,37 @@ WHERE mr.id IN ({placeholders})
 # 同 _PLATFORM_QUERY，但不做 per-platform 去重，保留同平台全部录音供 singer fallback 遍历
 _ALL_PLATFORM_RECORDS_QUERY = _PLATFORM_QUERY
 
+# records2 只接收具备完整录音元数据的关联关系。
+_RECORDS2_RELATIONS_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 sar.song_id IN ({placeholders})
+  AND mr.platform IN ('1','2','4')
+  AND mr.platform_unique_key IS NOT NULL
+  AND TRIM(mr.platform_unique_key) != ''
+  AND mr.deleted = 0
+  AND mr.record_name IS NOT NULL AND TRIM(mr.record_name) != ''
+  AND mr.duration IS NOT NULL AND mr.duration > 0
+  AND mr.singer_name IS NOT NULL AND TRIM(mr.singer_name) != ''
+  AND mr.platform_index_url IS NOT NULL AND TRIM(mr.platform_index_url) != ''
+  AND (
+      (mr.storage_url IS NOT NULL AND TRIM(mr.storage_url) != '')
+      OR (mr.platform_play_url IS NOT NULL AND TRIM(mr.platform_play_url) != '')
+  )
+  AND mr.pub_time IS NOT NULL
+ORDER BY sar.song_id, mr.id
+"""
+
 
 def iter_hk_songs_batches(
     conn: pymysql.Connection,
@@ -196,11 +227,11 @@ def fetch_all_platform_records(source_conn: pymysql.Connection, song_ids: list[i
 
 
 def fetch_all_song_record_relations(source_conn: pymysql.Connection, song_ids: list[int]) -> list[dict]:
-    """返回指定 song_id 的全部有效 (song_id, record_id, platform) 关联。"""
+    """返回指定 song_id 的全部必填字段完整的 (song_id, record_id, platform) 关联。"""
     if not song_ids:
         return []
     placeholders = ','.join(['%s'] * len(song_ids))
-    query = _ALL_PLATFORM_RECORDS_QUERY.format(placeholders=placeholders)
+    query = _RECORDS2_RELATIONS_QUERY.format(placeholders=placeholders)
     with source_conn.cursor() as cur:
         cur.execute(query, song_ids)
         rows = cur.fetchall()

@@ -8,7 +8,6 @@ from .config import PLATFORM_QQ, PLATFORM_KUGOU, PLATFORM_NETEASE, BATCH_SIZE, B
 from .connections import get_hk_songs_conn, get_source_conn, get_spider_conn, get_pg_conn, get_oss_bucket, refresh_conn, close_all_pools
 from .reader import (
     iter_hk_songs_batches,
-    iter_hk_songs_records2_batches,
     fetch_hk_songs_by_source_ids,
     mark_hk_songs_deleted,
     fetch_platform_records,
@@ -30,6 +29,7 @@ from .writer import (
     insert_yinyan_song_records,
     insert_yinyan_song_records2,
     delete_yinyan_song_records2_existing_relations,
+    fetch_yinyan_song_ids,
     update_yinyan_record_platforms,
     upsert_yinyan_song_records,
     fetch_existing_yinyan_song_ids,
@@ -1247,26 +1247,26 @@ def initialize_yinyan_song_records(platforms: list[str], max_batches: int | None
 
 
 def initialize_yinyan_song_records2(platforms: list[str], max_batches: int | None = None) -> None:
-    """写入所有合格歌曲关联，再删除 records1 已存在的关联。"""
-    hk_conn = get_hk_songs_conn()
+    """以 records1 全量 song_id 为范围，写入补充录音关联并去重。"""
     src_conn = get_source_conn()
     pg_conn = get_pg_conn()
     total_inserted = 0
 
     try:
-        for index, batch in enumerate(
-            tqdm(iter_hk_songs_records2_batches(hk_conn, BATCH_SIZE), desc='init-yinyan-records2')
-        ):
+        with pg_conn.cursor() as pg_cur:
+            song_ids = fetch_yinyan_song_ids(pg_cur)
+        log.info('records2 source scope: records1 song_ids=%d', len(song_ids))
+
+        for index, start in enumerate(tqdm(range(0, len(song_ids), BATCH_SIZE), desc='init-yinyan-records2')):
             # 长任务连接可能断开，每批次开始前刷新
-            hk_conn = refresh_conn(hk_conn, 'hk_songs')
             src_conn = refresh_conn(src_conn, 'source')
             pg_conn = refresh_conn(pg_conn, 'pg')
 
             if max_batches is not None and index >= max_batches:
                 break
 
-            song_ids = list({int(row['source_song_id']) for row in batch})
-            relations = fetch_all_song_record_relations(src_conn, song_ids)
+            batch_song_ids = song_ids[start:start + BATCH_SIZE]
+            relations = fetch_all_song_record_relations(src_conn, batch_song_ids)
             rows = [
                 {
                     'song_id': int(relation['source_song_id']),

@@ -70,6 +70,18 @@ def delete_yinyan_song_records2_existing_relations(cur) -> int:
     return cur.rowcount
 
 
+def fetch_yinyan_song_ids(cur) -> list[int]:
+    """返回 records1 全量 song_id，作为 records2 补充范围。"""
+    cur.execute(
+        """
+        SELECT DISTINCT song_id
+        FROM yinyan_song_records
+        ORDER BY song_id
+        """
+    )
+    return [int(row[0]) for row in cur.fetchall()]
+
+
 def fetch_existing_yinyan_song_ids(cur) -> set[int]:
     """返回已完成初始化 platform 的 song_id 集合。"""
     cur.execute("SELECT DISTINCT song_id FROM yinyan_song_records WHERE platform IS NOT NULL")

@@ -98,6 +98,15 @@ def test_fetch_all_song_record_relations_keeps_all_records_for_a_song():
 
     result = fetch_all_song_record_relations(conn, [10])
 
+    sql, params = conn.cursor.return_value.execute.call_args[0]
+    assert 'mr.record_name IS NOT NULL' in sql
+    assert 'mr.duration IS NOT NULL AND mr.duration > 0' in sql
+    assert 'mr.singer_name IS NOT NULL' in sql
+    assert 'mr.platform_index_url IS NOT NULL' in sql
+    assert 'mr.storage_url IS NOT NULL' in sql
+    assert 'mr.platform_play_url IS NOT NULL' in sql
+    assert 'mr.pub_time IS NOT NULL' in sql
+    assert params == [10]
     assert [(row['record_id'], row['platform']) for row in result] == [
         (100, '1'), (101, '1'), (200, '2'),
     ]

@@ -263,12 +263,10 @@ def test_initialize_yinyan_song_records2_writes_all_relations_then_deduplicates(
     inserted = []
     dedupe_calls = []
 
-    monkeypatch.setattr(runner, 'get_hk_songs_conn', lambda: _Connection())
     monkeypatch.setattr(runner, 'get_source_conn', lambda: _Connection())
     monkeypatch.setattr(runner, 'get_pg_conn', lambda: pg_conn)
-    monkeypatch.setattr(runner, 'iter_hk_songs_records2_batches', lambda conn, batch_size: [[
-        {'id': 50, 'source_song_id': 10},
-    ]])
+    monkeypatch.setattr(runner, 'refresh_conn', lambda conn, name: conn)
+    monkeypatch.setattr(runner, 'fetch_yinyan_song_ids', lambda cur: [10])
     monkeypatch.setattr(runner, 'fetch_all_song_record_relations', lambda conn, song_ids: [
         {'source_song_id': 10, 'record_id': 100, 'platform': '1'},
         {'source_song_id': 10, 'record_id': 200, 'platform': '2'},

@@ -5,6 +5,7 @@ from etl_to_crawler.writer import (
     insert_yinyan_song_records,
     insert_yinyan_song_records2,
     delete_yinyan_song_records2_existing_relations,
+    fetch_yinyan_song_ids,
     update_yinyan_record_platforms,
     upsert_kugou_albums,
     upsert_kugou_singers,
@@ -148,6 +149,19 @@ def test_delete_yinyan_song_records2_existing_relations_uses_records1_as_dedup_s
     assert 'USING yinyan_song_records AS ysr1' in sql
     assert 'ysr2.song_id = ysr1.song_id' in sql
     assert 'ysr2.record_id = ysr1.record_id' in sql
+    assert 'is_yinyan_push' not in sql
+
+
+def test_fetch_yinyan_song_ids_reads_all_records1_song_ids():
+    cur = MagicMock()
+    cur.fetchall.return_value = [(10,), (11,)]
+
+    song_ids = fetch_yinyan_song_ids(cur)
+
+    sql = cur.execute.call_args[0][0]
+    assert 'SELECT DISTINCT song_id' in sql
+    assert 'WHERE' not in sql
+    assert song_ids == [10, 11]
 
 
 def test_fetch_yinyan_records_missing_platform_reads_null_platform_rows():