feat(etl): 添加回填三平台空singers字段歌曲记录功能

- run_etl.py中新增--backfill-empty-singers参数支持回填操作
- runner.py新增backfill_empty_singers主流程函数及qq、酷狗、网易分平台回填实现
- writer.py新增查询缺失singers的歌曲及更新singers字段的数据库操作函数
- 回填过程中处理歌手头像的转存及歌手与歌曲关联关系的维护
- 支持批量分批次处理，避免长时间阻塞与重复无效查询
- 完整回填流程整合蜘蛛库和pg库连接资源管理，并日志记录处理结果

@@ -34,6 +34,8 @@ from .writer import (
     upsert_kugou_singer_songs, upsert_kugou_singer_albums,
     upsert_netease_singers, upsert_netease_albums, upsert_netease_songs,
     upsert_netease_singer_songs, upsert_netease_singer_albums,
+    fetch_qq_songs_missing_singers, fetch_kugou_songs_missing_singers, fetch_netease_songs_missing_singers,
+    update_qq_song_singers, update_kugou_song_singers, update_netease_song_singers,
 )
 from .oss import transfer_url, transfer_url_with_md5, build_oss_key
 from .utils import split_title_version, upload_plain_lyric_to_bucket
@@ -1105,6 +1107,239 @@ def backfill_yinyan_record_platforms(max_batches: int | None = None) -> None:
     log.info("Backfilled yinyan_song_records platform rows=%d, skipped=%d", total, skipped)
 
 
+def _backfill_qq_singers(spider_conn, pg_conn, bucket, base_url, max_batches):
+    batch_index = 0
+    total_ok = total_skip = 0
+    pbar = tqdm(desc='backfill-qq-singers')
+    while max_batches is None or batch_index < max_batches:
+        with pg_conn.cursor() as pg_cur:
+            rows = fetch_qq_songs_missing_singers(pg_cur, BATCH_SIZE)
+        if not rows:
+            break
+
+        mids = [row['mid'] for row in rows]
+        qq_songs_map = fetch_qq_songs(spider_conn, mids)
+        qq_db_ids = [v['id'] for v in qq_songs_map.values()]
+        qq_singers_map = fetch_qq_singers(spider_conn, qq_db_ids) if qq_db_ids else {}
+
+        singer_by_id: dict[int, dict] = {}
+        for song_data in qq_songs_map.values():
+            for sg in qq_singers_map.get(song_data['id'], []):
+                singer_by_id[sg['singer_id']] = sg
+
+        avatar_tasks = {
+            f"avatar:{sg['singer_id']}": (
+                lambda sg=sg: _safe_transfer(
+                    sg.get('avatar', ''),
+                    build_oss_key('qq', 'singer', sg['mid'] + '.jpg'),
+                    bucket, base_url,
+                )
+            )
+            for sg in singer_by_id.values()
+        }
+        avatar_assets = _run_io_tasks(avatar_tasks)
+
+        singer_rows = [{
+            **sg,
+            'id': sg['singer_id'],
+            'avatar': avatar_assets.get(f"avatar:{sg['singer_id']}", sg.get('avatar', '')),
+            'provider_name': PROVIDER_YINYAN,
+            'crawler_source_data': _source_json(sg),
+        } for sg in singer_by_id.values()]
+
+        song_updates = []
+        singer_song_pairs = []
+        for row in rows:
+            song_data = qq_songs_map.get(row['mid'])
+            if not song_data:
+                total_skip += 1
+                continue
+            singer_list = qq_singers_map.get(song_data['id'], [])
+            if not singer_list:
+                total_skip += 1
+                continue
+            singers_json = json.dumps([{
+                'name': sg['name'],
+                'singer_id': sg['singer_id'],
+                'platform_singer_id': sg['mid'],
+            } for sg in singer_list], ensure_ascii=False)
+            song_updates.append({'platform_song_id': row['platform_song_id'], 'singers_json': singers_json})
+            singer_song_pairs.extend([(sg['singer_id'], row['id']) for sg in singer_list])
+            total_ok += 1
+
+        if not song_updates:
+            log.warning("QQ singers backfill: no singer data found for this batch, stopping to avoid retry loop")
+            break
+
+        with pg_conn.cursor() as pg_cur:
+            upsert_qq_singers(pg_cur, singer_rows)
+            update_qq_song_singers(pg_cur, song_updates)
+            upsert_qq_singer_songs(pg_cur, singer_song_pairs)
+        pg_conn.commit()
+
+        batch_index += 1
+        pbar.update(1)
+    pbar.close()
+    log.info("QQ singers backfill: ok=%d skip=%d", total_ok, total_skip)
+
+
+def _backfill_kugou_singers(spider_conn, pg_conn, bucket, base_url, max_batches):
+    batch_index = 0
+    total_ok = total_skip = 0
+    pbar = tqdm(desc='backfill-kugou-singers')
+    while max_batches is None or batch_index < max_batches:
+        with pg_conn.cursor() as pg_cur:
+            rows = fetch_kugou_songs_missing_singers(pg_cur, BATCH_SIZE)
+        if not rows:
+            break
+
+        song_ids = [row['platform_song_id'] for row in rows]
+        kugou_singers_map = fetch_kugou_singers(spider_conn, song_ids)
+
+        singer_by_id: dict[int, dict] = {}
+        for sgs in kugou_singers_map.values():
+            for sg in sgs:
+                singer_by_id[sg['singer_id']] = sg
+
+        avatar_tasks = {
+            f"avatar:{sg['singer_id']}": (
+                lambda sg=sg: _safe_transfer(
+                    sg.get('avatar', ''),
+                    build_oss_key('kugou', 'singer', str(sg['singer_id']) + '.jpg'),
+                    bucket, base_url,
+                )
+            )
+            for sg in singer_by_id.values()
+        }
+        avatar_assets = _run_io_tasks(avatar_tasks)
+
+        singer_rows = [{
+            **sg,
+            'id': sg['singer_id'],
+            'avatar': avatar_assets.get(f"avatar:{sg['singer_id']}", sg.get('avatar', '')),
+            'provider_name': PROVIDER_YINYAN,
+            'crawler_source_data': _source_json(sg),
+        } for sg in singer_by_id.values()]
+
+        song_updates = []
+        singer_song_pairs = []
+        for row in rows:
+            singer_list = kugou_singers_map.get(row['platform_song_id'], [])
+            if not singer_list:
+                total_skip += 1
+                continue
+            singers_json = json.dumps([{
+                'name': sg['name'],
+                'singer_id': sg['singer_id'],
+                'platform_singer_id': str(sg['singer_id']),
+            } for sg in singer_list], ensure_ascii=False)
+            song_updates.append({'platform_song_id': row['platform_song_id'], 'singers_json': singers_json})
+            singer_song_pairs.extend([(sg['singer_id'], row['id']) for sg in singer_list])
+            total_ok += 1
+
+        if not song_updates:
+            log.warning("Kugou singers backfill: no singer data found for this batch, stopping to avoid retry loop")
+            break
+
+        with pg_conn.cursor() as pg_cur:
+            upsert_kugou_singers(pg_cur, singer_rows)
+            update_kugou_song_singers(pg_cur, song_updates)
+            upsert_kugou_singer_songs(pg_cur, singer_song_pairs)
+        pg_conn.commit()
+
+        batch_index += 1
+        pbar.update(1)
+    pbar.close()
+    log.info("Kugou singers backfill: ok=%d skip=%d", total_ok, total_skip)
+
+
+def _backfill_netease_singers(spider_conn, pg_conn, bucket, base_url, max_batches):
+    batch_index = 0
+    total_ok = total_skip = 0
+    pbar = tqdm(desc='backfill-netease-singers')
+    while max_batches is None or batch_index < max_batches:
+        with pg_conn.cursor() as pg_cur:
+            rows = fetch_netease_songs_missing_singers(pg_cur, BATCH_SIZE)
+        if not rows:
+            break
+
+        song_ids = [row['platform_song_id'] for row in rows]
+        netease_singers_map = fetch_netease_singers(spider_conn, song_ids)
+
+        singer_by_id: dict[int, dict] = {}
+        for sgs in netease_singers_map.values():
+            for sg in sgs:
+                singer_by_id[sg['singer_id']] = sg
+
+        avatar_tasks = {
+            f"avatar:{sg['singer_id']}": (
+                lambda sg=sg: _safe_transfer(
+                    sg.get('avatar', ''),
+                    build_oss_key('netease', 'singer', str(sg['singer_id']) + '.jpg'),
+                    bucket, base_url,
+                )
+            )
+            for sg in singer_by_id.values()
+        }
+        avatar_assets = _run_io_tasks(avatar_tasks)
+
+        singer_rows = [{
+            **sg,
+            'id': sg['singer_id'],
+            'avatar': avatar_assets.get(f"avatar:{sg['singer_id']}", sg.get('avatar', '')),
+            'provider_name': PROVIDER_YINYAN,
+            'crawler_source_data': _source_json(sg),
+        } for sg in singer_by_id.values()]
+
+        song_updates = []
+        singer_song_pairs = []
+        for row in rows:
+            singer_list = netease_singers_map.get(row['platform_song_id'], [])
+            if not singer_list:
+                total_skip += 1
+                continue
+            singers_json = json.dumps([{
+                'name': sg['name'],
+                'singer_id': sg['singer_id'],
+                'platform_singer_id': str(sg['singer_id']),
+            } for sg in singer_list], ensure_ascii=False)
+            song_updates.append({'platform_song_id': row['platform_song_id'], 'singers_json': singers_json})
+            singer_song_pairs.extend([(sg['singer_id'], row['id']) for sg in singer_list])
+            total_ok += 1
+
+        if not song_updates:
+            log.warning("Netease singers backfill: no singer data found for this batch, stopping to avoid retry loop")
+            break
+
+        with pg_conn.cursor() as pg_cur:
+            upsert_netease_singers(pg_cur, singer_rows)
+            update_netease_song_singers(pg_cur, song_updates)
+            upsert_netease_singer_songs(pg_cur, singer_song_pairs)
+        pg_conn.commit()
+
+        batch_index += 1
+        pbar.update(1)
+    pbar.close()
+    log.info("Netease singers backfill: ok=%d skip=%d", total_ok, total_skip)
+
+
+def backfill_empty_singers(platforms: list[str], max_batches: int | None = None) -> None:
+    spider_conn = get_spider_conn()
+    pg_conn = get_pg_conn()
+    bucket = get_oss_bucket()
+    base_url = OSS_CONFIG['base_url']
+    try:
+        if PLATFORM_QQ in platforms:
+            _backfill_qq_singers(spider_conn, pg_conn, bucket, base_url, max_batches)
+        if PLATFORM_KUGOU in platforms:
+            _backfill_kugou_singers(spider_conn, pg_conn, bucket, base_url, max_batches)
+        if PLATFORM_NETEASE in platforms:
+            _backfill_netease_singers(spider_conn, pg_conn, bucket, base_url, max_batches)
+    finally:
+        spider_conn.close()
+        pg_conn.close()
+
+
 def run(
     platforms: list[str],
     max_batches: int | None = None,

@@ -118,6 +118,108 @@ def upsert_yinyan_song_records(cur, records: list[dict]) -> None:
     )
 
 
+# ─── Backfill helpers ────────────────────────────────────────────────────────
+
+def fetch_qq_songs_missing_singers(cur, limit: int) -> list[dict]:
+    cur.execute(
+        """
+        SELECT id, platform_song_id, mid
+        FROM crawler_qqmusic_songs
+        WHERE singers IS NULL OR jsonb_array_length(singers) = 0
+        ORDER BY platform_song_id
+        LIMIT %s
+        """,
+        (limit,),
+    )
+    rows = cur.fetchall()
+    return [{'id': str(row[0]), 'platform_song_id': row[1], 'mid': row[2]} for row in rows]
+
+
+def fetch_kugou_songs_missing_singers(cur, limit: int) -> list[dict]:
+    cur.execute(
+        """
+        SELECT id, platform_song_id
+        FROM crawler_kugou_songs
+        WHERE singers IS NULL OR jsonb_array_length(singers) = 0
+        ORDER BY platform_song_id
+        LIMIT %s
+        """,
+        (limit,),
+    )
+    rows = cur.fetchall()
+    return [{'id': str(row[0]), 'platform_song_id': row[1]} for row in rows]
+
+
+def fetch_netease_songs_missing_singers(cur, limit: int) -> list[dict]:
+    cur.execute(
+        """
+        SELECT id, platform_song_id
+        FROM crawler_netease_songs
+        WHERE singers IS NULL OR jsonb_array_length(singers) = 0
+        ORDER BY platform_song_id
+        LIMIT %s
+        """,
+        (limit,),
+    )
+    rows = cur.fetchall()
+    return [{'id': str(row[0]), 'platform_song_id': row[1]} for row in rows]
+
+
+def update_qq_song_singers(cur, updates: list[dict]) -> None:
+    """updates: [{'platform_song_id': int, 'singers_json': str}, ...]"""
+    if not updates:
+        return
+    values_sql = ', '.join(['(%s::bigint, %s::jsonb)'] * len(updates))
+    params = []
+    for u in updates:
+        params.extend([u['platform_song_id'], u['singers_json']])
+    cur.execute(
+        f"""
+        UPDATE crawler_qqmusic_songs AS s
+        SET singers = v.singers, updated_at = NOW()
+        FROM (VALUES {values_sql}) AS v(platform_song_id, singers)
+        WHERE s.platform_song_id = v.platform_song_id
+        """,
+        tuple(params),
+    )
+
+
+def update_kugou_song_singers(cur, updates: list[dict]) -> None:
+    if not updates:
+        return
+    values_sql = ', '.join(['(%s::bigint, %s::jsonb)'] * len(updates))
+    params = []
+    for u in updates:
+        params.extend([u['platform_song_id'], u['singers_json']])
+    cur.execute(
+        f"""
+        UPDATE crawler_kugou_songs AS s
+        SET singers = v.singers, updated_at = NOW()
+        FROM (VALUES {values_sql}) AS v(platform_song_id, singers)
+        WHERE s.platform_song_id = v.platform_song_id
+        """,
+        tuple(params),
+    )
+
+
+def update_netease_song_singers(cur, updates: list[dict]) -> None:
+    if not updates:
+        return
+    values_sql = ', '.join(['(%s::bigint, %s::jsonb)'] * len(updates))
+    params = []
+    for u in updates:
+        params.extend([u['platform_song_id'], u['singers_json']])
+    cur.execute(
+        f"""
+        UPDATE crawler_netease_songs AS s
+        SET singers = v.singers, updated_at = NOW()
+        FROM (VALUES {values_sql}) AS v(platform_song_id, singers)
+        WHERE s.platform_song_id = v.platform_song_id
+        """,
+        tuple(params),
+    )
+
+
 # ─── QQ Music ────────────────────────────────────────────────────────────────
 
 def upsert_qq_singers(cur, singers: list[dict]) -> None:

 #!/usr/bin/env python3
 import argparse
 from etl_to_crawler.config import PLATFORM_QQ, PLATFORM_KUGOU, PLATFORM_NETEASE, PLATFORMS
-from etl_to_crawler.runner import backfill_yinyan_record_platforms, initialize_yinyan_song_records, run
+from etl_to_crawler.runner import backfill_yinyan_record_platforms, initialize_yinyan_song_records, run, backfill_empty_singers
 
 PLATFORM_MAP = {
     'qq': PLATFORM_QQ,
@@ -20,6 +20,8 @@ if __name__ == '__main__':
                         help='只初始化 yinyan_song_records 待导入状态表，不执行 crawler 导入')
     parser.add_argument('--backfill-yinyan-platforms', action='store_true',
                         help='只回填 yinyan_song_records 中为空的 platform 平台代码')
+    parser.add_argument('--backfill-empty-singers', action='store_true',
+                        help='回填三平台 singers 为空的歌曲记录')
     args = parser.parse_args()
 
     if args.platform == 'all':
@@ -32,5 +34,7 @@ if __name__ == '__main__':
         backfill_yinyan_record_platforms(max_batches=args.max_batches)
     elif args.init_yinyan_records:
         initialize_yinyan_song_records(platforms, max_batches=args.max_batches)
+    elif args.backfill_empty_singers:
+        backfill_empty_singers(platforms, max_batches=args.max_batches)
     else:
         run(platforms, max_batches=args.max_batches)