浅念
356 lines
15 KiB
Python
356 lines
15 KiB
Python
"""存量导入脚本(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()
|