"""存量导入脚本(Phase 4 Task 2): 老单租户库 → 多租户 tenants/<slug>.db。

老 jc/auto 项目与本项目五表同源(conversations/messages/customer_profiles/
evaluations/lead_push_logs), 本脚本把老库整库搬进目标租户库, 保 ID/
session_id/status/page_token 原样(逐字 INSERT OR IGNORE, 幂等键=主键),
切换前未完成的 pending 推送日志转 failed 并追加备注。

纪律:
- 老库**只读**打开(ATTACH ...?mode=ro, 写入即 OperationalError), 绝不回写;
- 不依赖 registry(部署脚本独立运行): 直连 tenants/<slug>.db; 租户须已
  bootstrap(platform.db 有该 slug), 否则报错提示先跑 bootstrap;
- 父先子后(conversations → messages → customer_profiles → evaluations →
  lead_push_logs), 单事务全量复制, 对账不一致 → 非零退出码;
- 老库结构假设防御: 关键列缺失/孤儿行 → 显性报错(不静默丢数据);
  新库多列 → 模型默认值填充并在报告标注; 老库多列(新库已废弃, 如 auto
  evaluations.correction_sent)→ 不迁移并标注。

用法(Windows 须 PYTHONUTF8=1, 输出含中文):
  PYTHONUTF8=1 python scripts/import_legacy.py --source <老库路径> \
      --tenant <目标slug> [--platform-db <路径>] [--tenants-dir <目录>]
"""
from __future__ import annotations

import argparse
import asyncio
import enum
import sqlite3
import sys
from pathlib import Path

# 仓库根入 sys.path(照 bootstrap_platform.py 锚定纪律: 脚本可从任意 CWD 跑)
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

import aiosqlite

from app.engine.models import Base, ensure_schema
from app.platform.registry import make_tenant_engine

REPO_ROOT = Path(__file__).resolve().parents[1]

# 迁移顺序: 父先子后(lead_push_logs 依赖 evaluations+conversations)
TABLE_ORDER = (
    "conversations", "messages", "customer_profiles", "evaluations",
    "lead_push_logs",
)

# 子表 → 父表 FK 依赖(老库孤儿行预检: 悬挂引用导入后炸应用, 必须显性报错)
FK_PARENTS = {
    "messages": [("conversation_id", "conversations")],
    "customer_profiles": [("conversation_id", "conversations")],
    "evaluations": [("conversation_id", "conversations")],
    "lead_push_logs": [("conversation_id", "conversations"),
                       ("evaluation_id", "evaluations")],
}

# 保真关键列: 缺失即硬错——默认值顶替 = 静默丢数据, 违背保 ID/session/
# status/外键承诺。lead_push_logs 的 status/error_message 也在内: 它俩是
# pending→failed 转换的输入, 缺了转换无从谈起。
CRITICAL_SOURCE_COLUMNS = {
    "conversations": ("id", "session_id", "status"),
    "messages": ("id", "conversation_id"),
    "customer_profiles": ("id", "conversation_id"),
    "evaluations": ("id", "conversation_id"),
    "lead_push_logs": ("id", "evaluation_id", "conversation_id",
                       "status", "error_message"),
}

PENDING_MARKER = "[迁移] 切换前未完成态"

# 报告里 conversation_id 原样性对照(max)的子表
CONV_ID_TABLES = ("messages", "customer_profiles", "evaluations",
                  "lead_push_logs")


def _ro_uri(path: Path) -> str:
    """mode=ro URI(只读打开最稳; 老库/jc-auto 真库演练零接触的兜底)。"""
    return path.resolve().as_uri() + "?mode=ro"


def _sql_literal(value) -> str:
    """Python 值 → SQL 字面量(缺列默认值填充用; 值源自本仓 models, 非用户输入)。"""
    if value is None:
        return "NULL"
    if isinstance(value, bool):
        return "1" if value else "0"
    if isinstance(value, enum.Enum):
        value = value.value
    if isinstance(value, (int, float)):
        return repr(value)
    return "'" + str(value).replace("'", "''") + "'"


def _fill_literal(table: str, column: str) -> str:
    """目标列缺于老库时的填充字面量: 可空 → NULL; 非空 → 模型默认值;
    非空且无默认 → 显性报错(不静默丢数据)。"""
    col = Base.metadata.tables[table].c[column]
    if col.nullable:
        return "NULL"
    default = col.default
    if default is None:
        raise SystemExit(
            f"老库 {table} 缺列 {column}: 非空且模型无默认值, "
            f"无法填充——显性报错, 不静默丢数据")
    arg = default.arg
    if callable(arg):                      # lambda: datetime.now(utc) 等
        arg = arg()
    return _sql_literal(arg)


def _tenant_bootstrapped(platform_db: Path, slug: str) -> None:
    """门禁: 租户须已在 platform.db 注册(先跑 bootstrap_platform.py)。"""
    if not platform_db.is_file():
        raise SystemExit(
            f"platform.db 不存在: {platform_db}\n"
            f"租户须先 bootstrap: 先跑 scripts/bootstrap_platform.py")
    try:
        con = sqlite3.connect(_ro_uri(platform_db), uri=True)
        try:
            row = con.execute(
                "SELECT 1 FROM tenants WHERE slug = ?", (slug,)).fetchone()
        finally:
            con.close()
    except sqlite3.Error as exc:
        raise SystemExit(
            f"platform.db 读取失败({exc}): 先跑 scripts/bootstrap_platform.py"
        ) from exc
    if row is None:
        raise SystemExit(
            f"租户 {slug} 未在 platform.db 注册: "
            f"先跑 scripts/bootstrap_platform.py 导入该租户档案")


async def _columns(conn, schema: str, table: str) -> list[str]:
    """指定 schema 下表的列名清单(PRAGMA table_info; 表不存在 → 空)。"""
    cur = await conn.execute(f"PRAGMA {schema}.table_info({table})")
    return [row[1] for row in await cur.fetchall()]


async def _scalar(conn, sql: str):
    cur = await conn.execute(sql)
    row = await cur.fetchone()
    return row[0]


def _select_exprs(table: str, src_cols: set[str],
                  dst_cols: list[str]) -> list[str]:
    """INSERT ... SELECT 的列表达式: 同名列按名直取, 缺列填默认字面量;
    lead_push_logs 的 status/error_message 在插入期完成 pending→failed。"""
    exprs = []
    for col in dst_cols:
        if col not in src_cols:
            exprs.append(_fill_literal(table, col))
        elif table == "lead_push_logs" and col == "status":
            exprs.append(
                "CASE WHEN src.lead_push_logs.status = 'pending' "
                "THEN 'failed' ELSE src.lead_push_logs.status END")
        elif table == "lead_push_logs" and col == "error_message":
            marker = _sql_literal(PENDING_MARKER)
            exprs.append(
                "CASE WHEN src.lead_push_logs.status = 'pending' THEN "
                "CASE WHEN src.lead_push_logs.error_message IS NULL "
                "OR src.lead_push_logs.error_message = '' "
                f"THEN {marker} "
                "ELSE src.lead_push_logs.error_message || ' ' || "
                f"{marker} END "
                "ELSE src.lead_push_logs.error_message END")
        else:
            exprs.append(f"src.{table}.{col}")
    return exprs


async def _validate_source(target, diff_notes: list[str]) -> dict:
    """列结构核对(表在/关键列在/缺列可填)+ 孤儿行预检。

    返回 {表: (源列集合, 目标列清单)}; 结构不符 → SystemExit 显性报列名。
    额外收集列差异备注(老库缺列→默认值 / 老库多列→不迁移)进 diff_notes。
    """
    info = {}
    for table in TABLE_ORDER:
        src_cols = set(await _columns(target, "src", table))
        if not src_cols:
            raise SystemExit(
                f"老库缺表: {table}(五表全量迁移, 拒绝部分导入)")
        dst_cols = await _columns(target, "main", table)

        missing_critical = set(CRITICAL_SOURCE_COLUMNS[table]) - src_cols
        if missing_critical:
            raise SystemExit(
                f"老库 {table} 缺关键列: "
                f"{', '.join(sorted(missing_critical))}"
                f"(保 ID/session/status/外键承诺, 拒绝静默丢数据)")

        for col in dst_cols:
            if col not in src_cols:
                diff_notes.append(
                    f"{table}: 老库缺列 {col} → "
                    f"填默认值 {_fill_literal(table, col)}")
        for col in sorted(src_cols - set(dst_cols)):
            diff_notes.append(
                f"{table}: 老库多列 {col} → 不迁移(新库已无此列)")
        info[table] = (src_cols, dst_cols)

    for table, deps in FK_PARENTS.items():
        for col, parent in deps:
            orphans = await _scalar(target, (
                f"SELECT COUNT(*) FROM src.{table} AS child "
                f"WHERE child.{col} IS NULL OR child.{col} NOT IN "
                f"(SELECT id FROM src.{parent})"))
            if orphans:
                raise SystemExit(
                    f"老库存在孤儿行: {table}.{col} 无对应 {parent} "
                    f"共 {orphans} 条——先修库再迁移, 不导入悬挂数据")
    return info


async def _migrate(target, source: Path, tenant_db: Path) -> None:
    # ---- 校验(结构防御; 全过才开始写) ------------------------------------
    diff_notes: list[str] = []
    info = await _validate_source(target, diff_notes)

    # ---- 复制(父先子后; INSERT OR IGNORE 幂等键=主键; 单事务) -------------
    stats = {}
    for table in TABLE_ORDER:
        src_cols, dst_cols = info[table]
        cols_sql = ", ".join(dst_cols)
        exprs_sql = ", ".join(_select_exprs(table, src_cols, dst_cols))
        before = target.total_changes
        await target.execute(
            f"INSERT OR IGNORE INTO main.{table} ({cols_sql}) "
            f"SELECT {exprs_sql} FROM src.{table}")
        imported = target.total_changes - before
        src_count = await _scalar(target, f"SELECT COUNT(*) FROM src.{table}")
        stats[table] = (src_count, imported, src_count - imported)
    await target.commit()

    # ---- 对账报告(stdout) -------------------------------------------------
    print(f"源库(只读): {source}")
    print(f"目标租户库: {tenant_db}")
    print("表对账(源行数/导入/跳过已存在/目标合计):")
    mismatched: list[str] = []
    for table in TABLE_ORDER:
        src_count, imported, skipped = stats[table]
        dst_count = await _scalar(target, f"SELECT COUNT(*) FROM main.{table}")
        ok = src_count == dst_count
        if not ok:
            mismatched.append(table)
        print(f"  {table}: 源={src_count} 导入={imported} "
              f"跳过={skipped} 目标={dst_count} {'一致' if ok else '不一致'}")

    if diff_notes:
        print("列差异标注:")
        for note in diff_notes:
            print(f"  {note}")

    pending_n = await _scalar(
        target, "SELECT COUNT(*) FROM src.lead_push_logs "
                "WHERE status = 'pending'")
    print(f"pending→failed: 转换={pending_n} 条"
          f"(error_message 追加\"{PENDING_MARKER}\")")

    # 抽查列取交集(老库缺列——如无 page_token 的旧结构——则少比一列,
    # 排序同理退到 id; 列差异已在上方标注, 不在此重复报)
    src_conv_cols = info["conversations"][0]
    spot_cols = [c for c in ("id", "session_id", "status", "page_token")
                 if c in src_conv_cols]
    order = ("ORDER BY created_at DESC, id DESC"
             if "created_at" in src_conv_cols else "ORDER BY id DESC")
    spot_sql = (f"SELECT {', '.join(spot_cols)} "
                f"FROM conversations {order} LIMIT 10")
    src_cur = await target.execute(
        spot_sql.replace("FROM conversations", "FROM src.conversations"))
    src_spot = await src_cur.fetchall()
    dst_cur = await target.execute(
        spot_sql.replace("FROM conversations", "FROM main.conversations"))
    dst_spot = await dst_cur.fetchall()
    print(f"最近 10 条会话对照({'/'.join(spot_cols)}):")
    spot_ok = src_spot == dst_spot
    for src_row, dst_row in zip(src_spot, dst_spot):
        same = src_row == dst_row
        print(f"  {src_row[0]} {src_row[1]} "
              f"源={src_row[2]} 目标={dst_row[2]} "
              f"{'一致' if same else '不一致'}")
    if not spot_ok:
        mismatched.append("最近会话抽查")

    print("conversation_id 原样性(各子表 max):")
    for table in CONV_ID_TABLES:
        src_max = await _scalar(
            target, f"SELECT MAX(conversation_id) FROM src.{table}")
        dst_max = await _scalar(
            target, f"SELECT MAX(conversation_id) FROM main.{table}")
        ok = src_max == dst_max
        if not ok:
            mismatched.append(f"{table}.conversation_id")
        print(f"  {table}: 源={src_max} 目标={dst_max} "
              f"{'一致' if ok else '不一致'}")

    if mismatched:
        raise SystemExit(
            "对账不一致: " + ", ".join(dict.fromkeys(mismatched)) +
            "(源≠目标合计或抽查偏差, 人工核查后再切换)")
    print("对账结论: 全部一致")


async def run(args) -> None:
    source = Path(args.source)
    if not source.is_file():
        raise SystemExit(f"老库不存在: {source}")
    platform_db = Path(args.platform_db)
    _tenant_bootstrapped(platform_db, args.tenant)

    tenants_dir = Path(args.tenants_dir)
    tenants_dir.mkdir(parents=True, exist_ok=True)
    tenant_db = tenants_dir / f"{args.tenant}.db"

    # 租户库 schema 就位(与 app 懒建同一 ensure_schema: 新文件建五表,
    # 旧文件补缺列; make_tenant_engine 同款 WAL 纪律)
    engine = make_tenant_engine(tenant_db)
    try:
        await ensure_schema(engine)
    finally:
        await engine.dispose()                  # Windows 文件锁纪律

    # 目标连接 uri=True 使 ATTACH 的 ?mode=ro 生效(老库只读的最稳形态)
    async with aiosqlite.connect(str(tenant_db), uri=True,
                                 timeout=30) as target:
        await target.execute("ATTACH DATABASE ? AS src", (_ro_uri(source),))
        try:
            await _migrate(target, source, tenant_db)
        finally:
            await target.execute("DETACH src")


def main() -> None:
    parser = argparse.ArgumentParser(
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--source", required=True,
                        help="老库路径(只读打开, 绝不回写)")
    parser.add_argument("--tenant", required=True,
                        help="目标租户 slug(须已 bootstrap 注册)")
    parser.add_argument("--platform-db",
                        default=str(REPO_ROOT / "data" / "platform.db"),
                        help="platform.db 路径(默认 <repo>/data/, 门禁查"
                             "租户是否已 bootstrap)")
    parser.add_argument("--tenants-dir",
                        default=str(REPO_ROOT / "tenants"),
                        help="租户库目录(默认 <repo>/tenants, 目标库为"
                             " <目录>/<slug>.db)")
    args = parser.parse_args()
    asyncio.run(run(args))


if __name__ == "__main__":
    main()