feat(etl): 新增 yinyan_song_records2 初始化功能及空字符清理支持

- 扩展 run_etl.py，添加 --init-yinyan-records2 参数支持初始化 yinyan_song_records2
- 提升 BATCH_SIZE 从 100 到 1000，提升批处理效率
- 实现 _NulSafePgConnection 及 _NulSafePgCursor，自动清理 PostgreSQL 参数中的 NUL 字符，防止插入错误
- 新增查询接口 iter_hk_songs_records2_batches 和 fetch_all_song_record_relations，用于提取满足 records2 条件的歌曲关联数据
- 在 runner.py 中实现 initialize_yinyan_song_records2，写入所有合格歌曲关系并统一去除 records1 已存在的重复关系
- 在 writer.py 中新增 insert_yinyan_song_records2 和 delete_yinyan_song_records2_existing_relations 函数，实现批量插入和基于 records1 的重复关系删除
- 新增单元测试覆盖 NUL 字符清理、records2 插入及去重逻辑，保障代码质量和功能稳定性

@@ -48,7 +48,7 @@ PLATFORM_KUGOU = '2'
 PLATFORM_NETEASE = '4'
 PLATFORMS = [PLATFORM_QQ, PLATFORM_KUGOU, PLATFORM_NETEASE]
 
-BATCH_SIZE = 100
+BATCH_SIZE = 1000
 BACKFILL_BATCH_SIZE = int(os.environ.get('BACKFILL_BATCH_SIZE', '5000'))
 HTTP_POOL_MAXSIZE = int(os.environ.get('HTTP_POOL_MAXSIZE', '128'))
 HTTP_TRANSFER_RETRIES = int(os.environ.get('HTTP_TRANSFER_RETRIES', '3'))

@@ -5,6 +5,58 @@ import oss2
 from .config import SOURCE_DB, HK_SONGS_DB, CRAWLER_DB, OSS_CONFIG, OSS_CONNECTION_POOL_SIZE
 
 
+def _strip_nul(value):
+    """PostgreSQL text/json 参数不接受 NUL 字符，递归清理全部绑定参数。"""
+    if isinstance(value, str):
+        return value.replace('\x00', '')
+    if isinstance(value, tuple):
+        return tuple(_strip_nul(item) for item in value)
+    if isinstance(value, list):
+        return [_strip_nul(item) for item in value]
+    if isinstance(value, dict):
+        return {key: _strip_nul(item) for key, item in value.items()}
+    return value
+
+
+class _NulSafePgCursor:
+    """在所有 PostgreSQL 参数化查询前统一移除文本中的 NUL 字符。"""
+
+    def __init__(self, cursor):
+        self._cursor = cursor
+
+    def __enter__(self):
+        self._cursor.__enter__()
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        return self._cursor.__exit__(exc_type, exc_value, traceback)
+
+    def execute(self, operation, parameters=None):
+        if parameters is None:
+            return self._cursor.execute(operation)
+        return self._cursor.execute(operation, _strip_nul(parameters))
+
+    def executemany(self, operation, parameter_sets):
+        return self._cursor.executemany(
+            operation,
+            [_strip_nul(parameters) for parameters in parameter_sets],
+        )
+
+    def __getattr__(self, name):
+        return getattr(self._cursor, name)
+
+
+class _NulSafePgConnection:
+    def __init__(self, connection):
+        self._connection = connection
+
+    def cursor(self, *args, **kwargs):
+        return _NulSafePgCursor(self._connection.cursor(*args, **kwargs))
+
+    def __getattr__(self, name):
+        return getattr(self._connection, name)
+
+
 def get_source_conn() -> pymysql.Connection:
     return pymysql.connect(**SOURCE_DB, cursorclass=pymysql.cursors.DictCursor)
 
@@ -19,10 +71,10 @@ def get_hk_songs_conn() -> pymysql.Connection:
     return pymysql.connect(**HK_SONGS_DB, cursorclass=pymysql.cursors.DictCursor)
 
 
-def get_pg_conn() -> pg8000.Connection:
+def get_pg_conn() -> _NulSafePgConnection:
     conn = pg8000.connect(**CRAWLER_DB)
     conn.run("SET TimeZone = 'Asia/Shanghai'")
-    return conn
+    return _NulSafePgConnection(conn)
 
 
 def get_oss_bucket() -> oss2.Bucket:

@@ -28,6 +28,24 @@ WHERE deleted = '0'
 ORDER BY id
 """
 
+# 用于补全 yinyan_song_records2 的歌曲必须具备所有下游所需字段。
+# 当前 hk_songs_test 中的 lyrics_url 即歌曲链接字段。
+_HK_SONGS_RECORDS2_QUERY = """
+SELECT id, source_song_id
+FROM hk_songs_test
+WHERE deleted = '0'
+  AND id > %s
+  AND source_song_id IS NOT NULL
+  AND name IS NOT NULL AND TRIM(name) != ''
+  AND song_time IS NOT NULL AND song_time > 0
+  AND singer IS NOT NULL AND TRIM(singer) != ''
+  AND lyrics_url IS NOT NULL AND TRIM(lyrics_url) != ''
+  AND audio_url IS NOT NULL AND TRIM(audio_url) != ''
+  AND issue_time IS NOT NULL
+ORDER BY id
+LIMIT %s
+"""
+
 _PLATFORM_QUERY = """
 SELECT
     sar.song_id AS source_song_id,
@@ -96,6 +114,25 @@ def iter_hk_songs_batches(
                 break
 
 
+def iter_hk_songs_records2_batches(
+    conn: pymysql.Connection,
+    batch_size: int,
+    start_after_id: int = 0,
+) -> Iterator[list[dict]]:
+    """遍历具备 records2 入库必填字段的歌曲。"""
+    last_id = start_after_id
+    with conn.cursor() as cur:
+        while True:
+            cur.execute(_HK_SONGS_RECORDS2_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_hk_songs_by_source_ids(conn: pymysql.Connection, source_song_ids: list[int]) -> dict[int, dict]:
     if not source_song_ids:
         return {}
@@ -158,6 +195,27 @@ def fetch_all_platform_records(source_conn: pymysql.Connection, song_ids: list[i
     return result
 
 
+def fetch_all_song_record_relations(source_conn: pymysql.Connection, song_ids: list[int]) -> list[dict]:
+    """返回指定 song_id 的全部有效 (song_id, record_id, platform) 关联。"""
+    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()
+
+    # 同一关联可能因源表脏数据重复出现；保留一条即可。
+    seen_relations: set[tuple[int, int]] = set()
+    result = []
+    for row in rows:
+        key = (int(row['source_song_id']), int(row['record_id']))
+        if key not in seen_relations:
+            seen_relations.add(key)
+            result.append(row)
+    return result
+
+
 def fetch_platform_records_by_record_ids(source_conn: pymysql.Connection, record_ids: list[int]) -> list[dict]:
     """按状态表指定的录音 id 批量查询录音，不再按 song_id 拉取全平台候选。"""
     if not record_ids:

@@ -8,10 +8,12 @@ from .config import PLATFORM_QQ, PLATFORM_KUGOU, PLATFORM_NETEASE, BATCH_SIZE, B
 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,
+    iter_hk_songs_records2_batches,
     fetch_hk_songs_by_source_ids,
     mark_hk_songs_deleted,
     fetch_platform_records,
     fetch_all_platform_records,
+    fetch_all_song_record_relations,
     fetch_platform_records_by_record_ids,
     fetch_record_platforms,
     select_primary_record,
@@ -26,6 +28,8 @@ from .writer import (
     fetch_pending_yinyan_song_records,
     fetch_yinyan_records_missing_platform,
     insert_yinyan_song_records,
+    insert_yinyan_song_records2,
+    delete_yinyan_song_records2_existing_relations,
     update_yinyan_record_platforms,
     upsert_yinyan_song_records,
     fetch_existing_yinyan_song_ids,
@@ -1239,6 +1243,55 @@ def initialize_yinyan_song_records(platforms: list[str], max_batches: int | None
     log.info("Initialized yinyan_song_records candidates=%d, skipped_batches=%d", total, skipped_batches)
 
 
+def initialize_yinyan_song_records2(platforms: list[str], max_batches: int | None = None) -> None:
+    """写入所有合格歌曲关联，再删除 records1 已存在的关联。"""
+    hk_conn = get_hk_songs_conn()
+    src_conn = get_source_conn()
+    pg_conn = get_pg_conn()
+    total_inserted = 0
+
+    try:
+        for index, batch in enumerate(
+            tqdm(iter_hk_songs_records2_batches(hk_conn, BATCH_SIZE), desc='init-yinyan-records2')
+        ):
+            if max_batches is not None and index >= max_batches:
+                break
+
+            song_ids = list({int(row['source_song_id']) for row in batch})
+            relations = fetch_all_song_record_relations(src_conn, song_ids)
+            rows = [
+                {
+                    'song_id': int(relation['source_song_id']),
+                    'record_id': int(relation['record_id']),
+                    'platform': str(relation['platform']),
+                }
+                for relation in relations
+                if str(relation['platform']) in platforms
+            ]
+            if not rows:
+                continue
+
+            with pg_conn.cursor() as pg_cur:
+                insert_yinyan_song_records2(pg_cur, rows)
+            pg_conn.commit()
+            total_inserted += len(rows)
+
+        # 必须在全部候选关系写入后，统一根据 records1 做去重。
+        with pg_conn.cursor() as pg_cur:
+            deleted = delete_yinyan_song_records2_existing_relations(pg_cur)
+        pg_conn.commit()
+    finally:
+        hk_conn.close()
+        src_conn.close()
+        pg_conn.close()
+
+    log.info(
+        "Initialized yinyan_song_records2 candidates=%d, removed_existing_records1_relations=%d",
+        total_inserted,
+        deleted,
+    )
+
+
 def backfill_yinyan_record_platforms(max_batches: int | None = None) -> None:
     src_conn = get_source_conn()
     pg_conn = get_pg_conn()

@@ -38,6 +38,38 @@ def insert_yinyan_song_records(cur, records: list[dict]) -> None:
     )
 
 
+def insert_yinyan_song_records2(cur, records: list[dict]) -> None:
+    """写入全部候选关联；与 records1 的去重由后续独立步骤完成。"""
+    if not records:
+        return
+    values_sql = ', '.join(['(%s::bigint, %s::bigint, %s::varchar)'] * len(records))
+    params = []
+    for record in records:
+        params.extend([record['song_id'], record['record_id'], record['platform']])
+    cur.execute(
+        f"""
+        INSERT INTO yinyan_song_records2 (song_id, record_id, platform)
+        VALUES {values_sql}
+        ON CONFLICT (song_id, record_id) DO UPDATE
+            SET platform = EXCLUDED.platform
+        """,
+        tuple(params),
+    )
+
+
+def delete_yinyan_song_records2_existing_relations(cur) -> int:
+    """删除 records2 中已存在于 records1 的 song-record 关联。"""
+    cur.execute(
+        """
+        DELETE FROM yinyan_song_records2 AS ysr2
+        USING yinyan_song_records AS ysr1
+        WHERE ysr2.song_id = ysr1.song_id
+          AND ysr2.record_id = ysr1.record_id
+        """
+    )
+    return cur.rowcount
+
+
 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")

 #!/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 backfill_yinyan_record_platforms, initialize_yinyan_song_records, run, backfill_empty_singers, backfill_qq_invalid_covers
+from etl_to_crawler.runner import backfill_yinyan_record_platforms, initialize_yinyan_song_records, initialize_yinyan_song_records2, run, backfill_empty_singers, backfill_qq_invalid_covers
 
 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('--init-yinyan-records2', action='store_true',
+                        help='写入全部合格 song-record 关联至 yinyan_song_records2，再排除 yinyan_song_records 已有关系')
     parser.add_argument('--backfill-yinyan-platforms', action='store_true',
                         help='只回填 yinyan_song_records 中为空的 platform 平台代码')
     parser.add_argument('--backfill-empty-singers', action='store_true',
@@ -36,6 +38,8 @@ if __name__ == '__main__':
         backfill_yinyan_record_platforms(max_batches=args.max_batches)
     elif args.init_yinyan_records:
         initialize_yinyan_song_records(platforms, max_batches=args.max_batches)
+    elif args.init_yinyan_records2:
+        initialize_yinyan_song_records2(platforms, max_batches=args.max_batches)
     elif args.backfill_empty_singers:
         backfill_empty_singers(platforms, max_batches=args.max_batches)
     elif args.backfill_qq_invalid_covers:

+from etl_to_crawler.connections import _NulSafePgConnection
+
+
+class _Cursor:
+    def __init__(self):
+        self.executed = []
+        self.executed_many = []
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        return False
+
+    def execute(self, operation, parameters=None):
+        self.executed.append((operation, parameters))
+
+    def executemany(self, operation, parameter_sets):
+        self.executed_many.append((operation, parameter_sets))
+
+
+class _Connection:
+    def __init__(self):
+        self.raw_cursor = _Cursor()
+
+    def cursor(self):
+        return self.raw_cursor
+
+
+def test_pg_connection_removes_nul_from_execute_parameters():
+    raw_connection = _Connection()
+    connection = _NulSafePgConnection(raw_connection)
+
+    with connection.cursor() as cur:
+        cur.execute('INSERT INTO example VALUES (%s, %s)', ('歌\x00词', {'text': '介\x00绍'}))
+
+    assert raw_connection.raw_cursor.executed == [
+        ('INSERT INTO example VALUES (%s, %s)', ('歌词', {'text': '介绍'})),
+    ]
+
+
+def test_pg_connection_removes_nul_from_every_executemany_row():
+    raw_connection = _Connection()
+    connection = _NulSafePgConnection(raw_connection)
+
+    with connection.cursor() as cur:
+        cur.executemany('INSERT INTO example VALUES (%s)', [('A\x00',), ('B\x00',)])
+
+    assert raw_connection.raw_cursor.executed_many == [
+        ('INSERT INTO example VALUES (%s)', [('A',), ('B',)]),
+    ]

 from unittest.mock import MagicMock
 
 from etl_to_crawler.reader import (
+    fetch_all_song_record_relations,
     fetch_platform_records_by_record_ids,
     fetch_platform_records,
     iter_hk_songs_batches,
@@ -86,6 +87,22 @@ def test_fetch_platform_records_by_record_ids_queries_exact_records_once():
     assert result[0]['platform'] == '2'
 
 
+def test_fetch_all_song_record_relations_keeps_all_records_for_a_song():
+    conn = MagicMock()
+    conn.cursor.return_value = _make_cursor([
+        {'source_song_id': 10, 'record_id': 100, 'platform': '1'},
+        {'source_song_id': 10, 'record_id': 101, 'platform': '1'},
+        {'source_song_id': 10, 'record_id': 200, 'platform': '2'},
+        {'source_song_id': 10, 'record_id': 101, 'platform': '1'},
+    ])
+
+    result = fetch_all_song_record_relations(conn, [10])
+
+    assert [(row['record_id'], row['platform']) for row in result] == [
+        (100, '1'), (101, '1'), (200, '2'),
+    ]
+
+
 def test_select_primary_record_prefers_main_version_then_is_high():
     rows = [
         {

@@ -258,6 +258,38 @@ def test_initialize_yinyan_song_records_inserts_primary_records(monkeypatch):
     assert pg_conn.commits == 1
 
 
+def test_initialize_yinyan_song_records2_writes_all_relations_then_deduplicates(monkeypatch):
+    pg_conn = _PgConnection()
+    inserted = []
+    dedupe_calls = []
+
+    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, 'iter_hk_songs_records2_batches', lambda conn, batch_size: [[
+        {'id': 50, 'source_song_id': 10},
+    ]])
+    monkeypatch.setattr(runner, 'fetch_all_song_record_relations', lambda conn, song_ids: [
+        {'source_song_id': 10, 'record_id': 100, 'platform': '1'},
+        {'source_song_id': 10, 'record_id': 200, 'platform': '2'},
+    ])
+    monkeypatch.setattr(runner, 'insert_yinyan_song_records2', lambda cur, rows: inserted.extend(rows))
+    monkeypatch.setattr(
+        runner,
+        'delete_yinyan_song_records2_existing_relations',
+        lambda cur: dedupe_calls.append(True) or 1,
+    )
+
+    runner.initialize_yinyan_song_records2(['1', '2'])
+
+    assert inserted == [
+        {'song_id': 10, 'record_id': 100, 'platform': '1'},
+        {'song_id': 10, 'record_id': 200, 'platform': '2'},
+    ]
+    assert dedupe_calls == [True]
+    assert pg_conn.commits == 2
+
+
 def test_backfill_yinyan_record_platforms_updates_missing_platform_rows(monkeypatch):
     pg_conn = _PgConnection()
     updated = []

@@ -3,6 +3,8 @@ from etl_to_crawler.writer import (
     fetch_pending_yinyan_song_records,
     fetch_yinyan_records_missing_platform,
     insert_yinyan_song_records,
+    insert_yinyan_song_records2,
+    delete_yinyan_song_records2_existing_relations,
     update_yinyan_record_platforms,
     upsert_kugou_albums,
     upsert_kugou_singers,
@@ -123,6 +125,31 @@ def test_insert_yinyan_song_records_initializes_unpushed_rows():
     assert rows == [(10, 100, '1'), (11, 101, '2')]
 
 
+def test_insert_yinyan_song_records2_writes_all_candidates_before_deduplication():
+    cur = MagicMock()
+    insert_yinyan_song_records2(cur, [
+        {'song_id': 10, 'record_id': 100, 'platform': '1'},
+        {'song_id': 10, 'record_id': 200, 'platform': '2'},
+    ])
+
+    sql, params = cur.execute.call_args[0]
+    assert 'INSERT INTO yinyan_song_records2' in sql
+    assert 'VALUES (%s::bigint, %s::bigint, %s::varchar), (%s::bigint, %s::bigint, %s::varchar)' in sql
+    assert 'ON CONFLICT (song_id, record_id) DO UPDATE' in sql
+    assert params == (10, 100, '1', 10, 200, '2')
+
+
+def test_delete_yinyan_song_records2_existing_relations_uses_records1_as_dedup_source():
+    cur = MagicMock()
+    delete_yinyan_song_records2_existing_relations(cur)
+
+    sql = cur.execute.call_args[0][0]
+    assert 'DELETE FROM yinyan_song_records2 AS ysr2' in sql
+    assert 'USING yinyan_song_records AS ysr1' in sql
+    assert 'ysr2.song_id = ysr1.song_id' in sql
+    assert 'ysr2.record_id = ysr1.record_id' in sql
+
+
 def test_fetch_yinyan_records_missing_platform_reads_null_platform_rows():
     cur = MagicMock()
     cur.fetchall.return_value = [(10, 100), (11, 101)]