refactor(runner): 移除 spider DB 歌手关联校验，简化 records2 初始化逻辑

沈秋雨
Commit 7a8fed26 ... 7a8fed26c310df80e77422db55d47c9077b986ca authored 2026-07-14 10:55:54 +0800 by 沈秋雨
Showing 2 changed files with 60 additions and 75 deletions
etl_to_crawler/runner.py
tests/test_runner.py
--- a/etl_to_crawler/runner.py
View file @7a8fed2
+++ b/etl_to_crawler/runner.py
View file @7a8fed2
@@ -1409,54 +1409,15 @@ def initialize_yinyan_song_records(platforms: list[str], max_batches: int | None
    log.info("Initialized yinyan_song_records candidates=%d, skipped_batches=%d", total, skipped_batches)
-def _row_singer_key(row: dict) -> str | int:
-    """返回一条 relation 用于匹配歌手的 key：QQ 用 mid（str），其他平台用 int id。"""
-    if row['platform'] == PLATFORM_QQ:
-        return row.get('platform_unique_key', '')
-    return int(row.get('platform_unique_key', 0))
-def _fetch_valid_singer_keys(spider_conn, rows: list[dict]) -> set:
-    """查询 spider DB，返回有有效歌手关联的 platform_unique_key 集合。"""
-    qq_mids = [r.get('platform_unique_key') for r in rows if r['platform'] == PLATFORM_QQ and r.get('platform_unique_key')]
-    kugou_ids = [int(r['platform_unique_key']) for r in rows if r['platform'] == PLATFORM_KUGOU and r.get('platform_unique_key')]
-    netease_ids = [int(r['platform_unique_key']) for r in rows if r['platform'] == PLATFORM_NETEASE and r.get('platform_unique_key')]
-    valid: set = set()
-    if qq_mids:
-        songs = fetch_qq_songs(spider_conn, qq_mids)
-        db_ids = [v['id'] for v in songs.values()]
-        singers = fetch_qq_singers(spider_conn, db_ids) if db_ids else {}
-        # singers 以 spider DB song_id 为 key，反查回 mid
-        id_to_mid = {v['id']: mid for mid, v in songs.items()}
-        for song_id in singers:
-            mid = id_to_mid.get(song_id)
-            if mid:
-                valid.add(mid)
-    if kugou_ids:
-        songs = fetch_kugou_songs(spider_conn, kugou_ids)
-        singers = fetch_kugou_singers(spider_conn, list(songs.keys()))
-        valid.update(int(k) for k in singers)
-    if netease_ids:
-        songs = fetch_netease_songs(spider_conn, netease_ids)
-        singers = fetch_netease_singers(spider_conn, list(songs.keys()))
-        valid.update(int(k) for k in singers)
-    return valid
 def initialize_yinyan_song_records2(platforms: list[str], max_batches: int | None = None) -> None:
    """以 records1 全量 song_id 为范围，写入补充录音关联并去重。
-    初始化阶段即校验 spider DB 歌手关联，过滤掉歌手数据缺失的录音。
+    初始化只校验音眼录音的必要元数据，不以 spider 歌手数据作为准入条件；
+    后续导入允许歌曲的 singers 为空。
    """
    src_conn = get_source_conn()
-    spider_conn = get_spider_conn()
    pg_conn = get_pg_conn()
    total_inserted = 0
-    total_filtered_no_singer = 0
    try:
        with pg_conn.cursor() as pg_cur:
@@ -1466,7 +1427,6 @@ def initialize_yinyan_song_records2(platforms: list[str], max_batches: int | Non
        for index, start in enumerate(tqdm(range(0, len(song_ids), BATCH_SIZE), desc='init-yinyan-records2')):
            # 长任务连接可能断开，每批次开始前刷新
            src_conn = refresh_conn(src_conn, 'source')
-            spider_conn = refresh_conn(spider_conn, 'spider')
            pg_conn = refresh_conn(pg_conn, 'pg')
            if max_batches is not None and index >= max_batches:
@@ -1479,7 +1439,6 @@ def initialize_yinyan_song_records2(platforms: list[str], max_batches: int | Non
                    'song_id': int(relation['source_song_id']),
                    'record_id': int(relation['record_id']),
                    'platform': str(relation['platform']),
-                    'platform_unique_key': relation.get('platform_unique_key', ''),
                }
                for relation in relations
                if str(relation['platform']) in platforms
@@ -1487,20 +1446,6 @@ def initialize_yinyan_song_records2(platforms: list[str], max_batches: int | Non
            if not rows:
                continue
-            # ── 校验 spider DB 歌手关联，过滤无效录音 ──
-            valid_keys = _fetch_valid_singer_keys(spider_conn, rows)
-            before = len(rows)
-            rows = [
-                r for r in rows
-                if _row_singer_key(r) in valid_keys
-            ]
-            filtered = before - len(rows)
-            total_filtered_no_singer += filtered
-            if filtered:
-                log.info("init-records2 batch %d: filtered %d rows without valid singers", index, filtered)
-            if not rows:
-                continue
            with pg_conn.cursor() as pg_cur:
                insert_yinyan_song_records2(pg_cur, rows)
            pg_conn.commit()
@@ -1514,11 +1459,9 @@ def initialize_yinyan_song_records2(platforms: list[str], max_batches: int | Non
        close_all_pools()
    log.info(
-        "Initialized yinyan_song_records2 candidates=%d, removed_existing_records1_relations=%d, "
+        "Initialized yinyan_song_records2 candidates=%d, removed_existing_records1_relations=%d",
-        "filtered_no_singer=%d",
        total_inserted,
        deleted,
-        total_filtered_no_singer,
    )
--- a/tests/test_runner.py
View file @7a8fed2
+++ b/tests/test_runner.py
View file @7a8fed2
@@ -244,6 +244,44 @@ def test_records2_selection_accepts_current_record_with_empty_cover_and_lyric():
    assert reason == 'ok'
+def test_records2_prepare_payload_allows_empty_singers(monkeypatch):
+    bucket = object()
+    preparer = MagicMock(return_value={
+        'platform': runner.PLATFORM_QQ,
+        'platform_song_id': 100,
+        'result': {'platform': 'qq', 'platform_song_id': 100},
+        'singers': [],
+        'albums': [],
+        'songs': [{'platform_song_id': 100, 'singers_json': '[]'}],
+        'singer_songs': [],
+        'singer_albums': [],
+    })
+    monkeypatch.setattr(runner, 'YINYAN_IMPORT_TABLE', 'yinyan_song_records2')
+    monkeypatch.setitem(runner._PREPARERS, runner.PLATFORM_QQ, preparer)
+    payload = runner._prepare_import_payload(
+        {'song_id': 10, 'record_id': 20, 'platform': runner.PLATFORM_QQ},
+        {'name': '歌曲'},
+        {'record_id': 20, 'platform_unique_key': 'qq-mid'},
+        bucket,
+        'https://oss.example.com',
+        {runner.PLATFORM_QQ: {'qq-mid': {'id': 100}}},
+        {runner.PLATFORM_QQ: {}},
+    )
+    assert payload is not None
+    assert payload['singers'] == []
+    assert payload['singer_songs'] == []
+    preparer.assert_called_once_with(
+        {'name': '歌曲'},
+        {'record_id': 20, 'platform_unique_key': 'qq-mid'},
+        bucket,
+        'https://oss.example.com',
+        {'id': 100},
+        [],
+    )
 @pytest.mark.parametrize(
    ('platform', 'songs_table', 'song_upsert_name', 'relation_upsert_name'),
    [
@@ -463,7 +501,6 @@ def test_initialize_yinyan_song_records2_writes_all_relations_then_deduplicates(
    dedupe_calls = []
    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, 'refresh_conn', lambda conn, name: conn)
    monkeypatch.setattr(runner, 'fetch_yinyan_song_ids', lambda cur: [10])
@@ -471,7 +508,6 @@ def test_initialize_yinyan_song_records2_writes_all_relations_then_deduplicates(
        {'source_song_id': 10, 'record_id': 100, 'platform': '1', 'platform_unique_key': 'mid100'},
        {'source_song_id': 10, 'record_id': 200, 'platform': '2', 'platform_unique_key': '200'},
    ])
-    monkeypatch.setattr(runner, '_fetch_valid_singer_keys', lambda conn, rows: {'mid100', 200})
    monkeypatch.setattr(runner, 'insert_yinyan_song_records2', lambda cur, rows: inserted.extend(rows))
    monkeypatch.setattr(
        runner,
@@ -482,19 +518,20 @@ def test_initialize_yinyan_song_records2_writes_all_relations_then_deduplicates(
    runner.initialize_yinyan_song_records2(['1', '2'])
    assert inserted == [
-        {'song_id': 10, 'record_id': 100, 'platform': '1', 'platform_unique_key': 'mid100'},
+        {'song_id': 10, 'record_id': 100, 'platform': '1'},
-        {'song_id': 10, 'record_id': 200, 'platform': '2', 'platform_unique_key': '200'},
+        {'song_id': 10, 'record_id': 200, 'platform': '2'},
    ]
    assert dedupe_calls == [True]
    assert pg_conn.commits == 2
-def test_initialize_yinyan_song_records2_filters_rows_without_valid_singers(monkeypatch):
+def test_initialize_yinyan_song_records2_keeps_rows_without_valid_singers(monkeypatch):
    pg_conn = _PgConnection()
    inserted = []
    monkeypatch.setattr(runner, 'get_source_conn', lambda: _Connection())
-    monkeypatch.setattr(runner, 'get_spider_conn', lambda: _Connection())
+    get_spider_conn = MagicMock(side_effect=AssertionError('records2 init must not query spider'))
+    monkeypatch.setattr(runner, 'get_spider_conn', get_spider_conn)
    monkeypatch.setattr(runner, 'get_pg_conn', lambda: pg_conn)
    monkeypatch.setattr(runner, 'refresh_conn', lambda conn, name: conn)
    monkeypatch.setattr(runner, 'fetch_yinyan_song_ids', lambda cur: [10])
@@ -503,8 +540,6 @@ def test_initialize_yinyan_song_records2_filters_rows_without_valid_singers(monk
        {'source_song_id': 10, 'record_id': 200, 'platform': '1', 'platform_unique_key': 'mid_bad'},
        {'source_song_id': 10, 'record_id': 300, 'platform': '2', 'platform_unique_key': '300'},
    ])
-    # mid_bad 没有歌手关联，300 也没有
-    monkeypatch.setattr(runner, '_fetch_valid_singer_keys', lambda conn, rows: {'mid_good'})
    monkeypatch.setattr(runner, 'insert_yinyan_song_records2', lambda cur, rows: inserted.extend(rows))
    monkeypatch.setattr(
        runner, 'delete_yinyan_song_records2_existing_relations', lambda cur: 0,
@@ -512,10 +547,13 @@ def test_initialize_yinyan_song_records2_filters_rows_without_valid_singers(monk
    runner.initialize_yinyan_song_records2(['1', '2'])
-    # 只有 mid_good 的录音被保留
+    # 初始化不查询 spider，所有满足音眼必要字段的录音都应保留。
    assert inserted == [
-        {'song_id': 10, 'record_id': 100, 'platform': '1', 'platform_unique_key': 'mid_good'},
+        {'song_id': 10, 'record_id': 100, 'platform': '1'},
+        {'song_id': 10, 'record_id': 200, 'platform': '1'},
+        {'song_id': 10, 'record_id': 300, 'platform': '2'},
    ]
+    get_spider_conn.assert_not_called()
 def test_backfill_yinyan_record_platforms_updates_missing_platform_rows(monkeypatch):
@@ -597,7 +635,11 @@ def test_process_qq_builds_album_json_for_song_insert(monkeypatch):
            'issue_time': '2019-01-01',
            'song_time': 120,
        },
-        {'platform_unique_key': 'qq-mid', 'platform_mid': '200'},
+        {
+            'platform_unique_key': 'qq-mid',
+            'platform_mid': '200',
+            'record_name': '化风行万里 (DJ默涵版)',
+        },
        spider_conn=object(),
        pg_cur=pg_cur,
        bucket=object(),
@@ -665,7 +707,7 @@ def test_process_netease_builds_album_json_for_song_insert(monkeypatch):
            'issue_time': '2019-01-01',
            'song_time': 120,
        },
-        {'platform_unique_key': '300'},
+        {'platform_unique_key': '300', 'record_name': '化风行万里 (DJ默涵版)'},
        spider_conn=object(),
        pg_cur=pg_cur,
        bucket=object(),
@@ -710,7 +752,7 @@ def test_process_netease_uses_prefetched_song_and_singers(monkeypatch):
            'issue_time': '2019-01-01',
            'song_time': 120,
        },
-        {'platform_unique_key': '300'},
+        {'platform_unique_key': '300', 'record_name': '录音标题'},
        spider_conn=object(),
        pg_cur=pg_cur,
        bucket=object(),
@@ -789,7 +831,7 @@ def test_process_netease_keeps_timestamped_lyric_and_uploads_plain_lyric(monkeyp
            'issue_time': '2019-01-01',
            'song_time': 120,
        },
-        {'platform_unique_key': '300'},
+        {'platform_unique_key': '300', 'record_name': '录音标题'},
        spider_conn=object(),
        pg_cur=pg_cur,
        bucket=Bucket(),