reader.py
2.69 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
from typing import Iterator
import pymysql
from .config import PLATFORMS
_HK_SONGS_QUERY = """
SELECT id, name, lyricist, composer, audio_url, lyrics_url,
cover_url, singer, issue_time, source_song_id, song_time
FROM hk_songs_test
WHERE deleted = '0'
AND name IS NOT NULL AND name != ''
AND audio_url IS NOT NULL AND audio_url != ''
AND singer IS NOT NULL AND singer != ''
ORDER BY id
LIMIT %s OFFSET %s
"""
_PLATFORM_QUERY = """
SELECT
sar.song_id AS source_song_id,
mr.id AS record_id,
mr.platform,
mr.platform_unique_key,
mr.platform_mid,
mr.album_audio_id,
sar.is_main_version,
mr.is_high,
mr.pub_time
FROM hk_song_and_record sar
JOIN hk_music_record mr ON mr.id = sar.record_id
WHERE sar.song_id IN ({placeholders})
AND mr.platform IN ('1','2','4')
AND mr.platform_unique_key IS NOT NULL
AND mr.platform_unique_key != ''
AND mr.deleted = 0
ORDER BY sar.song_id,
COALESCE(sar.is_main_version, 0) DESC,
COALESCE(mr.is_high, 0) DESC,
(mr.pub_time IS NULL) ASC,
mr.pub_time ASC,
mr.id ASC
"""
def iter_hk_songs_batches(conn: pymysql.Connection, batch_size: int) -> Iterator[list[dict]]:
offset = 0
with conn.cursor() as cur:
while True:
cur.execute(_HK_SONGS_QUERY, (batch_size, offset))
rows = cur.fetchall()
if not rows:
break
yield rows
if len(rows) < batch_size:
break
offset += batch_size
def fetch_platform_records(source_conn: pymysql.Connection, song_ids: list[int]) -> list[dict]:
if not song_ids:
return []
placeholders = ','.join(['%s'] * len(song_ids))
query = _PLATFORM_QUERY.format(placeholders=placeholders)
with source_conn.cursor() as cur:
cur.execute(query, song_ids)
rows = cur.fetchall()
return select_platform_records(rows)
def select_platform_records(rows: list[dict]) -> list[dict]:
# 每个 (source_song_id, platform) 保留一条，用于继续导入多个平台的录音数据。
seen = {}
for row in sorted(rows, key=_record_priority):
key = (row['source_song_id'], row['platform'])
if key not in seen:
seen[key] = row
return list(seen.values())
def select_primary_record(rows: list[dict]) -> dict | None:
if not rows:
return None
return sorted(rows, key=_record_priority)[0]
def _record_priority(row: dict) -> tuple:
pub_time = row.get('pub_time')
return (
row['source_song_id'],
-(int(row.get('is_main_version') or 0)),
-(int(row.get('is_high') or 0)),
pub_time is None,
pub_time or '',
int(row.get('record_id') or 0),
)