From 3a3292c2392057a0124741e6f1288409221c4fb9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Wed, 14 Oct 2026 09:23:42 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=95=B0=E6=8D=AE=E5=BD=92=E6=A1=A3?= =?UTF-8?q?=E8=84=9A=E6=9C=AC=EF=BC=88=E5=AF=BC=E5=87=BA/=E6=B8=85?= =?UTF-8?q?=E7=90=86/VACUUM/dry-run=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Code --- scripts/archive.py | 261 +++++++++++++++++++++++++ tests/test_archive.py | 441 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 702 insertions(+) create mode 100644 scripts/archive.py create mode 100644 tests/test_archive.py diff --git a/scripts/archive.py b/scripts/archive.py new file mode 100644 index 0000000..df5dfd5 --- /dev/null +++ b/scripts/archive.py @@ -0,0 +1,261 @@ +"""数据归档脚本(Phase 4 Task 3): 租户库老会话 导出→清理→VACUUM。 + +conversations.created_at 早于 N 天(默认 365)的会话整链归档: 五表行先按 +会话分组导出 JSON 到 data/archives//<日期>-d.json, 落盘成功后 +才删库(倒 FK: lead_push_logs → evaluations → customer_profiles → +messages → conversations, 单事务), 删了行才 VACUUM。导出/删除集同源于 +同一截止串, 无漂移。 + +纪律: +- 门禁同 import_legacy: 租户须已在 platform.db bootstrap, 否则报错提示 + 先跑 bootstrap, 绝不摸租户库; +- 截止时刻 Python 侧算好 ISO 串再入 SQL(不依赖 SQLite datetime() 解析): + ORM 行存 'YYYY-MM-DD HH:MM:SS.ffffff'(无 tz 后缀), 老库导入行可能带 + '+00:00' 后缀——同族字符串字典序比较对两者皆单调(差异位在 '.'/'+', + 仅整秒边界有微差, 天级截止无实际影响); +- 导出先落盘后删除: 文件写失败 → 数据一字节不动(多一份孤儿导出无害, + 反之是不可逆丢失); 目录懒建(无可归档/失败时不产生任何文件); +- --dry-run 只列不动: 库零变化、不导出、不 VACUUM; +- archives 目录 = tenants 目录的兄弟 data/archives(同 registry + uploads_root 的派生纪律: 生产 /tenants → /data/archives, + 测试注入 tmp tenants → tmp 内隔离; data/ 在 .gitignore 内天然覆盖)。 + +用法(Windows 须 PYTHONUTF8=1, 输出含中文): + PYTHONUTF8=1 python scripts/archive.py --tenant \ + [--days 365] [--dry-run] [--platform-db <路径>] [--tenants-dir <目录>] +""" +from __future__ import annotations + +import argparse +import asyncio +import json +import sqlite3 +import sys +from datetime import datetime, timedelta, timezone +from pathlib import Path + +# 仓库根入 sys.path(照 import_legacy.py 锚定纪律: 脚本可从任意 CWD 跑) +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +import aiosqlite + +REPO_ROOT = Path(__file__).resolve().parents[1] + +# 导出分组顺序(父先子后, 与 import_legacy 的 TABLE_ORDER 一致) +TABLE_ORDER = ( + "conversations", "messages", "customer_profiles", "evaluations", + "lead_push_logs", +) +# 删除序 = 倒 FK(子先父后; SQLite 默认不开 FK 强制, 显式按序删) +DELETE_ORDER = tuple(reversed(TABLE_ORDER)) +CHILD_TABLES = TABLE_ORDER[1:] # 以 conversation_id 关联的子表 + +# 各表确定性排序(导出顺序稳定; 注意 customer_profiles 无 created_at +# 只有 updated_at, evaluations 是 evaluated_at——五表时间列不齐名) +TABLE_ORDER_BY = { + "conversations": "created_at, id", + "messages": "created_at, id", + "customer_profiles": "updated_at, id", + "evaluations": "evaluated_at, id", + "lead_push_logs": "created_at, id", +} + + +def _tenant_bootstrapped(platform_db: Path, slug: str) -> None: + """门禁(同 import_legacy): 租户须已在 platform.db 注册; + 只读打开(先看后动, 门禁绝不写 platform.db)。""" + if not platform_db.is_file(): + raise SystemExit( + f"platform.db 不存在: {platform_db}\n" + f"租户须先 bootstrap: 先跑 scripts/bootstrap_platform.py") + try: + con = sqlite3.connect( + platform_db.resolve().as_uri() + "?mode=ro", 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 导入该租户档案") + + +def _cutoff(days: int) -> str: + """截止 ISO 串(Python 侧算, 见模块 docstring 的时间形态论证)。""" + cut = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(days=days) + return cut.strftime("%Y-%m-%d %H:%M:%S.%f") + + +async def _columns(conn, table: str) -> list[str]: + cur = await conn.execute(f"PRAGMA table_info({table})") + cols = [row[1] for row in await cur.fetchall()] + if not cols: + raise SystemExit( + f"租户库缺表: {table}(schema 未就位, 先跑应用或 import_legacy)") + return cols + + +async def _scalar(conn, sql: str, params=()): + cur = await conn.execute(sql, params) + row = await cur.fetchone() + return row[0] + + +async def _old_conversations(conn, cutoff: str) -> list: + """待归档会话 (id, session_id, created_at), 时间序。""" + cur = await conn.execute( + "SELECT id, session_id, created_at FROM conversations " + "WHERE created_at < ? ORDER BY created_at, id", (cutoff,)) + return list(await cur.fetchall()) + + +async def _export_document(conn, cutoff: str) -> dict: + """按会话分组的五表完整行(conversations 键=会话行; 子表键=会话 id → + 行列表; 无该子表行的会话无键)。SELECT * 全列, 值原样读出(JSON 列即 + 库内存储文本, 字节级保真)。""" + conn.row_factory = aiosqlite.Row + doc: dict = {t: {} for t in TABLE_ORDER} + for table in TABLE_ORDER: + where_id = table == "conversations" + sql = (f"SELECT * FROM {table} WHERE " + f"{'id' if where_id else 'conversation_id'} IN " + f"(SELECT id FROM conversations WHERE created_at < ?) " + f"ORDER BY {TABLE_ORDER_BY[table]}") + cur = await conn.execute(sql, (cutoff,)) + for row in await cur.fetchall(): + fields = dict(row) + if where_id: + doc[table][row["id"]] = fields + else: + doc[table].setdefault(row["conversation_id"], + []).append(fields) + return doc + + +def _write_export(path: Path, doc: dict) -> None: + """先落盘后删库的'落盘'端: 目录懒建 + 原子替换(临时文件 + replace, + 中断不留半份导出)。""" + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(".json.tmp") + tmp.write_text( + json.dumps(doc, ensure_ascii=False, indent=2), encoding="utf-8") + tmp.replace(path) + + +async def run(args) -> None: + days: int = args.days + if days < 0: + raise SystemExit(f"--days 不得为负: {days}") + platform_db = Path(args.platform_db) + _tenant_bootstrapped(platform_db, args.tenant) # 门禁先于一切 + + tenants_dir = Path(args.tenants_dir) + tenant_db = tenants_dir / f"{args.tenant}.db" + if not tenant_db.is_file(): + raise SystemExit(f"租户库不存在: {tenant_db}") + archives_dir = tenants_dir.parent / "data" / "archives" + cutoff = _cutoff(days) + size_before = tenant_db.stat().st_size + + async with aiosqlite.connect(str(tenant_db), timeout=30) as conn: + for table in TABLE_ORDER: # 缺表显性报错 + await _columns(conn, table) + old = await _old_conversations(conn, cutoff) + old_ids = [row[0] for row in old] + + print(f"租户库: {tenant_db}") + print(f"归档策略: created_at < {cutoff}({days} 天前)") + + if args.dry_run: + print("模式: dry-run(只列不动, 库零变化)") + print(f"待归档会话: {len(old_ids)} 条") + for cid, session_id, created_at in old: + print(f" {cid} session={session_id} " + f"created_at={created_at}") + if old_ids: + print("将删除的各表行数(未动):") + for table in ("conversations", *CHILD_TABLES): + n = len(old_ids) if table == "conversations" else \ + await _scalar(conn, ( + f"SELECT COUNT(*) FROM {table} " + "WHERE conversation_id IN (SELECT id FROM " + "conversations WHERE created_at < ?)"), (cutoff,)) + print(f" {table}: {n}") + print("dry-run 结论: 零改动(未导出/未删除/未 VACUUM)") + return + + export_path = archives_dir / args.tenant / ( + f"{datetime.now(timezone.utc).strftime('%Y%m%d')}-{days}d.json") + deleted: dict[str, int] = {} + if not old_ids: + print("归档会话: 0 条") + print("结论: 无可归档会话, 零改动(未导出/未删除/未 VACUUM)") + print(f"VACUUM: 未执行(无删除行)") + print(f"库文件: {size_before} → {tenant_db.stat().st_size} 字节") + return + + # ---- 导出(先落盘, 失败即中止, 数据不动) --------------------------- + doc = {"tenant": args.tenant, + "generated_at": datetime.now(timezone.utc).isoformat(), + "days": days, "cutoff": cutoff, + "conversation_count": len(old_ids)} + doc.update(await _export_document(conn, cutoff)) + _write_export(export_path, doc) + + # ---- 清库(倒 FK, 单事务; 子查询同源截止串) ------------------------ + for table in DELETE_ORDER: + col = "id" if table == "conversations" else "conversation_id" + cur = await conn.execute( + f"DELETE FROM {table} WHERE {col} IN " + f"(SELECT id FROM conversations WHERE created_at < ?)", + (cutoff,)) + deleted[table] = cur.rowcount + await conn.commit() + + # ---- VACUUM(删了行才跑; 已 commit, 事务外) ------------------------ + pages_before = await _scalar(conn, "PRAGMA page_count") + await conn.execute("VACUUM") + pages_after = await _scalar(conn, "PRAGMA page_count") + + # ---- 报告 ---------------------------------------------------------- + print(f"归档会话: {len(old_ids)} 条; 各表删除行数(倒 FK 序):") + for table in DELETE_ORDER: + print(f" {table}: {deleted[table]}") + print(f"导出: {export_path}") + size_after = tenant_db.stat().st_size + print(f"库文件: {size_before} → {size_after} 字节") + print(f"VACUUM: 已执行(page_count {pages_before} → {pages_after})") + print("归档结论: 导出已落盘, 删除已提交, VACUUM 完成") + + +def main() -> None: + parser = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter) + parser.add_argument("--tenant", required=True, + help="租户 slug(须已 bootstrap 注册)") + parser.add_argument("--days", type=int, default=365, + help="归档阈值: created_at 早于 N 天前的会话" + "(默认 365)") + parser.add_argument("--dry-run", action="store_true", + help="只列不动: 不导出/不删除/不 VACUUM, 库零变化") + 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; 导出落其" + "兄弟目录 data/archives//)") + args = parser.parse_args() + asyncio.run(run(args)) + + +if __name__ == "__main__": + main() diff --git a/tests/test_archive.py b/tests/test_archive.py new file mode 100644 index 0000000..e68da00 --- /dev/null +++ b/tests/test_archive.py @@ -0,0 +1,441 @@ +"""scripts/archive.py(Phase 4 Task 3): 租户库老会话归档(导出→清理→VACUUM)。 + +契约(计划原文): +1. >N 天会话(含子表行)→ 导出 JSON 到 data/archives// → 清库 → + 五表清空(旧会话相关); 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 _ids(db, table, col="id") -> set: + con = _ro(db) + try: + return {r[0] for r in con.execute(f"SELECT {col} FROM {table}")} + finally: + con.close() + + +def _full_rows(db, table) -> list[dict]: + """全列行(PRAGMA 列序), 导出完整性对照的采集端。""" + con = _ro(db) + con.row_factory = sqlite3.Row + try: + return [dict(r) for r in con.execute( + f"SELECT * FROM {table} ORDER BY id")] + finally: + con.close() + + +def _page_count(db) -> int: + con = _ro(db) + try: + return con.execute("PRAGMA page_count").fetchone()[0] + finally: + con.close() + + +def _sha256(path) -> str: + return hashlib.sha256(Path(path).read_bytes()).hexdigest() + + +# ---- 数据形态 ---------------------------------------------------------------- + +OLD1 = "11111111-1111-1111-1111-111111111111" # 2020-01-01(ORM 形态) +OLD2 = "22222222-2222-2222-2222-222222222222" # 2020-06-15(+00:00 后缀) +MID = "44444444-4444-4444-4444-444444444444" # now-40d(365 天内, 30 天外) +REC1 = "33333333-3333-3333-3333-333333333333" # now-1d +E1 = "eeeeeeee-0000-0000-0000-000000000001" +E2 = "eeeeeeee-0000-0000-0000-000000000002" +E3 = "eeeeeeee-0000-0000-0000-000000000003" +M1 = "aaaa0000-0000-0000-0000-000000000001" +M2 = "aaaa0000-0000-0000-0000-000000000002" +M3 = "aaaa0000-0000-0000-0000-000000000003" +M4 = "aaaa0000-0000-0000-0000-000000000004" +P1 = "bbbb0000-0000-0000-0000-000000000001" +P2 = "bbbb0000-0000-0000-0000-000000000002" +L1 = "dddd0000-0000-0000-0000-000000000001" +L2 = "dddd0000-0000-0000-0000-000000000002" +L3 = "dddd0000-0000-0000-0000-000000000003" +L4 = "dddd0000-0000-0000-0000-000000000004" + +NOW = datetime.now(timezone.utc) + + +def _seed(db: Path) -> None: + """4 会话: OLD1(全子表)+ OLD2(无 profile, +00:00 时间形态)+ MID(裸, + 40 天)+ REC1(now-1d, 全子表)。""" + engine = create_engine(f"sqlite:///{db}") + Base.metadata.create_all(engine) + with Session(engine) as s: + s.add(Conversation(id=OLD1, session_id="sess-old1", + status=ConversationStatus.completed, + source=ConversationSource.widget, + client_info={"ip": "1.2.3.4"}, + created_at=datetime(2020, 1, 1, 10, tzinfo=timezone.utc), + updated_at=datetime(2020, 1, 1, 10, 5, + tzinfo=timezone.utc))) + s.add(Conversation(id=MID, session_id="sess-mid", + status=ConversationStatus.active, + source=ConversationSource.standalone, + created_at=NOW - timedelta(days=40))) + s.add(Conversation(id=REC1, session_id="sess-rec1", + status=ConversationStatus.active, + source=ConversationSource.widget, + created_at=NOW - timedelta(days=1))) + s.add_all([ + Message(id=M1, conversation_id=OLD1, role=MessageRole.user, + content="old1 问价", + created_at=datetime(2020, 1, 1, 10, 1, tzinfo=timezone.utc)), + Message(id=M2, conversation_id=OLD1, role=MessageRole.assistant, + content="old1 答", + created_at=datetime(2020, 1, 1, 10, 2, tzinfo=timezone.utc)), + Message(id=M3, conversation_id=OLD2, role=MessageRole.user, + content="old2 留了电话", + created_at=datetime(2020, 6, 15, 10, 1, tzinfo=timezone.utc)), + Message(id=M4, conversation_id=REC1, role=MessageRole.user, + content="recent 消息", created_at=NOW - timedelta(days=1)), + ]) + s.add(CustomerProfile(id=P1, conversation_id=OLD1, + raw_info={"name": "张三"}, summary="老客", + updated_at=datetime(2020, 1, 1, 10, 3, + tzinfo=timezone.utc))) + s.add(CustomerProfile(id=P2, conversation_id=REC1, raw_info={}, + summary="新客", updated_at=NOW)) + s.add_all([ + Evaluation(id=E1, conversation_id=OLD1, + intent_level=IntentLevel.interested, reasoning="明确", + contact_info={"phone": "13800000000"}, email_sent=True, + correction_count=1, + evaluated_at=datetime(2020, 1, 1, 10, 4, + tzinfo=timezone.utc)), + Evaluation(id=E2, conversation_id=OLD2, + intent_level=IntentLevel.pending, reasoning="", + contact_info={}, + evaluated_at=datetime(2020, 6, 15, 10, 2, + tzinfo=timezone.utc)), + Evaluation(id=E3, conversation_id=REC1, + intent_level=IntentLevel.interested, reasoning="热", + contact_info={"email": "a@b.c"}, email_sent=True), + ]) + s.add_all([ + LeadPushLog(id=L1, evaluation_id=E1, conversation_id=OLD1, + push_type="lead", status="success", attempts=1, + response_status=200, outcome="accepted", + created_at=datetime(2020, 1, 1, 10, 5, + tzinfo=timezone.utc)), + LeadPushLog(id=L2, evaluation_id=E1, conversation_id=OLD1, + push_type="correction", status="failed", attempts=3, + error_message="HTTP 500", + created_at=datetime(2020, 1, 1, 10, 6, + tzinfo=timezone.utc)), + LeadPushLog(id=L3, evaluation_id=E2, conversation_id=OLD2, + push_type="returning", status="pending", attempts=0), + LeadPushLog(id=L4, evaluation_id=E3, conversation_id=REC1, + push_type="lead", status="success", attempts=1, + response_status=200, outcome="accepted"), + ]) + s.commit() + engine.dispose() + + # OLD2: 裸 SQL 插入, created_at 带 +00:00 后缀(老库导入保真的时间形态) + con = sqlite3.connect(db) + con.execute( + "INSERT INTO conversations VALUES " + f"('{OLD2}', 'sess-old2', NULL, 'completed', 'widget', NULL, " + "'2020-06-15 10:00:00+00:00', '2020-06-15 10:00:00+00:00')") + con.commit() + con.close() + + +# 归档后各表期望保留(默认 365 天: OLD1/OLD2 走, MID/REC1 留)——键取 +# CONV_ID_COL 列的取值集: conversations 看会话 id, 子表看 conversation_id +KEPT = { + "conversations": {MID, REC1}, "messages": {REC1}, + "customer_profiles": {REC1}, "evaluations": {REC1}, + "lead_push_logs": {REC1}, +} + + +class Env: + """tmp 演练环境: platform.db(bootstrap 过 jiancai)+ tenants 目录; + archives 目录取 tenants 兄弟 data/archives(tmp 内隔离)。""" + + def __init__(self, tmp_path: Path, monkeypatch): + self.tmp_path = tmp_path + self.platform_db = tmp_path / "platform.db" + self.tenants_dir = tmp_path / "tenants" + self.tenants_dir.mkdir(parents=True) # sqlite 不建父目录, 种子先建 + run_bootstrap(monkeypatch, self.platform_db, "jiancai") + + def archive_args(self, *extra: str, tenant: str = "jiancai"): + return ("--tenant", tenant, + "--platform-db", str(self.platform_db), + "--tenants-dir", str(self.tenants_dir), *extra) + + @property + def tenant_db(self) -> Path: + return self.tenants_dir / "jiancai.db" + + @property + def archives_dir(self) -> Path: + return self.tmp_path / "data" / "archives" + + def export_path(self, days: int) -> Path: + date = datetime.now(timezone.utc).strftime("%Y%m%d") + return self.archives_dir / "jiancai" / f"{date}-{days}d.json" + + +# ---- 1. 归档: 旧会话五表清空, 近期不动 ------------------------------------------ + +def test_archive_old_conversations(tmp_path, monkeypatch, capsys): + env = Env(tmp_path, monkeypatch) + _seed(env.tenant_db) + + run_archive(monkeypatch, *env.archive_args()) # 默认 --days 365 + out = capsys.readouterr().out + + assert "归档会话: 2 条" in out + for table in FIVE_TABLES: # 五表: 旧的清空, 近期原样 + assert _ids(env.tenant_db, table, CONV_ID_COL[table]) == KEPT[table], \ + table + assert {OLD1, OLD2} & _ids(env.tenant_db, "conversations") == set() + assert env.export_path(365).is_file() # 导出落 data/archives// + assert "导出: " in out + + +# ---- 2. dry-run: 只列不动 ------------------------------------------------------ + +def test_dry_run_lists_without_touching(tmp_path, monkeypatch, capsys): + env = Env(tmp_path, monkeypatch) + _seed(env.tenant_db) + snapshot = {t: _full_rows(env.tenant_db, t) for t in FIVE_TABLES} + + run_archive(monkeypatch, *env.archive_args("--days", "30", "--dry-run")) + out = capsys.readouterr().out + + assert "dry-run" in out + assert "待归档会话: 3 条" in out # OLD1/OLD2/MID(40d > 30d) + for cid in (OLD1, OLD2, MID): + assert cid in out, "dry-run 应列出待归档会话 id" + assert {t: _full_rows(env.tenant_db, t) for t in FIVE_TABLES} == snapshot + assert not env.archives_dir.exists() # 不导出 → 目录懒建不发生 + assert "VACUUM: 已执行" not in out + assert "零改动" in out + + +# ---- 3. VACUUM 收缩 ------------------------------------------------------------ + +def test_vacuum_shrinks(tmp_path, monkeypatch, capsys): + env = Env(tmp_path, monkeypatch) + engine = create_engine(f"sqlite:///{env.tenant_db}") + Base.metadata.create_all(engine) + with Session(engine) as s: # 150 会话×4 消息, 均为 2020 年 + s.add_all([ + Conversation( + id=f"cccc0000-0000-0000-0000-{i:012d}", + session_id=f"sess-{i}", status=ConversationStatus.completed, + source=ConversationSource.widget, + created_at=datetime(2020, 1, 1, tzinfo=timezone.utc)) + for i in range(150)]) + s.add_all([ + Message(id=f"aaaa0000-0000-0000-{c:04d}-{m:08d}", + conversation_id=f"cccc0000-0000-0000-0000-{c:012d}", + role=MessageRole.user, content="价" * 600, + created_at=datetime(2020, 1, 1, 10, 0, tzinfo=timezone.utc)) + for c in range(150) for m in range(4)]) + s.commit() + engine.dispose() + + pages_before = _page_count(env.tenant_db) + size_before = env.tenant_db.stat().st_size + + run_archive(monkeypatch, *env.archive_args()) + out = capsys.readouterr().out + + assert "归档会话: 150 条" in out + assert "VACUUM: 已执行" in out + assert _count(env.tenant_db, "conversations") == 0 + assert _page_count(env.tenant_db) < pages_before # 页数收缩 + assert env.tenant_db.stat().st_size < size_before # 文件也随之变小 + + +# ---- 4. 导出完整性: 被删行完整字段, 五表按会话分组 ------------------------------- + +def test_export_complete(tmp_path, monkeypatch, capsys): + env = Env(tmp_path, monkeypatch) + _seed(env.tenant_db) + before = {t: _full_rows(env.tenant_db, t) for t in FIVE_TABLES} + + run_archive(monkeypatch, *env.archive_args()) + capsys.readouterr() + path = env.export_path(365) + + doc = json.loads(path.read_text(encoding="utf-8")) + assert doc["tenant"] == "jiancai" + assert doc["days"] == 365 + assert doc["cutoff"] + + def rows(table, **flt): + return sorted((r for r in before[table] if + all(r[k] == v for k, v in flt.items())), + key=lambda r: r["id"]) + + by_id = lambda table, i: next(r for r in before[table] if r["id"] == i) + + # 完整字段(全列 dict 逐字相等, 含 JSON 列原文/NULL/整型布尔) + assert doc["conversations"][OLD1] == by_id("conversations", OLD1) + assert doc["conversations"][OLD2]["created_at"] == \ + "2020-06-15 10:00:00+00:00" # +00:00 形态原样保真 + assert sorted(doc["messages"][OLD1], key=lambda r: r["id"]) == \ + rows("messages", conversation_id=OLD1) + assert doc["messages"][OLD2] == rows("messages", conversation_id=OLD2) + assert doc["customer_profiles"][OLD1] == \ + rows("customer_profiles", conversation_id=OLD1) + assert OLD2 not in doc["customer_profiles"] # 无 profile 的会话无键 + assert doc["evaluations"][OLD2] == \ + rows("evaluations", conversation_id=OLD2) + assert sorted(doc["lead_push_logs"][OLD1], key=lambda r: r["id"]) == \ + rows("lead_push_logs", conversation_id=OLD1) + # 近期会话不出现在任何分组 + for table in FIVE_TABLES: + assert REC1 not in doc[table], table + assert MID not in doc[table], table + + +# ---- 门禁与空跑 ----------------------------------------------------------------- + +def test_unbootstrapped_tenant_refused(tmp_path, monkeypatch): + db = tmp_path / "tenants" / "jiancai.db" + db.parent.mkdir(parents=True) + db.write_bytes(b"") + + # platform.db 不存在 + with pytest.raises(SystemExit, match="bootstrap"): + run_archive(monkeypatch, "--tenant", "jiancai", + "--platform-db", str(tmp_path / "nope" / "platform.db"), + "--tenants-dir", str(tmp_path / "tenants")) + + # platform.db 存在但只注册了 auto + platform_db = tmp_path / "platform.db" + run_bootstrap(monkeypatch, platform_db, "auto") + with pytest.raises(SystemExit, match="bootstrap"): + run_archive(monkeypatch, "--tenant", "jiancai", + "--platform-db", str(platform_db), + "--tenants-dir", str(tmp_path / "tenants")) + + +def test_tenant_db_missing_refused(tmp_path, monkeypatch): + env = Env(tmp_path, monkeypatch) + assert not env.tenant_db.exists() + with pytest.raises(SystemExit, match="租户库不存在"): + run_archive(monkeypatch, *env.archive_args()) + assert not env.archives_dir.exists() + + +def test_no_old_conversations_noop(tmp_path, monkeypatch, capsys): + env = Env(tmp_path, monkeypatch) + engine = create_engine(f"sqlite:///{env.tenant_db}") + Base.metadata.create_all(engine) + with Session(engine) as s: # 只有 1 天内的新会话 + s.add(Conversation(id=REC1, session_id="sess-rec1", + status=ConversationStatus.active, + source=ConversationSource.widget, + created_at=NOW - timedelta(days=1))) + s.commit() + engine.dispose() + sha = _sha256(env.tenant_db) + + run_archive(monkeypatch, *env.archive_args()) + out = capsys.readouterr().out + + assert "归档会话: 0 条" in out + assert "VACUUM: 未执行" in out + assert not env.export_path(365).exists() + assert not env.archives_dir.exists() + assert _sha256(env.tenant_db) == sha # 库文件字节级零变化