From ac9a4da51d2587f2d8c1a63167c036f74506f339 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Wed, 14 Oct 2026 09:00:39 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AD=98=E9=87=8F=E5=AF=BC=E5=85=A5?= =?UTF-8?q?=E8=84=9A=E6=9C=AC=EF=BC=88=E4=BF=9DID/session/status=E3=80=81p?= =?UTF-8?q?ending=E2=86=92failed=E3=80=81=E5=B9=82=E7=AD=89=E3=80=81?= =?UTF-8?q?=E5=AF=B9=E8=B4=A6=E6=8A=A5=E5=91=8A=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Code --- scripts/import_legacy.py | 355 +++++++++++++++++++++++ tests/test_import_legacy.py | 551 ++++++++++++++++++++++++++++++++++++ 2 files changed, 906 insertions(+) create mode 100644 scripts/import_legacy.py create mode 100644 tests/test_import_legacy.py diff --git a/scripts/import_legacy.py b/scripts/import_legacy.py new file mode 100644 index 0000000..deac2ca --- /dev/null +++ b/scripts/import_legacy.py @@ -0,0 +1,355 @@ +"""存量导入脚本(Phase 4 Task 2): 老单租户库 → 多租户 tenants/.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/.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 路径(默认 /data/, 门禁查" + "租户是否已 bootstrap)") + parser.add_argument("--tenants-dir", + default=str(REPO_ROOT / "tenants"), + help="租户库目录(默认 /tenants, 目标库为" + " <目录>/.db)") + args = parser.parse_args() + asyncio.run(run(args)) + + +if __name__ == "__main__": + main() diff --git a/tests/test_import_legacy.py b/tests/test_import_legacy.py new file mode 100644 index 0000000..9ab30b3 --- /dev/null +++ b/tests/test_import_legacy.py @@ -0,0 +1,551 @@ +"""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