feat(etl): 优化录音数据选择并新增引擎歌曲记录写入

- 调整批处理大小为3以适应新需求
- 扩展fetch_platform_records查询字段，增加录音优先级排序机制
- 新增select_primary_record方法以选出最优录音记录
- 修改runner确保每首歌只写入一个主要引擎录音关联
- 新增upsert_yinyan_song_records接口，实现引擎歌曲记录的批量写入和替换
- 添加对应单元测试覆盖录音选择逻辑和写入功能

@@ -43,4 +43,4 @@ PLATFORM_KUGOU = '2'
 PLATFORM_NETEASE = '4'
 PLATFORM_NETEASE = '4'
 PLATFORMS = [PLATFORM_QQ, PLATFORM_KUGOU, PLATFORM_NETEASE]
 PLATFORMS = [PLATFORM_QQ, PLATFORM_KUGOU, PLATFORM_NETEASE]
 
 
-BATCH_SIZE = 500
+BATCH_SIZE = 3

@@ -18,11 +18,14 @@ LIMIT %s OFFSET %s
 _PLATFORM_QUERY = """
 _PLATFORM_QUERY = """
 SELECT
 SELECT
     sar.song_id AS source_song_id,
     sar.song_id AS source_song_id,
+    mr.id AS record_id,
     mr.platform,
     mr.platform,
     mr.platform_unique_key,
     mr.platform_unique_key,
     mr.platform_mid,
     mr.platform_mid,
     mr.album_audio_id,
     mr.album_audio_id,
-    sar.is_main_version
+    sar.is_main_version,
+    mr.is_high,
+    mr.pub_time
 FROM hk_song_and_record sar
 FROM hk_song_and_record sar
 JOIN hk_music_record mr ON mr.id = sar.record_id
 JOIN hk_music_record mr ON mr.id = sar.record_id
 WHERE sar.song_id IN ({placeholders})
 WHERE sar.song_id IN ({placeholders})
@@ -30,7 +33,12 @@ WHERE sar.song_id IN ({placeholders})
   AND mr.platform_unique_key IS NOT NULL
   AND mr.platform_unique_key IS NOT NULL
   AND mr.platform_unique_key != ''
   AND mr.platform_unique_key != ''
   AND mr.deleted = 0
   AND mr.deleted = 0
-ORDER BY sar.song_id, mr.platform, sar.is_main_version DESC
+ORDER BY sar.song_id,
+         COALESCE(sar.is_main_version, 0) DESC,
+         COALESCE(mr.is_high, 0) DESC,
+         (mr.pub_time IS NULL) ASC,
+         mr.pub_time ASC,
+         mr.id ASC
 """
 """
 
 
 
 
@@ -57,10 +65,32 @@ def fetch_platform_records(source_conn: pymysql.Connection, song_ids: list[int])
         cur.execute(query, song_ids)
         cur.execute(query, song_ids)
         rows = cur.fetchall()
         rows = cur.fetchall()
 
 
-    # 每个 (source_song_id, platform) 只保留 is_main_version 最高的一条
+    return select_platform_records(rows)
+
+
+def select_platform_records(rows: list[dict]) -> list[dict]:
+    # 每个 (source_song_id, platform) 保留一条，用于继续导入多个平台的录音数据。
     seen = {}
     seen = {}
-    for row in rows:
+    for row in sorted(rows, key=_record_priority):
         key = (row['source_song_id'], row['platform'])
         key = (row['source_song_id'], row['platform'])
         if key not in seen:
         if key not in seen:
             seen[key] = row
             seen[key] = row
     return list(seen.values())
     return list(seen.values())
+
+
+def select_primary_record(rows: list[dict]) -> dict | None:
+    if not rows:
+        return None
+    return sorted(rows, key=_record_priority)[0]
+
+
+def _record_priority(row: dict) -> tuple:
+    pub_time = row.get('pub_time')
+    return (
+        row['source_song_id'],
+        -(int(row.get('is_main_version') or 0)),
+        -(int(row.get('is_high') or 0)),
+        pub_time is None,
+        pub_time or '',
+        int(row.get('record_id') or 0),
+    )

@@ -5,13 +5,14 @@ from tqdm import tqdm
 
 
 from .config import PLATFORM_QQ, PLATFORM_KUGOU, PLATFORM_NETEASE, BATCH_SIZE, OSS_CONFIG
 from .config import PLATFORM_QQ, PLATFORM_KUGOU, PLATFORM_NETEASE, BATCH_SIZE, OSS_CONFIG
 from .connections import get_hk_songs_conn, get_source_conn, get_spider_conn, get_pg_conn, get_oss_bucket
 from .connections import get_hk_songs_conn, get_source_conn, get_spider_conn, get_pg_conn, get_oss_bucket
-from .reader import iter_hk_songs_batches, fetch_platform_records
+from .reader import iter_hk_songs_batches, fetch_platform_records, select_primary_record
 from .spider import (
 from .spider import (
     fetch_qq_songs, fetch_qq_singers,
     fetch_qq_songs, fetch_qq_singers,
     fetch_kugou_songs, fetch_kugou_singers,
     fetch_kugou_songs, fetch_kugou_singers,
     fetch_netease_songs, fetch_netease_singers,
     fetch_netease_songs, fetch_netease_singers,
 )
 )
 from .writer import (
 from .writer import (
+    upsert_yinyan_song_records,
     upsert_qq_singers, upsert_qq_albums, upsert_qq_songs,
     upsert_qq_singers, upsert_qq_albums, upsert_qq_songs,
     upsert_qq_singer_songs, upsert_qq_singer_albums,
     upsert_qq_singer_songs, upsert_qq_singer_albums,
     upsert_kugou_singers, upsert_kugou_albums, upsert_kugou_songs,
     upsert_kugou_singers, upsert_kugou_albums, upsert_kugou_songs,
@@ -336,6 +337,8 @@ def run(platforms: list[str], max_batches: int | None = None) -> None:
                     src_id = int(hk_row['source_song_id']) if hk_row.get('source_song_id') else None
                     src_id = int(hk_row['source_song_id']) if hk_row.get('source_song_id') else None
                     if not src_id or src_id not in pr_by_song:
                     if not src_id or src_id not in pr_by_song:
                         continue
                         continue
+                    primary_record = select_primary_record(pr_by_song[src_id])
+                    wrote_yinyan_record = False
                     for pr in pr_by_song[src_id]:
                     for pr in pr_by_song[src_id]:
                         processor = _PROCESSORS.get(pr['platform'])
                         processor = _PROCESSORS.get(pr['platform'])
                         if not processor:
                         if not processor:
@@ -343,6 +346,9 @@ def run(platforms: list[str], max_batches: int | None = None) -> None:
                         pg_cur.execute('SAVEPOINT sp_song')
                         pg_cur.execute('SAVEPOINT sp_song')
                         try:
                         try:
                             result = processor(hk_row, pr, spider_conn, pg_cur, bucket, base_url)
                             result = processor(hk_row, pr, spider_conn, pg_cur, bucket, base_url)
+                            if result and primary_record and not wrote_yinyan_record:
+                                upsert_yinyan_song_records(pg_cur, [(src_id, int(primary_record['record_id']))])
+                                wrote_yinyan_record = True
                             pg_cur.execute('RELEASE SAVEPOINT sp_song')
                             pg_cur.execute('RELEASE SAVEPOINT sp_song')
                             total_ok += 1
                             total_ok += 1
                             if result:
                             if result:

 import uuid
 import uuid
 
 
 
 
+def upsert_yinyan_song_records(cur, pairs: list[tuple[int, int]]) -> None:
+    """pairs: [(source_song_id, hk_music_record_id), ...]"""
+    if not pairs:
+        return
+    cur.executemany(
+        "DELETE FROM yinyan_song_records WHERE song_id = %s",
+        [(song_id,) for song_id, _ in pairs],
+    )
+    cur.executemany(
+        """
+        INSERT INTO yinyan_song_records (song_id, record_id)
+        VALUES (%s, %s)
+        """,
+        pairs,
+    )
+
+
 # ─── QQ Music ────────────────────────────────────────────────────────────────
 # ─── QQ Music ────────────────────────────────────────────────────────────────
 
 
 def upsert_qq_singers(cur, singers: list[dict]) -> None:
 def upsert_qq_singers(cur, singers: list[dict]) -> None:

+from unittest.mock import MagicMock
+
+from etl_to_crawler.reader import fetch_platform_records, select_primary_record
+
+
+def _make_cursor(rows):
+    cur = MagicMock()
+    cur.__enter__ = MagicMock(return_value=cur)
+    cur.__exit__ = MagicMock(return_value=False)
+    cur.fetchall.return_value = rows
+    return cur
+
+
+def test_fetch_platform_records_keeps_one_record_per_song_and_platform():
+    conn = MagicMock()
+    conn.cursor.return_value = _make_cursor([
+        {
+            'source_song_id': 10,
+            'record_id': 100,
+            'platform': '1',
+            'platform_unique_key': 'qq-mid',
+            'platform_mid': '100',
+            'album_audio_id': None,
+            'is_main_version': 0,
+            'is_high': 1,
+            'pub_time': '2020-01-01',
+        },
+        {
+            'source_song_id': 10,
+            'record_id': 101,
+            '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',
+        },
+        {
+            'source_song_id': 10,
+            'record_id': 102,
+            'platform': '4',
+            'platform_unique_key': '300',
+            'platform_mid': None,
+            'album_audio_id': None,
+            'is_main_version': 1,
+            'is_high': 1,
+            'pub_time': '2022-01-01',
+        },
+    ])
+
+    result = fetch_platform_records(conn, [10])
+
+    assert len(result) == 3
+    assert {row['record_id'] for row in result} == {100, 101, 102}
+
+
+def test_select_primary_record_prefers_main_version_then_is_high():
+    rows = [
+        {
+            'source_song_id': 10,
+            'record_id': 100,
+            'platform': '1',
+            'platform_unique_key': 'qq-mid',
+            'platform_mid': '100',
+            'album_audio_id': None,
+            'is_main_version': 0,
+            'is_high': 1,
+            'pub_time': '2020-01-01',
+        },
+        {
+            'source_song_id': 10,
+            'record_id': 101,
+            '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',
+        },
+        {
+            'source_song_id': 10,
+            'record_id': 102,
+            'platform': '4',
+            'platform_unique_key': '300',
+            'platform_mid': None,
+            'album_audio_id': None,
+            'is_main_version': 1,
+            'is_high': 1,
+            'pub_time': '2022-01-01',
+        },
+    ]
+
+    assert select_primary_record(rows)['record_id'] == 102
+
+
+def test_select_primary_record_uses_earliest_pub_time_when_priority_ties():
+    rows = [
+        {
+            'source_song_id': 10,
+            'record_id': 100,
+            'platform': '1',
+            'platform_unique_key': 'qq-mid',
+            'platform_mid': '100',
+            'album_audio_id': None,
+            'is_main_version': 0,
+            'is_high': 0,
+            'pub_time': '2020-01-01',
+        },
+        {
+            'source_song_id': 10,
+            'record_id': 101,
+            'platform': '2',
+            'platform_unique_key': '200',
+            'platform_mid': 'kg-hash',
+            'album_audio_id': None,
+            'is_main_version': 0,
+            'is_high': 0,
+            'pub_time': '2019-01-01',
+        },
+    ]
+
+    assert select_primary_record(rows)['record_id'] == 101

+from unittest.mock import MagicMock
+
+from etl_to_crawler import runner
+
+
+class _Cursor:
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        return False
+
+    def execute(self, *args, **kwargs):
+        return None
+
+
+class _PgConnection:
+    def __init__(self):
+        self.cur = _Cursor()
+        self.commits = 0
+        self.closed = False
+
+    def cursor(self):
+        return self.cur
+
+    def commit(self):
+        self.commits += 1
+
+    def close(self):
+        self.closed = True
+
+
+class _Connection:
+    def close(self):
+        return None
+
+
+def test_run_imports_all_platform_records_but_writes_one_primary_yinyan_relation(monkeypatch):
+    pg_conn = _PgConnection()
+    processors = {
+        '1': MagicMock(return_value={'platform': 'qq', 'platform_song_id': 100, 'mid': 'qq-mid', 'title': '歌'}),
+        '2': MagicMock(return_value={'platform': 'kugou', 'platform_song_id': 200, 'hash': 'kg-hash', 'title': '歌'}),
+    }
+    yinyan_writer = MagicMock()
+
+    monkeypatch.setattr(runner, 'get_hk_songs_conn', lambda: _Connection())
+    monkeypatch.setattr(runner, 'get_source_conn', lambda: _Connection())
+    monkeypatch.setattr(runner, 'get_spider_conn', lambda: _Connection())
+    monkeypatch.setattr(runner, 'get_pg_conn', lambda: pg_conn)
+    monkeypatch.setattr(runner, 'get_oss_bucket', lambda: object())
+    monkeypatch.setattr(runner, 'iter_hk_songs_batches', lambda conn, batch_size: [[{
+        'source_song_id': 10,
+        'name': '歌',
+        'audio_url': 'https://example.com/a.mp3',
+        'singer': '歌手',
+    }]])
+    monkeypatch.setattr(runner, 'fetch_platform_records', lambda conn, song_ids: [
+        {
+            'source_song_id': 10,
+            'record_id': 100,
+            'platform': '1',
+            'platform_unique_key': 'qq-mid',
+            'platform_mid': '100',
+            'album_audio_id': None,
+            'is_main_version': 0,
+            'is_high': 1,
+            'pub_time': '2020-01-01',
+        },
+        {
+            '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',
+        },
+    ])
+    monkeypatch.setattr(runner, '_PROCESSORS', processors)
+    monkeypatch.setattr(runner, 'upsert_yinyan_song_records', yinyan_writer)
+
+    runner.run(['1', '2'])
+
+    processors['1'].assert_called_once()
+    processors['2'].assert_called_once()
+    yinyan_writer.assert_called_once_with(pg_conn.cur, [(10, 200)])
+    assert pg_conn.commits == 1

 from unittest.mock import MagicMock, call
 from unittest.mock import MagicMock, call
-from etl_to_crawler.writer import upsert_qq_singers, upsert_qq_singer_songs
+from etl_to_crawler.writer import (
+    upsert_qq_singers,
+    upsert_qq_singer_songs,
+    upsert_yinyan_song_records,
+)
 
 
 
 
 def test_upsert_qq_singers_executes_insert():
 def test_upsert_qq_singers_executes_insert():
@@ -27,3 +31,17 @@ def test_upsert_qq_singer_songs():
     assert cur.executemany.called
     assert cur.executemany.called
     sql = cur.executemany.call_args[0][0]
     sql = cur.executemany.call_args[0][0]
     assert 'crawler_qqmusic_singer_songs' in sql
     assert 'crawler_qqmusic_singer_songs' in sql
+
+
+def test_upsert_yinyan_song_records_replaces_existing_song_relation():
+    cur = MagicMock()
+    upsert_yinyan_song_records(cur, [(10, 100), (11, 101)])
+
+    assert cur.executemany.call_count == 2
+    delete_sql, delete_rows = cur.executemany.call_args_list[0][0]
+    insert_sql, insert_rows = cur.executemany.call_args_list[1][0]
+    assert 'DELETE FROM yinyan_song_records' in delete_sql
+    assert 'WHERE song_id = %s' in delete_sql
+    assert delete_rows == [(10,), (11,)]
+    assert 'INSERT INTO yinyan_song_records' in insert_sql
+    assert insert_rows == [(10, 100), (11, 101)]