feat(etl): 实现批次编排与 CLI 入口，支持 --max-batches 冒烟测试参数

+import uuid
+import json
+import logging
+from tqdm import tqdm
+
+from .config import PLATFORM_QQ, PLATFORM_KUGOU, PLATFORM_NETEASE, BATCH_SIZE, OSS_CONFIG
+from .connections import get_hk_songs_conn, get_source_conn, get_spider_conn, get_pg_conn, get_oss_bucket
+from .reader import iter_hk_songs_batches, fetch_platform_records
+from .spider import (
+    fetch_qq_songs, fetch_qq_singers,
+    fetch_kugou_songs, fetch_kugou_singers,
+    fetch_netease_songs, fetch_netease_singers,
+)
+from .writer import (
+    upsert_qq_singers, upsert_qq_albums, upsert_qq_songs,
+    upsert_qq_singer_songs, upsert_qq_singer_albums,
+    upsert_kugou_singers, upsert_kugou_albums, upsert_kugou_songs,
+    upsert_kugou_singer_songs, upsert_kugou_singer_albums,
+    upsert_netease_singers, upsert_netease_albums, upsert_netease_songs,
+    upsert_netease_singer_songs, upsert_netease_singer_albums,
+)
+from .oss import transfer_url, build_oss_key
+from .lyric import strip_timestamps
+
+logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
+log = logging.getLogger(__name__)
+
+
+def _safe_transfer(url, oss_key, bucket, base_url):
+    try:
+        return transfer_url(url, oss_key, bucket, base_url)
+    except Exception as e:
+        log.warning("OSS transfer failed for %s: %s", url, e)
+        return url  # 失败时保留原 URL，不阻断流程
+
+
+def _process_qq(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_url):
+    mid = pr['platform_unique_key']
+    songs_map = fetch_qq_songs(spider_conn, [mid])
+    if mid not in songs_map:
+        return
+    sp = songs_map[mid]
+    song_id_int = sp['id']
+
+    singers_map = fetch_qq_singers(spider_conn, [song_id_int])
+    singer_list = singers_map.get(song_id_int, [])
+
+    # OSS 转移
+    audio_url = _safe_transfer(
+        hk_row['audio_url'],
+        build_oss_key('qq', 'audio', mid + '.mp3'),
+        bucket, base_url
+    )
+    cover_url = _safe_transfer(
+        hk_row.get('cover_url') or sp.get('cover', ''),
+        build_oss_key('qq', 'cover', str(song_id_int) + '.jpg'),
+        bucket, base_url
+    )
+    album_cover = ''
+    if sp.get('album_id') and sp.get('album_cover'):
+        album_cover = _safe_transfer(
+            sp['album_cover'],
+            build_oss_key('qq', 'album', str(sp['album_id']) + '.jpg'),
+            bucket, base_url
+        )
+
+    # 歌手头像转移 + 写入 singers
+    singer_rows = []
+    for sg in singer_list:
+        avatar = _safe_transfer(
+            sg.get('avatar', ''),
+            build_oss_key('qq', 'singer', sg['mid'] + '.jpg'),
+            bucket, base_url
+        )
+        singer_rows.append({**sg, 'id': sg['singer_id'], 'avatar': avatar})
+    upsert_qq_singers(pg_cur, singer_rows)
+
+    # singers JSONB
+    singers_json = json.dumps([{
+        'name': sg['name'],
+        'singer_id': sg['singer_id'],
+        'platform_singer_id': sg['mid'],
+    } for sg in singer_list], ensure_ascii=False)
+
+    # 写入 album
+    album_id = None
+    if sp.get('album_id'):
+        album_id = sp['album_id']
+        upsert_qq_albums(pg_cur, [{
+            'id': sp['album_id'], 'mid': sp.get('album_mid') or '',
+            'cover': album_cover, 'title': sp.get('album_title') or '',
+            'intro': sp.get('album_intro'), 'type': sp.get('album_type') or '',
+            'company_id': sp.get('company_id') or 0, 'company': sp.get('company') or '',
+            'is_owner': sp.get('is_owner') or 0, 'published_at': sp.get('album_published_at'),
+        }])
+
+    # 写入 song
+    platform_song_id = int(pr['platform_mid']) if pr.get('platform_mid') else song_id_int
+    song_uuid = str(uuid.uuid4())
+    upsert_qq_songs(pg_cur, [{
+        'song_uuid': song_uuid,
+        'platform_song_id': platform_song_id,
+        'mid': mid,
+        'album_id': album_id,
+        'cover': cover_url,
+        'title': sp.get('title', hk_row['name']),
+        'name': hk_row['name'],
+        'duration': sp.get('duration', 0) or 0,
+        'lyric': strip_timestamps(sp.get('lyric')),
+        'composer_name': sp.get('composer_name') or hk_row.get('composer'),
+        'lyricist_name': sp.get('lyricist_name') or hk_row.get('lyricist'),
+        'url': audio_url,
+        'lyric_url': hk_row.get('lyrics_url'),
+        'platform_index_url': sp.get('platform_index_url'),
+        'published_at': sp.get('published_at') or hk_row.get('issue_time'),
+        'singers_json': singers_json,
+    }])
+    # 查询实际 UUID（ON CONFLICT DO NOTHING 时使用已有 UUID）
+    pg_cur.execute('SELECT id FROM crawler_qqmusic_songs WHERE platform_song_id = %s', (platform_song_id,))
+    row = pg_cur.fetchone()
+    if row:
+        song_uuid = str(row[0])
+
+    # singer_songs / singer_albums
+    upsert_qq_singer_songs(pg_cur, [(sg['singer_id'], song_uuid) for sg in singer_list])
+    if album_id:
+        upsert_qq_singer_albums(pg_cur, [(sg['singer_id'], album_id) for sg in singer_list])
+
+
+def _process_kugou(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_url):
+    song_id = int(pr['platform_unique_key'])
+    songs_map = fetch_kugou_songs(spider_conn, [song_id])
+    if song_id not in songs_map:
+        return
+    sp = songs_map[song_id]
+
+    singers_map = fetch_kugou_singers(spider_conn, [song_id])
+    singer_list = singers_map.get(song_id, [])
+
+    audio_url = _safe_transfer(
+        hk_row['audio_url'],
+        build_oss_key('kugou', 'audio', str(song_id) + '.mp3'),
+        bucket, base_url
+    )
+    cover_url = _safe_transfer(
+        hk_row.get('cover_url') or sp.get('cover', ''),
+        build_oss_key('kugou', 'cover', str(song_id) + '.jpg'),
+        bucket, base_url
+    )
+    album_cover = ''
+    if sp.get('album_id') and sp.get('album_cover'):
+        album_cover = _safe_transfer(
+            sp['album_cover'],
+            build_oss_key('kugou', 'album', str(sp['album_id']) + '.jpg'),
+            bucket, base_url
+        )
+
+    singer_rows = []
+    for sg in singer_list:
+        avatar = _safe_transfer(
+            sg.get('avatar', ''),
+            build_oss_key('kugou', 'singer', str(sg['singer_id']) + '.jpg'),
+            bucket, base_url
+        )
+        singer_rows.append({**sg, 'id': sg['singer_id'], 'avatar': avatar})
+    upsert_kugou_singers(pg_cur, singer_rows)
+
+    singers_json = json.dumps([{
+        'name': sg['name'],
+        'singer_id': sg['singer_id'],
+        'platform_singer_id': str(sg['singer_id']),
+    } for sg in singer_list], ensure_ascii=False)
+
+    album_id = None
+    if sp.get('album_id'):
+        album_id = sp['album_id']
+        upsert_kugou_albums(pg_cur, [{
+            'id': sp['album_id'], 'cover': album_cover,
+            'title': sp.get('album_title') or '', 'intro': sp.get('album_intro'),
+            'type': sp.get('album_type') or '', 'company_id': sp.get('company_id') or 0,
+            'company': sp.get('company') or '', 'is_owner': sp.get('is_owner') or 0,
+            'published_at': sp.get('album_published_at'),
+        }])
+
+    song_uuid = str(uuid.uuid4())
+    upsert_kugou_songs(pg_cur, [{
+        'song_uuid': song_uuid,
+        'platform_song_id': song_id,
+        'hash': sp.get('hid', pr.get('platform_mid', '')),
+        'album_audio_id': sp.get('album_audio_id') or pr.get('album_audio_id') or 0,
+        'album_id': album_id,
+        'cover': cover_url,
+        'title': sp.get('title', hk_row['name']),
+        'name': hk_row['name'],
+        'duration': sp.get('duration', 0) or 0,
+        'lyric': strip_timestamps(sp.get('lyric')),
+        'composer_name': sp.get('composer_name') or hk_row.get('composer'),
+        'lyricist_name': sp.get('lyricist_name') or hk_row.get('lyricist'),
+        'url': audio_url,
+        'lyric_url': hk_row.get('lyrics_url'),
+        'platform_index_url': sp.get('platform_index_url'),
+        'published_at': sp.get('published_at') or hk_row.get('issue_time'),
+        'singers_json': singers_json,
+    }])
+    # 查询实际 UUID（ON CONFLICT DO NOTHING 时使用已有 UUID）
+    pg_cur.execute('SELECT id FROM crawler_kugou_songs WHERE platform_song_id = %s', (song_id,))
+    row = pg_cur.fetchone()
+    if row:
+        song_uuid = str(row[0])
+
+    upsert_kugou_singer_songs(pg_cur, [(sg['singer_id'], song_uuid) for sg in singer_list])
+    if album_id:
+        upsert_kugou_singer_albums(pg_cur, [(sg['singer_id'], album_id) for sg in singer_list])
+
+
+def _process_netease(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_url):
+    song_id = int(pr['platform_unique_key'])
+    songs_map = fetch_netease_songs(spider_conn, [song_id])
+    if song_id not in songs_map:
+        return
+    sp = songs_map[song_id]
+
+    singers_map = fetch_netease_singers(spider_conn, [song_id])
+    singer_list = singers_map.get(song_id, [])
+
+    audio_url = _safe_transfer(
+        hk_row['audio_url'],
+        build_oss_key('netease', 'audio', str(song_id) + '.mp3'),
+        bucket, base_url
+    )
+    cover_url = _safe_transfer(
+        hk_row.get('cover_url') or sp.get('cover', ''),
+        build_oss_key('netease', 'cover', str(song_id) + '.jpg'),
+        bucket, base_url
+    )
+    album_cover = ''
+    if sp.get('album_id') and sp.get('album_cover'):
+        album_cover = _safe_transfer(
+            sp['album_cover'],
+            build_oss_key('netease', 'album', str(sp['album_id']) + '.jpg'),
+            bucket, base_url
+        )
+
+    singer_rows = []
+    for sg in singer_list:
+        avatar = _safe_transfer(
+            sg.get('avatar', ''),
+            build_oss_key('netease', 'singer', str(sg['singer_id']) + '.jpg'),
+            bucket, base_url
+        )
+        singer_rows.append({**sg, 'id': sg['singer_id'], 'avatar': avatar})
+    upsert_netease_singers(pg_cur, singer_rows)
+
+    singers_json = json.dumps([{
+        'name': sg['name'],
+        'singer_id': sg['singer_id'],
+        'platform_singer_id': str(sg['singer_id']),
+    } for sg in singer_list], ensure_ascii=False)
+
+    album_id = None
+    if sp.get('album_id'):
+        album_id = sp['album_id']
+        upsert_netease_albums(pg_cur, [{
+            'id': sp['album_id'], 'cover': album_cover,
+            'title': sp.get('album_title') or '', 'intro': sp.get('album_intro'),
+            'type': sp.get('album_type') or '', 'company_id': sp.get('company_id') or 0,
+            'company': sp.get('company') or '', 'is_owner': sp.get('is_owner') or 0,
+            'published_at': sp.get('album_published_at'),
+        }])
+
+    song_uuid = str(uuid.uuid4())
+    upsert_netease_songs(pg_cur, [{
+        'song_uuid': song_uuid,
+        'platform_song_id': song_id,
+        'album_id': album_id,
+        'cover': cover_url,
+        'title': sp.get('title', hk_row['name']),
+        'name': hk_row['name'],
+        'duration': sp.get('duration', 0) or 0,
+        'lyric': strip_timestamps(sp.get('lyric')),
+        'composer_name': sp.get('composer_name') or hk_row.get('composer'),
+        'lyricist_name': sp.get('lyricist_name') or hk_row.get('lyricist'),
+        'url': audio_url,
+        'lyric_url': hk_row.get('lyrics_url'),
+        'platform_index_url': sp.get('platform_index_url'),
+        'published_at': sp.get('published_at') or hk_row.get('issue_time'),
+        'singers_json': singers_json,
+    }])
+    # 查询实际 UUID（ON CONFLICT DO NOTHING 时使用已有 UUID）
+    pg_cur.execute('SELECT id FROM crawler_netease_songs WHERE platform_song_id = %s', (song_id,))
+    row = pg_cur.fetchone()
+    if row:
+        song_uuid = str(row[0])
+
+    upsert_netease_singer_songs(pg_cur, [(sg['singer_id'], song_uuid) for sg in singer_list])
+    if album_id:
+        upsert_netease_singer_albums(pg_cur, [(sg['singer_id'], album_id) for sg in singer_list])
+
+
+_PROCESSORS = {
+    PLATFORM_QQ: _process_qq,
+    PLATFORM_KUGOU: _process_kugou,
+    PLATFORM_NETEASE: _process_netease,
+}
+
+
+def run(platforms: list[str], max_batches: int | None = None) -> None:
+    hk_conn = get_hk_songs_conn()
+    src_conn = get_source_conn()
+    spider_conn = get_spider_conn()
+    pg_conn = get_pg_conn()
+    bucket = get_oss_bucket()
+    base_url = OSS_CONFIG['base_url']
+
+    total_ok = total_err = 0
+
+    try:
+        for i, batch in enumerate(tqdm(iter_hk_songs_batches(hk_conn, BATCH_SIZE), desc='batches')):
+            if max_batches is not None and i >= max_batches:
+                break
+            song_ids = [int(r['source_song_id']) for r in batch if r.get('source_song_id')]
+            platform_records = fetch_platform_records(src_conn, song_ids)
+
+            # index platform records by source_song_id
+            pr_by_song: dict[int, list] = {}
+            for pr in platform_records:
+                if pr['platform'] in platforms:
+                    pr_by_song.setdefault(int(pr['source_song_id']), []).append(pr)
+
+            with pg_conn.cursor() as pg_cur:
+                for hk_row in batch:
+                    src_id = int(hk_row['source_song_id']) if hk_row.get('source_song_id') else None
+                    if not src_id or src_id not in pr_by_song:
+                        continue
+                    for pr in pr_by_song[src_id]:
+                        processor = _PROCESSORS.get(pr['platform'])
+                        if not processor:
+                            continue
+                        pg_cur.execute('SAVEPOINT sp_song')
+                        try:
+                            processor(hk_row, pr, spider_conn, pg_cur, bucket, base_url)
+                            pg_cur.execute('RELEASE SAVEPOINT sp_song')
+                            total_ok += 1
+                        except Exception as e:
+                            pg_cur.execute('ROLLBACK TO SAVEPOINT sp_song')
+                            pg_cur.execute('RELEASE SAVEPOINT sp_song')
+                            log.error("Error processing song %s platform %s: %s",
+                                      hk_row.get('name'), pr['platform'], e)
+                            total_err += 1
+                pg_conn.commit()
+
+    finally:
+        hk_conn.close()
+        src_conn.close()
+        spider_conn.close()
+        pg_conn.close()
+
+    log.info("Done. OK=%d ERR=%d", total_ok, total_err)

+#!/usr/bin/env python3
+import argparse
+from etl_to_crawler.config import PLATFORM_QQ, PLATFORM_KUGOU, PLATFORM_NETEASE, PLATFORMS
+from etl_to_crawler.runner import run
+
+PLATFORM_MAP = {
+    'qq': PLATFORM_QQ,
+    'kugou': PLATFORM_KUGOU,
+    'netease': PLATFORM_NETEASE,
+}
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser(description='ETL: hk_songs_test → crawler_dev')
+    parser.add_argument('--platform', default='all',
+                        choices=['qq', 'kugou', 'netease', 'all'],
+                        help='要导入的平台（默认 all）')
+    parser.add_argument('--max-batches', type=int, default=None,
+                        help='最多处理多少批次（冒烟测试用）')
+    args = parser.parse_args()
+
+    if args.platform == 'all':
+        platforms = PLATFORMS
+    else:
+        platforms = [PLATFORM_MAP[args.platform]]
+
+    print(f"Starting ETL for platforms: {platforms}")
+    run(platforms, max_batches=args.max_batches)