"""Versioned project facade with durable jobs and resource-scoped operations.""" from copy import deepcopy import json from pathlib import Path from services.assistant_store import transaction, initialize, get, put, listing, event, uid, now, digest, Conflict from services import component_registry as registry from services.assistant_materials import demo_material, demo_assets, material, apply_candidate, FIELDS from services.assistant_data import load_catalog from services.assistant_model import configuration from services.assistant_settings import flags from services.assistant_documents import DocumentsMixin from services.assistant_experiences import ExperiencesMixin from services.skill_executor import ratio, number, execute_skill FIXTURE_PATH = Path(__file__).resolve().parents[1] / "fixtures" / "xingbang_demo" / "projects.json" DB_PATH = Path(__file__).resolve().parents[1] / "xingbang_data" / "project_assistant.sqlite3" FORMULA_VERSION = "lease-value-over-facility@1.0.0" TERMINAL = {"completed", "partial", "failed", "cancelled"} class ProjectAssistantService(DocumentsMixin, ExperiencesMixin): def __init__(self, db_path=None, fixture_path=None): self.db_path = Path(db_path or DB_PATH) self.fixture_path = Path(fixture_path or FIXTURE_PATH) initialize(self.db_path) with self.connect() as db: db.execute("BEGIN IMMEDIATE") if db.execute("SELECT 1 FROM assistant_meta WHERE key='schema_v2'").fetchone(): return registry.seed(db) fixtures = json.loads(self.fixture_path.read_text(encoding="utf-8")) legacy = "projects" in {r[0] for r in db.execute("SELECT name FROM sqlite_master WHERE type='table'")} prior = {r[0]: json.loads(r[1]) for r in db.execute("SELECT id,body FROM projects")} if legacy else {} for item in fixtures["projects"]: body = deepcopy(item) if item["id"] in prior: body["facts"].update(prior[item["id"]]["facts"]) body["migration_note"] = "沿用旧项目人工数据；旧版证据按未知来源归档，待重新核查" self._seed_project(db, body, fixtures["data_domain"]) if legacy: for table in ("runs", "snapshots", "findings", "scenarios", "reports", "experiences"): for row in db.execute("SELECT id,body FROM " + table): old = json.loads(row[1]) put(db, "legacy_archive", {"id": table + ":" + row[0], "project_id": old.get("project_id", ""), "kind": table, "legacy_id": row[0], "body": old, "provenance": "legacy_unknown"}, immutable=True) event(db, "", "migration.v2", "system", {"projects": list(prior), "legacy_tables": "retained_read_only"}) db.execute("INSERT INTO assistant_meta VALUES ('schema_v2',?)", (now(),)) def connect(self): return transaction(self.db_path) def _seed_project(self, db, item, domain): project = {k: deepcopy(v) for k, v in item.items() if k != "documents"} project.update(data_domain=domain, data_revision=1, period="2025", scope="合并", values={}, sources={}, selected_sources={}, document_ids=[], pending_candidates=[]) document = demo_material(project) put(db, "document", document, immutable=True) project["document_ids"].append(document["id"]) for candidate in document["candidates"]: apply_candidate(project, candidate, "demo-fixture", "演示源文件初始化" if not project.get("migration_note") else "旧数据迁移，待重新核查") assets = demo_assets(project) put(db, "document", assets, immutable=True) project["document_ids"].append(assets["id"]) for candidate in assets["candidates"]: apply_candidate(project, candidate, "demo-fixture", "按设备唯一编号去重计算，重复项仍保留核查") put(db, "project", project) def resource(self, kind, object_id, project_id=None): with self.connect() as db: body = get(db, kind, object_id) if project_id is not None and body.get("project_id") != project_id: raise KeyError("资源不属于该项目") return body @staticmethod def public_run(run): return {k: v for k, v in run.items() if k not in ("identity", "components", "model_config", "experience_versions")} def list_projects(self): with self.connect() as db: projects = listing(db, "project") for project in projects: runs = listing(db, "run", project["id"]) project["latest_run"] = self.public_run(runs[0]) if runs else None project["open_findings"] = sum(f["status"] not in ("已解决", "不适用") for f in listing(db, "finding", project["id"])) return sorted(projects, key=lambda p: p["id"]) def get_project(self, project_id): with self.connect() as db: project = get(db, "project", project_id) for plural, kind in (("runs", "run"), ("findings", "finding"), ("scenarios", "scenario"), ("reports", "report"), ("experiences", "experience"), ("feedback", "feedback"), ("experience_applications", "experience_application")): project[plural] = listing(db, kind, project_id) project["runs"] = [self.public_run(r) for r in project["runs"]] project["documents"] = [{k: v for k, v in get(db, "document", key).items() if k not in ("raw_base64", "chunks")} for key in project["document_ids"]] project["legacy_records"] = [{"id": x["id"], "kind": x["kind"], "provenance": x["provenance"]} for x in listing(db, "legacy_archive", project_id)] project["events"] = [{"seq": r[0], **json.loads(r[1])} for r in db.execute("SELECT seq,body FROM assistant_events WHERE project_id=? ORDER BY seq DESC LIMIT 100", (project_id,))] project["plans"] = [{"id": v["component_id"], "name": v["name"], "version": v["version"]} for v in listing(db, "component_version") if v["kind"] == "plan" and v["scope"] == project["data_domain"] and get(db, "release", v["component_id"])["version_id"] == v["id"]] return project def enqueue(self, project_id, identity, requirement="", plan_id="corporate.default", idempotency_key="", expected=None, feedback_id=None): catalog, model = load_catalog(), configuration() request_hash = digest({"project": project_id, "requirement": requirement, "plan": plan_id, "expected": expected, "feedback_id": feedback_id}) with self.connect() as db: db.execute("BEGIN IMMEDIATE") if idempotency_key: prior = db.execute("SELECT request_hash,object_id FROM assistant_idempotency WHERE actor=? AND operation=? AND key=?", (identity["id"], "run", idempotency_key)).fetchone() if prior: if prior[0] != request_hash: raise Conflict("幂等键已用于不同请求") return self.public_run(get(db, "run", prior[1])) project = get(db, "project", project_id) feedback = get(db, "feedback", feedback_id) if feedback_id else None if feedback and (feedback["project_id"] != project_id or feedback["category"] in ("expression", "factual", "data")): raise ValueError("此反馈不能用于本项目的核查经验提炼") if expected is not None and project["revision"] != expected: raise Conflict("项目已更新，请刷新后重新检查") components = registry.freeze(db, plan_id, catalog) if components["plan"]["scope"] != project["data_domain"]: raise ValueError("方案与项目数据域不一致") documents = [{k: v for k, v in get(db, "document", key).items() if k != "raw_base64"} for key in project["document_ids"]] for document in documents: for index, chunk in enumerate(document["chunks"]): chunk.setdefault("evidence_id", "evdoc-" + digest([document["id"], document["content_hash"], index, chunk["locator"]])[:32]) snapshot = dict(deepcopy(project), id=uid("input"), project_id=project_id, project_revision=project["revision"], documents=documents, created_at=now()) put(db, "input_snapshot", snapshot, immutable=True) run = {"id": uid("run"), "project_id": project_id, "actor_id": identity["id"], "identity": identity, "status": "queued", "created_at": now(), "requirement": requirement, "input_snapshot_id": snapshot["id"], "snapshot_id": None, "components": components, "model_config": model, "steps": [], "summary": {}, "evidence": [], "feature_flags": flags(), "feedback_context": feedback, "component_versions": [{"id": s["id"], "version": s["version"], "content_hash": s["content_hash"]} for s in [components["plan"]] + components["skills"] + components["rules"] + [components["template"]]]} run = put(db, "run", run) db.execute("INSERT INTO assistant_jobs(id,project_id,state) VALUES (?,?, 'queued')", (run["id"], project_id)) if idempotency_key: db.execute("INSERT INTO assistant_idempotency VALUES (?,?,?,?,?)", (identity["id"], "run", idempotency_key, request_hash, run["id"])) event(db, project_id, "run.queued", identity["id"], {"input_snapshot_id": snapshot["id"], "component_hash": components["content_hash"], "requirement": requirement}, run["id"]) return self.public_run(run) def cancel(self, project_id, run_id, actor): with self.connect() as db: db.execute("BEGIN IMMEDIATE") run = get(db, "run", run_id) if run["project_id"] != project_id: raise KeyError("运行不存在") if run["status"] in TERMINAL: raise Conflict("运行已结束") run.update(status="cancelled", ended_at=now()) put(db, "run", run) db.execute("UPDATE assistant_jobs SET state='cancelled',owner=NULL,lease_until=0 WHERE id=?", (run_id,)) event(db, project_id, "run.cancelled", actor, run_id=run_id) return self.public_run(run) def upload(self, project_id, filename, content, actor): document = material(project_id, filename, content, actor) with self.connect() as db: db.execute("BEGIN IMMEDIATE") project = get(db, "project", project_id) prior = listing(db, "document", project_id) duplicate = next((d for d in prior if d["content_hash"] == document["content_hash"]), None) if duplicate: return {k: v for k, v in duplicate.items() if k != "raw_base64"} document["version"] = 1 + max((d["version"] for d in prior if d["name"] == document["name"]), default=0) put(db, "document", document, immutable=True) project["document_ids"].append(document["id"]) project.setdefault("pending_candidates", []).extend(c["id"] for c in document["candidates"]) project["data_revision"] += 1 put(db, "project", project) self.mark_stale(db, project_id, ["*"]) event(db, project_id, "document.uploaded", actor, {"document_id": document["id"], "hash": document["content_hash"]}) return {k: v for k, v in document.items() if k != "raw_base64"} def supplement(self, project_id, actor, changes, finding_ids, expected, reason, candidate_ids=None): if not reason.strip() or (not changes and not candidate_ids): raise ValueError("请填写变更字段或材料候选以及确认依据") if set(changes) - (set(FIELDS) - {"facility_amount", "lease_asset_value"}): raise ValueError("补件字段不受支持；融资金额应使用情景采纳") with self.connect() as db: db.execute("BEGIN IMMEDIATE") project = get(db, "project", project_id) if project["revision"] != expected: raise Conflict("项目数据已更新，请刷新") findings = {f["id"]: f for f in listing(db, "finding", project_id)} if set(finding_ids) - set(findings): raise ValueError("关联问题不属于该项目") candidates = [c for d in listing(db, "document", project_id) for c in d["candidates"] if c["id"] in (candidate_ids or [])] if len(candidates) != len(set(candidate_ids or [])): raise ValueError("材料候选不存在或不属于项目") for field, value in changes.items(): standard = FIELDS[field] source = next((r for r in project["sources"].get(standard, []) if r["id"] == project["selected_sources"].get(standard)), {}) candidates.append({"field": standard, "value": value, "source": "人工补充", "locator": "补件说明：" + reason, **{k: source.get(k) for k in ("period", "scope", "unit", "currency")}}) if len({c["field"] for c in candidates}) != len(candidates): raise ValueError("同一字段有多个候选，请明确唯一来源") before = deepcopy(project["values"]) for candidate in candidates: apply_candidate(project, candidate, actor, reason) if candidate.get("id") in project.get("pending_candidates", []): project["pending_candidates"].remove(candidate["id"]) project["data_revision"] += 1 project = put(db, "project", project, expected=expected) changed = [k for k, v in project["values"].items() if before.get(k) != v] self.mark_stale(db, project_id, changed or ["*"]) event(db, project_id, "supplement.accepted", actor, {"fields": changed, "finding_ids": finding_ids, "reason": reason, "data_revision": project["data_revision"]}) return {"project_id": project_id, "revision": project["revision"], "changed_fields": changed, "status": "待重新检查"} def finding_action(self, project_id, finding_id, actor, status, note, expected): if status not in ("待处理", "待补件", "待复核", "不适用") or not note.strip(): raise ValueError("需选择有效状态并填写说明；已解决仅由有依据的检查确认") with self.connect() as db: db.execute("BEGIN IMMEDIATE") finding = get(db, "finding", finding_id) if finding["project_id"] != project_id: raise KeyError("问题不存在") finding["history"].append({"time": now(), "actor": actor, "from": finding["status"], "to": status, "note": note, "type": "human"}) finding.update(status=status, human_status=status, human_note=note) result = put(db, "finding", finding, expected=expected) self.mark_stale(db, project_id, ["*"]) event(db, project_id, "finding.action", actor, {"finding_id": finding_id, "status": status, "note": note}) return result def compare_scenario(self, project_id, amount, actor): with self.connect() as db: project = get(db, "project", project_id) source = project["values"] selected = {f: [r for r in refs if r["id"] == project["selected_sources"].get(f)] for f, refs in project["sources"].items()} skill = {"id": "scenario.coverage", "version": "1.0.0", "engine": "ratio", "numerator": "lease.total_value", "denominator": "facility.amount", "output": "metrics.coverage"} baseline = execute_skill(skill, source, selected)[0]["metrics.coverage"] proposed = execute_skill(skill, dict(source, **{"facility.amount": amount}), selected)[0]["metrics.coverage"] scenario = {"id": uid("scenario"), "project_id": project_id, "status": "comparison", "actor_id": actor, "created_at": now(), "project_revision": project["revision"], "data_revision": project["data_revision"], "formula_version": FORMULA_VERSION, "sources": deepcopy(project["sources"]), "baseline": {"facility_amount": source["facility.amount"], "coverage": baseline}, "scenario": {"facility_amount": str(number(amount)), "coverage": proposed}, "delta": str(number(proposed) - number(baseline)), "unresolved": [f["id"] for f in listing(db, "finding", project_id) if f["status"] not in ("已解决", "不适用") and f["code"] != "lease.coverage.below_demo_threshold"], "assumptions": ["租赁物价值与评估时点不变", "金额单位万元，币种人民币", "120%仅为演示阈值，非正式制度"], "data_domain": project["data_domain"]} result = put(db, "scenario", scenario, immutable=True) event(db, project_id, "scenario.compared", actor, {"scenario_id": result["id"]}) return result def adopt_scenario(self, project_id, scenario_id, actor, expected, reason): if not reason.strip(): raise ValueError("请填写采纳理由") with self.connect() as db: db.execute("BEGIN IMMEDIATE") scenario = get(db, "scenario", scenario_id) project = get(db, "project", project_id) if scenario["project_id"] != project_id: raise KeyError("情景不存在") if scenario["project_revision"] != project["revision"] or expected != project["revision"]: raise Conflict("情景基线已过期，请重新比较") selected = next(r for r in project["sources"]["facility.amount"] if r["id"] == project["selected_sources"]["facility.amount"]) candidate = {**selected, "field": "facility.amount", "value": scenario["scenario"]["facility_amount"], "source": "采纳融资情景", "locator": scenario_id} apply_candidate(project, candidate, actor, reason) project["data_revision"] += 1 result = put(db, "project", project, expected=expected) self.mark_stale(db, project_id, ["facility.amount", "metrics.coverage"]) event(db, project_id, "scenario.adopted", actor, {"scenario_id": scenario_id, "reason": reason}) return {"project_id": project_id, "revision": result["revision"], "status": "待重新检查"} def reset_demo(self, actor, confirmation): if confirmation != "RESET-DEMO-A-B-C": raise ValueError("请输入 RESET-DEMO-A-B-C 确认仅重置三个演示实例") fixtures = json.loads(self.fixture_path.read_text(encoding="utf-8")) with self.connect() as db: db.execute("BEGIN IMMEDIATE") for item in fixtures["projects"]: current = get(db, "project", item["id"]) if current["data_domain"] != "demo": raise ValueError("禁止重置正式项目") records = [dict(r) for r in db.execute("SELECT * FROM assistant_objects WHERE project_id=? OR (kind='project' AND id=?)", (item["id"], item["id"]))] put(db, "reset_archive", {"id": uid("reset"), "source_project_id": item["id"], "records": records, "actor": actor, "created_at": now()}, immutable=True) db.execute("DELETE FROM assistant_jobs WHERE project_id=?", (item["id"],)) db.execute("DELETE FROM assistant_idempotency WHERE object_id IN (SELECT id FROM assistant_objects WHERE project_id=?)", (item["id"],)) db.execute("DELETE FROM assistant_objects WHERE project_id=? OR (kind='project' AND id=?)", (item["id"], item["id"])) self._seed_project(db, item, "demo") event(db, item["id"], "demo.reset", actor, {"recoverable": True}) return {"status": "reset", "data_domain": "demo", "projects": len(fixtures["projects"]), "recoverable": True}