xinyouhe/ai-kefu-admin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 存量导入脚本（保ID/session/status、pending→failed、幂等、对账报告）

Browse Source 
Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
浅念
2026-10-14 09:00:39 +08:00
parent 6100b4eb15
commit ac9a4da51d
2 changed files with 906 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/import_legacy.py
+355
View File
@@ -0,0 +1,355 @@
"""存量导入脚本(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()
tests/test_import_legacy.py
+551
View File
@@ -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