import uuid

_SINGER_INDEX_VALUES = set('ABCDEFGHIJKLMNOPQRSTUVWXYZ#')
_SINGER_SEX_VALUES = {'M', 'F', 'C', 'U'}
_SINGER_AREA_VALUES = {'华语', '欧美', '韩国', '日本', '其他'}
# 资源转存失败哨兵值：platform_song_id 设为 -1 表示 OSS 失败，区别于 NULL（待导入）和正整数（成功）
PLATFORM_SONG_ID_RESOURCE_FAILED = -1


def _singer_index(value) -> str:
    text = str(value or '').strip().upper()
    return text if text in _SINGER_INDEX_VALUES else '#'


def _singer_sex(value) -> str:
    text = str(value or '').strip().upper()
    return text if text in _SINGER_SEX_VALUES else 'U'


def _singer_area(value) -> str:
    text = str(value or '').strip()
    return text if text in _SINGER_AREA_VALUES else '其他'


def insert_yinyan_song_records(cur, records: list[dict]) -> None:
    """Initialize yinyan_song_records rows before crawler import."""
    if not records:
        return
    cur.executemany(
        """
        INSERT INTO yinyan_song_records (song_id, record_id, platform, is_yinyan_push)
        VALUES (%s, %s, %s, FALSE)
        ON CONFLICT (song_id, record_id) DO UPDATE
            SET platform = EXCLUDED.platform
            WHERE yinyan_song_records.platform IS NULL
        """,
        [(r['song_id'], r['record_id'], r['platform']) for r in records],
    )


def insert_yinyan_song_records2(cur, records: list[dict]) -> None:
    """写入全部候选关联；与 records1 的去重由后续独立步骤完成。"""
    if not records:
        return
    values_sql = ', '.join(['(%s::bigint, %s::bigint, %s::varchar)'] * len(records))
    params = []
    for record in records:
        params.extend([record['song_id'], record['record_id'], record['platform']])
    cur.execute(
        f"""
        INSERT INTO yinyan_song_records2 (song_id, record_id, platform)
        VALUES {values_sql}
        ON CONFLICT (song_id, record_id) DO UPDATE
            SET platform = EXCLUDED.platform
        """,
        tuple(params),
    )


def delete_yinyan_song_records2_existing_relations(cur) -> int:
    """删除 records2 中已存在于 records1 的 song-record 关联。"""
    cur.execute(
        """
        DELETE FROM yinyan_song_records2 AS ysr2
        USING yinyan_song_records AS ysr1
        WHERE ysr2.song_id = ysr1.song_id
          AND ysr2.record_id = ysr1.record_id
        """
    )
    return cur.rowcount


def fetch_existing_yinyan_song_ids(cur) -> set[int]:
    """返回已完成初始化 platform 的 song_id 集合。"""
    cur.execute("SELECT DISTINCT song_id FROM yinyan_song_records WHERE platform IS NOT NULL")
    return {row[0] for row in cur.fetchall()}


def fetch_pending_yinyan_song_records(cur, limit: int) -> list[dict]:
    cur.execute(
        """
        SELECT song_id, record_id, platform
        FROM yinyan_song_records
        WHERE is_yinyan_push = FALSE
          AND platform IS NOT NULL
          AND platform_song_id IS NULL
        ORDER BY song_id
        LIMIT %s
        """,
        (limit,),
    )
    rows = cur.fetchall()
    return [{'song_id': row[0], 'record_id': row[1], 'platform': row[2]} for row in rows]


def fetch_yinyan_records_missing_platform(cur, limit: int) -> list[dict]:
    cur.execute(
        """
        SELECT song_id, record_id
        FROM yinyan_song_records
        WHERE platform IS NULL
        ORDER BY song_id
        LIMIT %s
        """,
        (limit,),
    )
    rows = cur.fetchall()
    return [{'song_id': row[0], 'record_id': row[1]} for row in rows]


def update_yinyan_record_platforms(cur, records: list[dict]) -> None:
    if not records:
        return
    values_sql = ', '.join(['(%s::bigint, %s::bigint, %s::varchar)'] * len(records))
    params = []
    for r in records:
        params.extend([r['song_id'], r['record_id'], r['platform']])
    cur.execute(
        f"""
        UPDATE yinyan_song_records AS ysr
        SET platform = v.platform
        FROM (VALUES {values_sql}) AS v(song_id, record_id, platform)
        WHERE ysr.song_id = v.song_id
          AND ysr.record_id = v.record_id
          AND ysr.platform IS NULL
        """,
        tuple(params),
    )


def upsert_yinyan_song_records(cur, records: list[dict]) -> None:
    """Mark pre-initialized yinyan song-record rows as pushed to crawler.
    Matches only on song_id so singer-fallback can use a different record_id than initialized.
    """
    if not records:
        return
    values_sql = ', '.join(['(%s::bigint, %s::bigint, %s::varchar, %s::bigint)'] * len(records))
    params = []
    for r in records:
        params.extend([r['song_id'], r['record_id'], r['platform'], r['platform_song_id']])
    cur.execute(
        f"""
        UPDATE yinyan_song_records AS ysr
        SET platform = v.platform,
            platform_song_id = v.platform_song_id,
            record_id = v.record_id,
            is_yinyan_push = TRUE
        FROM (VALUES {values_sql}) AS v(song_id, record_id, platform, platform_song_id)
        WHERE ysr.song_id = v.song_id
          AND ysr.is_yinyan_push = FALSE
        """,
        tuple(params),
    )


# ─── Backfill helpers ────────────────────────────────────────────────────────

def fetch_qq_songs_missing_singers(cur, limit: int) -> list[dict]:
    cur.execute(
        """
        SELECT id, platform_song_id, mid
        FROM crawler_qqmusic_songs
        WHERE singers IS NULL OR jsonb_array_length(singers) = 0
        ORDER BY platform_song_id
        LIMIT %s
        """,
        (limit,),
    )
    rows = cur.fetchall()
    return [{'id': str(row[0]), 'platform_song_id': row[1], 'mid': row[2]} for row in rows]


def fetch_qq_songs_with_invalid_covers(cur, invalid_cover: str, limit: int) -> list[dict]:
    """返回已导入、但仍使用 QQ 无专辑占位封面的歌曲及其源歌曲状态。"""
    cur.execute(
        """
        SELECT qs.id, qs.platform_song_id, qs.mid, qs.name,
               ysr.song_id, ysr.record_id
        FROM crawler_qqmusic_songs qs
        JOIN yinyan_song_records ysr
          ON ysr.platform = '1' AND ysr.platform_song_id = qs.platform_song_id
        WHERE qs.cover = %s
          AND qs.album_id IS NULL
          AND ysr.is_yinyan_push = TRUE
        ORDER BY ysr.song_id
        LIMIT %s
        """,
        (invalid_cover, limit),
    )
    return [
        {
            'id': str(row[0]), 'platform_song_id': int(row[1]), 'mid': row[2],
            'name': row[3], 'song_id': int(row[4]), 'record_id': int(row[5]),
        }
        for row in cur.fetchall()
    ]


def replace_qq_song(cur, old_platform_song_id: int, song: dict) -> None:
    """用同一源歌曲下的另一条 QQ 录音更新既有 crawler 歌曲。"""
    cur.execute(
        """
        UPDATE crawler_qqmusic_songs
        SET platform_song_id = %s, mid = %s, album_id = %s, cover = %s,
            title = %s, name = %s, duration = %s, lyric = %s,
            composer_name = %s, lyricist_name = %s, url = %s, audio_md5 = %s,
            lyric_url = %s, platform_index_url = %s, published_at = %s,
            album = %s::json, singers = %s::jsonb, version = %s,
            provider_name = %s, crawler_source_data = %s::json, updated_at = NOW()
        WHERE platform_song_id = %s
        """,
        (
            song['platform_song_id'], song['mid'], song.get('album_id'), song.get('cover', ''),
            song.get('title', ''), song.get('name', ''), song.get('duration', 0) or 0,
            song.get('lyric'), song.get('composer_name'), song.get('lyricist_name'),
            song.get('url', ''), song.get('audio_md5'), song.get('lyric_url'),
            song.get('platform_index_url'), song.get('published_at'), song.get('album_json'),
            song.get('singers_json', '[]'), song.get('version'), song.get('provider_name'),
            song.get('crawler_source_data'), old_platform_song_id,
        ),
    )


def replace_yinyan_song_record(cur, song_id: int, old_platform_song_id: int, record_id: int, platform_song_id: int) -> None:
    cur.execute(
        """
        UPDATE yinyan_song_records
        SET record_id = %s, platform_song_id = %s, platform = '1'
        WHERE song_id = %s AND platform = '1' AND platform_song_id = %s
        """,
        (record_id, platform_song_id, song_id, old_platform_song_id),
    )


def delete_qq_song_and_yinyan_record(cur, song_id: int, platform_song_id: int, song_uuid: str) -> None:
    """移除无法找到替代录音的 crawler 数据及其导入状态。"""
    cur.execute("DELETE FROM crawler_qqmusic_singer_songs WHERE song_id = %s", (song_uuid,))
    cur.execute("DELETE FROM crawler_qqmusic_songs WHERE platform_song_id = %s", (platform_song_id,))
    cur.execute(
        "DELETE FROM yinyan_song_records WHERE song_id = %s AND platform = '1' AND platform_song_id = %s",
        (song_id, platform_song_id),
    )


def mark_yinyan_record_resource_failed(cur, song_id: int, record_id: int, platform: str) -> None:
    """保留未导入状态，并将 platform_song_id 置为哨兵值 -1，避免后续批次重复处理。"""
    cur.execute(
        """
        UPDATE yinyan_song_records
        SET is_yinyan_push = FALSE, platform_song_id = %s
        WHERE song_id = %s AND record_id = %s AND platform = %s AND is_yinyan_push = FALSE
        """,
        (PLATFORM_SONG_ID_RESOURCE_FAILED, song_id, record_id, platform),
    )


def fetch_kugou_songs_missing_singers(cur, limit: int) -> list[dict]:
    cur.execute(
        """
        SELECT id, platform_song_id
        FROM crawler_kugou_songs
        WHERE singers IS NULL OR jsonb_array_length(singers) = 0
        ORDER BY platform_song_id
        LIMIT %s
        """,
        (limit,),
    )
    rows = cur.fetchall()
    return [{'id': str(row[0]), 'platform_song_id': row[1]} for row in rows]


def fetch_netease_songs_missing_singers(cur, limit: int) -> list[dict]:
    cur.execute(
        """
        SELECT id, platform_song_id
        FROM crawler_netease_songs
        WHERE singers IS NULL OR jsonb_array_length(singers) = 0
        ORDER BY platform_song_id
        LIMIT %s
        """,
        (limit,),
    )
    rows = cur.fetchall()
    return [{'id': str(row[0]), 'platform_song_id': row[1]} for row in rows]


def update_qq_song_singers(cur, updates: list[dict]) -> None:
    """updates: [{'platform_song_id': int, 'singers_json': str}, ...]"""
    if not updates:
        return
    values_sql = ', '.join(['(%s::bigint, %s::jsonb)'] * len(updates))
    params = []
    for u in updates:
        params.extend([u['platform_song_id'], u['singers_json']])
    cur.execute(
        f"""
        UPDATE crawler_qqmusic_songs AS s
        SET singers = v.singers, updated_at = NOW()
        FROM (VALUES {values_sql}) AS v(platform_song_id, singers)
        WHERE s.platform_song_id = v.platform_song_id
        """,
        tuple(params),
    )


def update_kugou_song_singers(cur, updates: list[dict]) -> None:
    if not updates:
        return
    values_sql = ', '.join(['(%s::bigint, %s::jsonb)'] * len(updates))
    params = []
    for u in updates:
        params.extend([u['platform_song_id'], u['singers_json']])
    cur.execute(
        f"""
        UPDATE crawler_kugou_songs AS s
        SET singers = v.singers, updated_at = NOW()
        FROM (VALUES {values_sql}) AS v(platform_song_id, singers)
        WHERE s.platform_song_id = v.platform_song_id
        """,
        tuple(params),
    )


def update_netease_song_singers(cur, updates: list[dict]) -> None:
    if not updates:
        return
    values_sql = ', '.join(['(%s::bigint, %s::jsonb)'] * len(updates))
    params = []
    for u in updates:
        params.extend([u['platform_song_id'], u['singers_json']])
    cur.execute(
        f"""
        UPDATE crawler_netease_songs AS s
        SET singers = v.singers, updated_at = NOW()
        FROM (VALUES {values_sql}) AS v(platform_song_id, singers)
        WHERE s.platform_song_id = v.platform_song_id
        """,
        tuple(params),
    )


# ─── QQ Music ────────────────────────────────────────────────────────────────

def upsert_qq_singers(cur, singers: list[dict]) -> None:
    if not singers:
        return
    sql = """
        INSERT INTO crawler_qqmusic_singers
            (id, mid, name, avatar, sex, area, "index", intro, home_url,
             created_at, updated_at, provider_name, crawler_source_data)
        VALUES (%s, %s, %s, %s, %s::singer_sex, %s::singer_area, %s::singer_index, %s, %s,
                NOW(), NOW(), %s, %s::json)
        ON CONFLICT (id) DO NOTHING
    """
    rows = [(
        s['id'], s['mid'], s['name'], s.get('avatar', ''),
        _singer_sex(s.get('sex')), _singer_area(s.get('area')), _singer_index(s.get('index')),
        s.get('intro'), s.get('home_url'),
        s.get('provider_name'), s.get('crawler_source_data'),
    ) for s in singers]
    cur.executemany(sql, rows)


def upsert_qq_albums(cur, albums: list[dict]) -> None:
    if not albums:
        return
    sql = """
        INSERT INTO crawler_qqmusic_albums
            (id, mid, cover, title, intro, type, company_id, company, is_owner, published_at,
             created_at, updated_at, provider_name, crawler_source_data)
        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW(), %s, %s::json)
        ON CONFLICT (id) DO NOTHING
    """
    rows = [(
        a['id'], a.get('mid', ''), a.get('cover', ''), a.get('title', ''),
        a.get('intro'), a.get('type', ''), a.get('company_id', 0) or 0,
        a.get('company', ''), a.get('is_owner', 0) or 0, a.get('published_at'),
        a.get('provider_name'), a.get('crawler_source_data'),
    ) for a in albums]
    cur.executemany(sql, rows)


def upsert_qq_songs(cur, songs: list[dict]) -> None:
    if not songs:
        return
    sql = """
        INSERT INTO crawler_qqmusic_songs
            (id, platform_song_id, mid, album_id, cover, title, name, duration,
             lyric, composer_name, lyricist_name, url, lyric_url,
             platform_index_url, published_at, album, singers, status, created_at, updated_at,
             version, audio_md5, provider_name, crawler_source_data)
        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s::json, %s::jsonb, 0, NOW(), NOW(), %s, %s, %s, %s::json)
        ON CONFLICT (platform_song_id) DO NOTHING
    """
    rows = [(
        s['song_uuid'], s['platform_song_id'], s['mid'], s.get('album_id'),
        s.get('cover', ''), s.get('title', ''), s.get('name', ''), s.get('duration', 0) or 0,
        s.get('lyric'), s.get('composer_name'), s.get('lyricist_name'),
        s.get('url', ''), s.get('lyric_url'),
        s.get('platform_index_url'), s.get('published_at'), s.get('album_json'), s.get('singers_json', '[]'),
        s.get('version'),
        s.get('audio_md5'),
        s.get('provider_name'), s.get('crawler_source_data'),
    ) for s in songs]
    cur.executemany(sql, rows)


def upsert_qq_singer_songs(cur, pairs: list[tuple]) -> None:
    """pairs: [(singer_id, song_uuid), ...]"""
    if not pairs:
        return
    sql = """
        INSERT INTO crawler_qqmusic_singer_songs (id, singer_id, song_id)
        VALUES (%s, %s, %s)
        ON CONFLICT (singer_id, song_id) DO NOTHING
    """
    rows = [(str(uuid.uuid4()), singer_id, song_uuid) for singer_id, song_uuid in pairs]
    cur.executemany(sql, rows)


def upsert_qq_singer_albums(cur, pairs: list[tuple]) -> None:
    """pairs: [(singer_id, album_id), ...]"""
    if not pairs:
        return
    sql = """
        INSERT INTO crawler_qqmusic_singer_albums (id, singer_id, album_id)
        VALUES (%s, %s, %s)
        ON CONFLICT (singer_id, album_id) DO NOTHING
    """
    rows = [(str(uuid.uuid4()), singer_id, album_id) for singer_id, album_id in pairs]
    cur.executemany(sql, rows)


# ─── Kugou ───────────────────────────────────────────────────────────────────

def upsert_kugou_singers(cur, singers: list[dict]) -> None:
    if not singers:
        return
    sql = """
        INSERT INTO crawler_kugou_singers
            (id, name, avatar, sex, area, "index", intro, home_url,
             created_at, updated_at, provider_name, crawler_source_data)
        VALUES (%s, %s, %s, %s::kugou_singer_sex, %s::kugou_singer_area, %s::kugou_singer_index, %s, %s,
                NOW(), NOW(), %s, %s::json)
        ON CONFLICT (id) DO NOTHING
    """
    rows = [(
        s['id'], s['name'], s.get('avatar', ''),
        _singer_sex(s.get('sex')), _singer_area(s.get('area')), _singer_index(s.get('index')),
        s.get('intro'), s.get('home_url', ''),
        s.get('provider_name'), s.get('crawler_source_data'),
    ) for s in singers]
    cur.executemany(sql, rows)


def upsert_kugou_albums(cur, albums: list[dict]) -> None:
    if not albums:
        return
    sql = """
        INSERT INTO crawler_kugou_albums
            (id, cover, title, intro, type, company_id, company, is_owner, published_at,
             created_at, updated_at, provider_name, crawler_source_data)
        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW(), %s, %s::json)
        ON CONFLICT (id) DO NOTHING
    """
    rows = [(
        a['id'], a.get('cover', ''), a.get('title', ''), a.get('intro'),
        a.get('type', ''), a.get('company_id', 0) or 0,
        a.get('company', ''), a.get('is_owner', 0) or 0, a.get('published_at'),
        a.get('provider_name'), a.get('crawler_source_data'),
    ) for a in albums]
    cur.executemany(sql, rows)


def upsert_kugou_songs(cur, songs: list[dict]) -> None:
    if not songs:
        return
    sql = """
        INSERT INTO crawler_kugou_songs
            (id, platform_song_id, hash, album_audio_id, album_id, cover, title, name, duration,
             lyric, composer_name, lyricist_name, url, lyric_url,
             platform_index_url, published_at, singers, status, created_at, updated_at,
             version, audio_md5, provider_name, crawler_source_data)
        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, 0, NOW(), NOW(), %s, %s, %s, %s::json)
        ON CONFLICT (platform_song_id) DO NOTHING
    """
    rows = [(
        s['song_uuid'], s['platform_song_id'], s.get('hash', ''),
        s.get('album_audio_id', 0) or 0, s.get('album_id'),
        s.get('cover', ''), s.get('title', ''), s.get('name', ''), s.get('duration', 0) or 0,
        s.get('lyric'), s.get('composer_name'), s.get('lyricist_name'),
        s.get('url', ''), s.get('lyric_url'),
        s.get('platform_index_url'), s.get('published_at'), s.get('singers_json', '[]'),
        s.get('version'),
        s.get('audio_md5'),
        s.get('provider_name'), s.get('crawler_source_data'),
    ) for s in songs]
    cur.executemany(sql, rows)


def upsert_kugou_singer_songs(cur, pairs: list[tuple]) -> None:
    if not pairs:
        return
    sql = """
        INSERT INTO crawler_kugou_singer_songs (id, singer_id, song_id)
        VALUES (%s, %s, %s)
        ON CONFLICT (singer_id, song_id) DO NOTHING
    """
    cur.executemany(sql, [(str(uuid.uuid4()), s, sg) for s, sg in pairs])


def upsert_kugou_singer_albums(cur, pairs: list[tuple]) -> None:
    if not pairs:
        return
    sql = """
        INSERT INTO crawler_kugou_singer_albums (id, singer_id, album_id)
        VALUES (%s, %s, %s)
        ON CONFLICT (singer_id, album_id) DO NOTHING
    """
    cur.executemany(sql, [(str(uuid.uuid4()), s, a) for s, a in pairs])


# ─── Netease ─────────────────────────────────────────────────────────────────

def upsert_netease_singers(cur, singers: list[dict]) -> None:
    if not singers:
        return
    sql = """
        INSERT INTO crawler_netease_singers
            (id, name, avatar, sex, area, "index", intro, home_url,
             created_at, updated_at, provider_name, crawler_source_data)
        VALUES (%s, %s, %s, %s::netease_singer_sex, %s::netease_singer_area, %s::netease_singer_index, %s, %s,
                NOW(), NOW(), %s, %s::json)
        ON CONFLICT (id) DO NOTHING
    """
    rows = [(
        s['id'], s['name'], s.get('avatar', ''),
        _singer_sex(s.get('sex')), _singer_area(s.get('area')), _singer_index(s.get('index')),
        s.get('intro'), s.get('home_url'),
        s.get('provider_name'), s.get('crawler_source_data'),
    ) for s in singers]
    cur.executemany(sql, rows)


def upsert_netease_albums(cur, albums: list[dict]) -> None:
    if not albums:
        return
    sql = """
        INSERT INTO crawler_netease_albums
            (id, cover, title, intro, type, company_id, company, is_owner, published_at,
             created_at, updated_at, provider_name, crawler_source_data)
        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW(), %s, %s::json)
        ON CONFLICT (id) DO NOTHING
    """
    rows = [(
        a['id'], a.get('cover', ''), a.get('title', ''), a.get('intro'),
        a.get('type', ''), a.get('company_id', 0) or 0,
        a.get('company', ''), a.get('is_owner', 0) or 0, a.get('published_at'),
        a.get('provider_name'), a.get('crawler_source_data'),
    ) for a in albums]
    cur.executemany(sql, rows)


def upsert_netease_songs(cur, songs: list[dict]) -> None:
    if not songs:
        return
    sql = """
        INSERT INTO crawler_netease_songs
            (id, platform_song_id, album_id, cover, title, name, duration,
             lyric, composer_name, lyricist_name, url, lyric_url,
             platform_index_url, published_at, album, singers, status, created_at, updated_at,
             version, audio_md5, provider_name, crawler_source_data)
        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s::json, %s::jsonb, 0, NOW(), NOW(), %s, %s, %s, %s::json)
        ON CONFLICT (platform_song_id) DO NOTHING
    """
    rows = [(
        s['song_uuid'], s['platform_song_id'], s.get('album_id'),
        s.get('cover', ''), s.get('title', ''), s.get('name', ''), s.get('duration', 0) or 0,
        s.get('lyric'), s.get('composer_name'), s.get('lyricist_name'),
        s.get('url', ''), s.get('lyric_url'),
        s.get('platform_index_url'), s.get('published_at'), s.get('album_json'), s.get('singers_json', '[]'),
        s.get('version'),
        s.get('audio_md5'),
        s.get('provider_name'), s.get('crawler_source_data'),
    ) for s in songs]
    cur.executemany(sql, rows)


def upsert_netease_singer_songs(cur, pairs: list[tuple]) -> None:
    if not pairs:
        return
    sql = """
        INSERT INTO crawler_netease_singer_songs (id, singer_id, song_id)
        VALUES (%s, %s, %s)
        ON CONFLICT (singer_id, song_id) DO NOTHING
    """
    cur.executemany(sql, [(str(uuid.uuid4()), s, sg) for s, sg in pairs])


def upsert_netease_singer_albums(cur, pairs: list[tuple]) -> None:
    if not pairs:
        return
    sql = """
        INSERT INTO crawler_netease_singer_albums (id, singer_id, album_id)
        VALUES (%s, %s, %s)
        ON CONFLICT (singer_id, album_id) DO NOTHING
    """
    cur.executemany(sql, [(str(uuid.uuid4()), s, a) for s, a in pairs])