import uuid
import json
import logging
from tqdm import tqdm

from .config import PLATFORM_QQ, PLATFORM_KUGOU, PLATFORM_NETEASE, 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_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,
)
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'


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 _process_qq(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_url):
    mid = pr['platform_unique_key']
    songs_map = fetch_qq_songs(spider_conn, [mid])
    if mid not in songs_map:
        return
    sp = songs_map[mid]
    song_id_int = sp['id']

    singers_map = fetch_qq_singers(spider_conn, [song_id_int])
    singer_list = singers_map.get(song_id_int, [])

    # OSS 转移
    audio_url, audio_md5 = _safe_transfer_audio(
        hk_row['audio_url'],
        build_oss_key('qq', 'audio', mid + '.mp3'),
        bucket, base_url
    )
    cover_url = _safe_transfer(
        hk_row.get('cover_url') or sp.get('cover', ''),
        build_oss_key('qq', 'cover', str(song_id_int) + '.jpg'),
        bucket, base_url
    )
    album_cover = ''
    if sp.get('album_id') and sp.get('album_cover'):
        album_cover = _safe_transfer(
            sp['album_cover'],
            build_oss_key('qq', 'album', str(sp['album_id']) + '.jpg'),
            bucket, base_url
        )
    # 专辑封面为空时回退使用歌曲封面
    if not album_cover and cover_url:
        album_cover = cover_url

    # 歌手头像转移 + 写入 singers
    singer_rows = []
    for sg in singer_list:
        avatar = _safe_transfer(
            sg.get('avatar', ''),
            build_oss_key('qq', 'singer', sg['mid'] + '.jpg'),
            bucket, base_url
        )
        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())
    raw_lyric = sp.get('lyric') or ''
    lyric_url = _safe_upload_lyric('qq', mid, raw_lyric, hk_row.get('lyrics_url'), bucket, base_url)
    title, version = split_title_version(sp.get('title') or hk_row['name'])
    upsert_qq_songs(pg_cur, [{
        'song_uuid': song_uuid,
        '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_id = int(pr['platform_unique_key'])
    songs_map = fetch_kugou_songs(spider_conn, [song_id])
    if song_id not in songs_map:
        return
    sp = songs_map[song_id]

    singers_map = fetch_kugou_singers(spider_conn, [song_id])
    singer_list = singers_map.get(song_id, [])

    audio_url, audio_md5 = _safe_transfer_audio(
        hk_row['audio_url'],
        build_oss_key('kugou', 'audio', str(song_id) + '.mp3'),
        bucket, base_url
    )
    cover_url = _safe_transfer(
        hk_row.get('cover_url') or sp.get('cover', ''),
        build_oss_key('kugou', 'cover', str(song_id) + '.jpg'),
        bucket, base_url
    )
    album_cover = ''
    if sp.get('album_id') and sp.get('album_cover'):
        album_cover = _safe_transfer(
            sp['album_cover'],
            build_oss_key('kugou', 'album', str(sp['album_id']) + '.jpg'),
            bucket, base_url
        )
    # 专辑封面为空时回退使用歌曲封面
    if not album_cover and cover_url:
        album_cover = cover_url

    singer_rows = []
    for sg in singer_list:
        avatar = _safe_transfer(
            sg.get('avatar', ''),
            build_oss_key('kugou', 'singer', str(sg['singer_id']) + '.jpg'),
            bucket, base_url
        )
        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())
    raw_lyric = sp.get('lyric') or ''
    lyric_url = _safe_upload_lyric('kugou', str(song_id), raw_lyric, hk_row.get('lyrics_url'), bucket, base_url)
    title, version = split_title_version(sp.get('title') or hk_row['name'])
    upsert_kugou_songs(pg_cur, [{
        'song_uuid': song_uuid,
        '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_id = int(pr['platform_unique_key'])
    songs_map = fetch_netease_songs(spider_conn, [song_id])
    if song_id not in songs_map:
        return
    sp = songs_map[song_id]

    singers_map = fetch_netease_singers(spider_conn, [song_id])
    singer_list = singers_map.get(song_id, [])

    audio_url, audio_md5 = _safe_transfer_audio(
        hk_row['audio_url'],
        build_oss_key('netease', 'audio', str(song_id) + '.mp3'),
        bucket, base_url
    )
    cover_url = _safe_transfer(
        hk_row.get('cover_url') or sp.get('cover', ''),
        build_oss_key('netease', 'cover', str(song_id) + '.jpg'),
        bucket, base_url
    )
    album_cover = ''
    if sp.get('album_id') and sp.get('album_cover'):
        album_cover = _safe_transfer(
            sp['album_cover'],
            build_oss_key('netease', 'album', str(sp['album_id']) + '.jpg'),
            bucket, base_url
        )
    # 专辑封面为空时回退使用歌曲封面
    if not album_cover and cover_url:
        album_cover = cover_url

    singer_rows = []
    for sg in singer_list:
        avatar = _safe_transfer(
            sg.get('avatar', ''),
            build_oss_key('netease', 'singer', str(sg['singer_id']) + '.jpg'),
            bucket, base_url
        )
        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())
    raw_lyric = sp.get('lyric') or ''
    lyric_url = _safe_upload_lyric('netease', str(song_id), raw_lyric, hk_row.get('lyrics_url'), bucket, base_url)
    title, version = split_title_version(sp.get('title') or hk_row['name'])
    upsert_netease_songs(pg_cur, [{
        'song_uuid': song_uuid,
        '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, 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 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]
            hk_by_song = fetch_hk_songs_by_source_ids(hk_conn, song_ids)
            # 全部录音（不去重），供 singer fallback 遍历
            all_platform_records = fetch_all_platform_records(src_conn, song_ids)

            # 按 (song_id, platform) 分组，每组已按优先级排好序
            pr_by_song_platform: dict[tuple, list] = {}
            for pr in all_platform_records:
                if pr['platform'] in platforms:
                    key = (int(pr['source_song_id']), pr['platform'])
                    pr_by_song_platform.setdefault(key, []).append(pr)

            # 每首歌：从各平台各选一条最优录音（有歌手优先）
            pr_by_song: dict[int, list] = {}
            for (src_id, _platform), records in pr_by_song_platform.items():
                chosen = _pick_record_with_singer(records, spider_conn)
                if chosen:
                    pr_by_song.setdefault(src_id, []).append(chosen)

            with pg_conn.cursor() as pg_cur:
                pushed_count = 0
                for src_id in song_ids:
                    hk_row = hk_by_song.get(src_id)
                    if not hk_row:
                        continue
                    if not src_id or src_id not in pr_by_song:
                        continue
                    wrote_yinyan_record = False
                    for pr in pr_by_song[src_id]:
                        processor = _PROCESSORS.get(pr['platform'])
                        if not processor:
                            continue
                        pg_cur.execute('SAVEPOINT sp_song')
                        try:
                            result = processor(hk_row, pr, spider_conn, pg_cur, bucket, base_url)
                            if result and not wrote_yinyan_record:
                                upsert_yinyan_song_records(pg_cur, [{
                                    'song_id': src_id,
                                    'record_id': int(pr['record_id']),
                                    'platform': pr['platform'],
                                    'platform_song_id': int(result['platform_song_id']),
                                }])
                                wrote_yinyan_record = True
                                pushed_count += 1
                            pg_cur.execute('RELEASE SAVEPOINT sp_song')
                            total_ok += 1
                            if result:
                                imported.append(result)
                        except Exception as e:
                            pg_cur.execute('ROLLBACK TO SAVEPOINT sp_song')
                            pg_cur.execute('RELEASE SAVEPOINT sp_song')
                            log.error("Error processing song %s platform %s: %s",
                                      hk_row.get('name'), pr['platform'], e)
                            total_err += 1
                pg_conn.commit()
                if pushed_count == 0:
                    log.error("No yinyan_song_records rows were marked pushed in this batch; stopping to avoid retry loop")
                    break
            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']}")