168 lines
12 KiB
Python
168 lines
12 KiB
Python
"""Isolated, versioned SQLite source -> real assistant worker demonstration.
|
||
|
||
Only LAB-001 in operations_demo.sqlite3 is writable here. No production SQL or
|
||
credentials are accepted. Source edits never rewrite frozen run snapshots.
|
||
"""
|
||
from copy import deepcopy
|
||
from decimal import Decimal
|
||
from pathlib import Path
|
||
import tempfile
|
||
import time
|
||
|
||
from services.project_assistant import ProjectAssistantService
|
||
from services.assistant_store import get, put, listing, event, digest, now, uid, Conflict
|
||
from services.assistant_materials import apply_candidate
|
||
from services import component_registry as registry
|
||
from services.skill_executor import number
|
||
from services.assistant_worker import DurableWorker
|
||
|
||
PROJECT = "LAB-001"
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
DB_PATH = ROOT / "xingbang_data" / "operations_demo.sqlite3"
|
||
FIXTURE = ROOT / "fixtures" / "operations_demo" / "project.json"
|
||
SKILLS = ["corporate.consistency", "corporate.cashflow", "corporate.assets", "corporate.coverage"]
|
||
|
||
|
||
class DemoService(ProjectAssistantService):
|
||
def __init__(self, db_path=None):
|
||
super().__init__(db_path or DB_PATH, FIXTURE)
|
||
with self.connect() as db:
|
||
db.executescript("""
|
||
CREATE TABLE IF NOT EXISTS lab_project (
|
||
id TEXT PRIMARY KEY, amount TEXT, revision INTEGER, enabled INTEGER, updated_at TEXT);
|
||
CREATE TABLE IF NOT EXISTS lab_finance (
|
||
project_id TEXT PRIMARY KEY, liabilities TEXT, dd_liabilities TEXT, cashflow TEXT);
|
||
CREATE TABLE IF NOT EXISTS lab_asset (
|
||
row_id INTEGER PRIMARY KEY, project_id TEXT, asset_key TEXT, value TEXT, active INTEGER);
|
||
""")
|
||
db.execute("BEGIN IMMEDIATE")
|
||
if not db.execute("SELECT 1 FROM lab_project WHERE id=?", (PROJECT,)).fetchone():
|
||
db.execute("INSERT INTO lab_project VALUES (?, '5000', 1, 1, ?)", (PROJECT, now()))
|
||
db.execute("INSERT INTO lab_finance VALUES (?, '13200', '12800', NULL)", (PROJECT,))
|
||
db.executemany("INSERT INTO lab_asset VALUES (?,?,?,?,1)", [(1, PROJECT, "EQ-01", "3000"), (2, PROJECT, "EQ-02", "2400"), (3, PROJECT, "EQ-01", "3000")])
|
||
for model in (False, True):
|
||
self._publish(db, {"id": "lab.model" if model else "lab.deterministic", "kind": "plan",
|
||
"name": "演示检查（含真实模型步骤）" if model else "演示检查（确定性 Skills）",
|
||
"skills": SKILLS + (["corporate.analysis"] if model else []), "rules": ["lab.coverage"],
|
||
"template": "template.due-diligence", "budget": {"timeout_seconds": 60, "max_tool_calls": 12}})
|
||
for version, threshold in (("1.0.0", "1.2"), ("2.0.0", "1.4")):
|
||
self._publish(db, {"id": "lab.coverage", "kind": "rule_set", "version": version,
|
||
"name": "隔离演示覆盖率规则", "conditions": [{"code": "lease.coverage.below_demo_threshold",
|
||
"field": "metrics.coverage", "operator": "lt", "threshold": threshold,
|
||
"title": "覆盖率低于演示阈值", "suggestion": "比较融资金额；此阈值非兴邦正式制度"}]})
|
||
put(db, "release", {"id": "lab.coverage", "version_id": "lab.coverage@1.0.0", "scope": "demo"})
|
||
event(db, PROJECT, "lab.initialized", "system-demo", {"tables": ["lab_project", "lab_finance", "lab_asset"]})
|
||
|
||
@staticmethod
|
||
def _publish(db, item):
|
||
body = registry.normalize(dict(item, status="published"), "system-demo")
|
||
registry.validate(body)
|
||
put(db, "component_version", dict(body, id=body["id"] + "@" + body["version"], component_id=body["id"]), immutable=True)
|
||
put(db, "release", {"id": body["id"], "version_id": body["id"] + "@" + body["version"], "scope": "demo"})
|
||
|
||
@staticmethod
|
||
def _source(db):
|
||
project = dict(db.execute("SELECT * FROM lab_project WHERE id=?", (PROJECT,)).fetchone())
|
||
finance = dict(db.execute("SELECT * FROM lab_finance WHERE project_id=?", (PROJECT,)).fetchone())
|
||
assets = [dict(r) for r in db.execute("SELECT * FROM lab_asset WHERE project_id=? ORDER BY row_id", (PROJECT,))]
|
||
return {"project": project, "finance": finance, "assets": assets, "content_hash": digest([project, finance, assets])}
|
||
|
||
def source(self):
|
||
with self.connect() as db:
|
||
return self._source(db)
|
||
|
||
def edit_source(self, actor, expected, amount, liabilities, dd_liabilities, cashflow, duplicate, enabled):
|
||
amount, liabilities, dd_liabilities = (str(number(v)) for v in (amount, liabilities, dd_liabilities))
|
||
cashflow = None if cashflow in (None, "") else str(number(cashflow))
|
||
if number(amount) <= 0 or number(liabilities) < 0 or number(dd_liabilities) < 0:
|
||
raise ValueError("融资金额须大于零，负债不能为负数")
|
||
with self.connect() as db:
|
||
db.execute("BEGIN IMMEDIATE")
|
||
before = self._source(db)
|
||
if before["project"]["revision"] != expected:
|
||
raise Conflict("演示源数据已更新，请刷新")
|
||
db.execute("UPDATE lab_project SET amount=?,revision=revision+1,enabled=?,updated_at=? WHERE id=?", (amount, int(enabled), now(), PROJECT))
|
||
db.execute("UPDATE lab_finance SET liabilities=?,dd_liabilities=?,cashflow=? WHERE project_id=?", (liabilities, dd_liabilities, cashflow, PROJECT))
|
||
db.execute("UPDATE lab_asset SET active=? WHERE row_id=3 AND project_id=?", (int(duplicate), PROJECT))
|
||
after = self._source(db)
|
||
event(db, PROJECT, "lab.source.updated", actor, {"before_hash": before["content_hash"], "after_hash": after["content_hash"], "source_revision": expected + 1})
|
||
return after
|
||
|
||
def ingest(self, actor, expected):
|
||
with self.connect() as db:
|
||
db.execute("BEGIN IMMEDIATE")
|
||
source = self._source(db)
|
||
if expected != source["project"]["revision"]:
|
||
raise Conflict("源数据版本已变化，请刷新后同步")
|
||
if not source["project"]["enabled"]:
|
||
raise ValueError("演示源已暂停：未同步、未创建运行；恢复数据源后重试")
|
||
project = get(db, "project", PROJECT)
|
||
if project.get("lab_source_hash") == source["content_hash"]:
|
||
return project
|
||
assets = [a for a in source["assets"] if a["active"]]
|
||
unique = {}
|
||
for asset in assets:
|
||
if asset["asset_key"] in unique and unique[asset["asset_key"]] != asset["value"]:
|
||
raise ValueError("同编号设备金额冲突，不能自动汇总")
|
||
unique[asset["asset_key"]] = asset["value"]
|
||
values = {
|
||
"facility.amount": (source["project"]["amount"], "lab_project", "amount", [PROJECT]),
|
||
"financial.total_liabilities": (source["finance"]["liabilities"], "lab_finance", "liabilities", [PROJECT]),
|
||
"due_diligence.liabilities": (source["finance"]["dd_liabilities"], "lab_finance", "dd_liabilities", [PROJECT]),
|
||
"financial.operating_cashflow": (source["finance"]["cashflow"], "lab_finance", "cashflow", [PROJECT]),
|
||
"lease.total_value": (str(sum((number(v) for v in unique.values()), Decimal(0))), "lab_asset", "value（按asset_key去重求和）", [a["row_id"] for a in assets]),
|
||
"lease.duplicate_count": (str(len(assets) - len(unique)), "lab_asset", "asset_key（重复数）", [a["row_id"] for a in assets]),
|
||
}
|
||
for field, (value, table, column, rows) in values.items():
|
||
apply_candidate(project, {"field": field, "value": value, "source": "SQLite 演示源 / " + table,
|
||
"locator": f"{table}.{column} / 行标识 {','.join(map(str, rows))}", "source_revision": expected,
|
||
"content_hash": source["content_hash"], "period": "2025", "scope": "合并", "currency": "CNY", "unit": "万元" if field != "lease.duplicate_count" else None}, actor, "人工确认同步隔离演示数据源")
|
||
project.update(lab_source_hash=source["content_hash"], lab_source_revision=expected, data_revision=project["data_revision"] + 1)
|
||
project = put(db, "project", project)
|
||
self.mark_stale(db, PROJECT, ["*"])
|
||
event(db, PROJECT, "lab.source.ingested", actor, {"source_revision": expected, "content_hash": source["content_hash"], "unique_assets": len(unique), "values": {k: v[0] for k, v in values.items()}})
|
||
return project
|
||
|
||
def switch_rule(self, actor, version, expected):
|
||
if version not in ("1.0.0", "2.0.0"):
|
||
raise ValueError("仅支持两个预置演示版本")
|
||
with self.connect() as db:
|
||
db.execute("BEGIN IMMEDIATE")
|
||
release = get(db, "release", "lab.coverage")
|
||
if release["revision"] != expected:
|
||
raise Conflict("配置已变化，请刷新")
|
||
result = put(db, "release", {**release, "version_id": "lab.coverage@" + version}, expected=expected)
|
||
event(db, PROJECT, "lab.rule.switched", actor, {"before": release["version_id"], "after": result["version_id"], "note": "仅影响后续演示任务，历史快照不变"})
|
||
return result
|
||
|
||
|
||
def regression(identity):
|
||
"""Real SQL + worker + frozen rules; four isolated fixed cases, no model mock."""
|
||
started = time.perf_counter()
|
||
rows = []
|
||
scenarios = [
|
||
("初始缺件与冲突", "5000", "12800", None, True, "1.0.0", "1.0800", {"financial.liabilities.conflict", "financial.cashflow.missing", "lease.asset.duplicate", "lease.coverage.below_demo_threshold"}),
|
||
("纠正负债并设备去重", "5000", "13200", None, False, "1.0.0", "1.0800", {"financial.cashflow.missing", "lease.coverage.below_demo_threshold"}),
|
||
("补齐现金流并调整金额", "4000", "13200", "900", False, "1.0.0", "1.3500", set()),
|
||
("同一数据切换严格规则", "4000", "13200", "900", False, "2.0.0", "1.3500", {"lease.coverage.below_demo_threshold"}),
|
||
]
|
||
with tempfile.TemporaryDirectory(prefix="xingbang-ops-eval-") as directory:
|
||
lab = DemoService(Path(directory) / "isolated.sqlite3")
|
||
worker = DurableWorker(lab)
|
||
for name, amount, dd, cash, duplicate, version, coverage, codes in scenarios:
|
||
source = lab.source()
|
||
source = lab.edit_source(identity["id"], source["project"]["revision"], amount, "13200", dd, cash, duplicate, True)
|
||
with lab.connect() as db:
|
||
release = get(db, "release", "lab.coverage")
|
||
lab.switch_rule(identity["id"], version, release["revision"])
|
||
project = lab.ingest(identity["id"], source["project"]["revision"])
|
||
run = lab.enqueue(PROJECT, identity, plan_id="lab.deterministic", expected=project["revision"])
|
||
worker.process_next()
|
||
run = lab.resource("run", run["id"], PROJECT)
|
||
actual = {c["code"] for c in run.get("checks", []) if c["result"] in ("issue", "unknown", "error")}
|
||
passed = actual == codes and run.get("summary", {}).get("coverage") == coverage and run["status"] == ("partial" if cash is None else "completed")
|
||
rows.append({"name": name, "passed": passed, "expected_codes": sorted(codes), "actual_codes": sorted(actual), "expected_coverage": coverage,
|
||
"actual_coverage": run.get("summary", {}).get("coverage"), "status": run["status"], "rule_version": version, "component_versions": run["component_versions"]})
|
||
return {"id": uid("evaluation"), "project_id": PROJECT, "created_at": now(), "passed": all(r["passed"] for r in rows),
|
||
"cases": rows, "duration_ms": round((time.perf_counter() - started) * 1000), "scope": "隔离临时库；确定性检查；不代表模型质量或真实业务验收"}
|