Skip to content

沈秋雨 / import_hk_songs

Go to a project
Toggle navigation
Toggle navigation pinning
Commit bc4f610a bc4f610a0326dd65c093f4e0b858cec9a3fbd744 by 沈秋雨
Options

feat(etl): runner 运行后打印导入记录列表

1 parent 62fc7695
Inline Side-by-side
Showing 1 changed file with 17 additions and 1 deletions
...@@ -125,6 +125,7 @@ def _process_qq(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_url): ...@@ -125,6 +125,7 @@ def _process_qq(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_url):
125 upsert_qq_singer_songs(pg_cur, [(sg['singer_id'], song_uuid) for sg in singer_list]) 125 upsert_qq_singer_songs(pg_cur, [(sg['singer_id'], song_uuid) for sg in singer_list])
126 if album_id: 126 if album_id:
127 upsert_qq_singer_albums(pg_cur, [(sg['singer_id'], album_id) for sg in singer_list]) 127 upsert_qq_singer_albums(pg_cur, [(sg['singer_id'], album_id) for sg in singer_list])
128 return {'platform': 'qq', 'platform_song_id': platform_song_id, 'mid': mid, 'title': hk_row['name']}
128 129
129 130
130 def _process_kugou(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_url): 131 def _process_kugou(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_url):
...@@ -211,6 +212,7 @@ def _process_kugou(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_url ...@@ -211,6 +212,7 @@ def _process_kugou(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_url
211 upsert_kugou_singer_songs(pg_cur, [(sg['singer_id'], song_uuid) for sg in singer_list]) 212 upsert_kugou_singer_songs(pg_cur, [(sg['singer_id'], song_uuid) for sg in singer_list])
212 if album_id: 213 if album_id:
213 upsert_kugou_singer_albums(pg_cur, [(sg['singer_id'], album_id) for sg in singer_list]) 214 upsert_kugou_singer_albums(pg_cur, [(sg['singer_id'], album_id) for sg in singer_list])
215 return {'platform': 'kugou', 'platform_song_id': song_id, 'hash': sp.get('hid', ''), 'title': hk_row['name']}
214 216
215 217
216 def _process_netease(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_url): 218 def _process_netease(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_url):
...@@ -295,6 +297,7 @@ def _process_netease(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_u ...@@ -295,6 +297,7 @@ def _process_netease(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_u
295 upsert_netease_singer_songs(pg_cur, [(sg['singer_id'], song_uuid) for sg in singer_list]) 297 upsert_netease_singer_songs(pg_cur, [(sg['singer_id'], song_uuid) for sg in singer_list])
296 if album_id: 298 if album_id:
297 upsert_netease_singer_albums(pg_cur, [(sg['singer_id'], album_id) for sg in singer_list]) 299 upsert_netease_singer_albums(pg_cur, [(sg['singer_id'], album_id) for sg in singer_list])
300 return {'platform': 'netease', 'platform_song_id': song_id, 'title': hk_row['name']}
298 301
299 302
300 _PROCESSORS = { 303 _PROCESSORS = {
...@@ -313,6 +316,7 @@ def run(platforms: list[str], max_batches: int | None = None) -> None: ...@@ -313,6 +316,7 @@ def run(platforms: list[str], max_batches: int | None = None) -> None:
313 base_url = OSS_CONFIG['base_url'] 316 base_url = OSS_CONFIG['base_url']
314 317
315 total_ok = total_err = 0 318 total_ok = total_err = 0
319 imported: list[dict] = []
316 320
317 try: 321 try:
318 for i, batch in enumerate(tqdm(iter_hk_songs_batches(hk_conn, BATCH_SIZE), desc='batches')): 322 for i, batch in enumerate(tqdm(iter_hk_songs_batches(hk_conn, BATCH_SIZE), desc='batches')):
...@@ -338,9 +342,11 @@ def run(platforms: list[str], max_batches: int | None = None) -> None: ...@@ -338,9 +342,11 @@ def run(platforms: list[str], max_batches: int | None = None) -> None:
338 continue 342 continue
339 pg_cur.execute('SAVEPOINT sp_song') 343 pg_cur.execute('SAVEPOINT sp_song')
340 try: 344 try:
341 processor(hk_row, pr, spider_conn, pg_cur, bucket, base_url) 345 result = processor(hk_row, pr, spider_conn, pg_cur, bucket, base_url)
342 pg_cur.execute('RELEASE SAVEPOINT sp_song') 346 pg_cur.execute('RELEASE SAVEPOINT sp_song')
343 total_ok += 1 347 total_ok += 1
348 if result:
349 imported.append(result)
344 except Exception as e: 350 except Exception as e:
345 pg_cur.execute('ROLLBACK TO SAVEPOINT sp_song') 351 pg_cur.execute('ROLLBACK TO SAVEPOINT sp_song')
346 pg_cur.execute('RELEASE SAVEPOINT sp_song') 352 pg_cur.execute('RELEASE SAVEPOINT sp_song')
...@@ -356,3 +362,13 @@ def run(platforms: list[str], max_batches: int | None = None) -> None: ...@@ -356,3 +362,13 @@ def run(platforms: list[str], max_batches: int | None = None) -> None:
356 pg_conn.close() 362 pg_conn.close()
357 363
358 log.info("Done. OK=%d ERR=%d", total_ok, total_err) 364 log.info("Done. OK=%d ERR=%d", total_ok, total_err)
365 if imported:
366 print(f"\n导入记录（共 {len(imported)} 条）：")
367 for r in imported:
368 p = r['platform']
369 if p == 'qq':
370 print(f" [QQ] platform_song_id={r['platform_song_id']} mid={r['mid']} title={r['title']}")
371 elif p == 'kugou':
372 print(f" [酷狗] platform_song_id={r['platform_song_id']} hash={r['hash']} title={r['title']}")
373 elif p == 'netease':
374 print(f" [网易] platform_song_id={r['platform_song_id']} title={r['title']}")
......