feat(etl): 新增 yinyan_song_records 平台回填与歌手优先录音选取

- 在 ETL 入口增加 --backfill-yinyan-platforms 参数，支持回填缺失平台代码 - 实现回填函数 backfill_yinyan_record_platforms，批量更新 platform 为空的记录 - fetch_all_platform_records 支持获取所有录音供歌手数据缺失的回退使用 - 通过 _pick_record_with_singer 函数优先选取有歌手数据的录音 - 初始化函数跳过已存在 song_id 的批次，避免重复处理 - yinyan_song_records 插入时允许更新 platform 字段，平台数据更完整 - upsert_yinyan_song_records 增加 record_id 更新支持，修正推送逻辑 - 所有平台歌曲标题拆分版本号存储 version 字段，支持特殊版本标记 - 在各平台封面缺失时使用歌曲封面回退，保证封面不为空 - spider 模块新增 probe_*_has_singers 函数校验录音是否有关联歌手 - writer 模块完善更新平台代码的操作及相关查询辅助函数 - utils 新增 split_title_version 助手，提取括号中版本信息 - upload_plain_lyric_to_bucket 上传时添加内容类型头，避免文本编码问题 - 测试覆盖添加回填功能及版本号拆分相关逻辑，提高稳定性维护性

feat(etl): 新增 yinyan_song_records 平台回填与歌手优先录音选取
- 在 ETL 入口增加 --backfill-yinyan-platforms 参数，支持回填缺失平台代码 - 实现回填函数 backfill_yinyan_record_platforms，批量更新 platform 为空的记录 - fetch_all_platform_records 支持获取所有录音供歌手数据缺失的回退使用 - 通过 _pick_record_with_singer 函数优先选取有歌手数据的录音 - 初始化函数跳过已存在 song_id 的批次，避免重复处理 - yinyan_song_records 插入时允许更新 platform 字段，平台数据更完整 - upsert_yinyan_song_records 增加 record_id 更新支持，修正推送逻辑 - 所有平台歌曲标题拆分版本号存储 version 字段，支持特殊版本标记 - 在各平台封面缺失时使用歌曲封面回退，保证封面不为空 - spider 模块新增 probe_*_has_singers 函数校验录音是否有关联歌手 - writer 模块完善更新平台代码的操作及相关查询辅助函数 - utils 新增 split_title_version 助手，提取括号中版本信息 - upload_plain_lyric_to_bucket 上传时添加内容类型头，避免文本编码问题 - 测试覆盖添加回填功能及版本号拆分相关逻辑，提高稳定性维护性
沈秋雨
Commit c403b29e ... c403b29eb0e01dabe288c2b8dde627a0738defb4 authored 2026-07-09 20:15:34 +0800 by 沈秋雨
Showing 10 changed files with 372 additions and 53 deletions
etl_to_crawler/config.py
etl_to_crawler/reader.py
etl_to_crawler/runner.py
etl_to_crawler/spider.py
etl_to_crawler/utils.py
etl_to_crawler/writer.py
run_etl.py
tests/test_runner.py
tests/test_utils.py
tests/test_writer.py
--- a/etl_to_crawler/config.py
View file @c403b29
+++ b/etl_to_crawler/config.py
View file @c403b29
@@ -43,4 +43,4 @@ PLATFORM_KUGOU = '2'
 PLATFORM_NETEASE = '4'
 PLATFORMS = [PLATFORM_QQ, PLATFORM_KUGOU, PLATFORM_NETEASE]
-BATCH_SIZE = 3
+BATCH_SIZE = 4
--- a/etl_to_crawler/reader.py
View file @c403b29
+++ b/etl_to_crawler/reader.py
View file @c403b29
@@ -54,6 +54,9 @@ ORDER BY sar.song_id,
         mr.id ASC
 """
+# 同 _PLATFORM_QUERY，但不做 per-platform 去重，保留同平台全部录音供 singer fallback 遍历
+_ALL_PLATFORM_RECORDS_QUERY = _PLATFORM_QUERY
 def iter_hk_songs_batches(
    conn: pymysql.Connection,
@@ -100,6 +103,44 @@ def fetch_platform_records(source_conn: pymysql.Connection, song_ids: list[int])
    return select_platform_records(rows)
+def fetch_all_platform_records(source_conn: pymysql.Connection, song_ids: list[int]) -> list[dict]:
+    """返回每首歌的全部录音（不做 per-platform 去重），供 singer fallback 遍历。"""
+    if not song_ids:
+        return []
+    placeholders = ','.join(['%s'] * len(song_ids))
+    query = _ALL_PLATFORM_RECORDS_QUERY.format(placeholders=placeholders)
+    with source_conn.cursor() as cur:
+        cur.execute(query, song_ids)
+        rows = cur.fetchall()
+    # 按优先级排序后按 record_id 去重（同一录音可能关联多次）
+    seen_record_ids: set = set()
+    result = []
+    for row in sorted(rows, key=_record_priority):
+        rid = row['record_id']
+        if rid not in seen_record_ids:
+            seen_record_ids.add(rid)
+            result.append(row)
+    return result
+def fetch_record_platforms(source_conn: pymysql.Connection, record_ids: list[int]) -> dict[int, str]:
+    """按录音 id 查询平台代码，用于回填 yinyan_song_records.platform。"""
+    if not record_ids:
+        return {}
+    placeholders = ','.join(['%s'] * len(record_ids))
+    query = f"""
+        SELECT id, platform
+        FROM hk_music_record
+        WHERE id IN ({placeholders})
+          AND platform IN ('1','2','4')
+          AND deleted = 0
+    """
+    with source_conn.cursor() as cur:
+        cur.execute(query, record_ids)
+        rows = cur.fetchall()
+    return {int(row['id']): str(row['platform']) for row in rows}
 def select_platform_records(rows: list[dict]) -> list[dict]:
    # 每个 (source_song_id, platform) 保留一条，用于继续导入多个平台的录音数据。
    seen = {}
--- a/etl_to_crawler/runner.py
View file @c403b29
+++ b/etl_to_crawler/runner.py
View file @c403b29
@@ -9,17 +9,23 @@ from .reader import (
    iter_hk_songs_batches,
    fetch_hk_songs_by_source_ids,
    fetch_platform_records,
+    fetch_all_platform_records,
+    fetch_record_platforms,
    select_primary_record,
 )
 from .spider import (
    fetch_qq_songs, fetch_qq_singers,
    fetch_kugou_songs, fetch_kugou_singers,
    fetch_netease_songs, fetch_netease_singers,
+    probe_qq_has_singers, probe_kugou_has_singers, probe_netease_has_singers,
 )
 from .writer import (
    fetch_pending_yinyan_song_records,
+    fetch_yinyan_records_missing_platform,
    insert_yinyan_song_records,
+    update_yinyan_record_platforms,
    upsert_yinyan_song_records,
+    fetch_existing_yinyan_song_ids,
    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,
@@ -28,7 +34,7 @@ from .writer import (
    upsert_netease_singer_songs, upsert_netease_singer_albums,
 )
 from .oss import transfer_url, transfer_url_with_md5, build_oss_key
-from .utils import upload_plain_lyric_to_bucket
+from .utils import split_title_version, upload_plain_lyric_to_bucket
 from .lyric import ensure_newlines
 logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
@@ -97,6 +103,9 @@ def _process_qq(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_url):
            build_oss_key('qq', 'album', str(sp['album_id']) + '.jpg'),
            bucket, base_url
        )
+    # 专辑封面为空时回退使用歌曲封面
+    if not album_cover and cover_url:
+        album_cover = cover_url
    # 歌手头像转移 + 写入 singers
    singer_rows = []
@@ -152,13 +161,15 @@ def _process_qq(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_url):
    song_uuid = str(uuid.uuid4())
    raw_lyric = sp.get('lyric') or ''
    lyric_url = _safe_upload_lyric('qq', mid, raw_lyric, hk_row.get('lyrics_url'), bucket, base_url)
+    title, version = split_title_version(sp.get('title') or hk_row['name'])
    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']),
+        'title': title,
+        'version': version,
        'name': hk_row['name'],
        'duration': sp.get('duration') or hk_row.get('song_time') or 0,
        'lyric': ensure_newlines(raw_lyric),
@@ -213,6 +224,9 @@ def _process_kugou(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_url
            build_oss_key('kugou', 'album', str(sp['album_id']) + '.jpg'),
            bucket, base_url
        )
+    # 专辑封面为空时回退使用歌曲封面
+    if not album_cover and cover_url:
+        album_cover = cover_url
    singer_rows = []
    for sg in singer_list:
@@ -262,6 +276,7 @@ def _process_kugou(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_url
    song_uuid = str(uuid.uuid4())
    raw_lyric = sp.get('lyric') or ''
    lyric_url = _safe_upload_lyric('kugou', str(song_id), raw_lyric, hk_row.get('lyrics_url'), bucket, base_url)
+    title, version = split_title_version(sp.get('title') or hk_row['name'])
    upsert_kugou_songs(pg_cur, [{
        'song_uuid': song_uuid,
        'platform_song_id': song_id,
@@ -269,7 +284,8 @@ def _process_kugou(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_url
        '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']),
+        'title': title,
+        'version': version,
        'name': hk_row['name'],
        'duration': sp.get('duration') or hk_row.get('song_time') or 0,
        'lyric': ensure_newlines(raw_lyric),
@@ -323,6 +339,9 @@ def _process_netease(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_u
            build_oss_key('netease', 'album', str(sp['album_id']) + '.jpg'),
            bucket, base_url
        )
+    # 专辑封面为空时回退使用歌曲封面
+    if not album_cover and cover_url:
+        album_cover = cover_url
    singer_rows = []
    for sg in singer_list:
@@ -385,13 +404,15 @@ def _process_netease(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_u
    song_uuid = str(uuid.uuid4())
    raw_lyric = sp.get('lyric') or ''
    lyric_url = _safe_upload_lyric('netease', str(song_id), raw_lyric, hk_row.get('lyrics_url'), bucket, base_url)
+    title, version = split_title_version(sp.get('title') or hk_row['name'])
    upsert_netease_songs(pg_cur, [{
        'song_uuid': song_uuid,
        'platform_song_id': song_id,
        'album_id': album_id,
        'album_json': album_json,
        'cover': cover_url,
-        'title': sp.get('title', hk_row['name']),
+        'title': title,
+        'version': version,
        'name': hk_row['name'],
        'duration': sp.get('duration') or hk_row.get('song_time') or 0,
        'lyric': ensure_newlines(raw_lyric),
@@ -424,18 +445,49 @@ _PROCESSORS = {
    PLATFORM_NETEASE: _process_netease,
 }
+_PROBERS = {
+    PLATFORM_QQ: lambda conn, pr: probe_qq_has_singers(conn, pr['platform_unique_key']),
+    PLATFORM_KUGOU: lambda conn, pr: probe_kugou_has_singers(conn, int(pr['platform_unique_key'])),
+    PLATFORM_NETEASE: lambda conn, pr: probe_netease_has_singers(conn, int(pr['platform_unique_key'])),
+}
+def _pick_record_with_singer(records: list[dict], spider_conn) -> dict | None:
+    """遍历录音列表，返回第一条在 spider DB 中有歌手数据的录音；全部无歌手时返回首条。"""
+    if not records:
+        return None
+    for pr in records:
+        prober = _PROBERS.get(pr['platform'])
+        if prober and prober(spider_conn, pr):
+            return pr
+    return records[0]  # fallback：保留优先级最高的录音，即使没有歌手
 def initialize_yinyan_song_records(platforms: list[str], max_batches: int | None = None) -> None:
    hk_conn = get_hk_songs_conn()
    src_conn = get_source_conn()
    pg_conn = get_pg_conn()
+    # 查询已存在的 song_id，用于跳过已处理的记录
+    with pg_conn.cursor() as pg_cur:
+        existing_song_ids = fetch_existing_yinyan_song_ids(pg_cur)
+    log.info("已存在的 yinyan_song_records: %d 条，将跳过这些记录", len(existing_song_ids))
    total = 0
+    skipped_batches = 0
    try:
        for i, batch in enumerate(tqdm(iter_hk_songs_batches(hk_conn, BATCH_SIZE), desc='init-yinyan')):
            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')]
+            # 过滤掉已存在的 song_id
+            song_ids = [
+                int(r['source_song_id']) for r in batch
+                if r.get('source_song_id') and int(r['source_song_id']) not in existing_song_ids
+            ]
+            if not song_ids:
+                skipped_batches += 1
+                continue
            platform_records = fetch_platform_records(src_conn, song_ids)
            pr_by_song: dict[int, list] = {}
@@ -455,6 +507,7 @@ def initialize_yinyan_song_records(platforms: list[str], max_batches: int | None
                        'record_id': int(primary_record['record_id']),
                        'platform': primary_record['platform'],
                    })
+                    existing_song_ids.add(src_id)  # 标记为已处理
            if init_rows:
                with pg_conn.cursor() as pg_cur:
@@ -466,7 +519,55 @@ def initialize_yinyan_song_records(platforms: list[str], max_batches: int | None
        src_conn.close()
        pg_conn.close()
-    log.info("Initialized yinyan_song_records candidates=%d", total)
+    log.info("Initialized yinyan_song_records candidates=%d, skipped_batches=%d", total, skipped_batches)
+def backfill_yinyan_record_platforms(max_batches: int | None = None) -> None:
+    src_conn = get_source_conn()
+    pg_conn = get_pg_conn()
+    total = 0
+    skipped = 0
+    try:
+        batch_index = 0
+        pbar = tqdm(desc='backfill-yinyan-platforms')
+        while max_batches is None or batch_index < max_batches:
+            with pg_conn.cursor() as pg_cur:
+                rows = fetch_yinyan_records_missing_platform(pg_cur, BATCH_SIZE)
+            if not rows:
+                break
+            record_ids = [int(row['record_id']) for row in rows if row.get('record_id')]
+            platforms_by_record = fetch_record_platforms(src_conn, record_ids)
+            updates = []
+            for row in rows:
+                platform = platforms_by_record.get(int(row['record_id']))
+                if not platform:
+                    skipped += 1
+                    continue
+                updates.append({
+                    'song_id': int(row['song_id']),
+                    'record_id': int(row['record_id']),
+                    'platform': platform,
+                })
+            if updates:
+                with pg_conn.cursor() as pg_cur:
+                    update_yinyan_record_platforms(pg_cur, updates)
+                pg_conn.commit()
+                total += len(updates)
+            else:
+                log.error("No yinyan_song_records platform rows were backfilled in this batch; stopping to avoid retry loop")
+                break
+            batch_index += 1
+            pbar.update(1)
+        pbar.close()
+    finally:
+        src_conn.close()
+        pg_conn.close()
+    log.info("Backfilled yinyan_song_records platform rows=%d, skipped=%d", total, skipped)
 def run(
@@ -492,15 +593,23 @@ def run(
                break
            song_ids = [int(r['song_id']) for r in pending_records]
-            pending_by_song = {int(r['song_id']): int(r['record_id']) for r in pending_records}
            hk_by_song = fetch_hk_songs_by_source_ids(hk_conn, song_ids)
-            platform_records = fetch_platform_records(src_conn, song_ids)
+            # 全部录音（不去重），供 singer fallback 遍历
+            all_platform_records = fetch_all_platform_records(src_conn, song_ids)
-            # index platform records by source_song_id
+            # 按 (song_id, platform) 分组，每组已按优先级排好序
-            pr_by_song: dict[int, list] = {}
+            pr_by_song_platform: dict[tuple, list] = {}
-            for pr in platform_records:
+            for pr in all_platform_records:
                if pr['platform'] in platforms:
-                    pr_by_song.setdefault(int(pr['source_song_id']), []).append(pr)
+                    key = (int(pr['source_song_id']), pr['platform'])
+                    pr_by_song_platform.setdefault(key, []).append(pr)
+            # 每首歌：从各平台各选一条最优录音（有歌手优先）
+            pr_by_song: dict[int, list] = {}
+            for (src_id, _platform), records in pr_by_song_platform.items():
+                chosen = _pick_record_with_singer(records, spider_conn)
+                if chosen:
+                    pr_by_song.setdefault(src_id, []).append(chosen)
            with pg_conn.cursor() as pg_cur:
                pushed_count = 0
@@ -511,7 +620,6 @@ def run(
                    if not src_id or src_id not in pr_by_song:
                        continue
                    wrote_yinyan_record = False
-                    pending_record_id = pending_by_song[src_id]
                    for pr in pr_by_song[src_id]:
                        processor = _PROCESSORS.get(pr['platform'])
                        if not processor:
@@ -519,15 +627,11 @@ def run(
                        pg_cur.execute('SAVEPOINT sp_song')
                        try:
                            result = processor(hk_row, pr, spider_conn, pg_cur, bucket, base_url)
-                            if (
+                            if result and not wrote_yinyan_record:
-                                result
-                                and not wrote_yinyan_record
-                                and int(pr['record_id']) == pending_record_id
-                            ):
                                upsert_yinyan_song_records(pg_cur, [{
                                    'song_id': src_id,
-                                    'record_id': pending_record_id,
+                                    'record_id': int(pr['record_id']),
-                                    'platform': result['platform'],
+                                    'platform': pr['platform'],
                                    'platform_song_id': int(result['platform_song_id']),
                                }])
                                wrote_yinyan_record = True
--- a/etl_to_crawler/spider.py
View file @c403b29
+++ b/etl_to_crawler/spider.py
View file @c403b29
@@ -46,6 +46,18 @@ def fetch_qq_singers(conn: pymysql.Connection, song_ids: list[int]) -> dict[int,
    return result
+def probe_qq_has_singers(conn: pymysql.Connection, mid: str) -> bool:
+    """检查 QQ 录音（按 mid）在 spider DB 中是否有歌手关联。"""
+    with conn.cursor() as cur:
+        cur.execute(
+            "SELECT 1 FROM media_tencent_songs s "
+            "JOIN media_tencent_singer_has_songs shs ON shs.song_id = s.id "
+            "WHERE s.mid = %s LIMIT 1",
+            (mid,),
+        )
+        return cur.fetchone() is not None
 # ─── Kugou ───────────────────────────────────────────────────────────────────
 _KUGOU_SONGS_SQL = """
@@ -81,6 +93,16 @@ def fetch_kugou_singers(conn: pymysql.Connection, song_ids: list[int]) -> dict[i
    return result
+def probe_kugou_has_singers(conn: pymysql.Connection, song_id: int) -> bool:
+    """检查酷狗录音在 spider DB 中是否有歌手关联。"""
+    with conn.cursor() as cur:
+        cur.execute(
+            "SELECT 1 FROM media_ku_gou_singer_has_songs WHERE song_id = %s LIMIT 1",
+            (song_id,),
+        )
+        return cur.fetchone() is not None
 # ─── Netease ─────────────────────────────────────────────────────────────────
 _NETEASE_SONGS_SQL = """
@@ -114,3 +136,13 @@ def fetch_netease_singers(conn: pymysql.Connection, song_ids: list[int]) -> dict
    for row in rows:
        result.setdefault(row['song_id'], []).append(row)
    return result
+def probe_netease_has_singers(conn: pymysql.Connection, song_id: int) -> bool:
+    """检查网易云录音在 spider DB 中是否有歌手关联。"""
+    with conn.cursor() as cur:
+        cur.execute(
+            "SELECT 1 FROM media_netease_singer_has_songs WHERE song_id = %s LIMIT 1",
+            (song_id,),
+        )
+        return cur.fetchone() is not None
--- a/etl_to_crawler/utils.py
View file @c403b29
+++ b/etl_to_crawler/utils.py
View file @c403b29
@@ -92,13 +92,29 @@ def extract_plain_lyric(lyric: str) -> str:
    return "\n".join(lines)
+def split_title_version(title: str | None) -> tuple[str, str]:
+    """拆分歌名末尾括号版本信息。
+    例如：化风行万里 (DJ默涵版) -> (化风行万里, DJ默涵版)
+    """
+    if not title:
+        return "", ""
+    text = title.strip()
+    match = re.match(r"^(?P<title>.+?)\s*[\(（](?P<version>[^()（）]+)[\)）]\s*$", text)
+    if not match:
+        return text, ""
+    clean_title = match.group("title").strip()
+    version = match.group("version").strip()
+    return clean_title or text, version
 def upload_plain_lyric_to_bucket(platform: str, unique_id: str, lyric: str, bucket, base_url: str) -> str:
    """将歌词去时间戳后上传到当前 ETL 使用的 OSS bucket"""
    plain_lyric = extract_plain_lyric(lyric)
    if not plain_lyric:
        return ""
    oss_key = f"crawler/{platform}/lyric/{unique_id}.txt"
-    bucket.put_object(oss_key, plain_lyric.encode("utf-8"))
+    bucket.put_object(oss_key, plain_lyric.encode("utf-8"), headers={'Content-Type': 'text/plain; charset=utf-8'})
    return f"{base_url.rstrip('/')}/{oss_key}"
--- a/etl_to_crawler/writer.py
View file @c403b29
+++ b/etl_to_crawler/writer.py
View file @c403b29
@@ -9,12 +9,20 @@ def insert_yinyan_song_records(cur, records: list[dict]) -> None:
        """
        INSERT INTO yinyan_song_records (song_id, record_id, platform, is_yinyan_push)
        VALUES (%s, %s, %s, FALSE)
-        ON CONFLICT (song_id, record_id) DO NOTHING
+        ON CONFLICT (song_id, record_id) DO UPDATE
+            SET platform = EXCLUDED.platform
+            WHERE yinyan_song_records.platform IS NULL
        """,
        [(r['song_id'], r['record_id'], r['platform']) for r in records],
    )
+def fetch_existing_yinyan_song_ids(cur) -> set[int]:
+    """返回已完成初始化 platform 的 song_id 集合。"""
+    cur.execute("SELECT DISTINCT song_id FROM yinyan_song_records WHERE platform IS NOT NULL")
+    return {row[0] for row in cur.fetchall()}
 def fetch_pending_yinyan_song_records(cur, limit: int) -> list[dict]:
    cur.execute(
        """
@@ -30,8 +38,40 @@ def fetch_pending_yinyan_song_records(cur, limit: int) -> list[dict]:
    return [{'song_id': row[0], 'record_id': row[1]} for row in rows]
+def fetch_yinyan_records_missing_platform(cur, limit: int) -> list[dict]:
+    cur.execute(
+        """
+        SELECT song_id, record_id
+        FROM yinyan_song_records
+        WHERE platform IS NULL
+        ORDER BY song_id
+        LIMIT %s
+        """,
+        (limit,),
+    )
+    rows = cur.fetchall()
+    return [{'song_id': row[0], 'record_id': row[1]} for row in rows]
+def update_yinyan_record_platforms(cur, records: list[dict]) -> None:
+    if not records:
+        return
+    cur.executemany(
+        """
+        UPDATE yinyan_song_records
+        SET platform = %s
+        WHERE song_id = %s
+          AND record_id = %s
+          AND platform IS NULL
+        """,
+        [(r['platform'], r['song_id'], r['record_id']) for r in records],
+    )
 def upsert_yinyan_song_records(cur, records: list[dict]) -> None:
-    """Mark pre-initialized yinyan song-record rows as pushed to crawler."""
+    """Mark pre-initialized yinyan song-record rows as pushed to crawler.
+    Matches only on song_id so singer-fallback can use a different record_id than initialized.
+    """
    if not records:
        return
    cur.executemany(
@@ -39,11 +79,12 @@ def upsert_yinyan_song_records(cur, records: list[dict]) -> None:
        UPDATE yinyan_song_records
        SET platform = %s,
            platform_song_id = %s,
+            record_id = %s,
            is_yinyan_push = TRUE
-        WHERE song_id = %s AND record_id = %s
+        WHERE song_id = %s AND is_yinyan_push = FALSE
        """,
        [
-            (r['platform'], r['platform_song_id'], r['song_id'], r['record_id'])
+            (r['platform'], r['platform_song_id'], r['record_id'], r['song_id'])
            for r in records
        ],
    )
@@ -98,8 +139,8 @@ def upsert_qq_songs(cur, songs: list[dict]) -> None:
            (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,
-             audio_md5, provider_name, crawler_source_data)
+             version, audio_md5, provider_name, crawler_source_data)
-        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, 0, NOW(), NOW(), %s, %s, %s::json)
+        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, 0, NOW(), NOW(), %s, %s, %s, %s::json)
        ON CONFLICT (platform_song_id) DO NOTHING
    """
    rows = [(
@@ -108,6 +149,7 @@ def upsert_qq_songs(cur, songs: list[dict]) -> None:
        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', '[]'),
+        s.get('version'),
        s.get('audio_md5'),
        s.get('provider_name'), s.get('crawler_source_data'),
    ) for s in songs]
@@ -189,8 +231,8 @@ def upsert_kugou_songs(cur, songs: list[dict]) -> None:
            (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,
-             audio_md5, provider_name, crawler_source_data)
+             version, audio_md5, provider_name, crawler_source_data)
-        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, 0, NOW(), NOW(), %s, %s, %s::json)
+        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, 0, NOW(), NOW(), %s, %s, %s, %s::json)
        ON CONFLICT (platform_song_id) DO NOTHING
    """
    rows = [(
@@ -200,6 +242,7 @@ def upsert_kugou_songs(cur, songs: list[dict]) -> None:
        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', '[]'),
+        s.get('version'),
        s.get('audio_md5'),
        s.get('provider_name'), s.get('crawler_source_data'),
    ) for s in songs]
@@ -277,8 +320,8 @@ def upsert_netease_songs(cur, songs: list[dict]) -> None:
            (id, platform_song_id, album_id, cover, title, name, duration,
             lyric, composer_name, lyricist_name, url, lyric_url,
             platform_index_url, published_at, album, singers, status, created_at, updated_at,
-             audio_md5, provider_name, crawler_source_data)
+             version, audio_md5, provider_name, crawler_source_data)
-        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s::json, %s::jsonb, 0, NOW(), NOW(), %s, %s, %s::json)
+        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s::json, %s::jsonb, 0, NOW(), NOW(), %s, %s, %s, %s::json)
        ON CONFLICT (platform_song_id) DO NOTHING
    """
    rows = [(
@@ -287,6 +330,7 @@ def upsert_netease_songs(cur, songs: list[dict]) -> None:
        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('album_json'), s.get('singers_json', '[]'),
+        s.get('version'),
        s.get('audio_md5'),
        s.get('provider_name'), s.get('crawler_source_data'),
    ) for s in songs]
--- a/run_etl.py
View file @c403b29
+++ b/run_etl.py
View file @c403b29
 #!/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 initialize_yinyan_song_records, run
+from etl_to_crawler.runner import backfill_yinyan_record_platforms, initialize_yinyan_song_records, run
 PLATFORM_MAP = {
    'qq': PLATFORM_QQ,
@@ -18,6 +18,8 @@ if __name__ == '__main__':
                        help='最多处理多少批次（冒烟测试用）')
    parser.add_argument('--init-yinyan-records', action='store_true',
                        help='只初始化 yinyan_song_records 待导入状态表，不执行 crawler 导入')
+    parser.add_argument('--backfill-yinyan-platforms', action='store_true',
+                        help='只回填 yinyan_song_records 中为空的 platform 平台代码')
    args = parser.parse_args()
    if args.platform == 'all':
@@ -26,7 +28,9 @@ if __name__ == '__main__':
        platforms = [PLATFORM_MAP[args.platform]]
    print(f"Starting ETL for platforms: {platforms}")
-    if args.init_yinyan_records:
+    if args.backfill_yinyan_platforms:
+        backfill_yinyan_record_platforms(max_batches=args.max_batches)
+    elif args.init_yinyan_records:
        initialize_yinyan_song_records(platforms, max_batches=args.max_batches)
    else:
        run(platforms, max_batches=args.max_batches)
--- a/tests/test_runner.py
View file @c403b29
+++ b/tests/test_runner.py
View file @c403b29
@@ -57,7 +57,7 @@ def test_run_imports_all_platform_records_but_writes_one_primary_yinyan_relation
        'audio_url': 'https://example.com/a.mp3',
        'singer': '歌手',
    }})
-    monkeypatch.setattr(runner, 'fetch_platform_records', lambda conn, song_ids: [
+    platform_records = [
        {
            'source_song_id': 10,
            'record_id': 100,
@@ -80,7 +80,10 @@ def test_run_imports_all_platform_records_but_writes_one_primary_yinyan_relation
            'is_high': 0,
            'pub_time': '2021-01-01',
        },
-    ])
+    ]
+    monkeypatch.setattr(runner, 'fetch_platform_records', lambda conn, song_ids: platform_records)
+    monkeypatch.setattr(runner, 'fetch_all_platform_records', lambda conn, song_ids: platform_records)
+    monkeypatch.setattr(runner, '_pick_record_with_singer', lambda records, spider_conn: records[0])
    monkeypatch.setattr(runner, '_PROCESSORS', processors)
    monkeypatch.setattr(runner, 'upsert_yinyan_song_records', yinyan_writer)
@@ -90,9 +93,9 @@ def test_run_imports_all_platform_records_but_writes_one_primary_yinyan_relation
    processors['2'].assert_called_once()
    yinyan_writer.assert_called_once_with(pg_conn.cur, [{
        'song_id': 10,
-        'record_id': 200,
+        'record_id': 100,
-        'platform': 'kugou',
+        'platform': '1',
-        'platform_song_id': 200,
+        'platform_song_id': 100,
    }])
    assert pg_conn.commits == 1
@@ -104,6 +107,7 @@ def test_initialize_yinyan_song_records_inserts_primary_records(monkeypatch):
    monkeypatch.setattr(runner, 'get_hk_songs_conn', lambda: _Connection())
    monkeypatch.setattr(runner, 'get_source_conn', lambda: _Connection())
    monkeypatch.setattr(runner, 'get_pg_conn', lambda: pg_conn)
+    monkeypatch.setattr(runner, 'fetch_existing_yinyan_song_ids', lambda cur: set())
    monkeypatch.setattr(runner, 'iter_hk_songs_batches', lambda conn, batch_size: [[
        {'id': 50, 'source_song_id': 10, 'name': '歌', 'audio_url': 'https://example.com/a.mp3', 'singer': '歌手'},
@@ -136,7 +140,32 @@ def test_initialize_yinyan_song_records_inserts_primary_records(monkeypatch):
    runner.initialize_yinyan_song_records(['1', '2'])
-    assert inserted == [{'song_id': 10, 'record_id': 200}]
+    assert inserted == [{'song_id': 10, 'record_id': 200, 'platform': '2'}]
+    assert pg_conn.commits == 1
+def test_backfill_yinyan_record_platforms_updates_missing_platform_rows(monkeypatch):
+    pg_conn = _PgConnection()
+    updated = []
+    monkeypatch.setattr(runner, 'get_source_conn', lambda: _Connection())
+    monkeypatch.setattr(runner, 'get_pg_conn', lambda: pg_conn)
+    monkeypatch.setattr(runner, 'fetch_yinyan_records_missing_platform', lambda cur, batch_size: [
+        {'song_id': 10, 'record_id': 100},
+        {'song_id': 11, 'record_id': 101},
+    ] if pg_conn.commits == 0 else [])
+    monkeypatch.setattr(runner, 'fetch_record_platforms', lambda conn, record_ids: {
+        100: '1',
+        101: '2',
+    })
+    monkeypatch.setattr(runner, 'update_yinyan_record_platforms', lambda cur, rows: updated.extend(rows))
+    runner.backfill_yinyan_record_platforms()
+    assert updated == [
+        {'song_id': 10, 'record_id': 100, 'platform': '1'},
+        {'song_id': 11, 'record_id': 101, 'platform': '2'},
+    ]
    assert pg_conn.commits == 1
@@ -158,7 +187,7 @@ def test_process_netease_builds_album_json_for_song_insert(monkeypatch):
            'is_owner': 1,
            'album_published_at': '2020-01-01',
            'cover': 'https://example.com/cover.jpg',
-            'title': '录音标题',
+            'title': '化风行万里 (DJ默涵版)',
            'duration': 180,
            'lyric': '[00:01.00]歌词',
            'composer_name': '曲作者',
@@ -197,6 +226,8 @@ def test_process_netease_builds_album_json_for_song_insert(monkeypatch):
    assert inserted_songs[0]['album_json']
    assert '"id": 20' in inserted_songs[0]['album_json']
    assert '"title": "专辑"' in inserted_songs[0]['album_json']
+    assert inserted_songs[0]['title'] == '化风行万里'
+    assert inserted_songs[0]['version'] == 'DJ默涵版'
 def test_process_netease_keeps_timestamped_lyric_and_uploads_plain_lyric(monkeypatch):
@@ -206,7 +237,7 @@ def test_process_netease_keeps_timestamped_lyric_and_uploads_plain_lyric(monkeyp
    uploaded = {}
    class Bucket:
-        def put_object(self, key, body):
+        def put_object(self, key, body, headers=None):
            uploaded['key'] = key
            uploaded['body'] = body
--- a/tests/test_utils.py 0 → 100644
View file @c403b29
+++ b/tests/test_utils.py 0 → 100644
View file @c403b29
+from etl_to_crawler.utils import split_title_version
+def test_split_title_version_extracts_parenthesized_suffix():
+    assert split_title_version("化风行万里 (DJ默涵版)") == ("化风行万里", "DJ默涵版")
+    assert split_title_version("化风行万里（DJ默涵版）") == ("化风行万里", "DJ默涵版")
+def test_split_title_version_leaves_plain_title_unchanged():
+    assert split_title_version("化风行万里") == ("化风行万里", "")
+    assert split_title_version("") == ("", "")
--- a/tests/test_writer.py
View file @c403b29
+++ b/tests/test_writer.py
View file @c403b29
 from unittest.mock import MagicMock, call
 from etl_to_crawler.writer import (
+    fetch_yinyan_records_missing_platform,
    insert_yinyan_song_records,
+    update_yinyan_record_platforms,
    upsert_kugou_albums,
    upsert_kugou_singers,
    upsert_kugou_songs,
@@ -45,32 +47,60 @@ def test_upsert_qq_singer_songs():
 def test_upsert_yinyan_song_records_marks_existing_relation_as_pushed():
    cur = MagicMock()
    upsert_yinyan_song_records(cur, [
-        {'song_id': 10, 'record_id': 100, 'platform': 'qq', 'platform_song_id': 1000},
+        {'song_id': 10, 'record_id': 100, 'platform': '1', 'platform_song_id': 1000},
-        {'song_id': 11, 'record_id': 101, 'platform': 'kugou', 'platform_song_id': 1001},
+        {'song_id': 11, 'record_id': 101, 'platform': '2', 'platform_song_id': 1001},
    ])
    sql, rows = cur.executemany.call_args[0]
    assert 'UPDATE yinyan_song_records' in sql
    assert 'platform = %s' in sql
    assert 'platform_song_id = %s' in sql
+    assert 'record_id = %s' in sql
    assert 'is_yinyan_push = TRUE' in sql
-    assert 'WHERE song_id = %s AND record_id = %s' in sql
+    assert 'WHERE song_id = %s AND is_yinyan_push = FALSE' in sql
-    assert rows == [('qq', 1000, 10, 100), ('kugou', 1001, 11, 101)]
+    assert rows == [('1', 1000, 100, 10), ('2', 1001, 101, 11)]
 def test_insert_yinyan_song_records_initializes_unpushed_rows():
    cur = MagicMock()
    insert_yinyan_song_records(cur, [
-        {'song_id': 10, 'record_id': 100},
+        {'song_id': 10, 'record_id': 100, 'platform': '1'},
-        {'song_id': 11, 'record_id': 101},
+        {'song_id': 11, 'record_id': 101, 'platform': '2'},
    ])
    sql, rows = cur.executemany.call_args[0]
    assert 'INSERT INTO yinyan_song_records' in sql
+    assert 'platform' in sql
    assert 'is_yinyan_push' in sql
    assert 'FALSE' in sql
-    assert 'ON CONFLICT (song_id, record_id) DO NOTHING' in sql
+    assert 'ON CONFLICT (song_id, record_id) DO UPDATE' in sql
-    assert rows == [(10, 100), (11, 101)]
+    assert 'WHERE yinyan_song_records.platform IS NULL' in sql
+    assert rows == [(10, 100, '1'), (11, 101, '2')]
+def test_fetch_yinyan_records_missing_platform_reads_null_platform_rows():
+    cur = MagicMock()
+    cur.fetchall.return_value = [(10, 100), (11, 101)]
+    rows = fetch_yinyan_records_missing_platform(cur, 500)
+    sql, params = cur.execute.call_args[0]
+    assert 'WHERE platform IS NULL' in sql
+    assert params == (500,)
+    assert rows == [{'song_id': 10, 'record_id': 100}, {'song_id': 11, 'record_id': 101}]
+def test_update_yinyan_record_platforms_fills_only_null_platform_rows():
+    cur = MagicMock()
+    update_yinyan_record_platforms(cur, [
+        {'song_id': 10, 'record_id': 100, 'platform': '1'},
+        {'song_id': 11, 'record_id': 101, 'platform': '2'},
+    ])
+    sql, rows = cur.executemany.call_args[0]
+    assert 'SET platform = %s' in sql
+    assert 'AND platform IS NULL' in sql
+    assert rows == [('1', 10, 100), ('2', 11, 101)]
 def test_upsert_netease_songs_writes_album_json_column():
@@ -123,13 +153,15 @@ def test_upsert_kugou_songs_writes_provider_and_source_data():
        'provider_name': 'yinyan',
        'crawler_source_data': '{"id": 200}',
        'audio_md5': 'md5-200',
+        'version': 'DJ默涵版',
    }])
    sql, rows = cur.executemany.call_args[0]
    assert 'audio_md5' in sql
+    assert 'version' in sql
    assert 'provider_name, crawler_source_data' in sql
    assert '%s::json' in sql
-    assert rows[0][-3:] == ('md5-200', 'yinyan', '{"id": 200}')
+    assert rows[0][-4:] == ('DJ默涵版', 'md5-200', 'yinyan', '{"id": 200}')
 def test_upsert_kugou_singers_writes_provider_and_source_data():
@@ -232,11 +264,13 @@ def test_upsert_qq_entities_write_provider_and_source_data():
        'provider_name': 'yinyan',
        'crawler_source_data': '{"id": 200}',
        'audio_md5': 'md5-200',
+        'version': 'DJ默涵版',
    }])
    sql, rows = cur.executemany.call_args[0]
    assert 'audio_md5' in sql
+    assert 'version' in sql
    assert 'provider_name, crawler_source_data' in sql
-    assert rows[0][-3:] == ('md5-200', 'yinyan', '{"id": 200}')
+    assert rows[0][-4:] == ('DJ默涵版', 'md5-200', 'yinyan', '{"id": 200}')
 def test_upsert_netease_entities_write_provider_and_source_data():
@@ -296,8 +330,10 @@ def test_upsert_netease_entities_write_provider_and_source_data():
        'provider_name': 'yinyan',
        'crawler_source_data': '{"id": 300}',
        'audio_md5': 'md5-300',
+        'version': 'DJ默涵版',
    }])
    sql, rows = cur.executemany.call_args[0]
    assert 'audio_md5' in sql
+    assert 'version' in sql
    assert 'provider_name, crawler_source_data' in sql
-    assert rows[0][-3:] == ('md5-300', 'yinyan', '{"id": 300}')
+    assert rows[0][-4:] == ('DJ默涵版', 'md5-300', 'yinyan', '{"id": 300}')