feat(check): 新增录音名与平台歌曲标题核对工具

+#!/usr/bin/env python3
+"""核对音眼录音名与 crawler 平台歌曲标题，并输出 Excel。
+
+数据链路：
+    CRAWLER_DB.yinyan_song_records.record_id
+      -> SOURCE_DB.hk_music_record.id -> record_name
+    CRAWLER_DB.yinyan_song_records.(platform, platform_song_id)
+      -> 对应 crawler_*_songs.platform_song_id -> title
+
+用法：
+    .venv/bin/python check_record_titles.py
+    .venv/bin/python check_record_titles.py --output output/record_title_check.xlsx
+"""
+
+import argparse
+import logging
+from pathlib import Path
+from typing import Iterable
+
+from openpyxl import Workbook
+from openpyxl.styles import Alignment, Font, PatternFill
+from openpyxl.utils import get_column_letter
+
+from etl_to_crawler.connections import close_all_pools, get_pg_conn, get_source_conn
+
+
+log = logging.getLogger(__name__)
+
+DEFAULT_OUTPUT = "record_title_check.xlsx"
+DEFAULT_BATCH_SIZE = 2000
+ALLOWED_IMPORT_TABLES = {"yinyan_song_records", "yinyan_song_records2"}
+
+HEADER_FILL = PatternFill(fill_type="solid", fgColor="4472C4")
+MATCH_FILL = PatternFill(fill_type="solid", fgColor="C6EFCE")
+HEADER_FONT = Font(color="FFFFFF", bold=True)
+
+
+def chunked(values: list[int], size: int) -> Iterable[list[int]]:
+    """将列表切成固定大小的批次。"""
+    for start in range(0, len(values), size):
+        yield values[start:start + size]
+
+
+def fetch_crawler_rows(pg_conn, import_table: str) -> list[dict]:
+    """读取关联记录，并按 platform 从对应平台歌曲表取得 title。"""
+    if import_table not in ALLOWED_IMPORT_TABLES:
+        raise ValueError(f"不支持的导入表: {import_table}")
+
+    sql = f"""
+        SELECT
+            ysr.record_id,
+            ysr.platform,
+            ysr.platform_song_id,
+            ysr.recording_id,
+            CASE ysr.platform
+                WHEN '1' THEN qq.title
+                WHEN '2' THEN kg.title
+                WHEN '4' THEN ne.title
+                ELSE NULL
+            END AS title
+        FROM {import_table} AS ysr
+        LEFT JOIN crawler_qqmusic_songs AS qq
+          ON ysr.platform = '1'
+         AND ysr.platform_song_id = qq.platform_song_id
+        LEFT JOIN crawler_kugou_songs AS kg
+          ON ysr.platform = '2'
+         AND ysr.platform_song_id = kg.platform_song_id
+        LEFT JOIN crawler_netease_songs AS ne
+          ON ysr.platform = '4'
+         AND ysr.platform_song_id = ne.platform_song_id
+        WHERE ysr.is_archive_push = TRUE
+        ORDER BY ysr.record_id, ysr.platform, ysr.platform_song_id
+    """
+    with pg_conn.cursor() as cursor:
+        cursor.execute(sql)
+        rows = cursor.fetchall()
+
+    return [
+        {
+            "record_id": row[0],
+            "platform": row[1],
+            "platform_song_id": row[2],
+            "recording_id": row[3],
+            "title": row[4],
+        }
+        for row in rows
+    ]
+
+
+def fetch_record_names(mysql_conn, record_ids: Iterable[int], batch_size: int) -> dict[int, str | None]:
+    """按 record_id 分批读取 hk_music_record.record_name。"""
+    unique_ids = sorted({int(record_id) for record_id in record_ids if record_id is not None})
+    names: dict[int, str | None] = {}
+
+    for batch in chunked(unique_ids, batch_size):
+        placeholders = ", ".join(["%s"] * len(batch))
+        sql = f"""
+            SELECT id, record_name
+            FROM hk_music_record
+            WHERE id IN ({placeholders})
+        """
+        with mysql_conn.cursor() as cursor:
+            cursor.execute(sql, tuple(batch))
+            for row in cursor.fetchall():
+                names[int(row["id"])] = row["record_name"]
+
+    return names
+
+
+def combine_rows(crawler_rows: list[dict], record_names: dict[int, str | None]) -> list[dict]:
+    """合并两个数据库的查询结果，并计算是否完全相同。"""
+    result = []
+    for row in crawler_rows:
+        record_id = row["record_id"]
+        record_name = record_names.get(int(record_id)) if record_id is not None else None
+        title = row["title"]
+        result.append(
+            {
+                "record_id": record_id,
+                "record_name": record_name,
+                "platform": row["platform"],
+                "platform_song_id": row["platform_song_id"],
+                "recording_id": row["recording_id"],
+                "title": title,
+                # 两边都必须有值；不做 trim、大小写或繁简转换，按库中原值精确比较。
+                "matched": record_name is not None and title is not None and record_name == title,
+            }
+        )
+    return result
+
+
+def write_excel(rows: list[dict], output_path: Path) -> None:
+    """写出三列表格；record_name 与 title 相同时整行标绿。"""
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+
+    workbook = Workbook()
+    worksheet = workbook.active
+    worksheet.title = "录音名与标题核对"
+    worksheet.freeze_panes = "A2"
+
+    headers = (
+        "record_id",
+        "record_name",
+        "title",
+        "platform",
+        "platform_song_id",
+        "recording_id",
+    )
+    for column, header in enumerate(headers, start=1):
+        cell = worksheet.cell(row=1, column=column, value=header)
+        cell.fill = HEADER_FILL
+        cell.font = HEADER_FONT
+        cell.alignment = Alignment(horizontal="center", vertical="center")
+
+    for row_number, row in enumerate(rows, start=2):
+        values = (
+            row["record_id"],
+            row["record_name"],
+            row["title"],
+            row["platform"],
+            row["platform_song_id"],
+            row["recording_id"],
+        )
+        for column, value in enumerate(values, start=1):
+            cell = worksheet.cell(row=row_number, column=column, value=value)
+            cell.alignment = Alignment(vertical="top", wrap_text=True)
+            if row["matched"]:
+                cell.fill = MATCH_FILL
+
+    for column, width in enumerate((16, 48, 48, 12, 20, 40), start=1):
+        worksheet.column_dimensions[get_column_letter(column)].width = width
+    worksheet.auto_filter.ref = f"A1:F{max(1, worksheet.max_row)}"
+
+    workbook.save(output_path)
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="核对 hk_music_record.record_name 与 crawler title")
+    parser.add_argument("--output", default=DEFAULT_OUTPUT, help=f"Excel 输出路径（默认: {DEFAULT_OUTPUT}）")
+    parser.add_argument(
+        "--table",
+        choices=sorted(ALLOWED_IMPORT_TABLES),
+        default="yinyan_song_records",
+        help="待核对的 CRAWLER_DB 关联表",
+    )
+    parser.add_argument("--batch-size", type=int, default=DEFAULT_BATCH_SIZE, help="MySQL 单批查询数量")
+    args = parser.parse_args()
+    if args.batch_size <= 0:
+        parser.error("--batch-size 必须大于 0")
+    return args
+
+
+def main() -> None:
+    args = parse_args()
+    output_path = Path(args.output).expanduser().resolve()
+
+    try:
+        pg_conn = get_pg_conn()
+        mysql_conn = get_source_conn()
+
+        crawler_rows = fetch_crawler_rows(pg_conn, args.table)
+        log.info("从 %s 读取 %d 条关联记录", args.table, len(crawler_rows))
+
+        record_names = fetch_record_names(
+            mysql_conn,
+            (row["record_id"] for row in crawler_rows),
+            args.batch_size,
+        )
+        log.info("从 hk_music_record 读取 %d 个录音名", len(record_names))
+
+        rows = combine_rows(crawler_rows, record_names)
+        write_excel(rows, output_path)
+
+        matched = sum(row["matched"] for row in rows)
+        missing_name = sum(row["record_name"] is None for row in rows)
+        missing_title = sum(row["title"] is None for row in rows)
+        log.info(
+            "核对完成：总数=%d，相同=%d，不同=%d，缺少 record_name=%d，缺少 title=%d",
+            len(rows), matched, len(rows) - matched, missing_name, missing_title,
+        )
+        log.info("Excel 已输出至 %s", output_path)
+    finally:
+        close_all_pools()
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
+    main()