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
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_hk_songs_by_source_ids,
    fetch_platform_records,
    fetch_all_platform_records,
    fetch_platform_records_by_record_ids,
    fetch_record_platforms,
    select_primary_record,
)
from .spider import (
    fetch_qq_songs, fetch_qq_singers,
    fetch_kugou_songs, fetch_kugou_singers,
    fetch_netease_songs, fetch_netease_singers,
    probe_qq_has_singers, probe_kugou_has_singers, probe_netease_has_singers,
)
from .writer import (
    fetch_pending_yinyan_song_records,
    fetch_yinyan_records_missing_platform,
    insert_yinyan_song_records,
    update_yinyan_record_platforms,
    upsert_yinyan_song_records,
    fetch_existing_yinyan_song_ids,
    upsert_qq_singers, upsert_qq_albums, upsert_qq_songs,
    upsert_qq_singer_songs, upsert_qq_singer_albums,
    upsert_kugou_singers, upsert_kugou_albums, upsert_kugou_songs,
    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
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):
    try:
        return transfer_url(url, oss_key, bucket, base_url)
    except Exception as e:
        log.warning("OSS transfer failed for %s: %s", url, e)
        return url  # 失败时保留原 URL，不阻断流程


def _safe_transfer_audio(url, oss_key, bucket, base_url) -> tuple[str, str]:
    try:
        return transfer_url_with_md5(url, oss_key, bucket, base_url)
    except Exception as e:
        log.warning("Audio transfer failed for %s: %s", url, e)
        return url or '', ''


def _safe_upload_lyric(platform: str, unique_id: str, lyric: str | None, fallback_url: str | None, bucket, base_url: str) -> str:
    try:
        return upload_plain_lyric_to_bucket(platform, unique_id, lyric or '', bucket, base_url) or (fallback_url or '')
    except Exception as e:
        log.warning("Lyric upload failed for %s/%s: %s", platform, unique_id, e)
        return fallback_url or ''


def _json_default(value):
    return str(value)


def _source_json(data: dict) -> str:
    return json.dumps(data, ensure_ascii=False, default=_json_default)


def _prepare_qq_payload(hk_row: dict, pr: dict, bucket, base_url, sp: dict, singer_list: list[dict]) -> dict:
    mid = pr['platform_unique_key']
    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)

    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'):
        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
        )
    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 = assets.get(f"singer_avatar:{sg['singer_id']}", sg.get('avatar', ''))
        singer_rows.append({
            **sg,
            'id': sg['singer_id'],
            'avatar': avatar,
            'provider_name': PROVIDER_YINYAN,
            'crawler_source_data': _source_json(sg),
        })
    upsert_qq_singers(pg_cur, singer_rows)

    # singers JSONB
    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)

    # 写入 album
    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)
        upsert_qq_albums(pg_cur, [{
            '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'),
            }),
        }])

    # 写入 song
    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'])
    upsert_qq_songs(pg_cur, [{
        '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),
    }])
    # 查询实际 UUID（ON CONFLICT DO NOTHING 时使用已有 UUID）
    pg_cur.execute('SELECT id FROM crawler_qqmusic_songs WHERE platform_song_id = %s', (platform_song_id,))
    row = pg_cur.fetchone()
    if row:
        song_uuid = str(row[0])

    # singer_songs / singer_albums
    upsert_qq_singer_songs(pg_cur, [(sg['singer_id'], song_uuid) for sg in singer_list])
    if album_id:
        upsert_qq_singer_albums(pg_cur, [(sg['singer_id'], album_id) for sg in singer_list])
    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,
    song_data: dict | None = None,
    singer_list: list[dict] | None = None,
):
    song_id = int(pr['platform_unique_key'])
    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'):
        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 = []
    for sg in singer_list:
        avatar = assets.get(f"singer_avatar:{sg['singer_id']}", sg.get('avatar', ''))
        singer_rows.append({
            **sg,
            'id': sg['singer_id'],
            'avatar': avatar,
            'provider_name': PROVIDER_YINYAN,
            'crawler_source_data': _source_json(sg),
        })
    upsert_kugou_singers(pg_cur, singer_rows)

    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_id = None
    if sp.get('album_id'):
        album_id = sp['album_id']
        upsert_kugou_albums(pg_cur, [{
            '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'])
    upsert_kugou_songs(pg_cur, [{
        '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),
    }])
    # 查询实际 UUID（ON CONFLICT DO NOTHING 时使用已有 UUID）
    pg_cur.execute('SELECT id FROM crawler_kugou_songs WHERE platform_song_id = %s', (song_id,))
    row = pg_cur.fetchone()
    if row:
        song_uuid = str(row[0])

    upsert_kugou_singer_songs(pg_cur, [(sg['singer_id'], song_uuid) for sg in singer_list])
    if album_id:
        upsert_kugou_singer_albums(pg_cur, [(sg['singer_id'], album_id) for sg in singer_list])
    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,
    song_data: dict | None = None,
    singer_list: list[dict] | None = None,
):
    song_id = int(pr['platform_unique_key'])
    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'):
        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 = []
    for sg in singer_list:
        avatar = assets.get(f"singer_avatar:{sg['singer_id']}", sg.get('avatar', ''))
        singer_rows.append({
            **sg,
            'id': sg['singer_id'],
            'avatar': avatar,
            'provider_name': PROVIDER_YINYAN,
            'crawler_source_data': _source_json(sg),
        })
    upsert_netease_singers(pg_cur, singer_rows)

    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_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)
        upsert_netease_albums(pg_cur, [{
            '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'])
    upsert_netease_songs(pg_cur, [{
        '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),
    }])
    # 查询实际 UUID（ON CONFLICT DO NOTHING 时使用已有 UUID）
    pg_cur.execute('SELECT id FROM crawler_netease_songs WHERE platform_song_id = %s', (song_id,))
    row = pg_cur.fetchone()
    if row:
        song_uuid = str(row[0])

    upsert_netease_singer_songs(pg_cur, [(sg['singer_id'], song_uuid) for sg in singer_list])
    if album_id:
        upsert_netease_singer_albums(pg_cur, [(sg['singer_id'], album_id) for sg in singer_list])
    return {'platform': 'netease', 'platform_song_id': song_id, 'title': hk_row['name']}


_PROCESSORS = {
    PLATFORM_QQ: _process_qq,
    PLATFORM_KUGOU: _process_kugou,
    PLATFORM_NETEASE: _process_netease,
}

_PROBERS = {
    PLATFORM_QQ: lambda conn, pr: probe_qq_has_singers(conn, pr['platform_unique_key']),
    PLATFORM_KUGOU: lambda conn, pr: probe_kugou_has_singers(conn, int(pr['platform_unique_key'])),
    PLATFORM_NETEASE: lambda conn, pr: probe_netease_has_singers(conn, int(pr['platform_unique_key'])),
}


def _pick_record_with_singer(records: list[dict], spider_conn) -> dict | None:
    """遍历录音列表，返回第一条在 spider DB 中有歌手数据的录音；全部无歌手时返回首条。"""
    if not records:
        return None
    for pr in records:
        prober = _PROBERS.get(pr['platform'])
        if prober and prober(spider_conn, pr):
            return pr
    return records[0]  # fallback：保留优先级最高的录音，即使没有歌手


def initialize_yinyan_song_records(platforms: list[str], max_batches: int | None = None) -> None:
    hk_conn = get_hk_songs_conn()
    src_conn = get_source_conn()
    pg_conn = get_pg_conn()

    # 查询已存在的 song_id，用于跳过已处理的记录
    with pg_conn.cursor() as pg_cur:
        existing_song_ids = fetch_existing_yinyan_song_ids(pg_cur)
    log.info("已存在的 yinyan_song_records: %d 条，将跳过这些记录", len(existing_song_ids))

    total = 0
    skipped_batches = 0
    try:
        for i, batch in enumerate(tqdm(iter_hk_songs_batches(hk_conn, BATCH_SIZE), desc='init-yinyan')):
            if max_batches is not None and i >= max_batches:
                break
            # 过滤掉已存在的 song_id
            song_ids = [
                int(r['source_song_id']) for r in batch
                if r.get('source_song_id') and int(r['source_song_id']) not in existing_song_ids
            ]
            if not song_ids:
                skipped_batches += 1
                continue

            platform_records = fetch_platform_records(src_conn, song_ids)

            pr_by_song: dict[int, list] = {}
            for pr in platform_records:
                if pr['platform'] in platforms:
                    pr_by_song.setdefault(int(pr['source_song_id']), []).append(pr)

            init_rows = []
            for hk_row in batch:
                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:
                    continue
                primary_record = select_primary_record(pr_by_song[src_id])
                if primary_record:
                    init_rows.append({
                        'song_id': src_id,
                        'record_id': int(primary_record['record_id']),
                        'platform': primary_record['platform'],
                    })
                    existing_song_ids.add(src_id)  # 标记为已处理

            if init_rows:
                with pg_conn.cursor() as pg_cur:
                    insert_yinyan_song_records(pg_cur, init_rows)
                pg_conn.commit()
                total += len(init_rows)
    finally:
        hk_conn.close()
        src_conn.close()
        pg_conn.close()

    log.info("Initialized yinyan_song_records candidates=%d, skipped_batches=%d", total, skipped_batches)


def backfill_yinyan_record_platforms(max_batches: int | None = None) -> None:
    src_conn = get_source_conn()
    pg_conn = get_pg_conn()
    total = 0
    skipped = 0

    try:
        batch_index = 0
        pbar = tqdm(desc='backfill-yinyan-platforms')
        while max_batches is None or batch_index < max_batches:
            with pg_conn.cursor() as pg_cur:
                rows = fetch_yinyan_records_missing_platform(pg_cur, BACKFILL_BATCH_SIZE)
            if not rows:
                break

            record_ids = [int(row['record_id']) for row in rows if row.get('record_id')]
            platforms_by_record = fetch_record_platforms(src_conn, record_ids)
            updates = []
            for row in rows:
                platform = platforms_by_record.get(int(row['record_id']))
                if not platform:
                    skipped += 1
                    continue
                updates.append({
                    'song_id': int(row['song_id']),
                    'record_id': int(row['record_id']),
                    'platform': platform,
                })

            if updates:
                with pg_conn.cursor() as pg_cur:
                    update_yinyan_record_platforms(pg_cur, updates)
                pg_conn.commit()
                total += len(updates)
            else:
                log.error("No yinyan_song_records platform rows were backfilled in this batch; stopping to avoid retry loop")
                break

            batch_index += 1
            pbar.update(1)
        pbar.close()
    finally:
        src_conn.close()
        pg_conn.close()

    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,
) -> None:
    hk_conn = get_hk_songs_conn()
    src_conn = get_source_conn()
    spider_conn = get_spider_conn()
    pg_conn = get_pg_conn()
    bucket = get_oss_bucket()
    base_url = OSS_CONFIG['base_url']
    total_ok = total_err = 0
    imported: list[dict] = []

    try:
        batch_index = 0
        pbar = tqdm(desc='batches')
        while max_batches is None or batch_index < max_batches:
            with pg_conn.cursor() as pg_cur:
                pending_records = fetch_pending_yinyan_song_records(pg_cur, BATCH_SIZE)
            if not pending_records:
                break

            song_ids = [int(r['song_id']) for r in pending_records]
            record_ids = [int(r['record_id']) for r in pending_records]
            hk_by_song = fetch_hk_songs_by_source_ids(hk_conn, song_ids)
            platform_records = fetch_platform_records_by_record_ids(src_conn, record_ids)
            pr_by_state_key: dict[tuple[int, int, str], dict] = {}
            for pr in platform_records:
                if pr['platform'] in platforms:
                    key = (int(pr['source_song_id']), int(pr['record_id']), pr['platform'])
                    pr_by_state_key[key] = pr

            # ── 批量预取 spider 数据（整批一次 IN 查询，不逐首调用）──────────
            qq_mids = list({pr['platform_unique_key'] for pr in platform_records if pr['platform'] == PLATFORM_QQ})
            kugou_ids = list({int(pr['platform_unique_key']) for pr in platform_records if pr['platform'] == PLATFORM_KUGOU})
            netease_ids = list({int(pr['platform_unique_key']) for pr in platform_records if pr['platform'] == PLATFORM_NETEASE})

            qq_songs_map = fetch_qq_songs(spider_conn, qq_mids) if qq_mids else {}
            kugou_songs_map = fetch_kugou_songs(spider_conn, kugou_ids) if kugou_ids else {}
            netease_songs_map = fetch_netease_songs(spider_conn, netease_ids) if netease_ids else {}

            qq_db_song_ids = [v['id'] for v in qq_songs_map.values()]
            qq_singers_map = fetch_qq_singers(spider_conn, qq_db_song_ids) if qq_db_song_ids else {}
            kugou_singers_map = fetch_kugou_singers(spider_conn, kugou_ids) if kugou_ids else {}
            netease_singers_map = fetch_netease_singers(spider_conn, netease_ids) if netease_ids else {}

            songs_maps = {
                PLATFORM_QQ: qq_songs_map,
                PLATFORM_KUGOU: kugou_songs_map,
                PLATFORM_NETEASE: netease_songs_map,
            }
            singers_maps = {
                PLATFORM_QQ: qq_singers_map,
                PLATFORM_KUGOU: kugou_singers_map,
                PLATFORM_NETEASE: netease_singers_map,
            }

            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()
                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()

    finally:
        hk_conn.close()
        src_conn.close()
        spider_conn.close()
        pg_conn.close()

    log.info("Done. OK=%d ERR=%d", total_ok, total_err)
    if imported:
        print(f"\n导入记录（共 {len(imported)} 条）：")
        for r in imported:
            p = r['platform']
            if p == 'qq':
                print(f"  [QQ]      platform_song_id={r['platform_song_id']}  mid={r['mid']}  title={r['title']}")
            elif p == 'kugou':
                print(f"  [酷狗]    platform_song_id={r['platform_song_id']}  hash={r['hash']}  title={r['title']}")
            elif p == 'netease':
                print(f"  [网易]    platform_song_id={r['platform_song_id']}  title={r['title']}")