"""核对报告（文档 12.3、12.5）：行数、逐记录逐字段、员工账号、来源明细剩余库存、客户 / 产品 / 批次三级汇总。

逐字段核对只允许以下有记录的转换，其余任何差异都算不一致：
- 日期时间：按 migration_legacy_value 里存档的原值与判定（utc / server / user_nz / pending / date）重新换算后应相等；
  没有存档的纯日期列（保质期等）只取日期部分。
- 合并的重复批次（batch_map）：被合并的批次不迁入，明细 BatchId 改指保留批次，保留批次的批号只去尾部空格。
- 改名的重复单号 / 条码（--rename-duplicates）：对旧数据副本重新执行同一改名规则得出应有新值，与新库实际值逐字比较；
  存档记录须与规则产生的记录一一对应。
"""
import datetime as dt
import uuid
from collections import defaultdict
from dataclasses import dataclass, field
from zoneinfo import ZoneInfo

from sqlalchemy import Boolean, Date, DateTime, Uuid, func, select
from sqlalchemy.orm import Session

from flyway.migration.dates import Decision, apply, apply_date
from flyway.migration.run import COPY_ORDER, rename_duplicate_codes
from flyway.migration.source import LegacySource
from flyway.models import ADMIN_ROLE, AppUser, LegacyValue, Transfer, TransferItem
from flyway.services.stock import compute_remaining, stock

MAX_LISTED = 200  # 报告里最多列出的字段差异条数（总数另计）


@dataclass
class Report:
    counts: dict = field(default_factory=dict)
    item_mismatches: list = field(default_factory=list)
    aggregate_mismatches: list = field(default_factory=list)
    field_mismatches: list = field(default_factory=list)  # (表, 主键, 列, 旧值, 新值)
    field_mismatch_count: int = 0

    @property
    def ok(self) -> bool:
        return (all(a == b for a, b in self.counts.values()) and not self.item_mismatches
                and not self.aggregate_mismatches and not self.field_mismatch_count)

    def mismatch(self, *row) -> None:
        self.field_mismatch_count += 1
        if len(self.field_mismatches) < MAX_LISTED:
            self.field_mismatches.append(tuple(str(x) for x in row))


def _norm(value, column):
    if value is None:
        return None
    if isinstance(column.type, Uuid):
        return str(value).lower()
    if isinstance(column.type, Boolean):
        return bool(value)
    return value


def _expected(table, row_id, column, value, archived, batch_map, server_tz, report):
    """旧值按允许的转换换算成新库应有的值；返回 None 表示该列已另行判定（不再比较）。"""
    arch = archived.get((table, row_id, column.name))
    if arch is not None and arch.resolution == "renamed":
        return (value,)  # value 已按改名规则换算（见 _check_renames），直接与新库值比较
    if isinstance(column.type, (DateTime, Date)) and value is not None:
        if isinstance(value, str):
            value = dt.datetime.fromisoformat(value)
        if arch is not None:
            if arch.raw_value != value.isoformat():
                report.mismatch(table, row_id, column.name, value, f"archived {arch.raw_value!r}")
            decision = Decision(arch.resolution, arch.reason)
            if isinstance(column.type, Date):
                return (apply_date(value, decision, server_tz),)
            return (apply(value, decision, server_tz),)
        if isinstance(column.type, Date):
            return (value.date() if isinstance(value, dt.datetime) else value,)
        report.mismatch(table, row_id, column.name, value, "converted without an archived decision")
        return None
    if table == "fw_ClientTransferItem" and column.name == "BatchId":
        return (batch_map.get(str(value).lower(), value),)
    if table == "fw_ProductBatch" and column.name == "Number" and row_id in batch_map.values():
        return ((value or "").rstrip(' '),)
    return (value,)


def _check_renames(legacy: dict, archived: dict, report: Report) -> None:
    """在旧数据副本上重放改名规则：改写副本里的单号 / 条码为应有新值，并核对存档与规则一致。"""
    stored = {k: v.raw_value for k, v in archived.items() if v.resolution == "renamed"}
    if not stored:
        return
    replayed: list[dict] = []
    rename_duplicate_codes(legacy, replayed)
    expected = {(r["table_name"], r["row_id"].lower(), r["column_name"]): r["raw_value"] for r in replayed}
    for key in stored.keys() | expected.keys():
        if stored.get(key) != expected.get(key):
            report.mismatch(key[0], key[1], key[2], f"rule {expected.get(key)!r}", f"archived {stored.get(key)!r}")


def _check_fields(session: Session, legacy: dict, batch_map: dict, server_tz: ZoneInfo, report: Report) -> None:
    archived = {(v.table_name, v.row_id.lower(), v.column_name): v for v in session.scalars(select(LegacyValue))}
    _check_renames(legacy, archived, report)
    for model in COPY_ORDER:
        table = model.__tablename__
        pk_cols = list(model.__table__.primary_key.columns)
        target = {tuple(_norm(r[c.name], c) for c in pk_cols): r
                  for r in session.execute(select(model.__table__)).mappings()}
        for old in legacy[table]:
            key = tuple(_norm(old[c.name], c) for c in pk_cols)
            row_id = str(key[0])
            if table == "fw_ProductBatch" and row_id in batch_map:
                continue  # 被合并，不迁入
            new = target.get(key)
            if new is None:
                report.mismatch(table, key, "*", "row", "missing")
                continue
            for column in model.__table__.columns:
                if column.name not in old:
                    continue  # 新库新增的列
                expected = _expected(table, row_id, column, old[column.name], archived, batch_map, server_tz, report)
                if expected is not None and _norm(expected[0], column) != _norm(new[column.name], column):
                    report.mismatch(table, row_id, column.name, expected[0], new[column.name])


def _check_users(session: Session, source: LegacySource, report: Report) -> None:
    """员工账号关键字段（12.3 非业务数据处理表）：用户名、邮箱、启用 / 锁定、创建与最后登录时间、管理员角色。"""
    members = {str(m["UserId"]).lower(): m for m in source.rows("aspnet_Membership")}
    roles = {str(r["RoleId"]).lower(): r["RoleName"] for r in source.rows("aspnet_Roles")}
    admins = {str(ur["UserId"]).lower() for ur in source.rows("aspnet_UsersInRoles")
              if roles.get(str(ur["RoleId"]).lower()) == ADMIN_ROLE}
    users = {str(u.id): u for u in session.scalars(select(AppUser))}

    def utc(v):
        return v.replace(tzinfo=dt.UTC) if v else None

    for old in source.rows("aspnet_Users"):
        key = str(old["UserId"]).lower()
        new, m = users.get(key), members.get(key, {})
        if new is None:
            report.mismatch("app_user", key, "*", "row", "missing")
            continue
        pairs = {"username": (old["UserName"], new.username), "email": (m.get("Email") or "", new.email),
                 "is_active": (bool(m.get("IsApproved", True)), new.is_active),
                 "is_locked_out": (bool(m.get("IsLockedOut", False)), new.is_locked_out),
                 "last_login_at": (utc(m.get("LastLoginDate")), new.last_login_at),
                 "last_activity_at": (utc(old.get("LastActivityDate")), new.last_activity_at),
                 "is_admin": (key in admins, new.is_admin)}
        if m.get("CreateDate"):
            pairs["created_at"] = (utc(m["CreateDate"]), new.created_at)
        for col, (expected, actual) in pairs.items():
            if expected != actual:
                report.mismatch("app_user", key, col, expected, actual)


def verify(session: Session, source: LegacySource, batch_map: dict[str, str] | None = None,
           server_tz: ZoneInfo = ZoneInfo("UTC")) -> Report:
    """batch_map：迁移时合并的批次（被合并 ID → 保留 ID），旧库一侧按同样映射汇总。server_tz 与迁移时相同。"""
    batch_map = batch_map or {}
    report = Report()
    legacy = {m.__tablename__: source.rows(m.__tablename__) for m in COPY_ORDER}
    for model in COPY_ORDER:
        target = session.scalar(select(func.count()).select_from(model.__table__))
        expected = len(legacy[model.__tablename__])
        if model.__tablename__ == "fw_ProductBatch":
            expected -= len(batch_map)
        report.counts[model.__tablename__] = (expected, target)
    _check_fields(session, legacy, batch_map, server_tz, report)
    _check_users(session, source, report)

    load_in_items = session.scalars(select(TransferItem).join(Transfer).where(Transfer.is_load_out.is_(False)))
    for item in load_in_items:
        expected = compute_remaining(session, item)
        if expected != item.remaining_stock:
            report.item_mismatches.append((str(item.id), item.remaining_stock, expected))

    transfers = {str(t["ID"]).lower(): t for t in legacy["fw_ClientTransfer"]}
    batches = {str(b["ID"]).lower(): b for b in legacy["fw_ProductBatch"]}
    sums = {"client_id": defaultdict(int), "product_id": defaultdict(int), "batch_id": defaultdict(int)}
    for it in legacy["fw_ClientTransferItem"]:
        t = transfers[str(it["TransferId"]).lower()]
        if t["IsLoadOut"] or t["IsDraft"]:
            continue
        batch_id = str(it["BatchId"]).lower()
        batch = batches[batch_id]
        sums["client_id"][str(t["ClientId"]).lower()] += it["RemainingStock"]
        sums["product_id"][str(batch["ProductId"]).lower()] += it["RemainingStock"]
        sums["batch_id"][batch_map.get(batch_id, batch_id)] += it["RemainingStock"]
    for level, values in sums.items():
        for key, expected in values.items():
            actual = stock(session, **{level: uuid.UUID(key)})
            if actual != expected:
                report.aggregate_mismatches.append((level, key, expected, actual))
    return report