数据库迁移最难的不是搬数据，是搬存储过程

"""迁移数据一致性校验：行数 → 校验和 → 抽样比对，三层逐步收紧。"""

from __future__ import annotations

import hashlib
from dataclasses import dataclass, field
from typing import Any, Sequence


@dataclass
class TableCheck:
    table: str
    key_column: str
    compare_columns: Sequence[str]
    source_rows: int = 0
    target_rows: int = 0
    checksum_match: bool = False
    sample_diffs: list[dict[str, Any]] = field(default_factory=list)

    @property
    def passed(self) -> bool:
        return (
            self.source_rows == self.target_rows
            and self.checksum_match
            and not self.sample_diffs
        )


def count_rows(conn, table: str) -> int:
    with conn.cursor() as cur:
        cur.execute(f"SELECT COUNT(*) FROM {table}")
        return cur.fetchone()[0]


def table_checksum(conn, table: str, columns: Sequence[str], key: str) -> str:
    """按主键排序后计算整表内容的 MD5，用于快速判断是否完全一致。

    注意：跨引擎的类型格式化差异会导致校验和不同，
    所以要统一转成文本并处理 NULL 与数值精度。
    """
    col_expr = ", ".join(f"COALESCE(CAST({c} AS CHAR), '~NULL~')" for c in columns)
    with conn.cursor() as cur:
        cur.execute(f"SELECT {col_expr} FROM {table} ORDER BY {key}")
        h = hashlib.md5()
        for row in cur:
            h.update("|".join(row).encode("utf-8"))
    return h.hexdigest()


def sample_compare(
    src_conn, tgt_conn, table: str, key: str,
    columns: Sequence[str], sample_size: int = 1000,
) -> list[dict[str, Any]]:
    """随机抽样逐行比对，捕捉校验和无法暴露的精度类问题。"""
    col_list = ", ".join(columns)
    with src_conn.cursor() as cur:
        cur.execute(
            f"SELECT {key}, {col_list} FROM {table} "
            f"ORDER BY RAND() LIMIT {sample_size}"
        )
        src_rows = {r[0]: r[1:] for r in cur.fetchall()}

    if not src_rows:
        return []

    keys = ", ".join(repr(k) for k in src_rows)
    with tgt_conn.cursor() as cur:
        cur.execute(
            f"SELECT {key}, {col_list} FROM {table} WHERE {key} IN ({keys})"
        )
        tgt_rows = {r[0]: r[1:] for r in cur.fetchall()}

    diffs: list[dict[str, Any]] = []
    for k, src_val in src_rows.items():
        tgt_val = tgt_rows.get(k)
        if tgt_val is None:
            diffs.append({"key": k, "issue": "目标端缺失"})
        elif tuple(map(str, src_val)) != tuple(map(str, tgt_val)):
            diffs.append({
                "key": k, "issue": "值不一致",
                "source": src_val, "target": tgt_val,
            })
    return diffs


def run_checks(src_conn, tgt_conn, checks: list[TableCheck]) -> tuple[bool, list[TableCheck]]:
    for chk in checks:
        chk.source_rows = count_rows(src_conn, chk.table)
        chk.target_rows = count_rows(tgt_conn, chk.table)

        if chk.source_rows != chk.target_rows:
            continue  # 行数就不对，没必要往下算

        src_sum = table_checksum(src_conn, chk.table, chk.compare_columns, chk.key_column)
        tgt_sum = table_checksum(tgt_conn, chk.table, chk.compare_columns, chk.key_column)
        chk.checksum_match = src_sum == tgt_sum

        if not chk.checksum_match:
            chk.sample_diffs = sample_compare(
                src_conn, tgt_conn, chk.table, chk.key_column, chk.compare_columns
            )

    all_passed = all(c.passed for c in checks)
    return all_passed, checks