"""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_files(self, days: int) -> list[Path]: """<日期>-<时分秒>-d.json(文件名带时分秒防同日重跑覆盖, 测试用 glob 取, 免跨秒竞态)。""" return sorted(self.archives_dir.glob(f"jiancai/*-{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() exports = env.export_files(365) assert len(exports) == 1 # 导出落 data/archives// assert re.match(r"^\d{8}-\d{6}-365d\.json$", exports[0].name), \ "文件名须为 <日期>-<时分秒>-d.json(同日重跑不覆盖)" assert "导出/删除对账: 一致" in out 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_files(365)[0] 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 env.export_files(365) == [] assert not env.archives_dir.exists() assert _sha256(env.tenant_db) == sha # 库文件字节级零变化 # ---- 导出写失败: 库零变化 ------------------------------------------------------- def test_export_write_failure_leaves_db_untouched(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} def boom(path, doc): raise OSError(28, "No space left on device") monkeypatch.setattr(archive, "_write_export", boom) with pytest.raises(SystemExit, match="导出写盘失败"): run_archive(monkeypatch, *env.archive_args()) assert {t: _full_rows(env.tenant_db, t) for t in FIVE_TABLES} == snapshot assert env.export_files(365) == [] assert not (env.archives_dir / "jiancai").exists()