浅念
471 lines
19 KiB
Python
471 lines
19 KiB
Python
"""scripts/archive.py(Phase 4 Task 3): 租户库老会话归档(导出→清理→VACUUM)。
|
|
|
|
契约(计划原文):
|
|
1. >N 天会话(含子表行)→ 导出 JSON 到 data/archives/<tenant>/ → 清库 →
|
|
五表清空(旧会话相关); <N 天不动;
|
|
2. --dry-run 只列不动: 库零变化、不导出、不 VACUUM;
|
|
3. 清理后 VACUUM → page_count 减少(PRAGMA page_count 断言, 比文件大小稳);
|
|
4. 导出 JSON 含被删行的完整字段(五表按会话分组);
|
|
5. 门禁同 import_legacy: 租户须已 bootstrap(platform.db 有 slug);
|
|
6. 导出写盘失败 → 显性中止, 库零变化(导出先于删除的底线);
|
|
导出/删除行数对账一致(文件名带时分秒, 同日重跑不覆盖)。
|
|
|
|
测试方式(照 test_import_legacy.py): import + sys.argv 注入, 非 subprocess;
|
|
夹具全在 tmp(archives 目录取 tenants 目录兄弟 data/ 派生, 同 registry
|
|
uploads_root 的布局纪律, tmp 内自动隔离)。时间形态双覆盖: ORM 行存
|
|
'...SS.ffffff'(无 tz 后缀), 另造 '+00:00' 后缀的老库行(导入保真形态),
|
|
字典序截止串比较两者皆单调。
|
|
"""
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import re
|
|
import sqlite3
|
|
import sys
|
|
from datetime import datetime, timedelta, 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"
|
|
FIVE_TABLES = (
|
|
"conversations", "messages", "customer_profiles", "evaluations",
|
|
"lead_push_logs",
|
|
)
|
|
# 子表 → conversation_id 列(清库/保留断言遍历用)
|
|
CONV_ID_COL = {
|
|
"conversations": "id", "messages": "conversation_id",
|
|
"customer_profiles": "conversation_id", "evaluations": "conversation_id",
|
|
"lead_push_logs": "conversation_id",
|
|
}
|
|
|
|
|
|
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 锚定, 无副作用)
|
|
archive = _load("archive")
|
|
bootstrap_platform = _load("bootstrap_platform")
|
|
|
|
|
|
def run_archive(monkeypatch, *argv):
|
|
monkeypatch.setattr(sys, "argv", ["archive.py", *argv])
|
|
archive.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 _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]:
|
|
"""<日期>-<时分秒>-<N>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/<tenant>/
|
|
assert re.match(r"^\d{8}-\d{6}-365d\.json$", exports[0].name), \
|
|
"文件名须为 <日期>-<时分秒>-<N>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()
|