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, refresh_conn, close_all_pools
from .reader import (
    iter_hk_songs_batches,
    fetch_hk_songs_by_source_ids,
    mark_hk_songs_deleted,
    fetch_platform_records,
    fetch_all_platform_records,
    fetch_all_song_record_relations,
    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,
    insert_yinyan_song_records2,
    delete_yinyan_song_records2_existing_relations,
    fetch_yinyan_song_ids,
    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,
    fetch_qq_songs_with_invalid_covers, replace_qq_song, replace_yinyan_song_record,
    delete_qq_song_and_yinyan_record, mark_yinyan_record_resource_failed,
)
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
QQ_MISSING_ALBUM_COVER = 'http://y.gtimg.cn/music/photo_new/T002R300x300M000.jpg'


class RequiredAssetTransferError(RuntimeError):
    """音频、封面或歌词未成功落到目标 OSS，当前歌曲不得入库。"""


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 ''  # 不将未转存的外链写入 crawler


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 '', ''  # 不将未转存的音频外链写入 crawler


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


def _require_primary_assets(assets: dict) -> None:
    """音频、歌曲封面、歌词均为入库必需资源，任一失败即拒绝整条记录。"""
    audio_url, _ = assets.get('audio', ('', ''))
    missing = [
        name for name, value in (
            ('audio', audio_url), ('cover', assets.get('cover')), ('lyric', assets.get('lyric')),
        ) if not value
    ]
    if missing:
        raise RequiredAssetTransferError(f"required asset transfer failed: {', '.join(missing)}")


def _json_default(value):
    return str(value)


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


def _is_usable_qq_cover(url: str | None) -> bool:
    return bool(url and str(url).strip() and str(url).strip() != QQ_MISSING_ALBUM_COVER)


def _qq_cover_source(hk_row: dict, song_data: dict) -> str:
    """HK 封面优先；QQ 的 M000 占位图视为无封面，改用当前录音的封面。"""
    hk_cover = hk_row.get('cover_url')
    return hk_cover if _is_usable_qq_cover(hk_cover) else (song_data.get('cover') or '')


def _has_usable_qq_lyric(song_data: dict) -> bool:
    return bool(str(song_data.get('lyric') or '').strip())


def _has_kugou_cover(hk_row: dict, song_data: dict) -> bool:
    """酷狗封面来源：hk_row.cover_url 优先，回退到 spider 录音封面。"""
    return bool((hk_row.get('cover_url') or '').strip() or (song_data.get('cover') or '').strip())


def _has_kugou_lyric(song_data: dict) -> bool:
    return bool(str(song_data.get('lyric') or '').strip())


def _has_netease_cover(hk_row: dict, song_data: dict) -> bool:
    """网易封面来源：hk_row.cover_url 优先，回退到 spider 录音封面。"""
    return bool((hk_row.get('cover_url') or '').strip() or (song_data.get('cover') or '').strip())


def _has_netease_lyric(song_data: dict) -> bool:
    return bool(str(song_data.get('lyric') or '').strip())


def _select_qq_import_record(
    hk_row: dict,
    current_record: dict,
    candidate_records: list[dict],
    songs_by_mid: dict[str, dict],
) -> tuple[dict | None, str]:
    """为导入选择 QQ 录音；当前录音的歌词或封面不可用时，改选同源的合格候选。
    返回 (record, reason)，record 为 None 时 reason 说明失败原因。"""
    current_song = songs_by_mid.get(current_record['platform_unique_key'])
    if not current_song:
        return None, 'spider data missing'
    needs_lyric = not _has_usable_qq_lyric(current_song)
    needs_cover = not _is_usable_qq_cover(_qq_cover_source(hk_row, current_song)) or not current_song.get('album_id')
    if not needs_lyric and not needs_cover:
        return current_record, 'ok'

    for candidate in candidate_records:
        song = songs_by_mid.get(candidate['platform_unique_key'])
        if not song:
            continue
        if needs_lyric and not _has_usable_qq_lyric(song):
            continue
        if needs_cover and (not song.get('album_id') or not _is_usable_qq_cover(song.get('cover'))):
            continue
        return candidate, 'ok'
    missing = []
    if needs_lyric:
        missing.append('lyric')
    if needs_cover:
        missing.append('cover')
    return None, f"no candidate with usable {' and '.join(missing)}"


def _select_kugou_import_record(
    hk_row: dict,
    current_record: dict,
    candidate_records: list[dict],
    songs_by_id: dict[int, dict],
) -> tuple[dict | None, str]:
    """为导入选择酷狗录音；当前录音的歌词或封面不可用时，改选同源的合格候选。
    返回 (record, reason)，record 为 None 时 reason 说明失败原因。"""
    song_id = int(current_record['platform_unique_key'])
    current_song = songs_by_id.get(song_id)
    if not current_song:
        return None, 'spider data missing'
    needs_lyric = not _has_kugou_lyric(current_song)
    needs_cover = not _has_kugou_cover(hk_row, current_song)
    if not needs_lyric and not needs_cover:
        return current_record, 'ok'

    for candidate in candidate_records:
        cid = int(candidate['platform_unique_key'])
        song = songs_by_id.get(cid)
        if not song:
            continue
        if needs_lyric and not _has_kugou_lyric(song):
            continue
        if needs_cover and not _has_kugou_cover(hk_row, song):
            continue
        return candidate, 'ok'
    missing = []
    if needs_lyric:
        missing.append('lyric')
    if needs_cover:
        missing.append('cover')
    return None, f"no candidate with usable {' and '.join(missing)}"


def _select_netease_import_record(
    hk_row: dict,
    current_record: dict,
    candidate_records: list[dict],
    songs_by_id: dict[int, dict],
) -> tuple[dict | None, str]:
    """为导入选择网易录音；当前录音的歌词或封面不可用时，改选同源的合格候选。
    返回 (record, reason)，record 为 None 时 reason 说明失败原因。"""
    song_id = int(current_record['platform_unique_key'])
    current_song = songs_by_id.get(song_id)
    if not current_song:
        return None, 'spider data missing'
    needs_lyric = not _has_netease_lyric(current_song)
    needs_cover = not _has_netease_cover(hk_row, current_song)
    if not needs_lyric and not needs_cover:
        return current_record, 'ok'

    for candidate in candidate_records:
        cid = int(candidate['platform_unique_key'])
        song = songs_by_id.get(cid)
        if not song:
            continue
        if needs_lyric and not _has_netease_lyric(song):
            continue
        if needs_cover and not _has_netease_cover(hk_row, song):
            continue
        return candidate, 'ok'
    missing = []
    if needs_lyric:
        missing.append('lyric')
    if needs_cover:
        missing.append('cover')
    return None, f"no candidate with usable {' and '.join(missing)}"


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(
            _qq_cover_source(hk_row, sp),
            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)
    _require_primary_assets(assets)
    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)
    _require_primary_assets(assets)
    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)
    _require_primary_assets(assets)
    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, [])
    if not singer_list:
        return None
    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(
            _qq_cover_source(hk_row, sp),
            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)
    _require_primary_assets(assets)
    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)
    _require_primary_assets(assets)
    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)
    _require_primary_assets(assets)
    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')):
            # 长任务连接可能断开，每批次开始前刷新
            hk_conn = refresh_conn(hk_conn, 'hk_songs')
            src_conn = refresh_conn(src_conn, 'source')
            pg_conn = refresh_conn(pg_conn, 'pg')

            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)

            # 本批次中无法匹配到任何录音的歌曲 → 软删 hk_songs_test
            matched_src_ids = {int(hk_row['source_song_id']) for hk_row in batch
                               if hk_row.get('source_song_id') and int(hk_row['source_song_id']) in pr_by_song}
            unmatched_ids = [
                int(hk_row['source_song_id']) for hk_row in batch
                if hk_row.get('source_song_id') and int(hk_row['source_song_id']) not in matched_src_ids
                and int(hk_row['source_song_id']) not in existing_song_ids
            ]
            if unmatched_ids:
                mark_hk_songs_deleted(hk_conn, unmatched_ids)
                hk_conn.commit()
                log.info("Init: marked %d unmatched songs as deleted", len(unmatched_ids))
    finally:
        close_all_pools()

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


def initialize_yinyan_song_records2(platforms: list[str], max_batches: int | None = None) -> None:
    """以 records1 全量 song_id 为范围，写入补充录音关联并去重。"""
    src_conn = get_source_conn()
    pg_conn = get_pg_conn()
    total_inserted = 0

    try:
        with pg_conn.cursor() as pg_cur:
            song_ids = fetch_yinyan_song_ids(pg_cur)
        log.info('records2 source scope: records1 song_ids=%d', len(song_ids))

        for index, start in enumerate(tqdm(range(0, len(song_ids), BATCH_SIZE), desc='init-yinyan-records2')):
            # 长任务连接可能断开，每批次开始前刷新
            src_conn = refresh_conn(src_conn, 'source')
            pg_conn = refresh_conn(pg_conn, 'pg')

            if max_batches is not None and index >= max_batches:
                break

            batch_song_ids = song_ids[start:start + BATCH_SIZE]
            relations = fetch_all_song_record_relations(src_conn, batch_song_ids)
            rows = [
                {
                    'song_id': int(relation['source_song_id']),
                    'record_id': int(relation['record_id']),
                    'platform': str(relation['platform']),
                }
                for relation in relations
                if str(relation['platform']) in platforms
            ]
            if not rows:
                continue

            with pg_conn.cursor() as pg_cur:
                insert_yinyan_song_records2(pg_cur, rows)
            pg_conn.commit()
            total_inserted += len(rows)

        # 必须在全部候选关系写入后，统一根据 records1 做去重。
        with pg_conn.cursor() as pg_cur:
            deleted = delete_yinyan_song_records2_existing_relations(pg_cur)
        pg_conn.commit()
    finally:
        close_all_pools()

    log.info(
        "Initialized yinyan_song_records2 candidates=%d, removed_existing_records1_relations=%d",
        total_inserted,
        deleted,
    )


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:
            # 长任务连接可能断开，每批次开始前刷新
            src_conn = refresh_conn(src_conn, 'source')
            pg_conn = refresh_conn(pg_conn, 'pg')

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

    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:
        # 长任务连接可能断开，每批次开始前刷新
        spider_conn = refresh_conn(spider_conn, 'spider')
        pg_conn = refresh_conn(pg_conn, 'pg')

        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:
        # 长任务连接可能断开，每批次开始前刷新
        spider_conn = refresh_conn(spider_conn, 'spider')
        pg_conn = refresh_conn(pg_conn, 'pg')

        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:
        # 长任务连接可能断开，每批次开始前刷新
        spider_conn = refresh_conn(spider_conn, 'spider')
        pg_conn = refresh_conn(pg_conn, 'pg')

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


def _is_target_oss_url(url: str | None, base_url: str) -> bool:
    return bool(url and str(url).startswith(base_url.rstrip('/') + '/'))


def backfill_qq_invalid_covers(max_batches: int | None = None) -> None:
    """替换 QQ 无专辑占位封面的录音；无候选时删除 crawler 记录并软删 HK 歌曲。"""
    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']
    batch_index = replaced = filtered = retry_later = 0
    pbar = tqdm(desc='backfill-qq-invalid-covers')
    try:
        while max_batches is None or batch_index < max_batches:
            # 长任务连接可能断开，每批次开始前刷新
            hk_conn = refresh_conn(hk_conn, 'hk_songs')
            src_conn = refresh_conn(src_conn, 'source')
            spider_conn = refresh_conn(spider_conn, 'spider')
            pg_conn = refresh_conn(pg_conn, 'pg')

            with pg_conn.cursor() as pg_cur:
                invalid_rows = fetch_qq_songs_with_invalid_covers(
                    pg_cur, QQ_MISSING_ALBUM_COVER, BATCH_SIZE,
                )
            if not invalid_rows:
                break

            source_ids = [row['song_id'] for row in invalid_rows]
            hk_by_song = fetch_hk_songs_by_source_ids(hk_conn, source_ids)
            records_by_song: dict[int, list[dict]] = {}
            for record in fetch_all_platform_records(src_conn, source_ids):
                if record['platform'] == PLATFORM_QQ:
                    records_by_song.setdefault(int(record['source_song_id']), []).append(record)

            mids = list({record['platform_unique_key'] for records in records_by_song.values() for record in records})
            songs_by_mid = fetch_qq_songs(spider_conn, mids) if mids else {}
            qq_song_ids = [song['id'] for song in songs_by_mid.values()]
            singers_by_song_id = fetch_qq_singers(spider_conn, qq_song_ids) if qq_song_ids else {}

            replacements: list[tuple[dict, dict, dict]] = []
            removals: list[dict] = []
            for row in invalid_rows:
                hk_row = hk_by_song.get(row['song_id'])
                candidates = [
                    record for record in records_by_song.get(row['song_id'], [])
                    if (song := songs_by_mid.get(record['platform_unique_key']))
                    and song.get('album_id')
                    and _is_usable_qq_cover(song.get('cover'))
                ]
                if not candidates:
                    removals.append(row)
                    continue
                if not hk_row:
                    log.warning('QQ cover backfill cannot find hk_songs_test row: source_song_id=%s', row['song_id'])
                    retry_later += 1
                    continue

                payload = None
                selected_candidate = None
                for candidate in candidates:
                    song = songs_by_mid[candidate['platform_unique_key']]
                    candidate_payload = _prepare_qq_payload(
                        hk_row, candidate, bucket, base_url, song,
                        singers_by_song_id.get(song['id'], []),
                    )
                    cover = candidate_payload['songs'][0]['cover']
                    if _is_target_oss_url(cover, base_url):
                        payload = candidate_payload
                        selected_candidate = candidate
                        break
                    log.warning(
                        'QQ cover replacement download failed: source_song_id=%s candidate_mid=%s',
                        row['song_id'], candidate['platform_unique_key'],
                    )
                if payload is None:
                    retry_later += 1
                    continue
                replacements.append((row, selected_candidate, payload))

            with pg_conn.cursor() as pg_cur:
                for row, candidate, payload in replacements:
                    song = payload['songs'][0]
                    upsert_qq_singers(pg_cur, payload['singers'])
                    upsert_qq_albums(pg_cur, payload['albums'])
                    replace_qq_song(pg_cur, row['platform_song_id'], song)
                    pg_cur.execute('DELETE FROM crawler_qqmusic_singer_songs WHERE song_id = %s', (row['id'],))
                    upsert_qq_singer_songs(pg_cur, [(sg['singer_id'], row['id']) for sg in payload['singers']])
                    upsert_qq_singer_albums(pg_cur, payload['singer_albums'])
                    replace_yinyan_song_record(
                        pg_cur, row['song_id'], row['platform_song_id'],
                        int(candidate['record_id']), int(song['platform_song_id']),
                    )
                for row in removals:
                    delete_qq_song_and_yinyan_record(pg_cur, row['song_id'], row['platform_song_id'], row['id'])
            pg_conn.commit()

            if removals:
                mark_hk_songs_deleted(hk_conn, [row['song_id'] for row in removals])
                hk_conn.commit()
            replaced += len(replacements)
            filtered += len(removals)
            batch_index += 1
            pbar.update(1)
    finally:
        pbar.close()
        close_all_pools()
    log.info('QQ invalid cover backfill: replaced=%d filtered=%d retry_later=%d', replaced, filtered, retry_later)


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:
            # 长任务连接可能断开，每批次开始前刷新
            hk_conn = refresh_conn(hk_conn, 'hk_songs')
            src_conn = refresh_conn(src_conn, 'source')
            spider_conn = refresh_conn(spider_conn, 'spider')
            pg_conn = refresh_conn(pg_conn, 'pg')

            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

            # 各平台在当前录音歌词或封面不可用时，需要同源的其他录音作为候选。
            all_source_ids = [int(pending['song_id']) for pending in pending_records]
            all_candidates = fetch_all_platform_records(src_conn, all_source_ids) if all_source_ids else []
            qq_records_by_song: dict[int, list[dict]] = {}
            kugou_records_by_song: dict[int, list[dict]] = {}
            netease_records_by_song: dict[int, list[dict]] = {}
            for candidate in all_candidates:
                sid = int(candidate['source_song_id'])
                if candidate['platform'] == PLATFORM_QQ:
                    qq_records_by_song.setdefault(sid, []).append(candidate)
                elif candidate['platform'] == PLATFORM_KUGOU:
                    kugou_records_by_song.setdefault(sid, []).append(candidate)
                elif candidate['platform'] == PLATFORM_NETEASE:
                    netease_records_by_song.setdefault(sid, []).append(candidate)

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

            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 = []
            rejected_pending: list[dict] = []
            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
                if platform == PLATFORM_QQ:
                    selected, reason = _select_qq_import_record(
                        hk_row, pr, qq_records_by_song.get(src_id, []), qq_songs_map,
                    )
                elif platform == PLATFORM_KUGOU:
                    selected, reason = _select_kugou_import_record(
                        hk_row, pr, kugou_records_by_song.get(src_id, []), kugou_songs_map,
                    )
                elif platform == PLATFORM_NETEASE:
                    selected, reason = _select_netease_import_record(
                        hk_row, pr, netease_records_by_song.get(src_id, []), netease_songs_map,
                    )
                else:
                    selected, reason = pr, 'ok'
                if selected is None:
                    log.warning(
                        'Skip import (%s): platform=%s source_song_id=%s record_id=%s',
                        reason, platform, src_id, pending['record_id'],
                    )
                    rejected_pending.append(pending)
                    continue
                if selected['record_id'] != pr['record_id']:
                    log.info(
                        'Replace import record: platform=%s source_song_id=%s old_record_id=%s new_record_id=%s',
                        platform, src_id, pr['record_id'], selected['record_id'],
                    )
                    pr = selected
                prepare_inputs.append((pending, hk_row, pr))

            payloads = []
            asset_failed_pending: list[dict] = []
            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
                                asset_failed_pending.append(pending)
                                log.warning(
                                    "Skip song because payload is empty: %s platform %s song_id=%s",
                                    hk_row.get('name'), pr['platform'], pending['song_id'],
                                )
                        except RequiredAssetTransferError as e:
                            total_err += 1
                            asset_failed_pending.append(pending)
                            log.warning(
                                "Skip song because required asset transfer failed: %s platform %s: %s",
                                hk_row.get('name'), pr['platform'], e,
                            )
                        except Exception as e:
                            total_err += 1
                            asset_failed_pending.append(pending)
                            log.error(
                                "Skip song because unexpected error: %s platform %s: %s",
                                hk_row.get('name'), pr['platform'], e,
                            )

            discarded_pending = rejected_pending + asset_failed_pending
            if payloads:
                with pg_conn.cursor() as pg_cur:
                    _write_import_payloads(pg_cur, payloads)
                    for pending in discarded_pending:
                        mark_yinyan_record_resource_failed(
                            pg_cur, int(pending['song_id']), int(pending['record_id']), str(pending['platform']),
                        )
                pg_conn.commit()
                if discarded_pending:
                    mark_hk_songs_deleted(hk_conn, [int(pending['song_id']) for pending in discarded_pending])
                    hk_conn.commit()
                imported.extend(payload['result'] for payload in payloads)
            elif discarded_pending:
                with pg_conn.cursor() as pg_cur:
                    for pending in discarded_pending:
                        mark_yinyan_record_resource_failed(
                            pg_cur, int(pending['song_id']), int(pending['record_id']), str(pending['platform']),
                        )
                pg_conn.commit()
                mark_hk_songs_deleted(hk_conn, [int(pending['song_id']) for pending in discarded_pending])
                hk_conn.commit()
            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:
        close_all_pools()

    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']}")