Merge branch 'feat/etl-hk-songs-to-crawler'

+import os
+from dotenv import load_dotenv
+
+load_dotenv()
+
+SOURCE_DB = {
+    'host': os.environ['SOURCE_DB_HOST'],
+    'port': int(os.environ['SOURCE_DB_PORT']),
+    'user': os.environ['SOURCE_DB_USER'],
+    'password': os.environ['SOURCE_DB_PASSWORD'],
+    'database': os.environ['SOURCE_DB_NAME'],
+    'charset': 'utf8mb4',
+}
+
+HK_SONGS_DB = {
+    'host': os.environ['TARGET_DB_HOST'],
+    'port': int(os.environ['TARGET_DB_PORT']),
+    'user': os.environ['TARGET_DB_USER'],
+    'password': os.environ['TARGET_DB_PASSWORD'],
+    'database': os.environ['TARGET_DB_NAME'],
+    'charset': 'utf8mb4',
+}
+
+CRAWLER_DB = {
+    'host': os.environ['CRAWLER_DB_HOST'],
+    'port': int(os.environ.get('CRAWLER_DB_PORT', '5432')),
+    'user': os.environ['CRAWLER_DB_USER'],
+    'password': os.environ['CRAWLER_DB_PASSWORD'],
+    'database': os.environ['CRAWLER_DB_NAME'],
+    'ssl_context': None if os.environ.get('CRAWLER_DB_SSL', 'false').lower() != 'true' else True,
+}
+
+OSS_CONFIG = {
+    'access_key_id': os.environ['OSS_ACCESS_KEY_ID'],
+    'access_key_secret': os.environ['OSS_ACCESS_KEY_SECRET'],
+    'endpoint': os.environ['OSS_ENDPOINT'],
+    'bucket_name': os.environ['OSS_BUCKET_NAME'],
+    'base_url': f"https://{os.environ['OSS_BUCKET_NAME']}.{os.environ['OSS_ENDPOINT']}",
+}
+
+PLATFORM_QQ = '1'
+PLATFORM_KUGOU = '2'
+PLATFORM_NETEASE = '4'
+PLATFORMS = [PLATFORM_QQ, PLATFORM_KUGOU, PLATFORM_NETEASE]
+
+BATCH_SIZE = 3

+import pymysql
+import pymysql.cursors
+import pg8000
+import oss2
+from .config import SOURCE_DB, HK_SONGS_DB, CRAWLER_DB, OSS_CONFIG
+
+
+def get_source_conn() -> pymysql.Connection:
+    return pymysql.connect(**SOURCE_DB, cursorclass=pymysql.cursors.DictCursor)
+
+
+def get_spider_conn() -> pymysql.Connection:
+    cfg = SOURCE_DB.copy()
+    cfg['database'] = 'hikoon-data-spider'
+    return pymysql.connect(**cfg, cursorclass=pymysql.cursors.DictCursor)
+
+
+def get_hk_songs_conn() -> pymysql.Connection:
+    return pymysql.connect(**HK_SONGS_DB, cursorclass=pymysql.cursors.DictCursor)
+
+
+def get_pg_conn() -> pg8000.Connection:
+    return pg8000.connect(**CRAWLER_DB)
+
+
+def get_oss_bucket() -> oss2.Bucket:
+    auth = oss2.Auth(OSS_CONFIG['access_key_id'], OSS_CONFIG['access_key_secret'])
+    return oss2.Bucket(auth, OSS_CONFIG['endpoint'], OSS_CONFIG['bucket_name'])

+import re
+
+_TIMESTAMP_RE = re.compile(r'\[\d{2}:\d{2}\.\d{2,3}\]')
+_META_TAG_RE = re.compile(r'\[[a-zA-Z]+:[^\]]*\]')
+
+
+def strip_timestamps(text: str | None) -> str:
+    if not text:
+        return ''
+    text = _META_TAG_RE.sub('', text)
+    text = _TIMESTAMP_RE.sub('', text)
+    lines = [line.strip() for line in text.splitlines() if line.strip()]
+    return '\n'.join(lines)

+import requests
+import oss2
+from urllib.parse import urlparse
+
+ARCHIVE_DEV_HOST = 'archive-dev.oss-cn-beijing.aliyuncs.com'
+
+
+def transfer_url(url: str | None, oss_key: str, bucket: oss2.Bucket, base_url: str) -> str:
+    """
+    将 url 指向的文件转移到 archive-dev OSS 的 oss_key 路径。
+    若 url 为空或已在 archive-dev，直接返回原 url（不上传）。
+    返回新的 archive-dev URL。
+    """
+    if not url:
+        return ''
+    if ARCHIVE_DEV_HOST in url:
+        return url
+    resp = requests.get(url, timeout=30)
+    resp.raise_for_status()
+    bucket.put_object(oss_key, resp.content)
+    return f"{base_url.rstrip('/')}/{oss_key}"
+
+
+def build_oss_key(platform: str, category: str, filename: str) -> str:
+    """
+    构造 archive-dev 内的存储路径。
+    platform: 'qq' | 'kugou' | 'netease'
+    category: 'audio' | 'cover' | 'singer' | 'album'
+    filename: 带扩展名的文件名
+    """
+    return f"crawler/{platform}/{category}/{filename}"

+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 id > %s
+  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
+"""
+
+_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,
+    start_after_id: int = 0,
+) -> Iterator[list[dict]]:
+    last_id = start_after_id
+    with conn.cursor() as cur:
+        while True:
+            cur.execute(_HK_SONGS_QUERY, (last_id, batch_size))
+            rows = cur.fetchall()
+            if not rows:
+                break
+            yield rows
+            last_id = int(rows[-1]['id'])
+            if len(rows) < batch_size:
+                break
+
+
+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),
+    )

+import uuid
+import json
+import logging
+from pathlib import Path
+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, select_primary_record
+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_yinyan_song_records,
+    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') or hk_row.get('song_time') 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') or f'https://y.qq.com/n/ryqq/songDetail/{mid}',
+        '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])
+    return {'platform': 'qq', 'platform_song_id': platform_song_id, 'mid': mid, 'title': hk_row['name']}
+
+
+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') or hk_row.get('song_time') 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') or f'http://www.kugou.com/song/#hash={sp.get("hid") or pr.get("platform_mid", "")}',
+        '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])
+    return {'platform': 'kugou', 'platform_song_id': song_id, 'hash': sp.get('hid', ''), 'title': hk_row['name']}
+
+
+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') or hk_row.get('song_time') 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') or f'https://music.163.com/#/song?id={song_id}',
+        '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])
+    return {'platform': 'netease', 'platform_song_id': song_id, 'title': hk_row['name']}
+
+
+_PROCESSORS = {
+    PLATFORM_QQ: _process_qq,
+    PLATFORM_KUGOU: _process_kugou,
+    PLATFORM_NETEASE: _process_netease,
+}
+
+
+DEFAULT_STATE_FILE = Path('output/etl_to_crawler_state.json')
+
+
+def _load_state(path: Path) -> dict:
+    if not path.exists():
+        return {}
+    with path.open('r', encoding='utf-8') as f:
+        return json.load(f)
+
+
+def _save_state(path: Path, state: dict) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open('w', encoding='utf-8') as f:
+        json.dump(state, f, ensure_ascii=False, indent=2, sort_keys=True)
+        f.write('\n')
+
+
+def _state_last_id(state: dict, key: str) -> int:
+    value = state.get(key, {}).get('last_hk_songs_id', 0)
+    return int(value or 0)
+
+
+def run(
+    platforms: list[str],
+    max_batches: int | None = None,
+    resume: bool = False,
+    state_file: str | Path = DEFAULT_STATE_FILE,
+    state_key: str = 'all',
+    start_after_id: int | None = None,
+    reset_state: bool = False,
+) -> 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']
+    state_path = Path(state_file)
+    state = _load_state(state_path) if resume and not reset_state else {}
+    resume_start_id = 0
+    if resume:
+        resume_start_id = _state_last_id(state, state_key)
+    if start_after_id is not None:
+        resume_start_id = start_after_id
+
+    total_ok = total_err = 0
+    imported: list[dict] = []
+
+    try:
+        for i, batch in enumerate(tqdm(
+            iter_hk_songs_batches(hk_conn, BATCH_SIZE, start_after_id=resume_start_id),
+            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
+                    primary_record = select_primary_record(pr_by_song[src_id])
+                    wrote_yinyan_record = False
+                    for pr in pr_by_song[src_id]:
+                        processor = _PROCESSORS.get(pr['platform'])
+                        if not processor:
+                            continue
+                        pg_cur.execute('SAVEPOINT sp_song')
+                        try:
+                            result = processor(hk_row, pr, spider_conn, pg_cur, bucket, base_url)
+                            if result and primary_record and not wrote_yinyan_record:
+                                upsert_yinyan_song_records(pg_cur, [(src_id, int(primary_record['record_id']))])
+                                wrote_yinyan_record = True
+                            pg_cur.execute('RELEASE SAVEPOINT sp_song')
+                            total_ok += 1
+                            if result:
+                                imported.append(result)
+                        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()
+                if resume:
+                    state.setdefault(state_key, {})['last_hk_songs_id'] = int(batch[-1]['id'])
+                    _save_state(state_path, state)
+
+    finally:
+        hk_conn.close()
+        src_conn.close()
+        spider_conn.close()
+        pg_conn.close()
+
+    log.info("Done. OK=%d ERR=%d", total_ok, total_err)
+    if imported:
+        print(f"\n导入记录（共 {len(imported)} 条）：")
+        for r in imported:
+            p = r['platform']
+            if p == 'qq':
+                print(f"  [QQ]      platform_song_id={r['platform_song_id']}  mid={r['mid']}  title={r['title']}")
+            elif p == 'kugou':
+                print(f"  [酷狗]    platform_song_id={r['platform_song_id']}  hash={r['hash']}  title={r['title']}")
+            elif p == 'netease':
+                print(f"  [网易]    platform_song_id={r['platform_song_id']}  title={r['title']}")

+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

+import uuid
+
+
+def upsert_yinyan_song_records(cur, pairs: list[tuple[int, int]]) -> None:
+    """pairs: [(source_song_id, hk_music_record_id), ...]"""
+    if not pairs:
+        return
+    cur.executemany(
+        "DELETE FROM yinyan_song_records WHERE song_id = %s",
+        [(song_id,) for song_id, _ in pairs],
+    )
+    cur.executemany(
+        """
+        INSERT INTO yinyan_song_records (song_id, record_id)
+        VALUES (%s, %s)
+        """,
+        pairs,
+    )
+
+
+# ─── QQ Music ────────────────────────────────────────────────────────────────
+
+def upsert_qq_singers(cur, singers: list[dict]) -> None:
+    if not singers:
+        return
+    sql = """
+        INSERT INTO crawler_qqmusic_singers
+            (id, mid, name, avatar, sex, area, "index", intro, home_url, created_at, updated_at)
+        VALUES (%s, %s, %s, %s, %s::singer_sex, %s::singer_area, %s::singer_index, %s, %s, NOW(), NOW())
+        ON CONFLICT (id) DO NOTHING
+    """
+    rows = [(
+        s['id'], s['mid'], s['name'], s.get('avatar', ''),
+        s.get('sex') or 'U', s.get('area') or '其他', s.get('index') or '#',
+        s.get('intro'), s.get('home_url'),
+    ) for s in singers]
+    cur.executemany(sql, rows)
+
+
+def upsert_qq_albums(cur, albums: list[dict]) -> None:
+    if not albums:
+        return
+    sql = """
+        INSERT INTO crawler_qqmusic_albums
+            (id, mid, cover, title, intro, type, company_id, company, is_owner, published_at, created_at, updated_at)
+        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW())
+        ON CONFLICT (id) DO NOTHING
+    """
+    rows = [(
+        a['id'], a.get('mid', ''), a.get('cover', ''), a.get('title', ''),
+        a.get('intro'), a.get('type', ''), a.get('company_id', 0) or 0,
+        a.get('company', ''), a.get('is_owner', 0) or 0, a.get('published_at'),
+    ) for a in albums]
+    cur.executemany(sql, rows)
+
+
+def upsert_qq_songs(cur, songs: list[dict]) -> None:
+    if not songs:
+        return
+    sql = """
+        INSERT INTO crawler_qqmusic_songs
+            (id, platform_song_id, mid, album_id, cover, title, name, duration,
+             lyric, composer_name, lyricist_name, url, lyric_url,
+             platform_index_url, published_at, singers, status, created_at, updated_at)
+        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, 0, NOW(), NOW())
+        ON CONFLICT (platform_song_id) DO NOTHING
+    """
+    rows = [(
+        s['song_uuid'], s['platform_song_id'], s['mid'], s.get('album_id'),
+        s.get('cover', ''), s.get('title', ''), s.get('name', ''), s.get('duration', 0) or 0,
+        s.get('lyric'), s.get('composer_name'), s.get('lyricist_name'),
+        s.get('url', ''), s.get('lyric_url'),
+        s.get('platform_index_url'), s.get('published_at'), s.get('singers_json', '[]'),
+    ) for s in songs]
+    cur.executemany(sql, rows)
+
+
+def upsert_qq_singer_songs(cur, pairs: list[tuple]) -> None:
+    """pairs: [(singer_id, song_uuid), ...]"""
+    if not pairs:
+        return
+    sql = """
+        INSERT INTO crawler_qqmusic_singer_songs (id, singer_id, song_id)
+        VALUES (%s, %s, %s)
+        ON CONFLICT (singer_id, song_id) DO NOTHING
+    """
+    rows = [(str(uuid.uuid4()), singer_id, song_uuid) for singer_id, song_uuid in pairs]
+    cur.executemany(sql, rows)
+
+
+def upsert_qq_singer_albums(cur, pairs: list[tuple]) -> None:
+    """pairs: [(singer_id, album_id), ...]"""
+    if not pairs:
+        return
+    sql = """
+        INSERT INTO crawler_qqmusic_singer_albums (id, singer_id, album_id)
+        VALUES (%s, %s, %s)
+        ON CONFLICT (singer_id, album_id) DO NOTHING
+    """
+    rows = [(str(uuid.uuid4()), singer_id, album_id) for singer_id, album_id in pairs]
+    cur.executemany(sql, rows)
+
+
+# ─── Kugou ───────────────────────────────────────────────────────────────────
+
+def upsert_kugou_singers(cur, singers: list[dict]) -> None:
+    if not singers:
+        return
+    sql = """
+        INSERT INTO crawler_kugou_singers
+            (id, name, avatar, sex, area, "index", intro, home_url, created_at, updated_at)
+        VALUES (%s, %s, %s, %s::kugou_singer_sex, %s::kugou_singer_area, %s::kugou_singer_index, %s, %s, NOW(), NOW())
+        ON CONFLICT (id) DO NOTHING
+    """
+    rows = [(
+        s['id'], s['name'], s.get('avatar', ''),
+        s.get('sex') or 'U', s.get('area') or '其他', s.get('index') or '#',
+        s.get('intro'), s.get('home_url', ''),
+    ) for s in singers]
+    cur.executemany(sql, rows)
+
+
+def upsert_kugou_albums(cur, albums: list[dict]) -> None:
+    if not albums:
+        return
+    sql = """
+        INSERT INTO crawler_kugou_albums
+            (id, cover, title, intro, type, company_id, company, is_owner, published_at, created_at, updated_at)
+        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW())
+        ON CONFLICT (id) DO NOTHING
+    """
+    rows = [(
+        a['id'], a.get('cover', ''), a.get('title', ''), a.get('intro'),
+        a.get('type', ''), a.get('company_id', 0) or 0,
+        a.get('company', ''), a.get('is_owner', 0) or 0, a.get('published_at'),
+    ) for a in albums]
+    cur.executemany(sql, rows)
+
+
+def upsert_kugou_songs(cur, songs: list[dict]) -> None:
+    if not songs:
+        return
+    sql = """
+        INSERT INTO crawler_kugou_songs
+            (id, platform_song_id, hash, album_audio_id, album_id, cover, title, name, duration,
+             lyric, composer_name, lyricist_name, url, lyric_url,
+             platform_index_url, published_at, singers, status, created_at, updated_at)
+        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, 0, NOW(), NOW())
+        ON CONFLICT (platform_song_id) DO NOTHING
+    """
+    rows = [(
+        s['song_uuid'], s['platform_song_id'], s.get('hash', ''),
+        s.get('album_audio_id', 0) or 0, s.get('album_id'),
+        s.get('cover', ''), s.get('title', ''), s.get('name', ''), s.get('duration', 0) or 0,
+        s.get('lyric'), s.get('composer_name'), s.get('lyricist_name'),
+        s.get('url', ''), s.get('lyric_url'),
+        s.get('platform_index_url'), s.get('published_at'), s.get('singers_json', '[]'),
+    ) for s in songs]
+    cur.executemany(sql, rows)
+
+
+def upsert_kugou_singer_songs(cur, pairs: list[tuple]) -> None:
+    if not pairs:
+        return
+    sql = """
+        INSERT INTO crawler_kugou_singer_songs (id, singer_id, song_id)
+        VALUES (%s, %s, %s)
+        ON CONFLICT (singer_id, song_id) DO NOTHING
+    """
+    cur.executemany(sql, [(str(uuid.uuid4()), s, sg) for s, sg in pairs])
+
+
+def upsert_kugou_singer_albums(cur, pairs: list[tuple]) -> None:
+    if not pairs:
+        return
+    sql = """
+        INSERT INTO crawler_kugou_singer_albums (id, singer_id, album_id)
+        VALUES (%s, %s, %s)
+        ON CONFLICT (singer_id, album_id) DO NOTHING
+    """
+    cur.executemany(sql, [(str(uuid.uuid4()), s, a) for s, a in pairs])
+
+
+# ─── Netease ─────────────────────────────────────────────────────────────────
+
+def upsert_netease_singers(cur, singers: list[dict]) -> None:
+    if not singers:
+        return
+    sql = """
+        INSERT INTO crawler_netease_singers
+            (id, name, avatar, sex, area, "index", intro, home_url, created_at, updated_at)
+        VALUES (%s, %s, %s, %s::netease_singer_sex, %s::netease_singer_area, %s::netease_singer_index, %s, %s, NOW(), NOW())
+        ON CONFLICT (id) DO NOTHING
+    """
+    rows = [(
+        s['id'], s['name'], s.get('avatar', ''),
+        s.get('sex') or 'U', s.get('area') or '其他', s.get('index') or '#',
+        s.get('intro'), s.get('home_url'),
+    ) for s in singers]
+    cur.executemany(sql, rows)
+
+
+def upsert_netease_albums(cur, albums: list[dict]) -> None:
+    if not albums:
+        return
+    sql = """
+        INSERT INTO crawler_netease_albums
+            (id, cover, title, intro, type, company_id, company, is_owner, published_at, created_at, updated_at)
+        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW())
+        ON CONFLICT (id) DO NOTHING
+    """
+    rows = [(
+        a['id'], a.get('cover', ''), a.get('title', ''), a.get('intro'),
+        a.get('type', ''), a.get('company_id', 0) or 0,
+        a.get('company', ''), a.get('is_owner', 0) or 0, a.get('published_at'),
+    ) for a in albums]
+    cur.executemany(sql, rows)
+
+
+def upsert_netease_songs(cur, songs: list[dict]) -> None:
+    if not songs:
+        return
+    sql = """
+        INSERT INTO crawler_netease_songs
+            (id, platform_song_id, album_id, cover, title, name, duration,
+             lyric, composer_name, lyricist_name, url, lyric_url,
+             platform_index_url, published_at, singers, status, created_at, updated_at)
+        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, 0, NOW(), NOW())
+        ON CONFLICT (platform_song_id) DO NOTHING
+    """
+    rows = [(
+        s['song_uuid'], s['platform_song_id'], s.get('album_id'),
+        s.get('cover', ''), s.get('title', ''), s.get('name', ''), s.get('duration', 0) or 0,
+        s.get('lyric'), s.get('composer_name'), s.get('lyricist_name'),
+        s.get('url', ''), s.get('lyric_url'),
+        s.get('platform_index_url'), s.get('published_at'), s.get('singers_json', '[]'),
+    ) for s in songs]
+    cur.executemany(sql, rows)
+
+
+def upsert_netease_singer_songs(cur, pairs: list[tuple]) -> None:
+    if not pairs:
+        return
+    sql = """
+        INSERT INTO crawler_netease_singer_songs (id, singer_id, song_id)
+        VALUES (%s, %s, %s)
+        ON CONFLICT (singer_id, song_id) DO NOTHING
+    """
+    cur.executemany(sql, [(str(uuid.uuid4()), s, sg) for s, sg in pairs])
+
+
+def upsert_netease_singer_albums(cur, pairs: list[tuple]) -> None:
+    if not pairs:
+        return
+    sql = """
+        INSERT INTO crawler_netease_singer_albums (id, singer_id, album_id)
+        VALUES (%s, %s, %s)
+        ON CONFLICT (singer_id, album_id) DO NOTHING
+    """
+    cur.executemany(sql, [(str(uuid.uuid4()), s, a) for s, a in pairs])

-pymysql
-python-dotenv
-oss2
-requests
-tqdm
-opencc-python-reimplemented
-pytest
+aliyun-python-sdk-core==2.16.0
+aliyun-python-sdk-kms==2.16.5
+asn1crypto==1.5.1
+certifi==2026.6.17
+cffi==2.0.0
+charset-normalizer==3.4.7
+cos_python_sdk_v5==1.9.44
+crcmod==1.7
+cryptography==49.0.0
+et_xmlfile==2.0.0
+exceptiongroup==1.3.1
+idna==3.18
+iniconfig==2.3.0
+jmespath==0.10.0
+OpenCC==1.4.0
+opencc-python-reimplemented==0.1.7
+openpyxl==3.1.5
+oss2==2.19.1
+packaging==26.2
+pg8000==1.31.5
+pluggy==1.6.0
+pycparser==3.0
+pycryptodome==3.23.0
+Pygments==2.20.0
+PyMySQL==1.2.0
+pytest==9.1.1
+python-dateutil==2.9.0.post0
+python-dotenv==1.2.2
+requests==2.34.2
+scramp==1.4.12
+six==1.17.0
+tomli==2.4.1
+tqdm==4.68.3
+typing_extensions==4.15.0
+urllib3==2.7.0
+xmltodict==1.0.4

+#!/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='最多处理多少批次（冒烟测试用）')
+    parser.add_argument('--resume', action='store_true',
+                        help='启用断点续传，按 hk_songs_test.id 从上次成功提交的批次后继续')
+    parser.add_argument('--state-file', default='output/etl_to_crawler_state.json',
+                        help='断点状态文件路径')
+    parser.add_argument('--start-after-id', type=int, default=None,
+                        help='手动指定从哪个 hk_songs_test.id 之后开始读取')
+    parser.add_argument('--reset-state', action='store_true',
+                        help='忽略已有断点状态，从头或 --start-after-id 指定位置重新开始')
+    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,
+        resume=args.resume,
+        state_file=args.state_file,
+        state_key=args.platform,
+        start_after_id=args.start_after_id,
+        reset_state=args.reset_state,
+    )

+from etl_to_crawler.lyric import strip_timestamps
+
+def test_strips_standard_timestamps():
+    lrc = "[00:12.34]第一行歌词\n[01:23.45]第二行歌词"
+    assert strip_timestamps(lrc) == "第一行歌词\n第二行歌词"
+
+def test_strips_millisecond_timestamps():
+    lrc = "[00:12.345]带三位毫秒"
+    assert strip_timestamps(lrc) == "带三位毫秒"
+
+def test_strips_meta_tags():
+    lrc = "[ti:歌曲名]\n[ar:歌手]\n[00:01.00]歌词"
+    result = strip_timestamps(lrc)
+    assert "歌词" in result
+    assert "[ti:" not in result
+
+def test_empty_and_none():
+    assert strip_timestamps(None) == ''
+    assert strip_timestamps('') == ''
+
+def test_plain_text_unchanged():
+    assert strip_timestamps("没有时间戳的歌词") == "没有时间戳的歌词"

+from unittest.mock import MagicMock, patch
+from etl_to_crawler.oss import transfer_url
+
+ARCHIVE_URL = "https://archive-dev.oss-cn-beijing.aliyuncs.com/some/path.mp3"
+OTHER_URL = "https://hikoon-data-platform.oss-cn-beijing.aliyuncs.com/qq-audio/abc.mp3"
+BASE_URL = "https://archive-dev.oss-cn-beijing.aliyuncs.com"
+
+def test_already_archive_dev_returns_as_is():
+    bucket = MagicMock()
+    result = transfer_url(ARCHIVE_URL, "any/key.mp3", bucket, BASE_URL)
+    assert result == ARCHIVE_URL
+    bucket.put_object.assert_not_called()
+
+def test_empty_url_returns_empty():
+    bucket = MagicMock()
+    result = transfer_url('', "any/key.mp3", bucket, BASE_URL)
+    assert result == ''
+    bucket.put_object.assert_not_called()
+
+def test_none_url_returns_empty():
+    bucket = MagicMock()
+    result = transfer_url(None, "any/key.mp3", bucket, BASE_URL)
+    assert result == ''
+
+def test_external_url_downloads_and_uploads():
+    bucket = MagicMock()
+    fake_content = b"audio_bytes"
+    with patch('etl_to_crawler.oss.requests.get') as mock_get:
+        mock_get.return_value.content = fake_content
+        mock_get.return_value.raise_for_status = MagicMock()
+        result = transfer_url(OTHER_URL, "crawler/qq/audio/abc.mp3", bucket, BASE_URL)
+    bucket.put_object.assert_called_once_with("crawler/qq/audio/abc.mp3", fake_content)
+    assert result == f"{BASE_URL}/crawler/qq/audio/abc.mp3"

+from unittest.mock import MagicMock
+
+from etl_to_crawler.reader import (
+    fetch_platform_records,
+    iter_hk_songs_batches,
+    select_primary_record,
+)
+
+
+def _make_cursor(rows):
+    cur = MagicMock()
+    cur.__enter__ = MagicMock(return_value=cur)
+    cur.__exit__ = MagicMock(return_value=False)
+    cur.fetchall.return_value = rows
+    return cur
+
+
+def test_fetch_platform_records_keeps_one_record_per_song_and_platform():
+    conn = MagicMock()
+    conn.cursor.return_value = _make_cursor([
+        {
+            'source_song_id': 10,
+            'record_id': 100,
+            'platform': '1',
+            'platform_unique_key': 'qq-mid',
+            'platform_mid': '100',
+            'album_audio_id': None,
+            'is_main_version': 0,
+            'is_high': 1,
+            'pub_time': '2020-01-01',
+        },
+        {
+            'source_song_id': 10,
+            'record_id': 101,
+            'platform': '2',
+            'platform_unique_key': '200',
+            'platform_mid': 'kg-hash',
+            'album_audio_id': None,
+            'is_main_version': 1,
+            'is_high': 0,
+            'pub_time': '2021-01-01',
+        },
+        {
+            'source_song_id': 10,
+            'record_id': 102,
+            'platform': '4',
+            'platform_unique_key': '300',
+            'platform_mid': None,
+            'album_audio_id': None,
+            'is_main_version': 1,
+            'is_high': 1,
+            'pub_time': '2022-01-01',
+        },
+    ])
+
+    result = fetch_platform_records(conn, [10])
+
+    assert len(result) == 3
+    assert {row['record_id'] for row in result} == {100, 101, 102}
+
+
+def test_select_primary_record_prefers_main_version_then_is_high():
+    rows = [
+        {
+            'source_song_id': 10,
+            'record_id': 100,
+            'platform': '1',
+            'platform_unique_key': 'qq-mid',
+            'platform_mid': '100',
+            'album_audio_id': None,
+            'is_main_version': 0,
+            'is_high': 1,
+            'pub_time': '2020-01-01',
+        },
+        {
+            'source_song_id': 10,
+            'record_id': 101,
+            'platform': '2',
+            'platform_unique_key': '200',
+            'platform_mid': 'kg-hash',
+            'album_audio_id': None,
+            'is_main_version': 1,
+            'is_high': 0,
+            'pub_time': '2021-01-01',
+        },
+        {
+            'source_song_id': 10,
+            'record_id': 102,
+            'platform': '4',
+            'platform_unique_key': '300',
+            'platform_mid': None,
+            'album_audio_id': None,
+            'is_main_version': 1,
+            'is_high': 1,
+            'pub_time': '2022-01-01',
+        },
+    ]
+
+    assert select_primary_record(rows)['record_id'] == 102
+
+
+def test_select_primary_record_uses_earliest_pub_time_when_priority_ties():
+    rows = [
+        {
+            'source_song_id': 10,
+            'record_id': 100,
+            'platform': '1',
+            'platform_unique_key': 'qq-mid',
+            'platform_mid': '100',
+            'album_audio_id': None,
+            'is_main_version': 0,
+            'is_high': 0,
+            'pub_time': '2020-01-01',
+        },
+        {
+            'source_song_id': 10,
+            'record_id': 101,
+            'platform': '2',
+            'platform_unique_key': '200',
+            'platform_mid': 'kg-hash',
+            'album_audio_id': None,
+            'is_main_version': 0,
+            'is_high': 0,
+            'pub_time': '2019-01-01',
+        },
+    ]
+
+    assert select_primary_record(rows)['record_id'] == 101
+
+
+def test_iter_hk_songs_batches_uses_id_cursor_instead_of_offset():
+    cur = _make_cursor([])
+    cur.fetchall.side_effect = [
+        [{'id': 10, 'source_song_id': 100}, {'id': 20, 'source_song_id': 200}],
+        [{'id': 25, 'source_song_id': 250}],
+    ]
+    conn = MagicMock()
+    conn.cursor.return_value = cur
+
+    batches = list(iter_hk_songs_batches(conn, batch_size=2, start_after_id=5))
+
+    assert batches == [
+        [{'id': 10, 'source_song_id': 100}, {'id': 20, 'source_song_id': 200}],
+        [{'id': 25, 'source_song_id': 250}],
+    ]
+    assert cur.execute.call_args_list[0][0][1] == (5, 2)
+    assert cur.execute.call_args_list[1][0][1] == (20, 2)

+from unittest.mock import MagicMock
+
+from etl_to_crawler import runner
+
+
+class _Cursor:
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        return False
+
+    def execute(self, *args, **kwargs):
+        return None
+
+
+class _PgConnection:
+    def __init__(self):
+        self.cur = _Cursor()
+        self.commits = 0
+        self.closed = False
+
+    def cursor(self):
+        return self.cur
+
+    def commit(self):
+        self.commits += 1
+
+    def close(self):
+        self.closed = True
+
+
+class _Connection:
+    def close(self):
+        return None
+
+
+def test_run_imports_all_platform_records_but_writes_one_primary_yinyan_relation(monkeypatch):
+    pg_conn = _PgConnection()
+    processors = {
+        '1': MagicMock(return_value={'platform': 'qq', 'platform_song_id': 100, 'mid': 'qq-mid', 'title': '歌'}),
+        '2': MagicMock(return_value={'platform': 'kugou', 'platform_song_id': 200, 'hash': 'kg-hash', 'title': '歌'}),
+    }
+    yinyan_writer = MagicMock()
+
+    monkeypatch.setattr(runner, 'get_hk_songs_conn', lambda: _Connection())
+    monkeypatch.setattr(runner, 'get_source_conn', lambda: _Connection())
+    monkeypatch.setattr(runner, 'get_spider_conn', lambda: _Connection())
+    monkeypatch.setattr(runner, 'get_pg_conn', lambda: pg_conn)
+    monkeypatch.setattr(runner, 'get_oss_bucket', lambda: object())
+    monkeypatch.setattr(runner, 'iter_hk_songs_batches', lambda conn, batch_size, start_after_id=0: [[{
+        'source_song_id': 10,
+        'name': '歌',
+        'audio_url': 'https://example.com/a.mp3',
+        'singer': '歌手',
+    }]])
+    monkeypatch.setattr(runner, 'fetch_platform_records', lambda conn, song_ids: [
+        {
+            'source_song_id': 10,
+            'record_id': 100,
+            'platform': '1',
+            'platform_unique_key': 'qq-mid',
+            'platform_mid': '100',
+            'album_audio_id': None,
+            'is_main_version': 0,
+            'is_high': 1,
+            'pub_time': '2020-01-01',
+        },
+        {
+            'source_song_id': 10,
+            'record_id': 200,
+            'platform': '2',
+            'platform_unique_key': '200',
+            'platform_mid': 'kg-hash',
+            'album_audio_id': None,
+            'is_main_version': 1,
+            'is_high': 0,
+            'pub_time': '2021-01-01',
+        },
+    ])
+    monkeypatch.setattr(runner, '_PROCESSORS', processors)
+    monkeypatch.setattr(runner, 'upsert_yinyan_song_records', yinyan_writer)
+
+    runner.run(['1', '2'])
+
+    processors['1'].assert_called_once()
+    processors['2'].assert_called_once()
+    yinyan_writer.assert_called_once_with(pg_conn.cur, [(10, 200)])
+    assert pg_conn.commits == 1
+
+
+def test_run_resume_starts_after_saved_id_and_updates_state_after_commit(monkeypatch, tmp_path):
+    pg_conn = _PgConnection()
+    state_file = tmp_path / 'etl_state.json'
+    state_file.write_text('{"all": {"last_hk_songs_id": 40}}', encoding='utf-8')
+    seen_start_ids = []
+
+    monkeypatch.setattr(runner, 'get_hk_songs_conn', lambda: _Connection())
+    monkeypatch.setattr(runner, 'get_source_conn', lambda: _Connection())
+    monkeypatch.setattr(runner, 'get_spider_conn', lambda: _Connection())
+    monkeypatch.setattr(runner, 'get_pg_conn', lambda: pg_conn)
+    monkeypatch.setattr(runner, 'get_oss_bucket', lambda: object())
+
+    def fake_batches(conn, batch_size, start_after_id=0):
+        seen_start_ids.append(start_after_id)
+        return iter([[
+            {'id': 50, 'source_song_id': 10, 'name': '歌', 'audio_url': 'https://example.com/a.mp3', 'singer': '歌手'},
+            {'id': 60, 'source_song_id': 20, 'name': '歌2', 'audio_url': 'https://example.com/b.mp3', 'singer': '歌手2'},
+        ]])
+
+    monkeypatch.setattr(runner, 'iter_hk_songs_batches', fake_batches)
+    monkeypatch.setattr(runner, 'fetch_platform_records', lambda conn, song_ids: [])
+
+    runner.run(['1', '2', '4'], resume=True, state_file=state_file, state_key='all')
+
+    assert seen_start_ids == [40]
+    assert pg_conn.commits == 1
+    assert '"last_hk_songs_id": 60' in state_file.read_text(encoding='utf-8')

+from unittest.mock import MagicMock, patch
+from etl_to_crawler.spider import fetch_qq_songs, fetch_qq_singers
+
+def _make_cursor(rows):
+    cur = MagicMock()
+    cur.__enter__ = MagicMock(return_value=cur)
+    cur.__exit__ = MagicMock(return_value=False)
+    cur.fetchall.return_value = rows
+    return cur
+
+def test_fetch_qq_songs_keys_by_mid():
+    conn = MagicMock()
+    conn.cursor.return_value = _make_cursor([
+        {'mid': 'abc123', 'id': 1, 'title': '歌曲A', 'duration': 200,
+         'lyric': None, 'composer_name': None, 'lyricist_name': None,
+         'platform_index_url': None, 'published_at': None,
+         'cover': '', 'album_id': None,
+         'album_mid': None, 'album_title': None, 'album_cover': None,
+         'album_intro': None, 'album_type': None, 'company_id': 0,
+         'company': None, 'is_owner': 0, 'album_published_at': None}
+    ])
+    result = fetch_qq_songs(conn, ['abc123'])
+    assert 'abc123' in result
+    assert result['abc123']['title'] == '歌曲A'
+
+def test_fetch_qq_songs_empty_list():
+    conn = MagicMock()
+    result = fetch_qq_songs(conn, [])
+    assert result == {}
+    conn.cursor.assert_not_called()
+
+def test_fetch_qq_singers_keys_by_song_id():
+    conn = MagicMock()
+    conn.cursor.return_value = _make_cursor([
+        {'song_id': 1, 'singer_id': 10, 'mid': 'sg1', 'name': '歌手A',
+         'avatar': '', 'sex': 'M', 'area': '华语', 'index': 'G',
+         'intro': None, 'home_url': None}
+    ])
+    result = fetch_qq_singers(conn, [1])
+    assert 1 in result
+    assert result[1][0]['name'] == '歌手A'

+from unittest.mock import MagicMock, call
+from etl_to_crawler.writer import (
+    upsert_qq_singers,
+    upsert_qq_singer_songs,
+    upsert_yinyan_song_records,
+)
+
+
+def test_upsert_qq_singers_executes_insert():
+    cur = MagicMock()
+    singers = [{
+        'id': 1, 'mid': 'abc', 'name': '歌手', 'avatar': 'https://archive-dev.oss-cn-beijing.aliyuncs.com/a.jpg',
+        'sex': 'M', 'area': '华语', 'index': 'G', 'intro': None, 'home_url': None
+    }]
+    upsert_qq_singers(cur, singers)
+    assert cur.executemany.called
+    args = cur.executemany.call_args
+    assert 'crawler_qqmusic_singers' in args[0][0]
+    assert 'ON CONFLICT' in args[0][0]
+
+
+def test_upsert_qq_singers_empty_does_nothing():
+    cur = MagicMock()
+    upsert_qq_singers(cur, [])
+    cur.executemany.assert_not_called()
+
+
+def test_upsert_qq_singer_songs():
+    cur = MagicMock()
+    upsert_qq_singer_songs(cur, [(1, 'uuid-abc'), (2, 'uuid-def')])
+    assert cur.executemany.called
+    sql = cur.executemany.call_args[0][0]
+    assert 'crawler_qqmusic_singer_songs' in sql
+
+
+def test_upsert_yinyan_song_records_replaces_existing_song_relation():
+    cur = MagicMock()
+    upsert_yinyan_song_records(cur, [(10, 100), (11, 101)])
+
+    assert cur.executemany.call_count == 2
+    delete_sql, delete_rows = cur.executemany.call_args_list[0][0]
+    insert_sql, insert_rows = cur.executemany.call_args_list[1][0]
+    assert 'DELETE FROM yinyan_song_records' in delete_sql
+    assert 'WHERE song_id = %s' in delete_sql
+    assert delete_rows == [(10,), (11,)]
+    assert 'INSERT INTO yinyan_song_records' in insert_sql
+    assert insert_rows == [(10, 100), (11, 101)]