spider.py
4.61 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
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