#!/usr/bin/env python3
"""用 hikoon-data-spider 平台原始标题修复 archive/crawler 标题字段。

数据链路：
    archive_crawler.yinyan_song_records (is_archive_push = TRUE)
      的 (platform, platform_song_id)
      -> hikoon-data-spider 对应 media_*_songs.id -> title
      -> title_norm.normalize_title(spider_title) -> (title_norm, version)

写入目标：
* archive_data.archive_recording：title/title_norm/version_type
* archive_data.archive_recording_platform_source：title/title_norm/version_type
* archive_crawler.crawler_*_songs：title/version

默认只执行 dry-run 统计，不写库；必须显式添加 ``--apply`` 才会提交。
"""

import argparse
import json
import logging
import os
from datetime import datetime, timezone
from dataclasses import dataclass
from pathlib import Path
from typing import Iterable, Sequence
from uuid import uuid4

from openpyxl import Workbook
from openpyxl.cell import WriteOnlyCell
from openpyxl.styles import Font, PatternFill
import title_norm as tn

from etl_to_crawler.connections import (
    close_all_pools,
    get_archive_conn,
    get_archive_crawler_conn,
    get_spider_conn,
)


log = logging.getLogger(__name__)

DEFAULT_BATCH_SIZE = 1000
DEFAULT_REPORT = "output/fix_record_titles_report.xlsx"
DEFAULT_BACKUP_DIR = Path("output/backups")
BACKUP_SCHEMA_VERSION = 1
MAX_VARCHAR_LEN = 50  # archive/crawler 表 title/title_norm/version_type 列最大长度

PLATFORMS = {
    "1": {
        "archive_name": "qqmusic",
        "spider_table": "media_tencent_songs",
        "crawler_table": "crawler_qqmusic_songs",
    },
    "2": {
        "archive_name": "kugou",
        "spider_table": "media_ku_gou_songs",
        "crawler_table": "crawler_kugou_songs",
    },
    "4": {
        "archive_name": "netease",
        "spider_table": "media_netease_songs",
        "crawler_table": "crawler_netease_songs",
    },
}


@dataclass(frozen=True)
class InputRow:
    record_id: int
    platform: str
    platform_song_id: int
    recording_id: str


@dataclass(frozen=True)
class FixRow(InputRow):
    spider_title: str
    title_norm: str
    version: str


def chunked(rows: Sequence, size: int) -> Iterable[Sequence]:
    for start in range(0, len(rows), size):
        yield rows[start:start + size]


def fetch_input_rows(crawler_conn, limit: int | None = None) -> tuple[list[InputRow], dict[str, int]]:
    """直接读取正式 yinyan_song_records 的全部 archive 推送关系。"""
    sql = """
        SELECT record_id, platform, platform_song_id, recording_id
        FROM yinyan_song_records
        WHERE is_archive_push = TRUE
        ORDER BY record_id, platform, platform_song_id
    """
    params = None
    if limit is not None:
        sql += " LIMIT %s"
        params = (limit,)
    with crawler_conn.cursor() as cursor:
        cursor.execute(sql, params)
        values_list = cursor.fetchall()

    stats = {"total": 0, "missing_key": 0, "unsupported_platform": 0}
    rows: list[InputRow] = []
    seen_recordings: set[str] = set()
    seen_platform_songs: set[tuple[str, int]] = set()
    for values in values_list:
        stats["total"] += 1
        record_id, platform_value, platform_song_id, recording_id = values
        platform = str(platform_value)
        if platform not in PLATFORMS:
            stats["unsupported_platform"] += 1
            continue
        if None in (platform_song_id, recording_id, record_id):
            stats["missing_key"] += 1
            continue

        row = InputRow(
            record_id=int(record_id),
            platform=platform,
            platform_song_id=int(platform_song_id),
            recording_id=str(recording_id),
        )
        recording_key = row.recording_id
        song_key = (row.platform, row.platform_song_id)
        if recording_key in seen_recordings:
            raise ValueError(f"Excel 中 recording_id 重复: {recording_key}")
        if song_key in seen_platform_songs:
            raise ValueError(f"Excel 中 platform_song_id 重复: {song_key}")
        seen_recordings.add(recording_key)
        seen_platform_songs.add(song_key)
        rows.append(row)
    return rows, stats


def fetch_spider_titles(spider_conn, rows: list[InputRow], batch_size: int) -> dict[tuple[str, int], str]:
    """按平台从 spider 歌曲表读取原始 title；表名严格来自白名单。"""
    titles: dict[tuple[str, int], str] = {}
    for platform, config in PLATFORMS.items():
        ids = sorted({row.platform_song_id for row in rows if row.platform == platform})
        table = config["spider_table"]
        for batch in chunked(ids, batch_size):
            placeholders = ", ".join(["%s"] * len(batch))
            with spider_conn.cursor() as cursor:
                cursor.execute(
                    f"SELECT id, title FROM {table} WHERE id IN ({placeholders})",
                    tuple(batch),
                )
                for result in cursor.fetchall():
                    title = result["title"]
                    if title is not None and str(title).strip() != "":
                        titles[(platform, int(result["id"]))] = str(title)
    return titles


def build_fix_rows(
    input_rows: list[InputRow],
    spider_titles: dict[tuple[str, int], str],
    long_only: bool = False,
    exclude_titles: set[str] | None = None,
) -> tuple[list[FixRow], dict[str, int]]:
    """以 spider title 为唯一修复源，并用 title_norm 提取归一标题和版本。

    默认模式下超过 ``MAX_VARCHAR_LEN`` 的记录会被跳过，避免写入时 varchar 溢出。
    当 ``long_only=True`` 时反转过滤：仅保留超长记录，用于 DBA 扩列后补跑。
    """
    fixes: list[FixRow] = []
    exclude_titles = exclude_titles or set()
    stats = {
        "missing_spider": 0,
        "excluded_title": 0,
        "skip_long_title": 0,
        "skip_long_title_norm": 0,
        "skip_long_version": 0,
        "skip_short": 0,
    }
    for row in input_rows:
        spider_title = spider_titles.get((row.platform, row.platform_song_id))
        if spider_title is None:
            stats["missing_spider"] += 1
            continue
        if spider_title in exclude_titles:
            stats["excluded_title"] += 1
            continue
        normalized_title, version = tn.normalize_title(spider_title)
        is_long = (
            len(spider_title) > MAX_VARCHAR_LEN
            or len(normalized_title) > MAX_VARCHAR_LEN
            or (version and len(version) > MAX_VARCHAR_LEN)
        )
        if long_only:
            if not is_long:
                stats["skip_short"] += 1
                continue
        else:
            if is_long:
                if len(spider_title) > MAX_VARCHAR_LEN:
                    stats["skip_long_title"] += 1
                elif len(normalized_title) > MAX_VARCHAR_LEN:
                    stats["skip_long_title_norm"] += 1
                else:
                    stats["skip_long_version"] += 1
                continue
        fixes.append(
            FixRow(
                **row.__dict__,
                spider_title=spider_title,
                title_norm=normalized_title,
                version=version,
            )
        )
    return fixes, stats


def _archive_recording_sql(batch_length: int, apply: bool, lock: bool = False) -> tuple[str, list[str]]:
    placeholders = ", ".join(["(%s, %s, %s, %s)"] * batch_length)
    values = "(recording_id, new_title, new_title_norm, new_version)"
    params = ["recording_id", "spider_title", "title_norm", "archive_version"]
    difference = """(
        target.title IS DISTINCT FROM source.new_title
        OR target.title_norm IS DISTINCT FROM source.new_title_norm
        OR target.version_type IS DISTINCT FROM source.new_version
    )"""
    if apply:
        sql = f"""
            UPDATE archive_recording AS target
            SET title = source.new_title,
                title_norm = source.new_title_norm,
                version_type = source.new_version,
                updated_at = NOW()
            FROM (VALUES {placeholders}) AS source{values}
            WHERE target.recording_id = source.recording_id
              AND {difference}
            RETURNING target.recording_id
        """
    else:
        sql = f"""
            SELECT target.recording_id,
                   target.title, source.new_title,
                   target.title_norm, source.new_title_norm,
                   target.version_type, source.new_version,
                   target.updated_at
            FROM archive_recording AS target
            JOIN (VALUES {placeholders}) AS source{values}
              ON target.recording_id = source.recording_id
            WHERE {difference}
            {"FOR UPDATE OF target" if lock else ""}
        """
    return sql, params


def _archive_source_sql(batch_length: int, apply: bool, lock: bool = False) -> tuple[str, list[str]]:
    placeholders = ", ".join(["(%s, %s, %s, %s, %s, %s)"] * batch_length)
    values = "(recording_id, platform, platform_song_id, new_title, new_title_norm, new_version)"
    params = [
        "recording_id", "archive_platform", "platform_song_id_text",
        "spider_title", "title_norm", "archive_version",
    ]
    difference = """(
        target.title IS DISTINCT FROM source.new_title
        OR target.title_norm IS DISTINCT FROM source.new_title_norm
        OR target.version_type IS DISTINCT FROM source.new_version
    )"""
    join = """target.recording_id = source.recording_id
        AND target.platform = source.platform
        AND target.platform_song_id = source.platform_song_id"""
    if apply:
        sql = f"""
            UPDATE archive_recording_platform_source AS target
            SET title = source.new_title,
                title_norm = source.new_title_norm,
                version_type = source.new_version,
                updated_at = NOW()
            FROM (VALUES {placeholders}) AS source{values}
            WHERE {join} AND {difference}
            RETURNING target.recording_id, target.platform, target.platform_song_id
        """
    else:
        sql = f"""
            SELECT target.recording_id, target.platform, target.platform_song_id,
                   target.title, source.new_title,
                   target.title_norm, source.new_title_norm,
                   target.version_type, source.new_version,
                   target.updated_at
            FROM archive_recording_platform_source AS target
            JOIN (VALUES {placeholders}) AS source{values} ON {join}
            WHERE {difference}
            {"FOR UPDATE OF target" if lock else ""}
        """
    return sql, params


def _crawler_sql(table: str, batch_length: int, apply: bool, lock: bool = False) -> tuple[str, list[str]]:
    placeholders = ", ".join(["(%s, %s, %s)"] * batch_length)
    values = "(platform_song_id, new_title, new_version)"
    params = ["platform_song_id", "spider_title", "version"]
    difference = """(
        target.title IS DISTINCT FROM source.new_title
        OR target.version IS DISTINCT FROM source.new_version
    )"""
    if apply:
        sql = f"""
            UPDATE {table} AS target
            SET title = source.new_title,
                version = source.new_version,
                updated_at = NOW()
            FROM (VALUES {placeholders}) AS source{values}
            WHERE target.platform_song_id = source.platform_song_id::bigint
              AND {difference}
            RETURNING target.platform_song_id
        """
    else:
        sql = f"""
            SELECT target.platform_song_id,
                   target.title, source.new_title,
                   target.version, source.new_version,
                   target.updated_at
            FROM {table} AS target
            JOIN (VALUES {placeholders}) AS source{values}
              ON target.platform_song_id = source.platform_song_id::bigint
            WHERE {difference}
            {"FOR UPDATE OF target" if lock else ""}
        """
    return sql, params


def _row_value(row: FixRow, name: str):
    values = {
        "recording_id": row.recording_id,
        "archive_platform": PLATFORMS[row.platform]["archive_name"],
        "platform_song_id_text": str(row.platform_song_id),
        "platform_song_id": row.platform_song_id,
        "spider_title": row.spider_title,
        "title_norm": row.title_norm,
        "archive_version": row.version or None,
        "version": row.version,
    }
    return values[name]


def execute_batches(conn, rows: list[FixRow], sql_builder, batch_size: int, apply: bool) -> list[tuple]:
    if not rows:
        return []
    matched_keys: list[tuple] = []
    for batch in chunked(rows, batch_size):
        sql, param_names = sql_builder(len(batch), apply)
        params = [_row_value(row, name) for row in batch for name in param_names]
        with conn.cursor() as cursor:
            cursor.execute(sql, tuple(params))
            matched_keys.extend(tuple(result) for result in cursor.fetchall())
    return matched_keys


def _changed_fields(pairs: tuple[tuple[str, object, object], ...]) -> str:
    return ",".join(field for field, old, new in pairs if old != new)


def build_report_rows(fixes: list[FixRow], preview: dict[str, list[tuple]], apply: bool) -> list[tuple]:
    """把更新前快照和目标值转换成可审计的前后对比明细。"""
    by_recording = {row.recording_id: row for row in fixes}
    by_platform_song = {(row.platform, row.platform_song_id): row for row in fixes}
    action = "updated" if apply else "would_update"
    report_rows: list[tuple] = []

    for result in preview.get("archive_recording", []):
        row = by_recording.get(str(result[0]))
        if row:
            fields = _changed_fields((
                ("title", result[1], result[2]),
                ("title_norm", result[3], result[4]),
                ("version_type", result[5], result[6]),
            ))
            report_rows.append((
                "archive_data", "archive_recording", row.record_id, row.recording_id,
                row.platform, row.platform_song_id, fields,
                result[1], result[2], result[3], result[4], result[5], result[6], action,
            ))

    for result in preview.get("archive_recording_platform_source", []):
        row = by_recording.get(str(result[0]))
        if row:
            fields = _changed_fields((
                ("title", result[3], result[4]),
                ("title_norm", result[5], result[6]),
                ("version_type", result[7], result[8]),
            ))
            report_rows.append((
                "archive_data", "archive_recording_platform_source", row.record_id,
                row.recording_id, row.platform, row.platform_song_id,
                fields, result[3], result[4], result[5], result[6],
                result[7], result[8], action,
            ))

    for platform, config in PLATFORMS.items():
        table = config["crawler_table"]
        for result in preview.get(table, []):
            row = by_platform_song.get((platform, int(result[0])))
            if row:
                fields = _changed_fields((
                    ("title", result[1], result[2]),
                    ("version", result[3], result[4]),
                ))
                report_rows.append((
                    "archive_crawler", table, row.record_id, row.recording_id,
                    row.platform, row.platform_song_id, fields,
                    result[1], result[2], None, None, result[3], result[4], action,
                ))
    return report_rows


def build_backup_records(
    fixes: list[FixRow],
    preview: dict[str, list[tuple]],
    run_id: str,
) -> list[dict]:
    """将加锁读取的更新前快照转换成可机器回退的记录。"""
    by_recording = {row.recording_id: row for row in fixes}
    by_platform_song = {(row.platform, row.platform_song_id): row for row in fixes}
    records: list[dict] = []

    for result in preview.get("archive_recording", []):
        row = by_recording[str(result[0])]
        records.append({
            "type": "change", "schema_version": BACKUP_SCHEMA_VERSION, "run_id": run_id,
            "database": "archive_data", "table": "archive_recording",
            "keys": {"recording_id": row.recording_id},
            "old": {
                "title": result[1], "title_norm": result[3],
                "version_type": result[5], "updated_at": result[7],
            },
            "new": {
                "title": result[2], "title_norm": result[4], "version_type": result[6],
            },
        })

    for result in preview.get("archive_recording_platform_source", []):
        records.append({
            "type": "change", "schema_version": BACKUP_SCHEMA_VERSION, "run_id": run_id,
            "database": "archive_data", "table": "archive_recording_platform_source",
            "keys": {
                "recording_id": str(result[0]), "platform": str(result[1]),
                "platform_song_id": str(result[2]),
            },
            "old": {
                "title": result[3], "title_norm": result[5],
                "version_type": result[7], "updated_at": result[9],
            },
            "new": {
                "title": result[4], "title_norm": result[6], "version_type": result[8],
            },
        })

    for platform, config in PLATFORMS.items():
        table = config["crawler_table"]
        for result in preview.get(table, []):
            row = by_platform_song[(platform, int(result[0]))]
            records.append({
                "type": "change", "schema_version": BACKUP_SCHEMA_VERSION, "run_id": run_id,
                "database": "archive_crawler", "table": table,
                "keys": {"platform_song_id": row.platform_song_id},
                "old": {
                    "title": result[1], "version": result[3], "updated_at": result[5],
                },
                "new": {"title": result[2], "version": result[4]},
            })
    return records


def _json_default(value):
    if isinstance(value, (datetime,)):
        return value.isoformat()
    raise TypeError(f"无法序列化 {type(value).__name__}")


def write_backup_atomic(path: Path, records: list[dict], run_id: str) -> None:
    """先 fsync 临时文件再无覆盖发布，保证 UPDATE 前备份已持久化。"""
    path.parent.mkdir(parents=True, exist_ok=True)
    if path.exists():
        raise FileExistsError(f"拒绝覆盖已有回退备份: {path}")
    temp_path = path.with_name(path.name + f".{uuid4().hex}.tmp")
    metadata = {
        "type": "metadata",
        "schema_version": BACKUP_SCHEMA_VERSION,
        "run_id": run_id,
        "created_at": datetime.now(timezone.utc).isoformat(),
        "change_count": len(records),
    }
    with temp_path.open("w", encoding="utf-8") as handle:
        handle.write(json.dumps(metadata, ensure_ascii=False) + "\n")
        for record in records:
            handle.write(json.dumps(record, ensure_ascii=False, default=_json_default) + "\n")
        handle.flush()
        os.fsync(handle.fileno())
    try:
        # 硬链接发布不会覆盖已存在目标；并发使用同一路径时也会安全失败。
        os.link(temp_path, path)
        directory_fd = os.open(path.parent, os.O_RDONLY)
        try:
            os.fsync(directory_fd)
        finally:
            os.close(directory_fd)
    finally:
        temp_path.unlink(missing_ok=True)


def load_backup(path: Path) -> tuple[dict, list[dict]]:
    with path.open("r", encoding="utf-8") as handle:
        lines = [json.loads(line) for line in handle if line.strip()]
    if not lines or lines[0].get("type") != "metadata":
        raise ValueError("无效的回退备份：缺少 metadata")
    metadata = lines[0]
    records = lines[1:]
    if metadata.get("schema_version") != BACKUP_SCHEMA_VERSION:
        raise ValueError("不支持的备份版本")
    if metadata.get("change_count") != len(records):
        raise ValueError("备份记录数校验失败")
    return metadata, records


def _rollback_sql(table: str, batch_length: int) -> tuple[str, list[str]]:
    """生成带新值保护的批量回退 SQL；表名只能来自内部白名单。"""
    if table == "archive_recording":
        placeholders = ", ".join(["(%s, %s, %s, %s, %s, %s, %s, %s)"] * batch_length)
        sql = f"""
            UPDATE archive_recording AS target
            SET title = source.old_title,
                title_norm = source.old_title_norm,
                version_type = source.old_version,
                updated_at = source.old_updated_at::timestamptz
            FROM (VALUES {placeholders}) AS source(
                recording_id, old_title, old_title_norm, old_version, old_updated_at,
                new_title, new_title_norm, new_version
            )
            WHERE target.recording_id = source.recording_id
              AND target.title IS NOT DISTINCT FROM source.new_title
              AND target.title_norm IS NOT DISTINCT FROM source.new_title_norm
              AND target.version_type IS NOT DISTINCT FROM source.new_version
            RETURNING target.recording_id
        """
        names = [
            "recording_id", "old_title", "old_title_norm", "old_version", "old_updated_at",
            "new_title", "new_title_norm", "new_version",
        ]
    elif table == "archive_recording_platform_source":
        placeholders = ", ".join(["(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"] * batch_length)
        sql = f"""
            UPDATE archive_recording_platform_source AS target
            SET title = source.old_title,
                title_norm = source.old_title_norm,
                version_type = source.old_version,
                updated_at = source.old_updated_at::timestamptz
            FROM (VALUES {placeholders}) AS source(
                recording_id, platform, platform_song_id,
                old_title, old_title_norm, old_version, old_updated_at,
                new_title, new_title_norm, new_version
            )
            WHERE target.recording_id = source.recording_id
              AND target.platform = source.platform
              AND target.platform_song_id = source.platform_song_id
              AND target.title IS NOT DISTINCT FROM source.new_title
              AND target.title_norm IS NOT DISTINCT FROM source.new_title_norm
              AND target.version_type IS NOT DISTINCT FROM source.new_version
            RETURNING target.recording_id, target.platform, target.platform_song_id
        """
        names = [
            "recording_id", "platform", "platform_song_id",
            "old_title", "old_title_norm", "old_version", "old_updated_at",
            "new_title", "new_title_norm", "new_version",
        ]
    elif table in {config["crawler_table"] for config in PLATFORMS.values()}:
        placeholders = ", ".join(["(%s, %s, %s, %s, %s, %s)"] * batch_length)
        sql = f"""
            UPDATE {table} AS target
            SET title = source.old_title,
                version = source.old_version,
                updated_at = source.old_updated_at::timestamp
            FROM (VALUES {placeholders}) AS source(
                platform_song_id, old_title, old_version, old_updated_at,
                new_title, new_version
            )
            WHERE target.platform_song_id = source.platform_song_id::bigint
              AND target.title IS NOT DISTINCT FROM source.new_title
              AND target.version IS NOT DISTINCT FROM source.new_version
            RETURNING target.platform_song_id
        """
        names = [
            "platform_song_id", "old_title", "old_version", "old_updated_at",
            "new_title", "new_version",
        ]
    else:
        raise ValueError(f"备份包含不允许回退的表: {table}")
    return sql, names


def _backup_value(record: dict, name: str):
    if name in record["keys"]:
        return record["keys"][name]
    if name.startswith("old_"):
        field = name[4:]
        if field == "version":
            field = "version_type" if "version_type" in record["old"] else "version"
        return record["old"].get(field)
    if name.startswith("new_"):
        field = name[4:]
        if field == "version":
            field = "version_type" if "version_type" in record["new"] else "version"
        return record["new"].get(field)
    raise KeyError(name)


def execute_rollback_batches(conn, table: str, records: list[dict], batch_size: int) -> list[tuple]:
    restored: list[tuple] = []
    for batch in chunked(records, batch_size):
        sql, names = _rollback_sql(table, len(batch))
        params = [_backup_value(record, name) for record in batch for name in names]
        with conn.cursor() as cursor:
            cursor.execute(sql, tuple(params))
            restored.extend(tuple(result) for result in cursor.fetchall())
    return restored


def rollback_backup(path: Path, batch_size: int) -> dict[str, int]:
    """按备份回退；当前值不再等于本次新值的记录会安全跳过。"""
    metadata, records = load_backup(path)
    groups: dict[tuple[str, str], list[dict]] = {}
    for record in records:
        if record.get("type") != "change" or record.get("run_id") != metadata.get("run_id"):
            raise ValueError("备份记录 run_id 或类型无效")
        key = (record.get("database"), record.get("table"))
        groups.setdefault(key, []).append(record)

    archive_conn = get_archive_conn()
    crawler_conn = get_archive_crawler_conn()
    restored: dict[str, int] = {}
    try:
        for (database, table), table_records in groups.items():
            if database == "archive_data":
                conn = archive_conn
            elif database == "archive_crawler":
                conn = crawler_conn
            else:
                raise ValueError(f"备份包含不允许回退的数据库: {database}")
            keys = execute_rollback_batches(conn, table, table_records, batch_size)
            restored[table] = len(keys)
        archive_conn.commit()
        crawler_conn.commit()
    except Exception:
        archive_conn.rollback()
        crawler_conn.rollback()
        raise
    finally:
        close_all_pools()

    log.info("已按 run_id=%s 执行回退", metadata["run_id"])
    for table, count in restored.items():
        log.info("%s：已回退 %d 行", table, count)
    return restored


def write_report(report_path: Path, rows: list[tuple]) -> None:
    report_path.parent.mkdir(parents=True, exist_ok=True)
    workbook = Workbook(write_only=True)
    worksheet = workbook.create_sheet()
    worksheet.title = "更新明细"
    worksheet.freeze_panes = "A2"
    headers = [
        "database", "table", "record_id", "recording_id", "platform",
        "platform_song_id", "changed_fields", "old_title", "new_title",
        "old_title_norm", "new_title_norm", "old_version", "new_version", "action",
    ]
    header_fill = PatternFill(fill_type="solid", fgColor="4472C4")
    header_font = Font(color="FFFFFF", bold=True)
    old_fill = PatternFill(fill_type="solid", fgColor="FFC7CE")
    new_fill = PatternFill(fill_type="solid", fgColor="C6EFCE")
    header_cells = []
    for header in headers:
        cell = WriteOnlyCell(worksheet, value=header)
        cell.fill = header_fill
        cell.font = header_font
        header_cells.append(cell)
    worksheet.append(header_cells)
    for values in rows:
        output_cells = []
        changed_fields = set(str(values[6]).split(","))
        changed_columns = set()
        if "title" in changed_fields:
            changed_columns.update((8, 9))
        if "title_norm" in changed_fields:
            changed_columns.update((10, 11))
        if "version" in changed_fields or "version_type" in changed_fields:
            changed_columns.update((12, 13))
        for column, value in enumerate(values, 1):
            if value is None:
                value = "<NULL>"
            elif value == "":
                value = "<EMPTY>"
            cell = WriteOnlyCell(worksheet, value=value)
            if column in changed_columns:
                cell.fill = old_fill if column in (8, 10, 12) else new_fill
            output_cells.append(cell)
        worksheet.append(output_cells)
    widths = (18, 42, 15, 24, 10, 20, 34, 48, 48, 40, 40, 24, 24, 16)
    for column, width in enumerate(widths, 1):
        worksheet.column_dimensions[chr(64 + column)].width = width
    worksheet.auto_filter.ref = f"A1:N{len(rows) + 1}"
    workbook.save(report_path)


def run(
    batch_size: int,
    apply: bool,
    limit: int | None,
    report_path: Path,
    backup_path: Path | None = None,
    long_only: bool = False,
    exclude_titles: set[str] | None = None,
) -> dict[str, int]:
    crawler_conn = get_archive_crawler_conn()
    input_rows, input_stats = fetch_input_rows(crawler_conn, limit)
    spider_conn = get_spider_conn()
    spider_titles = fetch_spider_titles(spider_conn, input_rows, batch_size)
    fixes, fix_stats = build_fix_rows(
        input_rows,
        spider_titles,
        long_only=long_only,
        exclude_titles=exclude_titles,
    )
    skip_long = fix_stats["skip_long_title"] + fix_stats["skip_long_title_norm"] + fix_stats["skip_long_version"]
    if long_only:
        log.info(
            "[long-only] yinyan archive记录=%d，有效键=%d，超长候选=%d，短标题跳过=%d，spider 缺失=%d",
            input_stats["total"], len(input_rows), len(fixes), fix_stats["skip_short"], fix_stats["missing_spider"],
        )
    else:
        log.info(
            "yinyan archive记录=%d，有效键=%d，spider 标题=%d，spider 缺失=%d，超长跳过=%d (title=%d, title_norm=%d, version=%d)",
            input_stats["total"], len(input_rows), len(fixes), fix_stats["missing_spider"],
            skip_long, fix_stats["skip_long_title"], fix_stats["skip_long_title_norm"], fix_stats["skip_long_version"],
        )
    if fix_stats["excluded_title"]:
        log.info("本次按 spider title 精确排除=%d 条，不进入预览、备份或更新", fix_stats["excluded_title"])

    archive_conn = get_archive_conn()
    preview: dict[str, list[tuple]] = {}
    actual: dict[str, list[tuple]] = {}
    run_id = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + "_" + uuid4().hex[:8]
    try:
        preview["archive_recording"] = execute_batches(
            archive_conn, fixes,
            lambda length, _: _archive_recording_sql(length, False, lock=apply),
            batch_size, False,
        )
        preview["archive_recording_platform_source"] = execute_batches(
            archive_conn, fixes,
            lambda length, _: _archive_source_sql(length, False, lock=apply),
            batch_size, False,
        )
        for platform, config in PLATFORMS.items():
            table = config["crawler_table"]
            platform_rows = [row for row in fixes if row.platform == platform]
            preview[table] = execute_batches(
                crawler_conn,
                platform_rows,
                lambda length, _, table=table: _crawler_sql(table, length, False, lock=apply),
                batch_size,
                False,
            )
        if apply:
            backup_records = build_backup_records(fixes, preview, run_id)
            if backup_path is None:
                backup_path = DEFAULT_BACKUP_DIR / f"fix_record_titles_{run_id}.jsonl"
            backup_path = backup_path.expanduser().resolve()
            write_backup_atomic(backup_path, backup_records, run_id)
            log.info("回退备份已持久化至 %s，共 %d 条", backup_path, len(backup_records))
            actual["archive_recording"] = execute_batches(
                archive_conn, fixes, _archive_recording_sql, batch_size, True,
            )
            actual["archive_recording_platform_source"] = execute_batches(
                archive_conn, fixes, _archive_source_sql, batch_size, True,
            )
            for platform, config in PLATFORMS.items():
                table = config["crawler_table"]
                platform_rows = [row for row in fixes if row.platform == platform]
                actual[table] = execute_batches(
                    crawler_conn,
                    platform_rows,
                    lambda length, do_apply, table=table: _crawler_sql(table, length, do_apply),
                    batch_size,
                    True,
                )
            archive_conn.commit()
            crawler_conn.commit()
        else:
            actual = preview
            archive_conn.rollback()
            crawler_conn.rollback()
    except Exception:
        archive_conn.rollback()
        crawler_conn.rollback()
        raise
    finally:
        close_all_pools()

    report_rows = build_report_rows(fixes, preview, apply)
    write_report(report_path, report_rows)
    results = {table: len(keys) for table, keys in actual.items()}
    for table, count in results.items():
        log.info("%s：%s %d 行", table, "已更新" if apply else "需更新", count)
    if not apply:
        log.info("当前为 dry-run，未修改数据库；确认后添加 --apply")
    log.info("更新明细已输出至 %s，共 %d 行", report_path, len(report_rows))
    return results


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="按 spider 原始标题修复 archive/crawler 标题")
    parser.add_argument("--batch-size", type=int, default=DEFAULT_BATCH_SIZE)
    parser.add_argument("--limit", type=int, help="最多处理多少条 yinyan 关系（默认全部）")
    parser.add_argument("--report", default=DEFAULT_REPORT, help=f"更新明细 Excel（默认: {DEFAULT_REPORT}）")
    parser.add_argument("--backup", help="--apply 前写入的回退 JSONL 路径（默认自动生成）")
    parser.add_argument("--long-only", action="store_true", help="仅处理超长标题记录（DBA 扩列后补跑）")
    parser.add_argument(
        "--exclude-title",
        action="append",
        default=[],
        help="本次运行不更新的 spider 完整标题（精确匹配，可重复传入）",
    )
    mode = parser.add_mutually_exclusive_group()
    mode.add_argument("--apply", action="store_true", help="备份成功后实际提交更新；默认仅 dry-run")
    mode.add_argument("--rollback", help="使用指定 JSONL 备份安全回退")
    args = parser.parse_args()
    if args.batch_size <= 0:
        parser.error("--batch-size 必须大于 0")
    if args.limit is not None and args.limit <= 0:
        parser.error("--limit 必须大于 0")
    if args.rollback and (args.limit is not None or args.backup or args.exclude_title):
        parser.error("--rollback 不能与 --limit、--backup 或 --exclude-title 同时使用")
    return args


def main() -> None:
    args = parse_args()
    if args.rollback:
        rollback_backup(Path(args.rollback).expanduser().resolve(), args.batch_size)
        return
    run(
        args.batch_size,
        args.apply,
        args.limit,
        Path(args.report).expanduser().resolve(),
        Path(args.backup) if args.backup else None,
        long_only=args.long_only,
        exclude_titles=set(args.exclude_title),
    )


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
    main()