84 lines
5.2 KiB
Python
84 lines
5.2 KiB
Python
"""Reproducible offline acceptance and latency evidence in an isolated database."""
|
|
import argparse
|
|
import base64
|
|
import json
|
|
import os
|
|
import statistics
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / "backend"))
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--output", type=Path, default=ROOT / "test-evidence" / ("assistant-acceptance-" + datetime.now().strftime("%Y%m%d-%H%M%S")))
|
|
args = parser.parse_args()
|
|
output = args.output.resolve()
|
|
if not output.is_relative_to(ROOT / "test-evidence"):
|
|
raise ValueError("验收输出必须位于当前仓库 test-evidence")
|
|
output.mkdir(parents=True, exist_ok=False)
|
|
os.environ.update(XINGBANG_LOCAL_MODEL_URL="", XINGBANG_AUTH_USERS="", XINGBANG_REQUIRE_AUTH="0")
|
|
from api import risk_agent_config as catalog
|
|
from services.project_assistant import ProjectAssistantService
|
|
from services.assistant_worker import DurableWorker
|
|
from services.assistant_store import listing
|
|
identity = {"id": "offline-acceptance", "role": "admin", "projects": ["*"], "preview": True}
|
|
records = []
|
|
with tempfile.TemporaryDirectory(dir=output, ignore_cleanup_errors=True) as tmp:
|
|
catalog.DB_PATH = Path(tmp) / "catalog.sqlite3"
|
|
service = ProjectAssistantService(Path(tmp) / "assistant.sqlite3")
|
|
worker = DurableWorker(service)
|
|
def run(project_id):
|
|
start = time.perf_counter()
|
|
queued = service.enqueue(project_id, identity, "核对现金流与来源")
|
|
queued_ms = (time.perf_counter() - start) * 1000
|
|
worker.process_next()
|
|
result = service.resource("run", queued["id"])
|
|
records.append({"project": project_id, "run_id": result["id"], "enqueue_ms": round(queued_ms, 2), "total_ms": round((time.perf_counter() - start) * 1000, 2), "status": result["status"], "summary": result["summary"]})
|
|
return result
|
|
first = run("DEMO-B")
|
|
assert first["summary"]["coverage"] == "1.0800" and first["summary"]["open"] == 4
|
|
project = service.resource("project", "DEMO-B")
|
|
service.supplement("DEMO-B", identity["id"], {"due_diligence_liabilities": "13200", "duplicate_asset_count": 0}, [], project["revision"], "离线验收补件")
|
|
second = run("DEMO-B")
|
|
assert len(second["change_summary"]["resolved"]) == 2
|
|
scenario = service.compare_scenario("DEMO-B", "4000", identity["id"])
|
|
assert scenario["scenario"]["coverage"] == "1.3500" and len(scenario["unresolved"]) == 1
|
|
feedback = service.feedback("DEMO-B", identity["id"], {"category": "rule_omission", "target_kind": "run", "target_id": second["id"], "note": "追加回款凭证检查"})
|
|
experience = service.save_experience("DEMO-B", identity["id"], {"title": "回款凭证检查", "trigger": "存在回款依赖", "action": "补充到账依据", "exceptions": "已有到账凭证", "scope": "team", "team": "demo", "feedback_id": feedback["id"], "conditions": {"collection.dependency": True}, "required_field": "collection.evidence", "exception_field": "collection.evidence", "expires_at": "2099-01-01T00:00:00+00:00"})
|
|
experience = service.experience_action("DEMO-B", experience["id"], identity, "evaluate", experience["revision"], "隔离正反例评测")
|
|
service.experience_action("DEMO-B", experience["id"], identity, "publish", experience["revision"], "本机模拟独立审核")
|
|
c = run("DEMO-C")
|
|
a = run("DEMO-A")
|
|
assert any(f["code"].startswith("experience.") for f in c["findings"])
|
|
assert not any(f["code"].startswith("experience.") for f in a["findings"])
|
|
reports = []
|
|
for template in ("template.due-diligence", "template.review", "template.committee"):
|
|
report = service.create_report("DEMO-B", second["id"], identity["id"], template)
|
|
target = output / (template + ".docx")
|
|
target.write_bytes(service.export_report("DEMO-B", report["id"]))
|
|
reports.append({"path": target.name, "template": template, "snapshot_id": report["snapshot_id"], "visual_qa": "pending-renderer"})
|
|
times = []
|
|
for _ in range(30):
|
|
start = time.perf_counter()
|
|
service.resource("run", second["id"])
|
|
times.append((time.perf_counter() - start) * 1000)
|
|
with service.connect() as db:
|
|
for document in listing(db, "document"):
|
|
(output / document["name"]).write_bytes(base64.b64decode(document["raw_base64"]))
|
|
summary = {"mode": "offline-deterministic-no-live-model", "records": records, "scenario": scenario,
|
|
"poll_ms_p95": round(sorted(times)[28], 2), "poll_ms_median": round(statistics.median(times), 2),
|
|
"reports": reports, "experience_positive_and_negative": "passed", "preserved_live_database": True}
|
|
(output / "summary.json").write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
print(json.dumps({"output": str(output), "mode": summary["mode"], "poll_ms_p95": summary["poll_ms_p95"], "records": records}, ensure_ascii=False, indent=2))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|