"""scripts/import_legacy.py(Phase 4 Task 2): 老单租户库 → 多租户租户库。

契约(计划原文):
1. 五表对账: 行数源/目标一致 + 最近 10 条会话逐字段抽查
   (id/session_id/status/page_token)+ conversation_id 原样(max 相等);
2. pending 推送日志 → failed, error_message 追加"[迁移] 切换前未完成态";
3. 幂等: 跑两遍第二遍零变化(对账报告 skipped==全部);
4. 老库结构假设防御: 关键列缺失 → 显性报错列名, 不静默丢数据;
   新库多列 → 默认值填充并在报告标注; 老库多列 → 不迁移并标注;
5. 对账不一致(源≠目标合计) → 非零退出码。

测试方式(照 test_bootstrap_script.py): import + sys.argv 注入, 非 subprocess。
真实老库用例经 env 指路(ALL_KEFU_TEST_LEGACY_JC / _AUTO), 缺失 skip(机器
相关); 用前 VACUUM INTO 拷到 tmp, 绝不直连原库——原库 sha256 前后比对
钉死"零接触"。合成库用例(结构防御/幂等/pending 转换)不依赖 env, 常跑。
"""
import hashlib
import importlib.util
import os
import sqlite3
import sys
from datetime import datetime, timezone
from pathlib import Path

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import Session

from app.engine.models import (
    Base,
    Conversation,
    ConversationSource,
    ConversationStatus,
    CustomerProfile,
    Evaluation,
    IntentLevel,
    LeadPushLog,
    Message,
    MessageRole,
)

REPO_ROOT = Path(__file__).resolve().parents[1]
SEED_DIR = REPO_ROOT / "seed_profiles"
LEGACY_TABLES = (
    "conversations", "messages", "customer_profiles", "evaluations",
    "lead_push_logs",
)
# 行级比对列(名取交集两边都有——jc/auto 老库实测均有这些列)
COMPARE_COLUMNS = {
    "conversations": ("id", "session_id", "page_token", "status", "source",
                      "client_info", "created_at", "updated_at"),
    "messages": ("id", "conversation_id", "role", "content", "created_at"),
    "customer_profiles": ("id", "conversation_id", "raw_info", "summary",
                          "updated_at"),
    "evaluations": ("id", "conversation_id", "intent_level", "reasoning",
                    "contact_info", "email_sent", "correction_count",
                    "evaluated_at"),
    "lead_push_logs": ("id", "evaluation_id", "conversation_id", "push_type",
                       "status", "attempts", "request_payload",
                       "response_status", "outcome", "error_message",
                       "created_at", "updated_at"),
}
PENDING_MARKER = "[迁移] 切换前未完成态"


def _load(name: str):
    spec = importlib.util.spec_from_file_location(
        name, REPO_ROOT / "scripts" / f"{name}.py")
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


# 模块级加载一次(脚本 import 期只做 sys.path 锚定, 无副作用)
import_legacy = _load("import_legacy")
bootstrap_platform = _load("bootstrap_platform")


def run_import(monkeypatch, *argv):
    monkeypatch.setattr(sys, "argv", ["import_legacy.py", *argv])
    import_legacy.main()


def run_bootstrap(monkeypatch, db: Path, *profiles: str):
    monkeypatch.setattr(
        sys, "argv",
        ["bootstrap_platform.py", "--db", str(db), "--seed", str(SEED_DIR),
         "--profiles", *profiles])
    bootstrap_platform.main()


# ---- 夹具/辅助 ---------------------------------------------------------------

def _ro(db) -> sqlite3.Connection:
    return sqlite3.connect(f"file:{db}?mode=ro", uri=True)


def _count(db, table) -> int:
    con = _ro(db)
    try:
        return con.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]
    finally:
        con.close()


def _rows(db, table, columns, order="id"):
    con = _ro(db)
    con.row_factory = sqlite3.Row
    try:
        return [dict(r) for r in con.execute(
            f"SELECT {', '.join(columns)} FROM {table} ORDER BY {order}")]
    finally:
        con.close()


def _spot_rows(db):
    """最近 10 条会话(id/session_id/status/page_token), 源目标同一排序。"""
    con = _ro(db)
    try:
        return con.execute(
            "SELECT id, session_id, status, page_token FROM conversations "
            "ORDER BY created_at DESC, id DESC LIMIT 10").fetchall()
    finally:
        con.close()


def _max_conv_id(db, table):
    con = _ro(db)
    try:
        return con.execute(
            f"SELECT MAX(conversation_id) FROM {table}").fetchone()[0]
    finally:
        con.close()


def _sha256(path) -> str:
    return hashlib.sha256(Path(path).read_bytes()).hexdigest()


def _expected_lpl(rows):
    """源 lead_push_logs 行 → 导入后期望形态(pending→failed+追加备注)。"""
    out = []
    for r in rows:
        r = dict(r)
        if r["status"] == "pending":
            r["status"] = "failed"
            old = r["error_message"]
            r["error_message"] = (old + " " + PENDING_MARKER) if old else \
                PENDING_MARKER
        out.append(r)
    return out


def _parse_report_line(out: str, table: str):
    """'  conversations: 源=3 导入=3 跳过=0 目标=3 一致' → (3,3,0,3)。"""
    import re
    m = re.search(rf"{table}: 源=(\d+) 导入=(\d+) 跳过=(\d+) 目标=(\d+)", out)
    assert m, f"报告缺 {table} 对账行:\n{out}"
    return tuple(int(g) for g in m.groups())


def _assert_tables_identical(source_db, target_db):
    """五表行级比对: 四表逐字相等, lead_push_logs 按转换后期望相等;
    附带计划点名的 conversation_id 原样性(max 相等)。"""
    for table in LEGACY_TABLES:
        src = _rows(source_db, table, COMPARE_COLUMNS[table])
        dst = _rows(target_db, table, COMPARE_COLUMNS[table])
        expected = _expected_lpl(src) if table == "lead_push_logs" else src
        assert dst == expected, f"{table} 行级不一致"
    for table in ("messages", "customer_profiles", "evaluations",
                  "lead_push_logs"):
        assert _max_conv_id(source_db, table) == _max_conv_id(target_db, table)


# ---- 合成老库(确定性, 不依赖 env)----------------------------------------------

C1 = "11111111-1111-1111-1111-111111111111"
C2 = "22222222-2222-2222-2222-222222222222"
C3 = "33333333-3333-3333-3333-333333333333"
E2 = "eeeeeeee-2222-2222-2222-222222222222"
E3 = "eeeeeeee-3333-3333-3333-333333333333"


def _make_legacy_db(path: Path) -> None:
    """3 会话小库: 状态三态齐备 + 2 条 pending 推送(一条带旧错误信息,
    验证追加语义)+ success/failed 各一(验证非 pending 不动)。"""
    engine = create_engine(f"sqlite:///{path}")
    Base.metadata.create_all(engine)
    with Session(engine) as s:
        def ts(day, minute=0):
            return datetime(2026, 9, day, 10, minute, tzinfo=timezone.utc)

        s.add_all([
            Conversation(id=C1, session_id="sess-c1", page_token="tok-c1",
                         status=ConversationStatus.completed,
                         source=ConversationSource.widget,
                         client_info={"ip": "1.2.3.4"},
                         created_at=ts(1), updated_at=ts(1, 5)),
            Conversation(id=C2, session_id="sess-c2", page_token=None,
                         status=ConversationStatus.active,
                         source=ConversationSource.standalone,
                         created_at=ts(2), updated_at=ts(2, 5)),
            Conversation(id=C3, session_id="sess-c3", page_token="tok-c3",
                         status=ConversationStatus.abandoned,
                         source=ConversationSource.embedded,
                         created_at=ts(3), updated_at=ts(3, 5)),
        ])
        msgs = []
        for i, (mid, cid, role, day) in enumerate([
                ("aaaa0000-0000-0000-0000-000000000001", C1,
                 MessageRole.user, 1),
                ("aaaa0000-0000-0000-0000-000000000002", C1,
                 MessageRole.assistant, 1),
                ("aaaa0000-0000-0000-0000-000000000003", C2,
                 MessageRole.user, 2),
                ("aaaa0000-0000-0000-0000-000000000004", C2,
                 MessageRole.assistant, 2),
                ("aaaa0000-0000-0000-0000-000000000005", C3,
                 MessageRole.user, 3)]):
            msgs.append(Message(id=mid, conversation_id=cid, role=role,
                                content=f"msg-{i}", created_at=ts(day, i)))
        s.add_all(msgs)
        s.add_all([
            CustomerProfile(id="bbbb0000-0000-0000-0000-000000000001",
                            conversation_id=C1,
                            raw_info={"name": "张三"}, summary="老客",
                            updated_at=ts(1, 6)),
            CustomerProfile(id="bbbb0000-0000-0000-0000-000000000002",
                            conversation_id=C2, raw_info={}, summary=None,
                            updated_at=ts(2, 6)),
            CustomerProfile(id="bbbb0000-0000-0000-0000-000000000003",
                            conversation_id=C3, raw_info=None,
                            summary="仅询价", updated_at=ts(3, 6)),
        ])
        s.add_all([
            Evaluation(id=E2, conversation_id=C2,
                       intent_level=IntentLevel.interested, reasoning="明确",
                       contact_info={"phone": "13800000000"}, email_sent=True,
                       correction_count=1, evaluated_at=ts(2, 7)),
            Evaluation(id=E3, conversation_id=C3,
                       intent_level=IntentLevel.pending, reasoning="",
                       contact_info={}, email_sent=False, correction_count=0,
                       evaluated_at=ts(3, 7)),
        ])
        s.add_all([
            LeadPushLog(id="dddd0000-0000-0000-0000-000000000001",
                        evaluation_id=E2, conversation_id=C2,
                        push_type="lead", status="success", attempts=1,
                        response_status=200, outcome="accepted",
                        created_at=ts(2, 8), updated_at=ts(2, 8)),
            LeadPushLog(id="dddd0000-0000-0000-0000-000000000002",
                        evaluation_id=E3, conversation_id=C3,
                        push_type="lead", status="pending", attempts=2,
                        error_message=None,
                        created_at=ts(3, 8), updated_at=ts(3, 8)),
            LeadPushLog(id="dddd0000-0000-0000-0000-000000000003",
                        evaluation_id=E3, conversation_id=C3,
                        push_type="correction", status="pending", attempts=1,
                        error_message="旧错误: 推送超时",
                        created_at=ts(3, 9), updated_at=ts(3, 9)),
            LeadPushLog(id="dddd0000-0000-0000-0000-000000000004",
                        evaluation_id=E2, conversation_id=C2,
                        push_type="returning", status="failed", attempts=3,
                        error_message="HTTP 500",
                        created_at=ts(2, 9), updated_at=ts(2, 9)),
        ])
        s.commit()
    engine.dispose()


LEGACY_COUNTS = {"conversations": 3, "messages": 5, "customer_profiles": 3,
                 "evaluations": 2, "lead_push_logs": 4}


class Env:
    """tmp 演练环境: platform.db(bootstrap 过 jiancai)+ tenants 目录。"""

    def __init__(self, tmp_path: Path, monkeypatch):
        self.platform_db = tmp_path / "platform.db"
        self.tenants_dir = tmp_path / "tenants"
        run_bootstrap(monkeypatch, self.platform_db, "jiancai")

    def import_args(self, source: Path, tenant: str = "jiancai"):
        return ("--source", str(source), "--tenant", tenant,
                "--platform-db", str(self.platform_db),
                "--tenants-dir", str(self.tenants_dir))

    @property
    def tenant_db(self) -> Path:
        return self.tenants_dir / "jiancai.db"


# ---- 合成库: 导入正确性 -------------------------------------------------------

def test_import_preserves_all_five_tables(tmp_path, monkeypatch, capsys):
    """五表对账全绿: 行数一致 + 行级逐字(四表)+ conversation_id 原样 +
    最近会话抽查逐字段 + 报告每表 一致。"""
    env = Env(tmp_path, monkeypatch)
    legacy = tmp_path / "legacy.db"
    _make_legacy_db(legacy)

    run_import(monkeypatch, *env.import_args(legacy))
    out = capsys.readouterr().out
    for table, n in LEGACY_COUNTS.items():
        src, imp, skip, dst = _parse_report_line(out, table)
        assert (src, imp, skip, dst) == (n, n, 0, n)
        assert _count(env.tenant_db, table) == n
    assert "对账结论: 全部一致" in out
    assert _spot_rows(legacy) == _spot_rows(env.tenant_db)
    _assert_tables_identical(legacy, env.tenant_db)


def test_pending_push_logs_converted_to_failed(tmp_path, monkeypatch, capsys):
    """pending → failed + 追加备注: 空 error 只剩备注; 旧错误为前缀追加;
    非 pending 行(status/error_message)逐字不动。"""
    env = Env(tmp_path, monkeypatch)
    legacy = tmp_path / "legacy.db"
    _make_legacy_db(legacy)

    run_import(monkeypatch, *env.import_args(legacy))
    capsys.readouterr()
    lpl = _rows(env.tenant_db, "lead_push_logs", COMPARE_COLUMNS["lead_push_logs"])
    by_id = {r["id"]: r for r in lpl}
    p1 = by_id["dddd0000-0000-0000-0000-000000000002"]
    assert p1["status"] == "failed"
    assert p1["error_message"] == PENDING_MARKER
    p2 = by_id["dddd0000-0000-0000-0000-000000000003"]
    assert p2["status"] == "failed"
    assert p2["error_message"] == f"旧错误: 推送超时 {PENDING_MARKER}"
    assert by_id["dddd0000-0000-0000-0000-000000000001"]["status"] == "success"
    assert by_id["dddd0000-0000-0000-0000-000000000001"]["error_message"] is None
    assert by_id["dddd0000-0000-0000-0000-000000000004"]["status"] == "failed"
    assert by_id["dddd0000-0000-0000-0000-000000000004"]["error_message"] == "HTTP 500"


def test_idempotent_second_run_zero_change(tmp_path, monkeypatch, capsys):
    """跑两遍: 第二遍每表 导入=0 跳过=源, DB 行级快照逐字不变。"""
    env = Env(tmp_path, monkeypatch)
    legacy = tmp_path / "legacy.db"
    _make_legacy_db(legacy)

    run_import(monkeypatch, *env.import_args(legacy))
    capsys.readouterr()
    snapshot = {t: _rows(env.tenant_db, t, COMPARE_COLUMNS[t])
                for t in LEGACY_TABLES}

    run_import(monkeypatch, *env.import_args(legacy))
    out = capsys.readouterr().out
    for table, n in LEGACY_COUNTS.items():
        assert _parse_report_line(out, table) == (n, 0, n, n)
    assert {t: _rows(env.tenant_db, t, COMPARE_COLUMNS[t])
            for t in LEGACY_TABLES} == snapshot


# ---- 合成库: 结构防御 ---------------------------------------------------------

def test_missing_critical_column_fails_naming_column(tmp_path, monkeypatch):
    """保真关键列(conversations.status)缺失 → 显性报列名, 不静默丢数据。"""
    env = Env(tmp_path, monkeypatch)
    legacy = tmp_path / "legacy.db"
    _make_legacy_db(legacy)
    con = sqlite3.connect(legacy)
    con.execute("ALTER TABLE conversations DROP COLUMN status")
    con.commit()
    con.close()

    with pytest.raises(SystemExit, match="status"):
        run_import(monkeypatch, *env.import_args(legacy))


def test_missing_optional_column_defaulted_and_noted(
        tmp_path, monkeypatch, capsys):
    """非关键列(page_token 可空)缺失 → 默认值 NULL 填充 + 报告标注,
    导入照常成功(对账仍绿)。"""
    env = Env(tmp_path, monkeypatch)
    legacy = tmp_path / "legacy.db"
    _make_legacy_db(legacy)
    con = sqlite3.connect(legacy)
    con.execute("DROP INDEX ix_conversations_page_token")
    con.execute("ALTER TABLE conversations DROP COLUMN page_token")
    con.commit()
    con.close()

    run_import(monkeypatch, *env.import_args(legacy))
    out = capsys.readouterr().out
    assert "老库缺列 page_token" in out
    assert _count(env.tenant_db, "conversations") == 3
    con = _ro(env.tenant_db)
    vals = [r[0] for r in con.execute("SELECT page_token FROM conversations")]
    con.close()
    assert vals == [None, None, None]


def test_extra_source_column_disclosed_not_fatal(tmp_path, monkeypatch, capsys):
    """老库多列(新库已废弃, 如 auto 真库的 correction_sent)→ 不迁移 +
    报告标注, 导入照常成功。"""
    env = Env(tmp_path, monkeypatch)
    legacy = tmp_path / "legacy.db"
    _make_legacy_db(legacy)
    con = sqlite3.connect(legacy)
    con.execute("ALTER TABLE evaluations ADD COLUMN legacy_note TEXT")
    con.execute("UPDATE evaluations SET legacy_note = '老数据'")
    con.commit()
    con.close()

    run_import(monkeypatch, *env.import_args(legacy))
    out = capsys.readouterr().out
    assert "老库多列 legacy_note" in out
    assert _count(env.tenant_db, "evaluations") == 2


def test_orphan_rows_refused(tmp_path, monkeypatch):
    """老库孤儿行(子表 conversation_id 无父)→ 显性报错, 不静默导入悬挂数据。"""
    env = Env(tmp_path, monkeypatch)
    legacy = tmp_path / "legacy.db"
    _make_legacy_db(legacy)
    con = sqlite3.connect(legacy)
    con.execute(
        "INSERT INTO messages VALUES "
        "('ffff0000-0000-0000-0000-000000000009', 'no-such-conv', "
        "'user', '孤儿消息', '2026-09-04 10:00:00+00:00')")
    con.commit()
    con.close()

    with pytest.raises(SystemExit, match="孤儿"):
        run_import(monkeypatch, *env.import_args(legacy))


# ---- 合成库: 门禁与退出码 -----------------------------------------------------

def test_unbootstrapped_tenant_refused(tmp_path, monkeypatch):
    """租户未 bootstrap: platform.db 缺失 / slug 未注册 → 显性提示先跑
    bootstrap, 绝不摸租户库。"""
    legacy = tmp_path / "legacy.db"
    _make_legacy_db(legacy)
    no_dir = tmp_path / "tenants"

    # platform.db 不存在
    with pytest.raises(SystemExit, match="bootstrap"):
        run_import(monkeypatch, "--source", str(legacy), "--tenant", "jiancai",
                   "--platform-db", str(tmp_path / "nope" / "platform.db"),
                   "--tenants-dir", str(no_dir))

    # platform.db 存在但只注册了 auto(目标 jiancai 未注册)
    platform_db = tmp_path / "platform.db"
    run_bootstrap(monkeypatch, platform_db, "auto")
    with pytest.raises(SystemExit, match="bootstrap"):
        run_import(monkeypatch, "--source", str(legacy), "--tenant", "jiancai",
                   "--platform-db", str(platform_db),
                   "--tenants-dir", str(no_dir))
    assert not no_dir.exists()          # 门禁未过绝不建租户库


def test_reconciliation_mismatch_nonzero_exit(tmp_path, monkeypatch, capsys):
    """目标已有源外数据(源≠目标合计)→ 打印对账报告后非零退出。
    形态: 先正常导入(目标库随之建立)→ 目标混入一条"原生"会话 → 第二遍
    导入对账炸(模拟切换前误跑过流量再补导的分歧场景)。"""
    env = Env(tmp_path, monkeypatch)
    legacy = tmp_path / "legacy.db"
    _make_legacy_db(legacy)
    run_import(monkeypatch, *env.import_args(legacy))
    capsys.readouterr()

    pre = sqlite3.connect(env.tenant_db)
    pre.execute(
        "INSERT INTO conversations VALUES "
        "('99999999-9999-9999-9999-999999999999', 'sess-native', NULL, "
        "'active', 'standalone', NULL, '2026-09-05 10:00:00+00:00', "
        "'2026-09-05 10:00:00+00:00')")
    pre.commit()
    pre.close()

    with pytest.raises(SystemExit, match="对账不一致"):
        run_import(monkeypatch, *env.import_args(legacy))
    out = capsys.readouterr().out
    assert "conversations" in out            # 报告先于退出码打印
    assert _parse_report_line(out, "conversations") == (3, 0, 3, 4)


def test_source_file_missing_refused(tmp_path, monkeypatch):
    with pytest.raises(SystemExit, match="老库"):
        run_import(monkeypatch, "--source", str(tmp_path / "ghost.db"),
                   "--tenant", "jiancai",
                   "--platform-db", str(tmp_path / "platform.db"),
                   "--tenants-dir", str(tmp_path / "tenants"))


# ---- 真实老库(env 指路, 缺失 skip)---------------------------------------------

def _legacy_copy(tmp_path: Path, env_var: str) -> tuple[Path, Path]:
    """(原库路径, VACUUM INTO 拷贝)——只读连原库做快照, 绝不直连写。"""
    raw = os.environ.get(env_var)
    if not raw:
        pytest.skip(f"未设 {env_var}(机器相关: 指向真实老库 data/*.db 才测)")
    src = Path(raw)
    assert src.is_file(), f"{env_var} 指向的老库不存在: {src}"
    dst = tmp_path / f"legacy-{env_var.split('_')[-1].lower()}.db"
    ro = sqlite3.connect(f"file:{src}?mode=ro", uri=True)
    try:
        ro.execute("VACUUM INTO ?", (str(dst),))
    finally:
        ro.close()
    return src, dst


def _real_rehearsal(tmp_path, monkeypatch, capsys, env_var, tenant):
    src, legacy = _legacy_copy(tmp_path, env_var)
    sha_before = _sha256(src)
    platform_db = tmp_path / "platform.db"
    run_bootstrap(monkeypatch, platform_db, tenant)
    tenants_dir = tmp_path / "tenants"
    tenant_db = tenants_dir / f"{tenant}.db"

    run_import(monkeypatch, "--source", str(legacy), "--tenant", tenant,
               "--platform-db", str(platform_db),
               "--tenants-dir", str(tenants_dir))
    out = capsys.readouterr().out
    assert "对账结论: 全部一致" in out

    # 五表对账 + 行级 + 抽查 + conversation_id 原样
    for table in LEGACY_TABLES:
        assert _count(legacy, table) == _count(tenant_db, table)
    _assert_tables_identical(legacy, tenant_db)
    assert _spot_rows(legacy) == _spot_rows(tenant_db)

    # 幂等第二遍: 全部跳过, 零变化
    run_import(monkeypatch, "--source", str(legacy), "--tenant", tenant,
               "--platform-db", str(platform_db),
               "--tenants-dir", str(tenants_dir))
    out2 = capsys.readouterr().out
    for table in LEGACY_TABLES:
        n = _count(legacy, table)
        assert _parse_report_line(out2, table) == (n, 0, n, n)

    # 原库零接触
    assert _sha256(src) == sha_before
    return out


def test_real_jc_legacy_rehearsal(tmp_path, monkeypatch, capsys):
    """jc 真库(拷贝上)全流程: 导入→对账全绿→幂等→原库 sha 不变。"""
    _real_rehearsal(tmp_path, monkeypatch, capsys,
                    "ALL_KEFU_TEST_LEGACY_JC", "jiancai")


def test_real_auto_legacy_rehearsal(tmp_path, monkeypatch, capsys):
    """auto 真库同理; 额外钉死 evaluations.correction_sent(新库已废弃)
    在报告中显性标注, 不静默丢列。"""
    out = _real_rehearsal(tmp_path, monkeypatch, capsys,
                          "ALL_KEFU_TEST_LEGACY_AUTO", "auto")
    assert "老库多列 correction_sent" in out