fix(etl_to_crawler): 解决同批次歌曲UUID冲突及关联关系处理问题

@@ -442,7 +442,8 @@ def _prepare_qq_payload(hk_row: dict, pr: dict, bucket, base_url, sp: dict, sing
         'singers': singer_rows,
         'singers': singer_rows,
         'albums': album_rows,
         'albums': album_rows,
         'songs': [song_row],
         'songs': [song_row],
-        'singer_songs': [(sg['singer_id'], song_uuid) for sg in singer_list],
+        # 关联阶段统一按平台歌曲 ID 解析数据库中的实际 UUID，兼容 songs 冲突跳过。
+        'singer_songs': [(sg['singer_id'], platform_song_id) for sg in singer_list],
         'singer_albums': [(sg['singer_id'], album_id) for sg in singer_list] if album_id else [],
         'singer_albums': [(sg['singer_id'], album_id) for sg in singer_list] if album_id else [],
     }
     }
 
 
@@ -554,7 +555,7 @@ def _prepare_kugou_payload(hk_row: dict, pr: dict, bucket, base_url, sp: dict, s
         'singers': singer_rows,
         'singers': singer_rows,
         'albums': album_rows,
         'albums': album_rows,
         'songs': [song_row],
         'songs': [song_row],
-        'singer_songs': [(sg['singer_id'], song_uuid) for sg in singer_list],
+        'singer_songs': [(sg['singer_id'], song_id) for sg in singer_list],
         'singer_albums': [(sg['singer_id'], album_id) for sg in singer_list] if album_id else [],
         'singer_albums': [(sg['singer_id'], album_id) for sg in singer_list] if album_id else [],
     }
     }
 
 
@@ -678,7 +679,7 @@ def _prepare_netease_payload(hk_row: dict, pr: dict, bucket, base_url, sp: dict,
         'singers': singer_rows,
         'singers': singer_rows,
         'albums': album_rows,
         'albums': album_rows,
         'songs': [song_row],
         'songs': [song_row],
-        'singer_songs': [(sg['singer_id'], song_uuid) for sg in singer_list],
+        'singer_songs': [(sg['singer_id'], song_id) for sg in singer_list],
         'singer_albums': [(sg['singer_id'], album_id) for sg in singer_list] if album_id else [],
         'singer_albums': [(sg['singer_id'], album_id) for sg in singer_list] if album_id else [],
     }
     }
 
 
@@ -731,6 +732,57 @@ def _extend_platform_payload(target: dict, payload: dict) -> None:
     target['singer_albums'].extend(payload.get('singer_albums', []))
     target['singer_albums'].extend(payload.get('singer_albums', []))
 
 
 
 
+_CRAWLER_SONG_TABLES = {
+    'crawler_qqmusic_songs',
+    'crawler_kugou_songs',
+    'crawler_netease_songs',
+}
+
+
+def _dedupe_songs_by_platform_id(songs: list[dict]) -> list[dict]:
+    """同批相同平台歌曲只尝试插入一次，保留首次出现的数据。"""
+    seen = set()
+    deduped = []
+    for song in songs:
+        platform_song_id = int(song['platform_song_id'])
+        if platform_song_id not in seen:
+            seen.add(platform_song_id)
+            deduped.append(song)
+    return deduped
+
+
+def _resolve_singer_song_pairs(
+    cur,
+    songs_table: str,
+    platform_pairs: list[tuple[int, int]],
+) -> list[tuple[int, str]]:
+    """将 (singer_id, platform_song_id) 解析为关联表需要的实际歌曲 UUID。"""
+    if not platform_pairs:
+        return []
+    if songs_table not in _CRAWLER_SONG_TABLES:
+        raise ValueError(f'unsupported crawler songs table: {songs_table}')
+
+    platform_song_ids = sorted({int(platform_song_id) for _, platform_song_id in platform_pairs})
+    placeholders = ', '.join(['%s'] * len(platform_song_ids))
+    cur.execute(
+        f'SELECT platform_song_id, id FROM {songs_table} '
+        f'WHERE platform_song_id IN ({placeholders})',
+        tuple(platform_song_ids),
+    )
+    actual_ids = {int(row[0]): str(row[1]) for row in cur.fetchall()}
+    missing = sorted(set(platform_song_ids) - actual_ids.keys())
+    if missing:
+        raise RuntimeError(
+            f'cannot resolve actual song UUIDs from {songs_table}: platform_song_ids={missing}'
+        )
+
+    # dict 保序去重，同一歌手与同一实际歌曲只写一次关系。
+    return list(dict.fromkeys(
+        (int(singer_id), actual_ids[int(platform_song_id)])
+        for singer_id, platform_song_id in platform_pairs
+    ))
+
+
 def _write_import_payloads(pg_cur, payloads: list[dict]) -> list[dict]:
 def _write_import_payloads(pg_cur, payloads: list[dict]) -> list[dict]:
     if not payloads:
     if not payloads:
         return []
         return []
@@ -747,24 +799,36 @@ def _write_import_payloads(pg_cur, payloads: list[dict]) -> list[dict]:
         imported.append(payload['result'])
         imported.append(payload['result'])
 
 
     qq = grouped[PLATFORM_QQ]
     qq = grouped[PLATFORM_QQ]
+    qq['songs'] = _dedupe_songs_by_platform_id(qq['songs'])
     upsert_qq_singers(pg_cur, qq['singers'])
     upsert_qq_singers(pg_cur, qq['singers'])
     upsert_qq_albums(pg_cur, qq['albums'])
     upsert_qq_albums(pg_cur, qq['albums'])
     upsert_qq_songs(pg_cur, qq['songs'])
     upsert_qq_songs(pg_cur, qq['songs'])
-    upsert_qq_singer_songs(pg_cur, qq['singer_songs'])
+    qq_singer_songs = _resolve_singer_song_pairs(
+        pg_cur, 'crawler_qqmusic_songs', qq['singer_songs'],
+    )
+    upsert_qq_singer_songs(pg_cur, qq_singer_songs)
     upsert_qq_singer_albums(pg_cur, qq['singer_albums'])
     upsert_qq_singer_albums(pg_cur, qq['singer_albums'])
 
 
     kugou = grouped[PLATFORM_KUGOU]
     kugou = grouped[PLATFORM_KUGOU]
+    kugou['songs'] = _dedupe_songs_by_platform_id(kugou['songs'])
     upsert_kugou_singers(pg_cur, kugou['singers'])
     upsert_kugou_singers(pg_cur, kugou['singers'])
     upsert_kugou_albums(pg_cur, kugou['albums'])
     upsert_kugou_albums(pg_cur, kugou['albums'])
     upsert_kugou_songs(pg_cur, kugou['songs'])
     upsert_kugou_songs(pg_cur, kugou['songs'])
-    upsert_kugou_singer_songs(pg_cur, kugou['singer_songs'])
+    kugou_singer_songs = _resolve_singer_song_pairs(
+        pg_cur, 'crawler_kugou_songs', kugou['singer_songs'],
+    )
+    upsert_kugou_singer_songs(pg_cur, kugou_singer_songs)
     upsert_kugou_singer_albums(pg_cur, kugou['singer_albums'])
     upsert_kugou_singer_albums(pg_cur, kugou['singer_albums'])
 
 
     netease = grouped[PLATFORM_NETEASE]
     netease = grouped[PLATFORM_NETEASE]
+    netease['songs'] = _dedupe_songs_by_platform_id(netease['songs'])
     upsert_netease_singers(pg_cur, netease['singers'])
     upsert_netease_singers(pg_cur, netease['singers'])
     upsert_netease_albums(pg_cur, netease['albums'])
     upsert_netease_albums(pg_cur, netease['albums'])
     upsert_netease_songs(pg_cur, netease['songs'])
     upsert_netease_songs(pg_cur, netease['songs'])
-    upsert_netease_singer_songs(pg_cur, netease['singer_songs'])
+    netease_singer_songs = _resolve_singer_song_pairs(
+        pg_cur, 'crawler_netease_songs', netease['singer_songs'],
+    )
+    upsert_netease_singer_songs(pg_cur, netease_singer_songs)
     upsert_netease_singer_albums(pg_cur, netease['singer_albums'])
     upsert_netease_singer_albums(pg_cur, netease['singer_albums'])
 
 
     upsert_yinyan_song_records(pg_cur, yinyan_records)
     upsert_yinyan_song_records(pg_cur, yinyan_records)

@@ -241,6 +241,79 @@ def test_records2_selection_accepts_current_record_with_empty_cover_and_lyric():
     assert reason == 'ok'
     assert reason == 'ok'
 
 
 
 
+@pytest.mark.parametrize(
+    ('platform', 'songs_table', 'song_upsert_name', 'relation_upsert_name'),
+    [
+        (runner.PLATFORM_QQ, 'crawler_qqmusic_songs', 'upsert_qq_songs', 'upsert_qq_singer_songs'),
+        (runner.PLATFORM_KUGOU, 'crawler_kugou_songs', 'upsert_kugou_songs', 'upsert_kugou_singer_songs'),
+        (runner.PLATFORM_NETEASE, 'crawler_netease_songs', 'upsert_netease_songs', 'upsert_netease_singer_songs'),
+    ],
+)
+def test_write_import_payloads_resolves_existing_song_uuid_for_all_platforms(
+    monkeypatch, platform, songs_table, song_upsert_name, relation_upsert_name,
+):
+    cur = MagicMock()
+    cur.fetchall.return_value = [(200, 'existing-song-uuid')]
+    song_upsert = MagicMock()
+    relation_upsert = MagicMock()
+    yinyan_upsert = MagicMock()
+    monkeypatch.setattr(runner, song_upsert_name, song_upsert)
+    monkeypatch.setattr(runner, relation_upsert_name, relation_upsert)
+    monkeypatch.setattr(runner, 'upsert_yinyan_song_records', yinyan_upsert)
+
+    runner._write_import_payloads(cur, [{
+        'platform': platform,
+        'platform_song_id': 200,
+        'result': {'platform_song_id': 200},
+        'yinyan_record': {
+            'song_id': 10, 'record_id': 20, 'platform': platform, 'platform_song_id': 200,
+        },
+        'singers': [],
+        'albums': [],
+        'songs': [
+            {'song_uuid': 'temporary-uuid-1', 'platform_song_id': 200},
+            {'song_uuid': 'temporary-uuid-2', 'platform_song_id': 200},
+        ],
+        'singer_songs': [(11, 200), (11, 200), (12, 200)],
+        'singer_albums': [],
+    }])
+
+    # 同批平台歌曲去重，且全部歌手关系使用数据库实际保留的歌曲 UUID。
+    assert len(song_upsert.call_args.args[1]) == 1
+    relation_upsert.assert_called_once_with(
+        cur, [(11, 'existing-song-uuid'), (12, 'existing-song-uuid')],
+    )
+    sql, params = cur.execute.call_args.args
+    assert f'FROM {songs_table}' in sql
+    assert params == (200,)
+    yinyan_upsert.assert_called_once()
+
+
+def test_write_import_payloads_does_not_mark_yinyan_success_when_song_uuid_is_missing(monkeypatch):
+    cur = MagicMock()
+    cur.fetchall.return_value = []
+    yinyan_upsert = MagicMock()
+    monkeypatch.setattr(runner, 'upsert_yinyan_song_records', yinyan_upsert)
+
+    with pytest.raises(RuntimeError, match=r'platform_song_ids=\[200\]'):
+        runner._write_import_payloads(cur, [{
+            'platform': runner.PLATFORM_KUGOU,
+            'platform_song_id': 200,
+            'result': {'platform_song_id': 200},
+            'yinyan_record': {
+                'song_id': 10, 'record_id': 20, 'platform': runner.PLATFORM_KUGOU,
+                'platform_song_id': 200,
+            },
+            'singers': [],
+            'albums': [],
+            'songs': [{'song_uuid': 'temporary-uuid', 'platform_song_id': 200}],
+            'singer_songs': [(11, 200)],
+            'singer_albums': [],
+        }])
+
+    yinyan_upsert.assert_not_called()
+
+
 def test_run_imports_only_pending_yinyan_platform_record(monkeypatch):
 def test_run_imports_only_pending_yinyan_platform_record(monkeypatch):
     pg_conn = _PgConnection()
     pg_conn = _PgConnection()
     prepare = MagicMock(return_value={
     prepare = MagicMock(return_value={