feat(etl): 实现 hikoon-data-spider 三平台数据获取

+import pymysql
+
+
+def _ids_query(query_template: str, ids: list, conn: pymysql.Connection) -> list[dict]:
+    if not ids:
+        return []
+    placeholders = ','.join(['%s'] * len(ids))
+    query = query_template.format(placeholders=placeholders)
+    with conn.cursor() as cur:
+        cur.execute(query, ids)
+        return cur.fetchall()
+
+
+# ─── QQ Music ────────────────────────────────────────────────────────────────
+
+_QQ_SONGS_SQL = """
+SELECT s.id, s.mid, s.album_id, s.cover, s.title, s.duration,
+       s.lyric, s.composer_name, s.lyricist_name, s.platform_index_url, s.published_at,
+       a.mid AS album_mid, a.cover AS album_cover, a.title AS album_title,
+       a.intro AS album_intro, a.type AS album_type, a.company_id,
+       a.company, a.is_owner, a.published_at AS album_published_at
+FROM media_tencent_songs s
+LEFT JOIN media_tencent_albums a ON a.id = s.album_id
+WHERE s.mid IN ({placeholders})
+"""
+
+_QQ_SINGERS_SQL = """
+SELECT shs.song_id, shs.singer_id, sg.mid, sg.name, sg.avatar,
+       sg.sex, sg.area, sg.`index`, sg.intro, sg.home_url
+FROM media_tencent_singer_has_songs shs
+JOIN media_tencent_singers sg ON sg.id = shs.singer_id
+WHERE shs.song_id IN ({placeholders})
+"""
+
+
+def fetch_qq_songs(conn: pymysql.Connection, mids: list[str]) -> dict[str, dict]:
+    rows = _ids_query(_QQ_SONGS_SQL, mids, conn)
+    return {row['mid']: row for row in rows}
+
+
+def fetch_qq_singers(conn: pymysql.Connection, song_ids: list[int]) -> dict[int, list[dict]]:
+    rows = _ids_query(_QQ_SINGERS_SQL, song_ids, conn)
+    result: dict[int, list] = {}
+    for row in rows:
+        result.setdefault(row['song_id'], []).append(row)
+    return result
+
+
+# ─── Kugou ───────────────────────────────────────────────────────────────────
+
+_KUGOU_SONGS_SQL = """
+SELECT s.id, s.hid, s.album_audio_id, s.album_id, s.cover, s.title, s.duration,
+       s.lyric, s.composer_name, s.lyricist_name, s.platform_index_url, s.published_at,
+       a.cover AS album_cover, a.title AS album_title, a.intro AS album_intro,
+       a.type AS album_type, a.company_id, a.company, a.is_owner,
+       a.published_at AS album_published_at
+FROM media_ku_gou_songs s
+LEFT JOIN media_ku_gou_albums a ON a.id = s.album_id
+WHERE s.id IN ({placeholders})
+"""
+
+_KUGOU_SINGERS_SQL = """
+SELECT shs.song_id, shs.singer_id, sg.name, sg.avatar,
+       sg.sex, sg.area, sg.`index`, sg.intro, sg.home_url
+FROM media_ku_gou_singer_has_songs shs
+JOIN media_ku_gou_singers sg ON sg.id = shs.singer_id
+WHERE shs.song_id IN ({placeholders})
+"""
+
+
+def fetch_kugou_songs(conn: pymysql.Connection, song_ids: list[int]) -> dict[int, dict]:
+    rows = _ids_query(_KUGOU_SONGS_SQL, song_ids, conn)
+    return {row['id']: row for row in rows}
+
+
+def fetch_kugou_singers(conn: pymysql.Connection, song_ids: list[int]) -> dict[int, list[dict]]:
+    rows = _ids_query(_KUGOU_SINGERS_SQL, song_ids, conn)
+    result: dict[int, list] = {}
+    for row in rows:
+        result.setdefault(row['song_id'], []).append(row)
+    return result
+
+
+# ─── Netease ─────────────────────────────────────────────────────────────────
+
+_NETEASE_SONGS_SQL = """
+SELECT s.id, s.album_id, s.cover, s.title, s.duration,
+       s.lyric, s.composer_name, s.lyricist_name, s.platform_index_url, s.published_at,
+       a.cover AS album_cover, a.title AS album_title, a.intro AS album_intro,
+       a.type AS album_type, a.company_id, a.company, a.is_owner,
+       a.published_at AS album_published_at
+FROM media_netease_songs s
+LEFT JOIN media_netease_albums a ON a.id = s.album_id
+WHERE s.id IN ({placeholders})
+"""
+
+_NETEASE_SINGERS_SQL = """
+SELECT shs.song_id, shs.singer_id, sg.name, sg.avatar,
+       sg.sex, sg.area, sg.`index`, sg.intro, sg.home_url
+FROM media_netease_singer_has_songs shs
+JOIN media_netease_singers sg ON sg.id = shs.singer_id
+WHERE shs.song_id IN ({placeholders})
+"""
+
+
+def fetch_netease_songs(conn: pymysql.Connection, song_ids: list[int]) -> dict[int, dict]:
+    rows = _ids_query(_NETEASE_SONGS_SQL, song_ids, conn)
+    return {row['id']: row for row in rows}
+
+
+def fetch_netease_singers(conn: pymysql.Connection, song_ids: list[int]) -> dict[int, list[dict]]:
+    rows = _ids_query(_NETEASE_SINGERS_SQL, song_ids, conn)
+    result: dict[int, list] = {}
+    for row in rows:
+        result.setdefault(row['song_id'], []).append(row)
+    return result

+from unittest.mock import MagicMock, patch
+from etl_to_crawler.spider import fetch_qq_songs, fetch_qq_singers
+
+def _make_cursor(rows):
+    cur = MagicMock()
+    cur.__enter__ = MagicMock(return_value=cur)
+    cur.__exit__ = MagicMock(return_value=False)
+    cur.fetchall.return_value = rows
+    return cur
+
+def test_fetch_qq_songs_keys_by_mid():
+    conn = MagicMock()
+    conn.cursor.return_value = _make_cursor([
+        {'mid': 'abc123', 'id': 1, 'title': '歌曲A', 'duration': 200,
+         'lyric': None, 'composer_name': None, 'lyricist_name': None,
+         'platform_index_url': None, 'published_at': None,
+         'cover': '', 'album_id': None,
+         'album_mid': None, 'album_title': None, 'album_cover': None,
+         'album_intro': None, 'album_type': None, 'company_id': 0,
+         'company': None, 'is_owner': 0, 'album_published_at': None}
+    ])
+    result = fetch_qq_songs(conn, ['abc123'])
+    assert 'abc123' in result
+    assert result['abc123']['title'] == '歌曲A'
+
+def test_fetch_qq_songs_empty_list():
+    conn = MagicMock()
+    result = fetch_qq_songs(conn, [])
+    assert result == {}
+    conn.cursor.assert_not_called()
+
+def test_fetch_qq_singers_keys_by_song_id():
+    conn = MagicMock()
+    conn.cursor.return_value = _make_cursor([
+        {'song_id': 1, 'singer_id': 10, 'mid': 'sg1', 'name': '歌手A',
+         'avatar': '', 'sex': 'M', 'area': '华语', 'index': 'G',
+         'intro': None, 'home_url': None}
+    ])
+    result = fetch_qq_singers(conn, [1])
+    assert 1 in result
+    assert result[1][0]['name'] == '歌手A'