from unittest.mock import MagicMock
import time
import pytest

from etl_to_crawler import runner


class _Cursor:
    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc, tb):
        return False

    def execute(self, *args, **kwargs):
        return None


class _PgConnection:
    def __init__(self):
        self.cur = _Cursor()
        self.commits = 0
        self.closed = False

    def cursor(self):
        return self.cur

    def commit(self):
        self.commits += 1

    def close(self):
        self.closed = True


class _Connection:
    def close(self):
        return None

    def cursor(self):
        return _Cursor()

    def commit(self):
        return None


def test_run_io_tasks_returns_named_results():
    def slow(value):
        time.sleep(0.01)
        return value

    result = runner._run_io_tasks({
        'audio': lambda: slow(('audio-url', 'md5')),
        'cover': lambda: slow('cover-url'),
    })

    assert result == {
        'audio': ('audio-url', 'md5'),
        'cover': 'cover-url',
    }


def test_safe_audio_transfer_returns_empty_url_after_final_failure(monkeypatch):
    monkeypatch.setattr(runner, 'transfer_url_with_md5', MagicMock(side_effect=TimeoutError('timeout')))
    assert runner._safe_transfer_audio('https://source.example/audio.mp3', 'audio.mp3', object(), 'https://archive.example') == ('', '')


def test_safe_transfer_returns_empty_url_after_final_failure(monkeypatch):
    monkeypatch.setattr(runner, 'transfer_url', MagicMock(side_effect=TimeoutError('timeout')))
    assert runner._safe_transfer('https://source.example/cover.jpg', 'cover.jpg', object(), 'https://archive.example') == ''


def test_required_asset_validation_rejects_partial_transfer():
    with pytest.raises(runner.RequiredAssetTransferError, match='audio, lyric'):
        runner._require_primary_assets({
            'audio': ('', ''),
            'cover': 'https://archive.example/cover.jpg',
            'lyric': '',
        })


def test_records_import_requires_audio_and_lyric_but_allows_empty_cover(monkeypatch):
    monkeypatch.setattr(runner, 'YINYAN_IMPORT_TABLE', 'yinyan_song_records')

    runner._require_primary_assets({
        'audio': ('https://archive.example/audio.mp3', 'md5'),
        'cover': '',
        'lyric': 'https://archive.example/lyric.txt',
    }, runner._required_import_assets())

    with pytest.raises(runner.RequiredAssetTransferError, match='lyric'):
        runner._require_primary_assets({
            'audio': ('https://archive.example/audio.mp3', 'md5'),
            'cover': '',
            'lyric': '',
        }, runner._required_import_assets())


def test_records2_import_allows_empty_cover_and_lyric(monkeypatch):
    monkeypatch.setattr(runner, 'YINYAN_IMPORT_TABLE', 'yinyan_song_records2')

    runner._require_primary_assets({
        'audio': ('https://archive.example/audio.mp3', 'md5'),
        'cover': '',
        'lyric': '',
    }, runner._required_import_assets())


def test_lyric_prepare_prefers_hk_text_for_both_url_and_text_column(monkeypatch):
    download = MagicMock(return_value='[00:01.00]HK lyric')
    monkeypatch.setattr(runner, 'download_text_url', download)
    upload = MagicMock(return_value='https://archive.example/crawler/lyric/qq/mid.txt')
    monkeypatch.setattr(runner, 'upload_plain_lyric_to_bucket', upload)
    bucket = object()

    result = runner._safe_prepare_lyric(
        'qq', 'mid', 'spider lyric', 'https://source.example/lyric.txt', bucket, 'https://archive.example',
    )

    assert result == (
        'https://archive.example/crawler/lyric/qq/mid.txt',
        '[00:01.00]HK lyric',
    )
    download.assert_called_once_with('https://source.example/lyric.txt', 'https://archive.example')
    upload.assert_called_once_with(
        'qq', 'mid', '[00:01.00]HK lyric', bucket, 'https://archive.example',
    )


def test_lyric_prepare_falls_back_to_spider_text_when_hk_download_fails(monkeypatch):
    monkeypatch.setattr(runner, 'download_text_url', MagicMock(side_effect=TimeoutError('timeout')))
    upload = MagicMock(return_value='https://archive.example/crawler/lyric/qq/mid.txt')
    monkeypatch.setattr(runner, 'upload_plain_lyric_to_bucket', upload)

    assert runner._safe_prepare_lyric(
        'qq', 'mid', 'spider lyric', 'https://source.example/lyric.txt', object(), 'https://archive.example',
    ) == ('https://archive.example/crawler/lyric/qq/mid.txt', 'spider lyric')
    assert upload.call_args.args[2] == 'spider lyric'


def test_qq_cover_source_ignores_missing_album_placeholder_from_hk():
    assert runner._qq_cover_source(
        {'cover_url': runner.QQ_MISSING_ALBUM_COVER},
        {'cover': 'https://example.com/valid-qq-cover.jpg'},
    ) == 'https://example.com/valid-qq-cover.jpg'


def test_qq_cover_source_keeps_valid_hk_cover():
    assert runner._qq_cover_source(
        {'cover_url': 'https://example.com/hk-cover.jpg'},
        {'cover': 'https://example.com/qq-cover.jpg'},
    ) == 'https://example.com/hk-cover.jpg'


def test_select_qq_import_record_replaces_missing_album_cover_with_same_source_candidate():
    current = {'record_id': 10, 'platform_unique_key': 'bad-mid'}
    candidate = {'record_id': 20, 'platform_unique_key': 'good-mid'}
    selected, reason = runner._select_qq_import_record(
        {'cover_url': runner.QQ_MISSING_ALBUM_COVER},
        current,
        [current, candidate],
        {
            'bad-mid': {'album_id': 0, 'cover': runner.QQ_MISSING_ALBUM_COVER, 'lyric': '歌词'},
            'good-mid': {'album_id': 100, 'cover': 'https://example.com/good.jpg', 'lyric': '歌词'},
        },
    )
    assert selected == candidate
    assert reason == 'ok'


def test_select_qq_import_record_replaces_missing_lyric_with_same_source_candidate():
    current = {'record_id': 10, 'platform_unique_key': 'without-lyric'}
    candidate = {'record_id': 20, 'platform_unique_key': 'with-lyric'}
    selected, reason = runner._select_qq_import_record(
        {'cover_url': 'https://example.com/hk-cover.jpg'},
        current,
        [current, candidate],
        {
            'without-lyric': {'album_id': 100, 'cover': 'https://example.com/current.jpg', 'lyric': ''},
            'with-lyric': {'album_id': 100, 'cover': 'https://example.com/candidate.jpg', 'lyric': '有效歌词'},
        },
    )
    assert selected == candidate
    assert reason == 'ok'


def test_select_qq_import_record_accepts_hk_lyric_when_spider_lyric_is_empty():
    current = {'record_id': 10, 'platform_unique_key': 'without-spider-lyric'}
    selected, reason = runner._select_qq_import_record(
        {'lyrics_url': 'https://archive-dev.example/lyrics/song.txt', 'cover_url': 'https://example.com/cover.jpg'},
        current,
        [current],
        {
            'without-spider-lyric': {
                'album_id': 100,
                'cover': 'https://example.com/cover.jpg',
                'lyric': '',
            },
        },
    )

    assert selected == current
    assert reason == 'ok'


def test_select_qq_import_record_returns_none_when_no_candidate_satisfies_missing_fields():
    current = {'record_id': 10, 'platform_unique_key': 'bad-mid'}
    result, reason = runner._select_qq_import_record(
        {'cover_url': runner.QQ_MISSING_ALBUM_COVER},
        current,
        [current],
        {'bad-mid': {'album_id': 0, 'cover': runner.QQ_MISSING_ALBUM_COVER, 'lyric': ''}},
    )
    assert result is None
    assert 'lyric' in reason and 'cover' in reason


def test_records_selection_does_not_reject_or_replace_for_missing_cover():
    current = {'record_id': 10, 'platform_unique_key': 'current-mid'}

    selected, reason = runner._select_qq_import_record(
        {'cover_url': '', 'lyrics_url': 'https://example.com/hk-lyric.lrc'},
        current,
        [current],
        {'current-mid': {'album_id': None, 'cover': '', 'lyric': ''}},
        required_assets=('audio', 'lyric'),
    )

    assert selected == current
    assert reason == 'ok'


def test_records2_selection_accepts_current_record_with_empty_cover_and_lyric():
    current = {'record_id': 10, 'platform_unique_key': 'current-mid'}

    selected, reason = runner._select_qq_import_record(
        {'cover_url': '', 'lyrics_url': ''},
        current,
        [current],
        {'current-mid': {'album_id': None, 'cover': '', 'lyric': ''}},
        required_assets=('audio',),
    )

    assert selected == current
    assert reason == 'ok'


@pytest.mark.parametrize(
    ('platform', 'songs_table', 'song_upsert_name', 'relation_upsert_name'),
    [
        (runner.PLATFORM_QQ, 'crawler_qqmusic_songs', 'upsert_qq_songs', 'upsert_qq_singer_songs'),
        (runner.PLATFORM_KUGOU, 'crawler_kugou_songs', 'upsert_kugou_songs', 'upsert_kugou_singer_songs'),
        (runner.PLATFORM_NETEASE, 'crawler_netease_songs', 'upsert_netease_songs', 'upsert_netease_singer_songs'),
    ],
)
def test_write_import_payloads_resolves_existing_song_uuid_for_all_platforms(
    monkeypatch, platform, songs_table, song_upsert_name, relation_upsert_name,
):
    cur = MagicMock()
    cur.fetchall.return_value = [(200, 'existing-song-uuid')]
    song_upsert = MagicMock()
    relation_upsert = MagicMock()
    yinyan_upsert = MagicMock()
    monkeypatch.setattr(runner, song_upsert_name, song_upsert)
    monkeypatch.setattr(runner, relation_upsert_name, relation_upsert)
    monkeypatch.setattr(runner, 'upsert_yinyan_song_records', yinyan_upsert)

    runner._write_import_payloads(cur, [{
        'platform': platform,
        'platform_song_id': 200,
        'result': {'platform_song_id': 200},
        'yinyan_record': {
            'song_id': 10, 'record_id': 20, 'platform': platform, 'platform_song_id': 200,
        },
        'singers': [],
        'albums': [],
        'songs': [
            {'song_uuid': 'temporary-uuid-1', 'platform_song_id': 200},
            {'song_uuid': 'temporary-uuid-2', 'platform_song_id': 200},
        ],
        'singer_songs': [(11, 200), (11, 200), (12, 200)],
        'singer_albums': [],
    }])

    # 同批平台歌曲去重，且全部歌手关系使用数据库实际保留的歌曲 UUID。
    assert len(song_upsert.call_args.args[1]) == 1
    relation_upsert.assert_called_once_with(
        cur, [(11, 'existing-song-uuid'), (12, 'existing-song-uuid')],
    )
    sql, params = cur.execute.call_args.args
    assert f'FROM {songs_table}' in sql
    assert params == (200,)
    yinyan_upsert.assert_called_once()


def test_write_import_payloads_does_not_mark_yinyan_success_when_song_uuid_is_missing(monkeypatch):
    cur = MagicMock()
    cur.fetchall.return_value = []
    yinyan_upsert = MagicMock()
    monkeypatch.setattr(runner, 'upsert_yinyan_song_records', yinyan_upsert)

    with pytest.raises(RuntimeError, match=r'platform_song_ids=\[200\]'):
        runner._write_import_payloads(cur, [{
            'platform': runner.PLATFORM_KUGOU,
            'platform_song_id': 200,
            'result': {'platform_song_id': 200},
            'yinyan_record': {
                'song_id': 10, 'record_id': 20, 'platform': runner.PLATFORM_KUGOU,
                'platform_song_id': 200,
            },
            'singers': [],
            'albums': [],
            'songs': [{'song_uuid': 'temporary-uuid', 'platform_song_id': 200}],
            'singer_songs': [(11, 200)],
            'singer_albums': [],
        }])

    yinyan_upsert.assert_not_called()


def test_run_imports_only_pending_yinyan_platform_record(monkeypatch):
    pg_conn = _PgConnection()
    prepare = MagicMock(return_value={
        'platform': '2',
        'platform_song_id': 200,
        'result': {'platform': 'kugou', 'platform_song_id': 200, 'hash': 'kg-hash', 'title': '歌'},
        'yinyan_record': {'song_id': 10, 'record_id': 200, 'platform': '2', 'platform_song_id': 200},
        'singers': [],
        'albums': [],
        'songs': [],
        'singer_songs': [],
        'singer_albums': [],
    })
    write_payloads = MagicMock(return_value=[
        {'platform': 'kugou', 'platform_song_id': 200, 'hash': 'kg-hash', 'title': '歌'},
    ])

    monkeypatch.setattr(runner, 'get_hk_songs_conn', lambda: _Connection())
    monkeypatch.setattr(runner, 'get_source_conn', lambda: _Connection())
    monkeypatch.setattr(runner, 'get_spider_conn', lambda: _Connection())
    monkeypatch.setattr(runner, 'get_pg_conn', lambda: pg_conn)
    monkeypatch.setattr(runner, 'get_oss_bucket', lambda: object())
    monkeypatch.setattr(runner, 'refresh_conn', lambda conn, name: conn)
    monkeypatch.setattr(runner, 'fetch_pending_yinyan_song_records', lambda cur, batch_size, platforms: [
        {'song_id': 10, 'record_id': 200, 'platform': '2'},
    ] if pg_conn.commits == 0 else [])
    monkeypatch.setattr(runner, 'fetch_hk_songs_by_source_ids', lambda conn, song_ids: {10: {
        'source_song_id': 10,
        'name': '歌',
        'audio_url': 'https://example.com/a.mp3',
        'singer': '歌手',
    }})
    platform_records = [{
        'source_song_id': 10,
        'record_id': 200,
        'platform': '2',
        'platform_unique_key': '200',
        'platform_mid': 'kg-hash',
        'album_audio_id': None,
        'is_main_version': 1,
        'is_high': 0,
        'pub_time': '2021-01-01',
    }]
    monkeypatch.setattr(runner, 'fetch_platform_records', lambda conn, song_ids: platform_records)
    monkeypatch.setattr(runner, 'fetch_all_platform_records', lambda conn, song_ids: [
        {
            'source_song_id': 10,
            'record_id': 200,
            'platform': '2',
            'platform_unique_key': '200',
            'platform_mid': 'kg-hash',
            'album_audio_id': None,
            'is_main_version': 1,
            'is_high': 0,
            'pub_time': '2021-01-01',
        },
    ])
    monkeypatch.setattr(runner, 'fetch_platform_records_by_record_ids', lambda conn, record_ids: platform_records)
    monkeypatch.setattr(runner, 'fetch_kugou_songs', lambda conn, song_ids: {
        200: {'id': 200, 'lyric': '歌词', 'cover': 'https://example.com/cover.jpg'},
    })
    monkeypatch.setattr(runner, 'fetch_kugou_singers', lambda conn, song_ids: {})
    monkeypatch.setattr(runner, '_prepare_import_payload', prepare)
    monkeypatch.setattr(runner, '_write_import_payloads', write_payloads)

    runner.run(['1', '2'])

    prepare.assert_called_once()
    write_payloads.assert_called_once()
    assert pg_conn.commits == 1


def test_run_retry_failed_resets_failures_and_restores_hk_sources(monkeypatch):
    pg_conn = _PgConnection()
    hk_conn = _Connection()
    reset = MagicMock(return_value=(2, [10, 11]))
    restore = MagicMock(return_value=2)

    monkeypatch.setattr(runner, 'get_hk_songs_conn', lambda: hk_conn)
    monkeypatch.setattr(runner, 'get_source_conn', lambda: _Connection())
    monkeypatch.setattr(runner, 'get_spider_conn', lambda: _Connection())
    monkeypatch.setattr(runner, 'get_pg_conn', lambda: pg_conn)
    monkeypatch.setattr(runner, 'get_oss_bucket', lambda: object())
    monkeypatch.setattr(runner, 'refresh_conn', lambda conn, name: conn)
    monkeypatch.setattr(runner, 'reset_failed_yinyan_song_records', reset)
    monkeypatch.setattr(runner, 'mark_hk_songs_active', restore)
    monkeypatch.setattr(runner, 'fetch_pending_yinyan_song_records', lambda cur, batch_size, platforms: [])

    runner.run(['1'], retry_failed=True)

    reset.assert_called_once_with(pg_conn.cur, ['1'])
    restore.assert_called_once_with(hk_conn, [10, 11])
    assert pg_conn.commits == 1


def test_initialize_yinyan_song_records_inserts_primary_records(monkeypatch):
    pg_conn = _PgConnection()
    inserted = []

    monkeypatch.setattr(runner, 'get_hk_songs_conn', lambda: _Connection())
    monkeypatch.setattr(runner, 'get_source_conn', lambda: _Connection())
    monkeypatch.setattr(runner, 'get_pg_conn', lambda: pg_conn)
    monkeypatch.setattr(runner, 'refresh_conn', lambda conn, name: conn)
    monkeypatch.setattr(runner, 'fetch_existing_yinyan_song_ids', lambda cur: set())

    monkeypatch.setattr(runner, 'iter_hk_songs_batches', lambda conn, batch_size: [[
        {'id': 50, 'source_song_id': 10, 'name': '歌', 'audio_url': 'https://example.com/a.mp3', 'singer': '歌手'},
    ]])
    monkeypatch.setattr(runner, 'fetch_platform_records', lambda conn, song_ids: [
        {
            'source_song_id': 10,
            'record_id': 100,
            'platform': '1',
            'platform_unique_key': 'qq-mid',
            'platform_mid': '100',
            'album_audio_id': None,
            'is_main_version': 0,
            'is_high': 0,
            'pub_time': '2020-01-01',
        },
        {
            'source_song_id': 10,
            'record_id': 200,
            'platform': '2',
            'platform_unique_key': '200',
            'platform_mid': 'kg-hash',
            'album_audio_id': None,
            'is_main_version': 1,
            'is_high': 0,
            'pub_time': '2021-01-01',
        },
    ])
    monkeypatch.setattr(runner, 'insert_yinyan_song_records', lambda cur, rows: inserted.extend(rows))

    runner.initialize_yinyan_song_records(['1', '2'])

    assert inserted == [{'song_id': 10, 'record_id': 200, 'platform': '2'}]
    assert pg_conn.commits == 1


def test_initialize_yinyan_song_records2_writes_all_relations_then_deduplicates(monkeypatch):
    pg_conn = _PgConnection()
    inserted = []
    dedupe_calls = []

    monkeypatch.setattr(runner, 'get_source_conn', lambda: _Connection())
    monkeypatch.setattr(runner, 'get_spider_conn', lambda: _Connection())
    monkeypatch.setattr(runner, 'get_pg_conn', lambda: pg_conn)
    monkeypatch.setattr(runner, 'refresh_conn', lambda conn, name: conn)
    monkeypatch.setattr(runner, 'fetch_yinyan_song_ids', lambda cur: [10])
    monkeypatch.setattr(runner, 'fetch_all_song_record_relations', lambda conn, song_ids: [
        {'source_song_id': 10, 'record_id': 100, 'platform': '1', 'platform_unique_key': 'mid100'},
        {'source_song_id': 10, 'record_id': 200, 'platform': '2', 'platform_unique_key': '200'},
    ])
    monkeypatch.setattr(runner, '_fetch_valid_singer_keys', lambda conn, rows: {'mid100', 200})
    monkeypatch.setattr(runner, 'insert_yinyan_song_records2', lambda cur, rows: inserted.extend(rows))
    monkeypatch.setattr(
        runner,
        'delete_yinyan_song_records2_existing_relations',
        lambda cur: dedupe_calls.append(True) or 1,
    )

    runner.initialize_yinyan_song_records2(['1', '2'])

    assert inserted == [
        {'song_id': 10, 'record_id': 100, 'platform': '1', 'platform_unique_key': 'mid100'},
        {'song_id': 10, 'record_id': 200, 'platform': '2', 'platform_unique_key': '200'},
    ]
    assert dedupe_calls == [True]
    assert pg_conn.commits == 2


def test_initialize_yinyan_song_records2_filters_rows_without_valid_singers(monkeypatch):
    pg_conn = _PgConnection()
    inserted = []

    monkeypatch.setattr(runner, 'get_source_conn', lambda: _Connection())
    monkeypatch.setattr(runner, 'get_spider_conn', lambda: _Connection())
    monkeypatch.setattr(runner, 'get_pg_conn', lambda: pg_conn)
    monkeypatch.setattr(runner, 'refresh_conn', lambda conn, name: conn)
    monkeypatch.setattr(runner, 'fetch_yinyan_song_ids', lambda cur: [10])
    monkeypatch.setattr(runner, 'fetch_all_song_record_relations', lambda conn, song_ids: [
        {'source_song_id': 10, 'record_id': 100, 'platform': '1', 'platform_unique_key': 'mid_good'},
        {'source_song_id': 10, 'record_id': 200, 'platform': '1', 'platform_unique_key': 'mid_bad'},
        {'source_song_id': 10, 'record_id': 300, 'platform': '2', 'platform_unique_key': '300'},
    ])
    # mid_bad 没有歌手关联，300 也没有
    monkeypatch.setattr(runner, '_fetch_valid_singer_keys', lambda conn, rows: {'mid_good'})
    monkeypatch.setattr(runner, 'insert_yinyan_song_records2', lambda cur, rows: inserted.extend(rows))
    monkeypatch.setattr(
        runner, 'delete_yinyan_song_records2_existing_relations', lambda cur: 0,
    )

    runner.initialize_yinyan_song_records2(['1', '2'])

    # 只有 mid_good 的录音被保留
    assert inserted == [
        {'song_id': 10, 'record_id': 100, 'platform': '1', 'platform_unique_key': 'mid_good'},
    ]


def test_backfill_yinyan_record_platforms_updates_missing_platform_rows(monkeypatch):
    pg_conn = _PgConnection()
    updated = []

    monkeypatch.setattr(runner, 'get_source_conn', lambda: _Connection())
    monkeypatch.setattr(runner, 'get_pg_conn', lambda: pg_conn)
    monkeypatch.setattr(runner, 'refresh_conn', lambda conn, name: conn)
    monkeypatch.setattr(runner, 'fetch_yinyan_records_missing_platform', lambda cur, batch_size: [
        {'song_id': 10, 'record_id': 100},
        {'song_id': 11, 'record_id': 101},
    ] if pg_conn.commits == 0 else [])
    monkeypatch.setattr(runner, 'fetch_record_platforms', lambda conn, record_ids: {
        100: '1',
        101: '2',
    })
    monkeypatch.setattr(runner, 'update_yinyan_record_platforms', lambda cur, rows: updated.extend(rows))

    runner.backfill_yinyan_record_platforms()

    assert updated == [
        {'song_id': 10, 'record_id': 100, 'platform': '1'},
        {'song_id': 11, 'record_id': 101, 'platform': '2'},
    ]
    assert pg_conn.commits == 1


def test_process_qq_builds_album_json_for_song_insert(monkeypatch):
    pg_cur = MagicMock()
    pg_cur.fetchone.return_value = ('song-uuid',)
    inserted_songs = []

    monkeypatch.setattr(runner, 'fetch_qq_songs', lambda conn, mids: {
        'qq-mid': {
            'id': 200,
            'mid': 'qq-mid',
            'album_id': 20,
            'album_mid': 'album-mid',
            'album_cover': 'https://example.com/album.jpg',
            'album_title': '专辑',
            'album_intro': '简介',
            'album_type': '专辑类型',
            'company_id': 7,
            'company': '唱片公司',
            'is_owner': 1,
            'album_published_at': '2020-01-01',
            'cover': 'https://example.com/cover.jpg',
            'title': '化风行万里 (DJ默涵版)',
            'duration': 180,
            'lyric': '[00:01.00]歌词',
            'composer_name': '曲作者',
            'lyricist_name': '词作者',
            'platform_index_url': None,
            'published_at': '2020-01-02',
        },
    })
    monkeypatch.setattr(runner, 'fetch_qq_singers', lambda conn, song_ids: {})
    monkeypatch.setattr(runner, '_safe_transfer', lambda url, oss_key, bucket, base_url: url)
    monkeypatch.setattr(runner, '_safe_transfer_audio', lambda url, oss_key, bucket, base_url: (url, 'audio-md5'))
    monkeypatch.setattr(
        runner, '_safe_prepare_lyric',
        lambda *args: ('https://bucket.example.com/lyric.txt', '[00:01.00]HK歌词'),
    )
    monkeypatch.setattr(runner, 'upsert_qq_singers', lambda cur, singers: None)
    monkeypatch.setattr(runner, 'upsert_qq_albums', lambda cur, albums: None)
    monkeypatch.setattr(runner, 'upsert_qq_songs', lambda cur, songs: inserted_songs.extend(songs))
    monkeypatch.setattr(runner, 'upsert_qq_singer_songs', lambda cur, pairs: None)
    monkeypatch.setattr(runner, 'upsert_qq_singer_albums', lambda cur, pairs: None)

    runner._process_qq(
        {
            'name': '词曲名',
            'audio_url': 'https://example.com/audio.mp3',
            'lyrics_url': 'https://example.com/lyric.lrc',
            'cover_url': '',
            'composer': '词曲曲作者',
            'lyricist': '词曲词作者',
            'issue_time': '2019-01-01',
            'song_time': 120,
        },
        {'platform_unique_key': 'qq-mid', 'platform_mid': '200'},
        spider_conn=object(),
        pg_cur=pg_cur,
        bucket=object(),
        base_url='https://bucket.example.com',
    )

    assert inserted_songs[0]['album_json']
    assert '"id": 20' in inserted_songs[0]['album_json']
    assert '"mid": "album-mid"' in inserted_songs[0]['album_json']
    assert '"title": "专辑"' in inserted_songs[0]['album_json']
    assert inserted_songs[0]['title'] == '化风行万里'
    assert inserted_songs[0]['version'] == 'DJ默涵版'
    assert inserted_songs[0]['lyric'] == '[00:01.00]HK歌词'
    assert inserted_songs[0]['lyric_url'] == 'https://bucket.example.com/lyric.txt'


def test_process_netease_builds_album_json_for_song_insert(monkeypatch):
    pg_cur = MagicMock()
    pg_cur.fetchone.return_value = ('song-uuid',)
    inserted_songs = []

    monkeypatch.setattr(runner, 'fetch_netease_songs', lambda conn, song_ids: {
        300: {
            'id': 300,
            'album_id': 20,
            'album_cover': 'https://example.com/album.jpg',
            'album_title': '专辑',
            'album_intro': '简介',
            'album_type': '专辑类型',
            'company_id': 7,
            'company': '唱片公司',
            'is_owner': 1,
            'album_published_at': '2020-01-01',
            'cover': 'https://example.com/cover.jpg',
            'title': '化风行万里 (DJ默涵版)',
            'duration': 180,
            'lyric': '[00:01.00]歌词',
            'composer_name': '曲作者',
            'lyricist_name': '词作者',
            'platform_index_url': None,
            'published_at': '2020-01-02',
        },
    })
    monkeypatch.setattr(runner, 'fetch_netease_singers', lambda conn, song_ids: {})
    monkeypatch.setattr(runner, '_safe_transfer', lambda url, oss_key, bucket, base_url: url)
    monkeypatch.setattr(runner, '_safe_transfer_audio', lambda url, oss_key, bucket, base_url: (url, 'audio-md5'))
    monkeypatch.setattr(
        runner, '_safe_prepare_lyric',
        lambda *args: ('https://bucket.example.com/lyric.txt', '[00:01.00]HK歌词'),
    )
    monkeypatch.setattr(runner, 'upsert_netease_singers', lambda cur, singers: None)
    monkeypatch.setattr(runner, 'upsert_netease_albums', lambda cur, albums: None)
    monkeypatch.setattr(runner, 'upsert_netease_songs', lambda cur, songs: inserted_songs.extend(songs))
    monkeypatch.setattr(runner, 'upsert_netease_singer_songs', lambda cur, pairs: None)
    monkeypatch.setattr(runner, 'upsert_netease_singer_albums', lambda cur, pairs: None)

    runner._process_netease(
        {
            'name': '词曲名',
            'audio_url': 'https://example.com/audio.mp3',
            'lyrics_url': 'https://example.com/lyric.lrc',
            'cover_url': '',
            'composer': '词曲曲作者',
            'lyricist': '词曲词作者',
            'issue_time': '2019-01-01',
            'song_time': 120,
        },
        {'platform_unique_key': '300'},
        spider_conn=object(),
        pg_cur=pg_cur,
        bucket=object(),
        base_url='https://bucket.example.com',
    )

    assert inserted_songs[0]['album_json']
    assert '"id": 20' in inserted_songs[0]['album_json']
    assert '"title": "专辑"' in inserted_songs[0]['album_json']
    assert inserted_songs[0]['title'] == '化风行万里'
    assert inserted_songs[0]['version'] == 'DJ默涵版'


def test_process_netease_uses_prefetched_song_and_singers(monkeypatch):
    pg_cur = MagicMock()
    pg_cur.fetchone.return_value = ('song-uuid',)
    inserted_songs = []
    fetch_songs = MagicMock()
    fetch_singers = MagicMock()

    monkeypatch.setattr(runner, 'fetch_netease_songs', fetch_songs)
    monkeypatch.setattr(runner, 'fetch_netease_singers', fetch_singers)
    monkeypatch.setattr(runner, '_safe_transfer', lambda url, oss_key, bucket, base_url: url)
    monkeypatch.setattr(runner, '_safe_transfer_audio', lambda url, oss_key, bucket, base_url: (url, 'audio-md5'))
    monkeypatch.setattr(
        runner, '_safe_prepare_lyric',
        lambda *args: ('https://bucket.example.com/lyric.txt', '[00:01.00]HK歌词'),
    )
    monkeypatch.setattr(runner, 'upsert_netease_singers', lambda cur, singers: None)
    monkeypatch.setattr(runner, 'upsert_netease_albums', lambda cur, albums: None)
    monkeypatch.setattr(runner, 'upsert_netease_songs', lambda cur, songs: inserted_songs.extend(songs))
    monkeypatch.setattr(runner, 'upsert_netease_singer_songs', lambda cur, pairs: None)

    runner._process_netease(
        {
            'name': '词曲名',
            'audio_url': 'https://example.com/audio.mp3',
            'lyrics_url': 'https://example.com/lyric.lrc',
            'cover_url': '',
            'composer': '词曲曲作者',
            'lyricist': '词曲词作者',
            'issue_time': '2019-01-01',
            'song_time': 120,
        },
        {'platform_unique_key': '300'},
        spider_conn=object(),
        pg_cur=pg_cur,
        bucket=object(),
        base_url='https://bucket.example.com',
        song_data={
            'id': 300,
            'album_id': None,
            'cover': 'https://example.com/cover.jpg',
            'title': '录音标题',
            'duration': 180,
            'lyric': '[00:01.00]歌词',
            'composer_name': '曲作者',
            'lyricist_name': '词作者',
            'platform_index_url': None,
            'published_at': '2020-01-02',
        },
        singer_list=[{
            'singer_id': 1,
            'name': '歌手',
            'avatar': '',
            'sex': 'U',
            'area': '其他',
            'index': '#',
            'intro': None,
            'home_url': None,
        }],
    )

    fetch_songs.assert_not_called()
    fetch_singers.assert_not_called()
    assert inserted_songs[0]['platform_song_id'] == 300
    assert '"name": "歌手"' in inserted_songs[0]['singers_json']


def test_process_netease_keeps_timestamped_lyric_and_uploads_plain_lyric(monkeypatch):
    pg_cur = MagicMock()
    pg_cur.fetchone.return_value = ('song-uuid',)
    inserted_songs = []
    uploaded = {}

    class Bucket:
        def put_object(self, key, body, headers=None):
            uploaded['key'] = key
            uploaded['body'] = body

    monkeypatch.setattr(runner, 'fetch_netease_songs', lambda conn, song_ids: {
        300: {
            'id': 300,
            'album_id': None,
            'cover': 'https://example.com/cover.jpg',
            'title': '录音标题',
            'duration': 180,
            'lyric': '[ti:歌名]\n[00:01.00]第一句\n[00:02.00]第二句',
            'composer_name': '曲作者',
            'lyricist_name': '词作者',
            'platform_index_url': None,
            'published_at': '2020-01-02',
        },
    })
    monkeypatch.setattr(runner, 'fetch_netease_singers', lambda conn, song_ids: {})
    monkeypatch.setattr(runner, '_safe_transfer', lambda url, oss_key, bucket, base_url: url)
    monkeypatch.setattr(runner, '_safe_transfer_audio', lambda url, oss_key, bucket, base_url: (url, 'audio-md5'))
    monkeypatch.setattr(runner, 'upsert_netease_singers', lambda cur, singers: None)
    monkeypatch.setattr(runner, 'upsert_netease_albums', lambda cur, albums: None)
    monkeypatch.setattr(runner, 'upsert_netease_songs', lambda cur, songs: inserted_songs.extend(songs))
    monkeypatch.setattr(runner, 'upsert_netease_singer_songs', lambda cur, pairs: None)

    runner._process_netease(
        {
            'name': '词曲名',
            'audio_url': 'https://example.com/audio.mp3',
            'lyrics_url': 'https://example.com/original.lrc',
            'cover_url': '',
            'composer': '词曲曲作者',
            'lyricist': '词曲词作者',
            'issue_time': '2019-01-01',
            'song_time': 120,
        },
        {'platform_unique_key': '300'},
        spider_conn=object(),
        pg_cur=pg_cur,
        bucket=Bucket(),
        base_url='https://bucket.example.com',
    )

    assert inserted_songs[0]['lyric'] == '[ti:歌名]\n[00:01.00]第一句\n[00:02.00]第二句'
    assert inserted_songs[0]['audio_md5'] == 'audio-md5'
    assert inserted_songs[0]['lyric_url'] == 'https://bucket.example.com/crawler/lyric/netease/300.txt'
    assert uploaded['body'].decode('utf-8') == '第一句\n第二句'