import datetime as dt
import uuid
from zoneinfo import ZoneInfo

import pytest
from sqlalchemy import func, select

from flyway.migration.run import MigrationBlocked, migrate
from flyway.migration.source import MemorySource
from flyway.migration.verify import verify
from flyway.models import AppUser, LegacyValue, Transfer

UTC = ZoneInfo("UTC")
T = dt.datetime(2024, 10, 8, 15, 0)


def _legacy():
    c, p, bt = uuid.uuid4(), uuid.uuid4(), uuid.uuid4()
    li, lo, src, out = uuid.uuid4(), uuid.uuid4(), uuid.uuid4(), uuid.uuid4()
    user = uuid.uuid4()
    log = "10/8/2024 3:00:01 PM <b>Sam(sam)</b> created the Load-In record LI"
    base = dict(IsLocal=False, PackingSlipId=None, DepartureDate=None, FromLocation=None, FromRMP=None,
                ToLocation=None, ToRMP=None, TransferFromId=None, TransportCompany=None,
                VehicleRegistration=None, ConditionOfVehicle=None, NonConformingProduct=False,
                ContainerNumber=None, SealNumber=None, CanNumber=None, AwbNumber=None, Operator="Sam",
                Notes=None, EdIdentification=None, ExportTo=None, ClientId=str(c))
    item = dict(NetWeight=None, RelId=None, SKU=None, Reference=None, EdLocation=None, Unit=1, BatchId=str(bt))
    return {
        "fw_Client": [dict(ID=str(c), Name="C", Email="", MobilePhone="", Phone=None, Phone1=None)],
        "fw_Product": [dict(ID=str(p), Title="P", DefaultUnit=1, AvailableUnit=None)],
        "fw_ProductBatch": [dict(ID=str(bt), ProductId=str(p), Number="B1", InitialStock=0, Date=T,
                                 UsedByDate=dt.datetime(2025, 1, 1))],
        "fw_ClientTransfer": [
            dict(base, ID=str(li), Code="LI", Date=T, IsLoadOut=False, IsDraft=False, Logs=log),
            dict(base, ID=str(lo), Code="LO", Date=T, IsLoadOut=True, IsDraft=False,
                 Logs="10/9/2024 9:00:00 AM <b>Sam(sam)</b> created the Load-Out record LO"),
        ],
        "fw_ClientTransferItem": [
            dict(item, ID=str(src), TransferId=str(li), Quantity=10, RelatedTransferItemId=None, RemainingStock=6),
            dict(item, ID=str(out), TransferId=str(lo), Quantity=4, RelatedTransferItemId=str(src), RemainingStock=0),
        ],
        "aspnet_Users": [dict(UserId=str(user), UserName="sam", LastActivityDate=T)],
        "aspnet_Membership": [dict(UserId=str(user), Email="s@x", IsApproved=True, IsLockedOut=False,
                                   CreateDate=T, LastLoginDate=T)],
        "aspnet_Roles": [dict(RoleId="r1", RoleName="Administrator")],
        "aspnet_UsersInRoles": [dict(UserId=str(user), RoleId="r1")],
        "aspnet_Profile": [],
    }


def test_migrate_converts_dates_keeps_raw_values_and_verifies(db):
    source = MemorySource(_legacy())
    result = migrate(db, source, server_tz=UTC, resolutions={}, allow_pending=False)
    assert result.pending == [] and result.duplicates == []
    li = db.scalar(select(Transfer).where(Transfer.code == "LI"))
    assert li.date.utcoffset() is not None
    sam = db.scalar(select(AppUser))
    assert sam.username == "sam" and sam.password_hash is None and sam.is_admin
    assert db.scalar(select(LegacyValue).where(LegacyValue.column_name == "Date",
                                                LegacyValue.table_name == "fw_ClientTransfer")).raw_value
    report = verify(db, source)
    assert report.ok, report


def test_pending_dates_block_unless_resolved(db):
    legacy = _legacy()
    legacy["fw_ClientTransfer"][0]["Logs"] = None
    with pytest.raises(MigrationBlocked) as e:
        migrate(db, MemorySource(legacy), server_tz=UTC, resolutions={}, allow_pending=False)
    # 入库单 Logs 为空：入库单日期与（靠该 Logs 时间判定的）批次日期都无法判定
    assert {(p["table"], p["column"]) for p in e.value.pending} == {
        ("fw_ClientTransfer", "Date"), ("fw_ProductBatch", "Date")}
    db.rollback()
    resolutions = {
        ("fw_ClientTransfer", legacy["fw_ClientTransfer"][0]["ID"], "Date"): "user_nz",
        ("fw_ProductBatch", legacy["fw_ProductBatch"][0]["ID"], "Date"): "server",
    }
    result = migrate(db, MemorySource(legacy), server_tz=UTC, resolutions=resolutions, allow_pending=False)
    assert result.pending == []


def test_duplicate_batches_can_be_merged_explicitly(db):
    legacy = _legacy()
    keeper = legacy["fw_ProductBatch"][0]
    dup = dict(keeper, ID=str(uuid.uuid4()), Number=" b1", Date=T + dt.timedelta(days=1))
    legacy["fw_ProductBatch"].append(dup)
    legacy["fw_ClientTransferItem"][0]["BatchId"] = dup["ID"]
    with pytest.raises(MigrationBlocked):
        migrate(db, MemorySource(legacy), server_tz=UTC, resolutions={}, allow_pending=False)
    db.rollback()
    source = MemorySource(legacy)
    result = migrate(db, source, server_tz=UTC, resolutions={}, allow_pending=False, merge_duplicate_batches=True)
    assert result.merged_batches == {dup["ID"].lower(): keeper["ID"].lower()}
    from flyway.models import ProductBatch, TransferItem
    assert db.scalar(select(func.count()).select_from(ProductBatch)) == 1
    assert {str(i.batch_id) for i in db.scalars(select(TransferItem))} == {keeper["ID"].lower()}
    assert db.scalar(select(LegacyValue.resolution).where(LegacyValue.row_id == dup["ID"])) == "merged"
    assert verify(db, source, batch_map=result.merged_batches).ok


def test_duplicates_block_migration(db):
    legacy = _legacy()
    legacy["fw_ClientTransfer"][1]["Code"] = " li "
    with pytest.raises(MigrationBlocked) as e:
        migrate(db, MemorySource(legacy), server_tz=UTC, resolutions={}, allow_pending=True)
    assert e.value.duplicates


def test_duplicate_codes_can_be_renamed_explicitly(db):
    legacy = _legacy()
    lo = legacy["fw_ClientTransfer"][1]
    lo["Code"], lo["Date"] = " li ", T + dt.timedelta(hours=1)  # 晚于入库单 → 被改名
    source = MemorySource(legacy)
    result = migrate(db, source, server_tz=UTC, resolutions={}, allow_pending=True, rename_duplicates=True)
    assert result.renamed and "-DUP1" in result.renamed[0]
    codes = set(db.scalars(select(Transfer.code)))
    assert codes == {"LI", "li-DUP1"}
    raw = db.scalar(select(LegacyValue.raw_value).where(LegacyValue.row_id == lo["ID"], LegacyValue.resolution == "renamed"))
    assert raw == " li "
    assert verify(db, source).ok
    # 改名后的新值被改错，核对必须发现
    db.scalar(select(Transfer).where(Transfer.code == "li-DUP1")).code = "li-DUP9"
    db.flush()
    report = verify(db, source)
    assert not report.ok and [(m[0], m[2], m[3], m[4]) for m in report.field_mismatches] == [
        ("fw_ClientTransfer", "Code", "li-DUP1", "li-DUP9")]


def test_verify_catches_changed_fields_and_accounts(db):
    legacy = _legacy()
    legacy["fw_Template"] = [dict(ID=str(uuid.uuid4()), Type="LoadInPrintHeader", Content="<p>{{PageNumber}}</p>")]
    source = MemorySource(legacy)
    migrate(db, source, server_tz=UTC, resolutions={}, allow_pending=False)
    assert verify(db, source).ok
    from flyway.models import Template
    db.scalar(select(Template)).content = "<p>changed</p>"
    db.scalar(select(Transfer).where(Transfer.code == "LO")).awb_number = "X"
    db.scalar(select(AppUser)).email = "other@x"
    db.flush()
    report = verify(db, source)
    assert not report.ok and report.field_mismatch_count == 3
    assert {(m[0], m[2]) for m in report.field_mismatches} == {
        ("fw_Template", "Content"), ("fw_ClientTransfer", "AwbNumber"), ("app_user", "email")}


def test_pending_list_shows_both_conversions(db):
    legacy = _legacy()
    legacy["fw_ClientTransfer"][0]["Logs"] = ""  # 入库单日期无法判定
    with pytest.raises(MigrationBlocked) as e:
        migrate(db, MemorySource(legacy), server_tz=UTC, resolutions={}, allow_pending=False)
    row = next(p for p in e.value.pending if p["table"] == "fw_ClientTransfer")
    # 2024-10-08 15:00：若是 UTC 服务器时间 → 新西兰 2024-10-09 04:00（夏令时 +13）；若是用户录入 → 原样
    assert (row["nz_if_server_time"], row["nz_if_user_input"]) == ("2024-10-09 04:00:00", "2024-10-08 15:00:00")