feat(etl): 实现 hk_songs_test 读取与 hikoon-data 关联
Showing
1 changed file
with
66 additions
and
0 deletions
etl_to_crawler/reader.py
0 → 100644
| 1 | from typing import Iterator | ||
| 2 | import pymysql | ||
| 3 | from .config import PLATFORMS | ||
| 4 | |||
| 5 | |||
| 6 | _HK_SONGS_QUERY = """ | ||
| 7 | SELECT id, name, lyricist, composer, audio_url, lyrics_url, | ||
| 8 | cover_url, singer, issue_time, source_song_id | ||
| 9 | FROM hk_songs_test | ||
| 10 | WHERE deleted = '0' | ||
| 11 | AND name IS NOT NULL AND name != '' | ||
| 12 | AND audio_url IS NOT NULL AND audio_url != '' | ||
| 13 | AND singer IS NOT NULL AND singer != '' | ||
| 14 | ORDER BY id | ||
| 15 | LIMIT %s OFFSET %s | ||
| 16 | """ | ||
| 17 | |||
| 18 | _PLATFORM_QUERY = """ | ||
| 19 | SELECT | ||
| 20 | sar.song_id AS source_song_id, | ||
| 21 | mr.platform, | ||
| 22 | mr.platform_unique_key, | ||
| 23 | mr.platform_mid, | ||
| 24 | mr.album_audio_id, | ||
| 25 | sar.is_main_version | ||
| 26 | FROM hk_song_and_record sar | ||
| 27 | JOIN hk_music_record mr ON mr.id = sar.record_id | ||
| 28 | WHERE sar.song_id IN ({placeholders}) | ||
| 29 | AND mr.platform IN ('1','2','4') | ||
| 30 | AND mr.platform_unique_key IS NOT NULL | ||
| 31 | AND mr.platform_unique_key != '' | ||
| 32 | AND mr.deleted = 0 | ||
| 33 | ORDER BY sar.song_id, mr.platform, sar.is_main_version DESC | ||
| 34 | """ | ||
| 35 | |||
| 36 | |||
| 37 | def iter_hk_songs_batches(conn: pymysql.Connection, batch_size: int) -> Iterator[list[dict]]: | ||
| 38 | offset = 0 | ||
| 39 | with conn.cursor() as cur: | ||
| 40 | while True: | ||
| 41 | cur.execute(_HK_SONGS_QUERY, (batch_size, offset)) | ||
| 42 | rows = cur.fetchall() | ||
| 43 | if not rows: | ||
| 44 | break | ||
| 45 | yield rows | ||
| 46 | if len(rows) < batch_size: | ||
| 47 | break | ||
| 48 | offset += batch_size | ||
| 49 | |||
| 50 | |||
| 51 | def fetch_platform_records(source_conn: pymysql.Connection, song_ids: list[int]) -> list[dict]: | ||
| 52 | if not song_ids: | ||
| 53 | return [] | ||
| 54 | placeholders = ','.join(['%s'] * len(song_ids)) | ||
| 55 | query = _PLATFORM_QUERY.format(placeholders=placeholders) | ||
| 56 | with source_conn.cursor() as cur: | ||
| 57 | cur.execute(query, song_ids) | ||
| 58 | rows = cur.fetchall() | ||
| 59 | |||
| 60 | # 每个 (source_song_id, platform) 只保留 is_main_version 最高的一条 | ||
| 61 | seen = {} | ||
| 62 | for row in rows: | ||
| 63 | key = (row['source_song_id'], row['platform']) | ||
| 64 | if key not in seen: | ||
| 65 | seen[key] = row | ||
| 66 | return list(seen.values()) |
-
Please register or sign in to post a comment