refactor(etl_to_crawler): 重构音频和歌手数据转存及导入流程

- 新增HTTP连接池配置，提升请求性能和资源复用
- 使用requests Session统一管理HTTP连接，减少请求开销
- 实现_url清洗函数，过滤无效或非法URL
- 并行执行音频、封面、歌词及歌手头像的OSS转存任务，提升效率
- 按平台统一构建导入数据payload，简化处理流程
- 批量写入歌曲、歌手、专辑及关联关系，提升数据库操作性能
- 替换原平台单独处理函数为统一预处理及写入方法
- 取消原单线程顺序处理，改为线程池并发执行数据准备和导入
- 获取平台歌曲及歌手信息支持传入缓存参数，减少重复查询
- 优化异常处理及日志记录，提高稳定性和可维护性
- 新增歌手索引和性别等字段辅助函数，完善歌手信息处理

@@ -50,3 +50,5 @@ PLATFORMS = [PLATFORM_QQ, PLATFORM_KUGOU, PLATFORM_NETEASE]
 
 BATCH_SIZE = 100
 BACKFILL_BATCH_SIZE = int(os.environ.get('BACKFILL_BATCH_SIZE', '5000'))
+HTTP_POOL_MAXSIZE = int(os.environ.get('HTTP_POOL_MAXSIZE', '128'))
+OSS_CONNECTION_POOL_SIZE = int(os.environ.get('OSS_CONNECTION_POOL_SIZE', str(HTTP_POOL_MAXSIZE)))

@@ -2,7 +2,7 @@ import pymysql
 import pymysql.cursors
 import pg8000
 import oss2
-from .config import SOURCE_DB, HK_SONGS_DB, CRAWLER_DB, OSS_CONFIG
+from .config import SOURCE_DB, HK_SONGS_DB, CRAWLER_DB, OSS_CONFIG, OSS_CONNECTION_POOL_SIZE
 
 
 def get_source_conn() -> pymysql.Connection:
@@ -26,5 +26,6 @@ def get_pg_conn() -> pg8000.Connection:
 
 
 def get_oss_bucket() -> oss2.Bucket:
+    oss2.defaults.connection_pool_size = OSS_CONNECTION_POOL_SIZE
     auth = oss2.Auth(OSS_CONFIG['access_key_id'], OSS_CONFIG['access_key_secret'])
     return oss2.Bucket(auth, OSS_CONFIG['endpoint'], OSS_CONFIG['bucket_name'])

 import requests
 import oss2
+import math
 from urllib.parse import urlparse
-from .config import OSS_CONFIG
+from requests.adapters import HTTPAdapter
+from .config import HTTP_POOL_MAXSIZE, OSS_CONFIG
 from .utils import compute_audio_md5
 
+_HTTP_SESSION = requests.Session()
+_HTTP_ADAPTER = HTTPAdapter(pool_connections=HTTP_POOL_MAXSIZE, pool_maxsize=HTTP_POOL_MAXSIZE)
+_HTTP_SESSION.mount('http://', _HTTP_ADAPTER)
+_HTTP_SESSION.mount('https://', _HTTP_ADAPTER)
+
+
+def _http_get(url: str, timeout: int = 30):
+    return _HTTP_SESSION.get(url, timeout=timeout)
+
+
+def _clean_url(url) -> str:
+    if url is None:
+        return ''
+    if isinstance(url, float) and math.isnan(url):
+        return ''
+    text = str(url).strip()
+    if not text or text.lower() in {'nan', 'none', 'null'}:
+        return ''
+    parsed = urlparse(text)
+    if parsed.scheme not in {'http', 'https'} or not parsed.netloc:
+        return ''
+    return text
+
 
 def _host(url: str | None) -> str:
     return urlparse(url or '').netloc.lower()
@@ -28,11 +53,12 @@ def transfer_url(url: str | None, oss_key: str, bucket: oss2.Bucket, base_url: s
     若 url 为空或已在目标 bucket，直接返回原 url（不上传）。
     返回新的公开访问 URL。
     """
+    url = _clean_url(url)
     if not url:
         return ''
     if _is_target_oss_url(url, base_url):
         return url
-    resp = requests.get(_download_url(url, base_url), timeout=30)
+    resp = _http_get(_download_url(url, base_url), timeout=30)
     resp.raise_for_status()
     bucket.put_object(oss_key, resp.content)
     return f"{base_url.rstrip('/')}/{oss_key}"
@@ -43,11 +69,12 @@ def transfer_url_with_md5(url: str | None, oss_key: str, bucket: oss2.Bucket, ba
     将音频 URL 转存到 OSS，并基于下载到的音频字节计算 MD5。
     已在目标 OSS 的 URL 无需重新下载，无法可靠计算 MD5，返回空 MD5。
     """
+    url = _clean_url(url)
     if not url:
         return '', ''
     if _is_target_oss_url(url, base_url):
         return url, ''
-    resp = requests.get(_download_url(url, base_url), timeout=30)
+    resp = _http_get(_download_url(url, base_url), timeout=30)
     resp.raise_for_status()
     audio_md5 = compute_audio_md5(resp.content)
     bucket.put_object(oss_key, resp.content)

 import uuid
 import json
 import logging
+from concurrent.futures import ThreadPoolExecutor, as_completed
 from tqdm import tqdm
 
 from .config import PLATFORM_QQ, PLATFORM_KUGOU, PLATFORM_NETEASE, BATCH_SIZE, BACKFILL_BATCH_SIZE, OSS_CONFIG
@@ -41,6 +42,20 @@ from .lyric import ensure_newlines
 logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
 log = logging.getLogger(__name__)
 PROVIDER_YINYAN = 'yinyan'
+MAX_RESOURCE_WORKERS = 8
+MAX_IMPORT_WORKERS = 16
+
+
+def _run_io_tasks(tasks: dict) -> dict:
+    if not tasks:
+        return {}
+    max_workers = min(MAX_RESOURCE_WORKERS, len(tasks))
+    results = {}
+    with ThreadPoolExecutor(max_workers=max_workers) as executor:
+        futures = {executor.submit(task): name for name, task in tasks.items()}
+        for future in as_completed(futures):
+            results[futures[future]] = future.result()
+    return results
 
 
 def _safe_transfer(url, oss_key, bucket, base_url):
@@ -75,47 +90,515 @@ def _source_json(data: dict) -> str:
     return json.dumps(data, ensure_ascii=False, default=_json_default)
 
 
-def _process_qq(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_url):
+def _prepare_qq_payload(hk_row: dict, pr: dict, bucket, base_url, sp: dict, singer_list: list[dict]) -> dict:
     mid = pr['platform_unique_key']
-    songs_map = fetch_qq_songs(spider_conn, [mid])
-    if mid not in songs_map:
-        return
-    sp = songs_map[mid]
     song_id_int = sp['id']
+    raw_lyric = sp.get('lyric') or ''
+    tasks = {
+        'audio': lambda: _safe_transfer_audio(
+            hk_row['audio_url'], build_oss_key('qq', 'audio', mid + '.mp3'), bucket, base_url
+        ),
+        'cover': lambda: _safe_transfer(
+            hk_row.get('cover_url') or sp.get('cover', ''),
+            build_oss_key('qq', 'cover', str(song_id_int) + '.jpg'),
+            bucket, base_url,
+        ),
+        'lyric': lambda: _safe_upload_lyric('qq', mid, raw_lyric, hk_row.get('lyrics_url'), bucket, base_url),
+    }
+    if sp.get('album_id') and sp.get('album_cover'):
+        tasks['album_cover'] = lambda: _safe_transfer(
+            sp['album_cover'], build_oss_key('qq', 'album', str(sp['album_id']) + '.jpg'), bucket, base_url
+        )
+    for sg in singer_list:
+        tasks[f"singer_avatar:{sg['singer_id']}"] = (
+            lambda sg=sg: _safe_transfer(
+                sg.get('avatar', ''),
+                build_oss_key('qq', 'singer', sg['mid'] + '.jpg'),
+                bucket, base_url,
+            )
+        )
+    assets = _run_io_tasks(tasks)
+    audio_url, audio_md5 = assets['audio']
+    cover_url = assets['cover']
+    lyric_url = assets['lyric']
+    album_cover = assets.get('album_cover') or cover_url
+
+    singer_rows = [{
+        **sg,
+        'id': sg['singer_id'],
+        'avatar': assets.get(f"singer_avatar:{sg['singer_id']}", sg.get('avatar', '')),
+        'provider_name': PROVIDER_YINYAN,
+        'crawler_source_data': _source_json(sg),
+    } for sg in singer_list]
+    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)
 
-    singers_map = fetch_qq_singers(spider_conn, [song_id_int])
-    singer_list = singers_map.get(song_id_int, [])
-
-    # OSS 转移
-    audio_url, audio_md5 = _safe_transfer_audio(
-        hk_row['audio_url'],
-        build_oss_key('qq', 'audio', mid + '.mp3'),
-        bucket, base_url
-    )
-    cover_url = _safe_transfer(
-        hk_row.get('cover_url') or sp.get('cover', ''),
-        build_oss_key('qq', 'cover', str(song_id_int) + '.jpg'),
-        bucket, base_url
-    )
-    album_cover = ''
+    album_rows = []
+    album_id = None
+    album_json = None
+    if sp.get('album_id'):
+        album_id = sp['album_id']
+        album_payload = {
+            'id': sp['album_id'],
+            'mid': sp.get('album_mid') or '',
+            'cover': album_cover,
+            'title': sp.get('album_title') or '',
+            'intro': sp.get('album_intro'),
+            'type': sp.get('album_type') or '',
+            'company_id': sp.get('company_id') or 0,
+            'company': sp.get('company') or '',
+            'is_owner': sp.get('is_owner') or 0,
+            'published_at': str(sp.get('album_published_at')) if sp.get('album_published_at') else None,
+        }
+        album_json = json.dumps(album_payload, ensure_ascii=False)
+        album_rows.append({
+            'id': sp['album_id'],
+            'mid': sp.get('album_mid') or '',
+            'cover': album_cover,
+            'title': sp.get('album_title') or '',
+            'intro': sp.get('album_intro'),
+            'type': sp.get('album_type') or '',
+            'company_id': sp.get('company_id') or 0,
+            'company': sp.get('company') or '',
+            'is_owner': sp.get('is_owner') or 0,
+            'published_at': sp.get('album_published_at'),
+            'provider_name': PROVIDER_YINYAN,
+            'crawler_source_data': _source_json({
+                'id': sp.get('album_id'),
+                'mid': sp.get('album_mid'),
+                'cover': sp.get('album_cover'),
+                'title': sp.get('album_title'),
+                'intro': sp.get('album_intro'),
+                'type': sp.get('album_type'),
+                'company_id': sp.get('company_id'),
+                'company': sp.get('company'),
+                'is_owner': sp.get('is_owner'),
+                'published_at': sp.get('album_published_at'),
+            }),
+        })
+
+    platform_song_id = int(pr['platform_mid']) if pr.get('platform_mid') else song_id_int
+    song_uuid = str(uuid.uuid4())
+    title, version = split_title_version(sp.get('title') or hk_row['name'])
+    song_row = {
+        'song_uuid': song_uuid,
+        'platform_song_id': platform_song_id,
+        'mid': mid,
+        'album_id': album_id,
+        'album_json': album_json,
+        'cover': cover_url,
+        'title': title,
+        'version': version,
+        'name': hk_row['name'],
+        'duration': sp.get('duration') or hk_row.get('song_time') or 0,
+        'lyric': ensure_newlines(raw_lyric),
+        'composer_name': sp.get('composer_name') or hk_row.get('composer'),
+        'lyricist_name': sp.get('lyricist_name') or hk_row.get('lyricist'),
+        'url': audio_url,
+        'audio_md5': audio_md5,
+        'lyric_url': lyric_url,
+        'platform_index_url': sp.get('platform_index_url') or f'https://y.qq.com/n/ryqq/songDetail/{mid}',
+        'published_at': sp.get('published_at') or hk_row.get('issue_time'),
+        'singers_json': singers_json,
+        'provider_name': PROVIDER_YINYAN,
+        'crawler_source_data': _source_json(sp),
+    }
+    return {
+        'platform': PLATFORM_QQ,
+        'display_platform': 'qq',
+        'platform_song_id': platform_song_id,
+        'result': {'platform': 'qq', 'platform_song_id': platform_song_id, 'mid': mid, 'title': hk_row['name']},
+        'singers': singer_rows,
+        'albums': album_rows,
+        'songs': [song_row],
+        'singer_songs': [(sg['singer_id'], song_uuid) for sg in singer_list],
+        'singer_albums': [(sg['singer_id'], album_id) for sg in singer_list] if album_id else [],
+    }
+
+
+def _prepare_kugou_payload(hk_row: dict, pr: dict, bucket, base_url, sp: dict, singer_list: list[dict]) -> dict:
+    song_id = int(pr['platform_unique_key'])
+    raw_lyric = sp.get('lyric') or ''
+    tasks = {
+        'audio': lambda: _safe_transfer_audio(
+            hk_row['audio_url'], build_oss_key('kugou', 'audio', str(song_id) + '.mp3'), bucket, base_url
+        ),
+        'cover': lambda: _safe_transfer(
+            hk_row.get('cover_url') or sp.get('cover', ''),
+            build_oss_key('kugou', 'cover', str(song_id) + '.jpg'),
+            bucket, base_url,
+        ),
+        'lyric': lambda: _safe_upload_lyric('kugou', str(song_id), raw_lyric, hk_row.get('lyrics_url'), bucket, base_url),
+    }
     if sp.get('album_id') and sp.get('album_cover'):
-        album_cover = _safe_transfer(
+        tasks['album_cover'] = lambda: _safe_transfer(
+            sp['album_cover'], build_oss_key('kugou', 'album', str(sp['album_id']) + '.jpg'), bucket, base_url
+        )
+    for sg in singer_list:
+        tasks[f"singer_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,
+            )
+        )
+    assets = _run_io_tasks(tasks)
+    audio_url, audio_md5 = assets['audio']
+    cover_url = assets['cover']
+    lyric_url = assets['lyric']
+    album_cover = assets.get('album_cover') or cover_url
+
+    singer_rows = [{
+        **sg,
+        'id': sg['singer_id'],
+        'avatar': assets.get(f"singer_avatar:{sg['singer_id']}", sg.get('avatar', '')),
+        'provider_name': PROVIDER_YINYAN,
+        'crawler_source_data': _source_json(sg),
+    } for sg in singer_list]
+    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)
+
+    album_rows = []
+    album_id = None
+    if sp.get('album_id'):
+        album_id = sp['album_id']
+        album_rows.append({
+            'id': sp['album_id'],
+            'cover': album_cover,
+            'title': sp.get('album_title') or '',
+            'intro': sp.get('album_intro'),
+            'type': sp.get('album_type') or '',
+            'company_id': sp.get('company_id') or 0,
+            'company': sp.get('company') or '',
+            'is_owner': sp.get('is_owner') or 0,
+            'published_at': sp.get('album_published_at'),
+            'provider_name': PROVIDER_YINYAN,
+            'crawler_source_data': _source_json({
+                'id': sp.get('album_id'),
+                'cover': sp.get('album_cover'),
+                'title': sp.get('album_title'),
+                'intro': sp.get('album_intro'),
+                'type': sp.get('album_type'),
+                'company_id': sp.get('company_id'),
+                'company': sp.get('company'),
+                'is_owner': sp.get('is_owner'),
+                'published_at': sp.get('album_published_at'),
+            }),
+        })
+
+    song_uuid = str(uuid.uuid4())
+    title, version = split_title_version(sp.get('title') or hk_row['name'])
+    song_row = {
+        'song_uuid': song_uuid,
+        'platform_song_id': song_id,
+        'hash': sp.get('hid', pr.get('platform_mid', '')),
+        'album_audio_id': sp.get('album_audio_id') or pr.get('album_audio_id') or 0,
+        'album_id': album_id,
+        'cover': cover_url,
+        'title': title,
+        'version': version,
+        'name': hk_row['name'],
+        'duration': sp.get('duration') or hk_row.get('song_time') or 0,
+        'lyric': ensure_newlines(raw_lyric),
+        'composer_name': sp.get('composer_name') or hk_row.get('composer'),
+        'lyricist_name': sp.get('lyricist_name') or hk_row.get('lyricist'),
+        'url': audio_url,
+        'audio_md5': audio_md5,
+        'lyric_url': lyric_url,
+        'platform_index_url': sp.get('platform_index_url') or f'http://www.kugou.com/song/#hash={sp.get("hid") or pr.get("platform_mid", "")}',
+        'published_at': sp.get('published_at') or hk_row.get('issue_time'),
+        'singers_json': singers_json,
+        'provider_name': PROVIDER_YINYAN,
+        'crawler_source_data': _source_json(sp),
+    }
+    return {
+        'platform': PLATFORM_KUGOU,
+        'display_platform': 'kugou',
+        'platform_song_id': song_id,
+        'result': {'platform': 'kugou', 'platform_song_id': song_id, 'hash': sp.get('hid', ''), 'title': hk_row['name']},
+        'singers': singer_rows,
+        'albums': album_rows,
+        'songs': [song_row],
+        'singer_songs': [(sg['singer_id'], song_uuid) for sg in singer_list],
+        'singer_albums': [(sg['singer_id'], album_id) for sg in singer_list] if album_id else [],
+    }
+
+
+def _prepare_netease_payload(hk_row: dict, pr: dict, bucket, base_url, sp: dict, singer_list: list[dict]) -> dict:
+    song_id = int(pr['platform_unique_key'])
+    raw_lyric = sp.get('lyric') or ''
+    tasks = {
+        'audio': lambda: _safe_transfer_audio(
+            hk_row['audio_url'], build_oss_key('netease', 'audio', str(song_id) + '.mp3'), bucket, base_url
+        ),
+        'cover': lambda: _safe_transfer(
+            hk_row.get('cover_url') or sp.get('cover', ''),
+            build_oss_key('netease', 'cover', str(song_id) + '.jpg'),
+            bucket, base_url,
+        ),
+        'lyric': lambda: _safe_upload_lyric('netease', str(song_id), raw_lyric, hk_row.get('lyrics_url'), bucket, base_url),
+    }
+    if sp.get('album_id') and sp.get('album_cover'):
+        tasks['album_cover'] = lambda: _safe_transfer(
+            sp['album_cover'], build_oss_key('netease', 'album', str(sp['album_id']) + '.jpg'), bucket, base_url
+        )
+    for sg in singer_list:
+        tasks[f"singer_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,
+            )
+        )
+    assets = _run_io_tasks(tasks)
+    audio_url, audio_md5 = assets['audio']
+    cover_url = assets['cover']
+    lyric_url = assets['lyric']
+    album_cover = assets.get('album_cover') or cover_url
+
+    singer_rows = [{
+        **sg,
+        'id': sg['singer_id'],
+        'avatar': assets.get(f"singer_avatar:{sg['singer_id']}", sg.get('avatar', '')),
+        'provider_name': PROVIDER_YINYAN,
+        'crawler_source_data': _source_json(sg),
+    } for sg in singer_list]
+    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)
+
+    album_rows = []
+    album_id = None
+    album_json = None
+    if sp.get('album_id'):
+        album_id = sp['album_id']
+        album_payload = {
+            'id': sp['album_id'],
+            'cover': album_cover,
+            'title': sp.get('album_title') or '',
+            'intro': sp.get('album_intro'),
+            'type': sp.get('album_type') or '',
+            'company_id': sp.get('company_id') or 0,
+            'company': sp.get('company') or '',
+            'is_owner': sp.get('is_owner') or 0,
+            'published_at': str(sp.get('album_published_at')) if sp.get('album_published_at') else None,
+        }
+        album_json = json.dumps(album_payload, ensure_ascii=False)
+        album_rows.append({
+            'id': sp['album_id'],
+            'cover': album_cover,
+            'title': sp.get('album_title') or '',
+            'intro': sp.get('album_intro'),
+            'type': sp.get('album_type') or '',
+            'company_id': sp.get('company_id') or 0,
+            'company': sp.get('company') or '',
+            'is_owner': sp.get('is_owner') or 0,
+            'published_at': sp.get('album_published_at'),
+            'provider_name': PROVIDER_YINYAN,
+            'crawler_source_data': _source_json({
+                'id': sp.get('album_id'),
+                'cover': sp.get('album_cover'),
+                'title': sp.get('album_title'),
+                'intro': sp.get('album_intro'),
+                'type': sp.get('album_type'),
+                'company_id': sp.get('company_id'),
+                'company': sp.get('company'),
+                'is_owner': sp.get('is_owner'),
+                'published_at': sp.get('album_published_at'),
+            }),
+        })
+
+    song_uuid = str(uuid.uuid4())
+    title, version = split_title_version(sp.get('title') or hk_row['name'])
+    song_row = {
+        'song_uuid': song_uuid,
+        'platform_song_id': song_id,
+        'album_id': album_id,
+        'album_json': album_json,
+        'cover': cover_url,
+        'title': title,
+        'version': version,
+        'name': hk_row['name'],
+        'duration': sp.get('duration') or hk_row.get('song_time') or 0,
+        'lyric': ensure_newlines(raw_lyric),
+        'composer_name': sp.get('composer_name') or hk_row.get('composer'),
+        'lyricist_name': sp.get('lyricist_name') or hk_row.get('lyricist'),
+        'url': audio_url,
+        'audio_md5': audio_md5,
+        'lyric_url': lyric_url,
+        'platform_index_url': sp.get('platform_index_url') or f'https://music.163.com/#/song?id={song_id}',
+        'published_at': sp.get('published_at') or hk_row.get('issue_time'),
+        'singers_json': singers_json,
+        'provider_name': PROVIDER_YINYAN,
+        'crawler_source_data': _source_json(sp),
+    }
+    return {
+        'platform': PLATFORM_NETEASE,
+        'display_platform': 'netease',
+        'platform_song_id': song_id,
+        'result': {'platform': 'netease', 'platform_song_id': song_id, 'title': hk_row['name']},
+        'singers': singer_rows,
+        'albums': album_rows,
+        'songs': [song_row],
+        'singer_songs': [(sg['singer_id'], song_uuid) for sg in singer_list],
+        'singer_albums': [(sg['singer_id'], album_id) for sg in singer_list] if album_id else [],
+    }
+
+
+_PREPARERS = {
+    PLATFORM_QQ: _prepare_qq_payload,
+    PLATFORM_KUGOU: _prepare_kugou_payload,
+    PLATFORM_NETEASE: _prepare_netease_payload,
+}
+
+
+def _prepare_import_payload(
+    pending: dict,
+    hk_row: dict,
+    pr: dict,
+    bucket,
+    base_url: str,
+    songs_maps: dict,
+    singers_maps: dict,
+) -> dict | None:
+    platform = str(pending['platform'])
+    preparer = _PREPARERS.get(platform)
+    if not preparer:
+        return None
+    platform_unique_id = pr['platform_unique_key']
+    song_key = platform_unique_id if platform == PLATFORM_QQ else int(platform_unique_id)
+    song_data = songs_maps.get(platform, {}).get(song_key)
+    if not song_data:
+        return None
+    singer_key = int(song_data['id']) if platform == PLATFORM_QQ else int(platform_unique_id)
+    singer_list = singers_maps.get(platform, {}).get(singer_key, [])
+    payload = preparer(hk_row, pr, bucket, base_url, song_data, singer_list)
+    payload['yinyan_record'] = {
+        'song_id': int(pending['song_id']),
+        'record_id': int(pr['record_id']),
+        'platform': platform,
+        'platform_song_id': int(payload['platform_song_id']),
+    }
+    return payload
+
+
+def _extend_platform_payload(target: dict, payload: dict) -> None:
+    target['singers'].extend(payload.get('singers', []))
+    target['albums'].extend(payload.get('albums', []))
+    target['songs'].extend(payload.get('songs', []))
+    target['singer_songs'].extend(payload.get('singer_songs', []))
+    target['singer_albums'].extend(payload.get('singer_albums', []))
+
+
+def _write_import_payloads(pg_cur, payloads: list[dict]) -> list[dict]:
+    if not payloads:
+        return []
+    grouped = {
+        PLATFORM_QQ: {'singers': [], 'albums': [], 'songs': [], 'singer_songs': [], 'singer_albums': []},
+        PLATFORM_KUGOU: {'singers': [], 'albums': [], 'songs': [], 'singer_songs': [], 'singer_albums': []},
+        PLATFORM_NETEASE: {'singers': [], 'albums': [], 'songs': [], 'singer_songs': [], 'singer_albums': []},
+    }
+    yinyan_records = []
+    imported = []
+    for payload in payloads:
+        _extend_platform_payload(grouped[payload['platform']], payload)
+        yinyan_records.append(payload['yinyan_record'])
+        imported.append(payload['result'])
+
+    qq = grouped[PLATFORM_QQ]
+    upsert_qq_singers(pg_cur, qq['singers'])
+    upsert_qq_albums(pg_cur, qq['albums'])
+    upsert_qq_songs(pg_cur, qq['songs'])
+    upsert_qq_singer_songs(pg_cur, qq['singer_songs'])
+    upsert_qq_singer_albums(pg_cur, qq['singer_albums'])
+
+    kugou = grouped[PLATFORM_KUGOU]
+    upsert_kugou_singers(pg_cur, kugou['singers'])
+    upsert_kugou_albums(pg_cur, kugou['albums'])
+    upsert_kugou_songs(pg_cur, kugou['songs'])
+    upsert_kugou_singer_songs(pg_cur, kugou['singer_songs'])
+    upsert_kugou_singer_albums(pg_cur, kugou['singer_albums'])
+
+    netease = grouped[PLATFORM_NETEASE]
+    upsert_netease_singers(pg_cur, netease['singers'])
+    upsert_netease_albums(pg_cur, netease['albums'])
+    upsert_netease_songs(pg_cur, netease['songs'])
+    upsert_netease_singer_songs(pg_cur, netease['singer_songs'])
+    upsert_netease_singer_albums(pg_cur, netease['singer_albums'])
+
+    upsert_yinyan_song_records(pg_cur, yinyan_records)
+    return imported
+
+
+def _process_qq(
+    hk_row: dict,
+    pr: dict,
+    spider_conn,
+    pg_cur,
+    bucket,
+    base_url,
+    song_data: dict | None = None,
+    singer_list: list[dict] | None = None,
+):
+    mid = pr['platform_unique_key']
+    sp = song_data
+    if sp is None:
+        songs_map = fetch_qq_songs(spider_conn, [mid])
+        if mid not in songs_map:
+            return
+        sp = songs_map[mid]
+    song_id_int = sp['id']
+
+    if singer_list is None:
+        singers_map = fetch_qq_singers(spider_conn, [song_id_int])
+        singer_list = singers_map.get(song_id_int, [])
+
+    raw_lyric = sp.get('lyric') or ''
+    tasks = {
+        'audio': lambda: _safe_transfer_audio(
+            hk_row['audio_url'],
+            build_oss_key('qq', 'audio', mid + '.mp3'),
+            bucket, base_url
+        ),
+        'cover': lambda: _safe_transfer(
+            hk_row.get('cover_url') or sp.get('cover', ''),
+            build_oss_key('qq', 'cover', str(song_id_int) + '.jpg'),
+            bucket, base_url
+        ),
+        'lyric': lambda: _safe_upload_lyric('qq', mid, raw_lyric, hk_row.get('lyrics_url'), bucket, base_url),
+    }
+    if sp.get('album_id') and sp.get('album_cover'):
+        tasks['album_cover'] = lambda: _safe_transfer(
             sp['album_cover'],
             build_oss_key('qq', 'album', str(sp['album_id']) + '.jpg'),
             bucket, base_url
         )
-    # 专辑封面为空时回退使用歌曲封面
-    if not album_cover and cover_url:
-        album_cover = cover_url
+    for sg in singer_list:
+        tasks[f"singer_avatar:{sg['singer_id']}"] = (
+            lambda sg=sg: _safe_transfer(
+                sg.get('avatar', ''),
+                build_oss_key('qq', 'singer', sg['mid'] + '.jpg'),
+                bucket, base_url
+            )
+        )
+    assets = _run_io_tasks(tasks)
+    audio_url, audio_md5 = assets['audio']
+    cover_url = assets['cover']
+    lyric_url = assets['lyric']
+    album_cover = assets.get('album_cover') or cover_url
 
     # 歌手头像转移 + 写入 singers
     singer_rows = []
     for sg in singer_list:
-        avatar = _safe_transfer(
-            sg.get('avatar', ''),
-            build_oss_key('qq', 'singer', sg['mid'] + '.jpg'),
-            bucket, base_url
-        )
+        avatar = assets.get(f"singer_avatar:{sg['singer_id']}", sg.get('avatar', ''))
         singer_rows.append({
             **sg,
             'id': sg['singer_id'],
@@ -174,8 +657,6 @@ def _process_qq(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_url):
     # 写入 song
     platform_song_id = int(pr['platform_mid']) if pr.get('platform_mid') else song_id_int
     song_uuid = str(uuid.uuid4())
-    raw_lyric = sp.get('lyric') or ''
-    lyric_url = _safe_upload_lyric('qq', mid, raw_lyric, hk_row.get('lyrics_url'), bucket, base_url)
     title, version = split_title_version(sp.get('title') or hk_row['name'])
     upsert_qq_songs(pg_cur, [{
         'song_uuid': song_uuid,
@@ -213,44 +694,65 @@ def _process_qq(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_url):
     return {'platform': 'qq', 'platform_song_id': platform_song_id, 'mid': mid, 'title': hk_row['name']}
 
 
-def _process_kugou(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_url):
+def _process_kugou(
+    hk_row: dict,
+    pr: dict,
+    spider_conn,
+    pg_cur,
+    bucket,
+    base_url,
+    song_data: dict | None = None,
+    singer_list: list[dict] | None = None,
+):
     song_id = int(pr['platform_unique_key'])
-    songs_map = fetch_kugou_songs(spider_conn, [song_id])
-    if song_id not in songs_map:
-        return
-    sp = songs_map[song_id]
-
-    singers_map = fetch_kugou_singers(spider_conn, [song_id])
-    singer_list = singers_map.get(song_id, [])
-
-    audio_url, audio_md5 = _safe_transfer_audio(
-        hk_row['audio_url'],
-        build_oss_key('kugou', 'audio', str(song_id) + '.mp3'),
-        bucket, base_url
-    )
-    cover_url = _safe_transfer(
-        hk_row.get('cover_url') or sp.get('cover', ''),
-        build_oss_key('kugou', 'cover', str(song_id) + '.jpg'),
-        bucket, base_url
-    )
-    album_cover = ''
+    sp = song_data
+    if sp is None:
+        songs_map = fetch_kugou_songs(spider_conn, [song_id])
+        if song_id not in songs_map:
+            return
+        sp = songs_map[song_id]
+
+    if singer_list is None:
+        singers_map = fetch_kugou_singers(spider_conn, [song_id])
+        singer_list = singers_map.get(song_id, [])
+
+    raw_lyric = sp.get('lyric') or ''
+    tasks = {
+        'audio': lambda: _safe_transfer_audio(
+            hk_row['audio_url'],
+            build_oss_key('kugou', 'audio', str(song_id) + '.mp3'),
+            bucket, base_url
+        ),
+        'cover': lambda: _safe_transfer(
+            hk_row.get('cover_url') or sp.get('cover', ''),
+            build_oss_key('kugou', 'cover', str(song_id) + '.jpg'),
+            bucket, base_url
+        ),
+        'lyric': lambda: _safe_upload_lyric('kugou', str(song_id), raw_lyric, hk_row.get('lyrics_url'), bucket, base_url),
+    }
     if sp.get('album_id') and sp.get('album_cover'):
-        album_cover = _safe_transfer(
+        tasks['album_cover'] = lambda: _safe_transfer(
             sp['album_cover'],
             build_oss_key('kugou', 'album', str(sp['album_id']) + '.jpg'),
             bucket, base_url
         )
-    # 专辑封面为空时回退使用歌曲封面
-    if not album_cover and cover_url:
-        album_cover = cover_url
+    for sg in singer_list:
+        tasks[f"singer_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
+            )
+        )
+    assets = _run_io_tasks(tasks)
+    audio_url, audio_md5 = assets['audio']
+    cover_url = assets['cover']
+    lyric_url = assets['lyric']
+    album_cover = assets.get('album_cover') or cover_url
 
     singer_rows = []
     for sg in singer_list:
-        avatar = _safe_transfer(
-            sg.get('avatar', ''),
-            build_oss_key('kugou', 'singer', str(sg['singer_id']) + '.jpg'),
-            bucket, base_url
-        )
+        avatar = assets.get(f"singer_avatar:{sg['singer_id']}", sg.get('avatar', ''))
         singer_rows.append({
             **sg,
             'id': sg['singer_id'],
@@ -290,8 +792,6 @@ def _process_kugou(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_url
         }])
 
     song_uuid = str(uuid.uuid4())
-    raw_lyric = sp.get('lyric') or ''
-    lyric_url = _safe_upload_lyric('kugou', str(song_id), raw_lyric, hk_row.get('lyrics_url'), bucket, base_url)
     title, version = split_title_version(sp.get('title') or hk_row['name'])
     upsert_kugou_songs(pg_cur, [{
         'song_uuid': song_uuid,
@@ -328,44 +828,65 @@ def _process_kugou(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_url
     return {'platform': 'kugou', 'platform_song_id': song_id, 'hash': sp.get('hid', ''), 'title': hk_row['name']}
 
 
-def _process_netease(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_url):
+def _process_netease(
+    hk_row: dict,
+    pr: dict,
+    spider_conn,
+    pg_cur,
+    bucket,
+    base_url,
+    song_data: dict | None = None,
+    singer_list: list[dict] | None = None,
+):
     song_id = int(pr['platform_unique_key'])
-    songs_map = fetch_netease_songs(spider_conn, [song_id])
-    if song_id not in songs_map:
-        return
-    sp = songs_map[song_id]
-
-    singers_map = fetch_netease_singers(spider_conn, [song_id])
-    singer_list = singers_map.get(song_id, [])
-
-    audio_url, audio_md5 = _safe_transfer_audio(
-        hk_row['audio_url'],
-        build_oss_key('netease', 'audio', str(song_id) + '.mp3'),
-        bucket, base_url
-    )
-    cover_url = _safe_transfer(
-        hk_row.get('cover_url') or sp.get('cover', ''),
-        build_oss_key('netease', 'cover', str(song_id) + '.jpg'),
-        bucket, base_url
-    )
-    album_cover = ''
+    sp = song_data
+    if sp is None:
+        songs_map = fetch_netease_songs(spider_conn, [song_id])
+        if song_id not in songs_map:
+            return
+        sp = songs_map[song_id]
+
+    if singer_list is None:
+        singers_map = fetch_netease_singers(spider_conn, [song_id])
+        singer_list = singers_map.get(song_id, [])
+
+    raw_lyric = sp.get('lyric') or ''
+    tasks = {
+        'audio': lambda: _safe_transfer_audio(
+            hk_row['audio_url'],
+            build_oss_key('netease', 'audio', str(song_id) + '.mp3'),
+            bucket, base_url
+        ),
+        'cover': lambda: _safe_transfer(
+            hk_row.get('cover_url') or sp.get('cover', ''),
+            build_oss_key('netease', 'cover', str(song_id) + '.jpg'),
+            bucket, base_url
+        ),
+        'lyric': lambda: _safe_upload_lyric('netease', str(song_id), raw_lyric, hk_row.get('lyrics_url'), bucket, base_url),
+    }
     if sp.get('album_id') and sp.get('album_cover'):
-        album_cover = _safe_transfer(
+        tasks['album_cover'] = lambda: _safe_transfer(
             sp['album_cover'],
             build_oss_key('netease', 'album', str(sp['album_id']) + '.jpg'),
             bucket, base_url
         )
-    # 专辑封面为空时回退使用歌曲封面
-    if not album_cover and cover_url:
-        album_cover = cover_url
+    for sg in singer_list:
+        tasks[f"singer_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
+            )
+        )
+    assets = _run_io_tasks(tasks)
+    audio_url, audio_md5 = assets['audio']
+    cover_url = assets['cover']
+    lyric_url = assets['lyric']
+    album_cover = assets.get('album_cover') or cover_url
 
     singer_rows = []
     for sg in singer_list:
-        avatar = _safe_transfer(
-            sg.get('avatar', ''),
-            build_oss_key('netease', 'singer', str(sg['singer_id']) + '.jpg'),
-            bucket, base_url
-        )
+        avatar = assets.get(f"singer_avatar:{sg['singer_id']}", sg.get('avatar', ''))
         singer_rows.append({
             **sg,
             'id': sg['singer_id'],
@@ -418,8 +939,6 @@ def _process_netease(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_u
         }])
 
     song_uuid = str(uuid.uuid4())
-    raw_lyric = sp.get('lyric') or ''
-    lyric_url = _safe_upload_lyric('netease', str(song_id), raw_lyric, hk_row.get('lyrics_url'), bucket, base_url)
     title, version = split_title_version(sp.get('title') or hk_row['name'])
     upsert_netease_songs(pg_cur, [{
         'song_uuid': song_uuid,
@@ -643,50 +1162,55 @@ def run(
                 PLATFORM_NETEASE: netease_singers_map,
             }
 
-            with pg_conn.cursor() as pg_cur:
-                pushed_count = 0
-                for pending in pending_records:
-                    src_id = int(pending['song_id'])
-                    platform = str(pending['platform'])
-                    hk_row = hk_by_song.get(src_id)
-                    if not hk_row:
-                        continue
-                    if platform not in platforms:
-                        continue
-                    pr = pr_by_state_key.get((src_id, int(pending['record_id']), platform))
-                    if not pr:
-                        log.warning(
-                            "Pending yinyan record not found in source records: song_id=%s record_id=%s platform=%s",
-                            src_id, pending['record_id'], platform,
-                        )
-                        continue
-                    processor = _PROCESSORS.get(platform)
-                    if not processor:
-                        continue
-                    pg_cur.execute('SAVEPOINT sp_song')
-                    try:
-                        result = processor(hk_row, pr, spider_conn, pg_cur, bucket, base_url)
-                        if result:
-                            upsert_yinyan_song_records(pg_cur, [{
-                                'song_id': src_id,
-                                'record_id': int(pr['record_id']),
-                                'platform': platform,
-                                'platform_song_id': int(result['platform_song_id']),
-                            }])
-                            pushed_count += 1
-                            imported.append(result)
-                        pg_cur.execute('RELEASE SAVEPOINT sp_song')
-                        total_ok += 1
-                    except Exception as e:
-                        pg_cur.execute('ROLLBACK TO SAVEPOINT sp_song')
-                        pg_cur.execute('RELEASE SAVEPOINT sp_song')
-                        log.error("Error processing song %s platform %s: %s",
-                                  hk_row.get('name'), platform, e)
-                        total_err += 1
+            prepare_inputs = []
+            for pending in pending_records:
+                src_id = int(pending['song_id'])
+                platform = str(pending['platform'])
+                hk_row = hk_by_song.get(src_id)
+                if not hk_row or platform not in platforms:
+                    continue
+                pr = pr_by_state_key.get((src_id, int(pending['record_id']), platform))
+                if not pr:
+                    log.warning(
+                        "Pending yinyan record not found in source records: song_id=%s record_id=%s platform=%s",
+                        src_id, pending['record_id'], platform,
+                    )
+                    continue
+                prepare_inputs.append((pending, hk_row, pr))
+
+            payloads = []
+            max_workers = min(MAX_IMPORT_WORKERS, len(prepare_inputs)) if prepare_inputs else 0
+            if max_workers:
+                with ThreadPoolExecutor(max_workers=max_workers) as executor:
+                    future_map = {
+                        executor.submit(
+                            _prepare_import_payload,
+                            pending, hk_row, pr, bucket, base_url, songs_maps, singers_maps,
+                        ): (pending, hk_row, pr)
+                        for pending, hk_row, pr in prepare_inputs
+                    }
+                    for future in as_completed(future_map):
+                        pending, hk_row, pr = future_map[future]
+                        try:
+                            payload = future.result()
+                            if payload:
+                                payloads.append(payload)
+                                total_ok += 1
+                            else:
+                                total_err += 1
+                        except Exception as e:
+                            total_err += 1
+                            log.error("Error preparing song %s platform %s: %s",
+                                      hk_row.get('name'), pr['platform'], e)
+
+            if payloads:
+                with pg_conn.cursor() as pg_cur:
+                    _write_import_payloads(pg_cur, payloads)
                 pg_conn.commit()
-                if pushed_count == 0:
-                    log.error("No yinyan_song_records rows were marked pushed in this batch; stopping to avoid retry loop")
-                    break
+                imported.extend(payload['result'] for payload in payloads)
+            else:
+                log.error("No import payloads were prepared in this batch; stopping to avoid retry loop")
+                break
             batch_index += 1
             pbar.update(1)
         pbar.close()

 import uuid
 
+_SINGER_INDEX_VALUES = set('ABCDEFGHIJKLMNOPQRSTUVWXYZ#')
+_SINGER_SEX_VALUES = {'M', 'F', 'C', 'U'}
+_SINGER_AREA_VALUES = {'华语', '欧美', '韩国', '日本', '其他'}
+
+
+def _singer_index(value) -> str:
+    text = str(value or '').strip().upper()
+    return text if text in _SINGER_INDEX_VALUES else '#'
+
+
+def _singer_sex(value) -> str:
+    text = str(value or '').strip().upper()
+    return text if text in _SINGER_SEX_VALUES else 'U'
+
+
+def _singer_area(value) -> str:
+    text = str(value or '').strip()
+    return text if text in _SINGER_AREA_VALUES else '其他'
+
 
 def insert_yinyan_song_records(cur, records: list[dict]) -> None:
     """Initialize yinyan_song_records rows before crawler import."""
@@ -114,7 +133,7 @@ def upsert_qq_singers(cur, singers: list[dict]) -> None:
     """
     rows = [(
         s['id'], s['mid'], s['name'], s.get('avatar', ''),
-        s.get('sex') or 'U', s.get('area') or '其他', s.get('index') or '#',
+        _singer_sex(s.get('sex')), _singer_area(s.get('area')), _singer_index(s.get('index')),
         s.get('intro'), s.get('home_url'),
         s.get('provider_name'), s.get('crawler_source_data'),
     ) for s in singers]
@@ -206,7 +225,7 @@ def upsert_kugou_singers(cur, singers: list[dict]) -> None:
     """
     rows = [(
         s['id'], s['name'], s.get('avatar', ''),
-        s.get('sex') or 'U', s.get('area') or '其他', s.get('index') or '#',
+        _singer_sex(s.get('sex')), _singer_area(s.get('area')), _singer_index(s.get('index')),
         s.get('intro'), s.get('home_url', ''),
         s.get('provider_name'), s.get('crawler_source_data'),
     ) for s in singers]
@@ -295,7 +314,7 @@ def upsert_netease_singers(cur, singers: list[dict]) -> None:
     """
     rows = [(
         s['id'], s['name'], s.get('avatar', ''),
-        s.get('sex') or 'U', s.get('area') or '其他', s.get('index') or '#',
+        _singer_sex(s.get('sex')), _singer_area(s.get('area')), _singer_index(s.get('index')),
         s.get('intro'), s.get('home_url'),
         s.get('provider_name'), s.get('crawler_source_data'),
     ) for s in singers]

 from unittest.mock import MagicMock, patch
+import math
 from etl_to_crawler.oss import transfer_url, transfer_url_with_md5
 
 ARCHIVE_URL = "https://archive-dev.oss-cn-beijing.aliyuncs.com/some/path.mp3"
@@ -22,10 +23,22 @@ def test_none_url_returns_empty():
     result = transfer_url(None, "any/key.mp3", bucket, BASE_URL)
     assert result == ''
 
+
+def test_invalid_nan_url_returns_empty_without_download():
+    bucket = MagicMock()
+    with patch('etl_to_crawler.oss._http_get') as mock_get:
+        assert transfer_url('nan', "any/key.mp3", bucket, BASE_URL) == ''
+        assert transfer_url(math.nan, "any/key.mp3", bucket, BASE_URL) == ''
+        assert transfer_url('not-a-url', "any/key.mp3", bucket, BASE_URL) == ''
+        assert transfer_url_with_md5('nan', "any/key.mp3", bucket, BASE_URL) == ('', '')
+    mock_get.assert_not_called()
+    bucket.put_object.assert_not_called()
+
+
 def test_external_url_downloads_and_uploads():
     bucket = MagicMock()
     fake_content = b"audio_bytes"
-    with patch('etl_to_crawler.oss.requests.get') as mock_get:
+    with patch('etl_to_crawler.oss._http_get') as mock_get:
         mock_get.return_value.content = fake_content
         mock_get.return_value.raise_for_status = MagicMock()
         result = transfer_url(OTHER_URL, "crawler/qq/audio/abc.mp3", bucket, BASE_URL)
@@ -42,7 +55,7 @@ def test_external_url_can_rewrite_download_base_to_internal_endpoint():
         'download_rewrite_from_base_url': 'https://source-bucket.oss-cn-hangzhou.aliyuncs.com',
         'download_base_url': internal_source_base,
     }):
-        with patch('etl_to_crawler.oss.requests.get') as mock_get:
+        with patch('etl_to_crawler.oss._http_get') as mock_get:
             mock_get.return_value.content = fake_content
             mock_get.return_value.raise_for_status = MagicMock()
             result = transfer_url(public_source, "crawler/qq/audio/abc.mp3", bucket, BASE_URL)
@@ -55,7 +68,7 @@ def test_external_url_can_rewrite_download_base_to_internal_endpoint():
 def test_transfer_url_with_md5_hashes_downloaded_content_before_upload():
     bucket = MagicMock()
     fake_content = b"audio_bytes"
-    with patch('etl_to_crawler.oss.requests.get') as mock_get:
+    with patch('etl_to_crawler.oss._http_get') as mock_get:
         mock_get.return_value.content = fake_content
         mock_get.return_value.raise_for_status = MagicMock()
         result, audio_md5 = transfer_url_with_md5(OTHER_URL, "crawler/qq/audio/abc.mp3", bucket, BASE_URL)

 from unittest.mock import MagicMock
+import time
 
 from etl_to_crawler import runner
 
@@ -35,13 +36,38 @@ class _Connection:
         return None
 
 
+def test_run_io_tasks_returns_named_results():
+    def slow(value):
+        time.sleep(0.01)
+        return value
+
+    result = runner._run_io_tasks({
+        'audio': lambda: slow(('audio-url', 'md5')),
+        'cover': lambda: slow('cover-url'),
+    })
+
+    assert result == {
+        'audio': ('audio-url', 'md5'),
+        'cover': 'cover-url',
+    }
+
+
 def test_run_imports_only_pending_yinyan_platform_record(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()
+    prepare = MagicMock(return_value={
+        'platform': '2',
+        'platform_song_id': 200,
+        'result': {'platform': 'kugou', 'platform_song_id': 200, 'hash': 'kg-hash', 'title': '歌'},
+        'yinyan_record': {'song_id': 10, 'record_id': 200, 'platform': '2', 'platform_song_id': 200},
+        'singers': [],
+        'albums': [],
+        'songs': [],
+        'singer_songs': [],
+        'singer_albums': [],
+    })
+    write_payloads = MagicMock(return_value=[
+        {'platform': 'kugou', 'platform_song_id': 200, 'hash': 'kg-hash', 'title': '歌'},
+    ])
 
     monkeypatch.setattr(runner, 'get_hk_songs_conn', lambda: _Connection())
     monkeypatch.setattr(runner, 'get_source_conn', lambda: _Connection())
@@ -57,47 +83,32 @@ def test_run_imports_only_pending_yinyan_platform_record(monkeypatch):
         'audio_url': 'https://example.com/a.mp3',
         'singer': '歌手',
     }})
-    platform_records = [
-        {
-            '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',
-        },
-    ]
+    platform_records = [{
+        '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, 'fetch_platform_records', lambda conn, song_ids: platform_records)
     monkeypatch.setattr(runner, 'fetch_all_platform_records', MagicMock())
     monkeypatch.setattr(runner, 'fetch_platform_records_by_record_ids', lambda conn, record_ids: platform_records)
-    monkeypatch.setattr(runner, '_PROCESSORS', processors)
-    monkeypatch.setattr(runner, 'upsert_yinyan_song_records', yinyan_writer)
+    monkeypatch.setattr(runner, 'fetch_kugou_songs', lambda conn, song_ids: {
+        200: {'id': 200},
+    })
+    monkeypatch.setattr(runner, 'fetch_kugou_singers', lambda conn, song_ids: {})
+    monkeypatch.setattr(runner, '_prepare_import_payload', prepare)
+    monkeypatch.setattr(runner, '_write_import_payloads', write_payloads)
 
     runner.run(['1', '2'])
 
-    processors['1'].assert_not_called()
-    processors['2'].assert_called_once()
+    prepare.assert_called_once()
+    write_payloads.assert_called_once()
     runner.fetch_all_platform_records.assert_not_called()
-    yinyan_writer.assert_called_once_with(pg_conn.cur, [{
-        'song_id': 10,
-        'record_id': 200,
-        'platform': '2',
-        'platform_song_id': 200,
-    }])
     assert pg_conn.commits == 1
 
 
@@ -295,6 +306,68 @@ def test_process_netease_builds_album_json_for_song_insert(monkeypatch):
     assert inserted_songs[0]['version'] == 'DJ默涵版'
 
 
+def test_process_netease_uses_prefetched_song_and_singers(monkeypatch):
+    pg_cur = MagicMock()
+    pg_cur.fetchone.return_value = ('song-uuid',)
+    inserted_songs = []
+    fetch_songs = MagicMock()
+    fetch_singers = MagicMock()
+
+    monkeypatch.setattr(runner, 'fetch_netease_songs', fetch_songs)
+    monkeypatch.setattr(runner, 'fetch_netease_singers', fetch_singers)
+    monkeypatch.setattr(runner, '_safe_transfer', lambda url, oss_key, bucket, base_url: url)
+    monkeypatch.setattr(runner, '_safe_transfer_audio', lambda url, oss_key, bucket, base_url: (url, 'audio-md5'))
+    monkeypatch.setattr(runner, 'upsert_netease_singers', lambda cur, singers: None)
+    monkeypatch.setattr(runner, 'upsert_netease_albums', lambda cur, albums: None)
+    monkeypatch.setattr(runner, 'upsert_netease_songs', lambda cur, songs: inserted_songs.extend(songs))
+    monkeypatch.setattr(runner, 'upsert_netease_singer_songs', lambda cur, pairs: None)
+
+    runner._process_netease(
+        {
+            'name': '词曲名',
+            'audio_url': 'https://example.com/audio.mp3',
+            'lyrics_url': 'https://example.com/lyric.lrc',
+            'cover_url': '',
+            'composer': '词曲曲作者',
+            'lyricist': '词曲词作者',
+            'issue_time': '2019-01-01',
+            'song_time': 120,
+        },
+        {'platform_unique_key': '300'},
+        spider_conn=object(),
+        pg_cur=pg_cur,
+        bucket=object(),
+        base_url='https://bucket.example.com',
+        song_data={
+            'id': 300,
+            'album_id': None,
+            'cover': 'https://example.com/cover.jpg',
+            'title': '录音标题',
+            'duration': 180,
+            'lyric': '[00:01.00]歌词',
+            'composer_name': '曲作者',
+            'lyricist_name': '词作者',
+            'platform_index_url': None,
+            'published_at': '2020-01-02',
+        },
+        singer_list=[{
+            'singer_id': 1,
+            'name': '歌手',
+            'avatar': '',
+            'sex': 'U',
+            'area': '其他',
+            'index': '#',
+            'intro': None,
+            'home_url': None,
+        }],
+    )
+
+    fetch_songs.assert_not_called()
+    fetch_singers.assert_not_called()
+    assert inserted_songs[0]['platform_song_id'] == 300
+    assert '"name": "歌手"' in inserted_songs[0]['singers_json']
+
+
 def test_process_netease_keeps_timestamped_lyric_and_uploads_plain_lyric(monkeypatch):
     pg_cur = MagicMock()
     pg_cur.fetchone.return_value = ('song-uuid',)

@@ -387,3 +387,42 @@ def test_upsert_netease_entities_write_provider_and_source_data():
     assert 'version' in sql
     assert 'provider_name, crawler_source_data' in sql
     assert rows[0][-4:] == ('DJ默涵版', 'md5-300', 'yinyan', '{"id": 300}')
+
+
+def test_upsert_singers_normalizes_invalid_enum_values():
+    cur = MagicMock()
+    upsert_netease_singers(cur, [{
+        'id': 1,
+        'name': '歌手',
+        'avatar': 'https://example.com/avatar.jpg',
+        'sex': '0',
+        'area': '未知地区',
+        'index': '0',
+        'intro': '简介',
+        'home_url': 'https://example.com/singer',
+        'provider_name': 'yinyan',
+        'crawler_source_data': '{"id": 1}',
+    }])
+
+    _, rows = cur.executemany.call_args[0]
+    assert rows[0][3:6] == ('U', '其他', '#')
+
+
+def test_upsert_singers_preserves_valid_group_sex_enum_value():
+    cur = MagicMock()
+    upsert_qq_singers(cur, [{
+        'id': 1,
+        'mid': 'singer-mid',
+        'name': '组合',
+        'avatar': 'https://example.com/avatar.jpg',
+        'sex': 'C',
+        'area': '华语',
+        'index': 'z',
+        'intro': '简介',
+        'home_url': 'https://example.com/singer',
+        'provider_name': 'yinyan',
+        'crawler_source_data': '{"id": 1}',
+    }])
+
+    _, rows = cur.executemany.call_args[0]
+    assert rows[0][4:7] == ('C', '华语', 'Z')