211 lines
10 KiB
Python
211 lines
10 KiB
Python
"""Read-only scoped monitoring and admin-only isolated demonstration actions."""
|
|
import json
|
|
import threading
|
|
import time
|
|
from datetime import datetime
|
|
from typing import Literal
|
|
from fastapi import APIRouter, Depends, HTTPException, Request
|
|
from pydantic import Field
|
|
from api.risk_agent_access import authorize, check_project
|
|
from api.project_assistant import Input, enabled, role, translate
|
|
from services.project_assistant import ProjectAssistantService
|
|
from services.operations_demo import DemoService, PROJECT, regression
|
|
from services.assistant_store import get, put, listing, event, now
|
|
from services.assistant_model import configuration
|
|
from services.assistant_settings import flags
|
|
from services.assistant_worker import LEASE_SECONDS
|
|
|
|
router = APIRouter(prefix="/api/xingbang/risk-agent/project-assistant/operations", dependencies=[Depends(authorize), Depends(enabled)], tags=["运行中心"])
|
|
EVALUATION_LOCK = threading.Lock()
|
|
SUBMIT_LOCK = threading.Lock()
|
|
|
|
|
|
def current_service():
|
|
return ProjectAssistantService()
|
|
|
|
|
|
def demo_service():
|
|
return DemoService()
|
|
|
|
|
|
def lab_guard(request, write=False):
|
|
if not flags()["demo"]:
|
|
raise HTTPException(403, "演示数据域已禁用")
|
|
check_project(request, PROJECT)
|
|
if write:
|
|
role(request, ("admin",))
|
|
|
|
|
|
def selected_service(request, domain):
|
|
if domain == "lab":
|
|
lab_guard(request)
|
|
return demo_service()
|
|
return current_service()
|
|
|
|
|
|
def visible(project, identity):
|
|
return ("*" in identity.get("projects", []) or project["id"] in identity.get("projects", [])) and (flags()["demo"] or project.get("data_domain") != "demo")
|
|
|
|
|
|
def seconds(start, end=None):
|
|
if not start:
|
|
return None
|
|
return round(max(0, (datetime.fromisoformat(end or now()) - datetime.fromisoformat(start)).total_seconds()), 3)
|
|
|
|
|
|
def summarize(run, job=None):
|
|
result = {k: run.get(k) for k in ("id", "project_id", "status", "created_at", "started_at", "ended_at", "attempt", "summary", "error", "superseded", "input_snapshot_id")}
|
|
result["elapsed_seconds"] = seconds(run.get("started_at"), run.get("ended_at"))
|
|
result["queue_seconds"] = seconds(run.get("created_at"), run.get("started_at") or run.get("ended_at"))
|
|
active = next((s for s in run["steps"] if s["status"] == "running"), None)
|
|
result["step"] = active["name"] if active else (run["steps"][-1]["name"] if run["steps"] else "等待工作进程领取")
|
|
result["lease_expired"] = bool(job and job["state"] == "running" and job["lease_until"] < time.time())
|
|
result["heartbeat_at"] = datetime.fromtimestamp(job["lease_until"] - LEASE_SECONDS).astimezone().isoformat() if job and job["state"] == "running" else None
|
|
return result
|
|
|
|
|
|
@router.get("/overview")
|
|
def overview(request: Request, domain: Literal["current", "lab"] = "current"):
|
|
service = selected_service(request, domain)
|
|
with service.connect() as db:
|
|
projects = [p for p in listing(db, "project") if visible(p, request.state.identity)]
|
|
ids = {p["id"] for p in projects}
|
|
runs = [r for r in listing(db, "run") if r["project_id"] in ids]
|
|
jobs = {r["id"]: dict(r) for r in db.execute("SELECT id,state,lease_until FROM assistant_jobs") if r["id"] in {x["id"] for x in runs}}
|
|
findings = [f for f in listing(db, "finding") if f["project_id"] in ids and f["status"] not in ("已解决", "不适用")]
|
|
stale = [r for r in listing(db, "report") if r["project_id"] in ids and r.get("status") == "stale"]
|
|
counts = {key: sum(r["status"] == key for r in runs) for key in ("queued", "running", "completed", "partial", "failed", "cancelled")}
|
|
data = {"domain": domain, "time": now(), "identity": request.state.identity, "counts": counts,
|
|
"projects": [{"id": p["id"], "name": p["name"], "data_revision": p["data_revision"]} for p in projects],
|
|
"runs": [summarize(r, jobs.get(r["id"])) for r in runs[:100]], "total_runs": len(runs),
|
|
"todo_count": len(findings), "stale_report_count": len(stale),
|
|
"todos": [{k: f.get(k) for k in ("id", "project_id", "title", "status", "stale", "suggested_action", "last_seen_run_id")} for f in findings[:50]],
|
|
"stale_reports": [{k: r.get(k) for k in ("id", "project_id", "status")} for r in stale[:30]],
|
|
"evaluations": [e for e in listing(db, "operations_evaluation") if e["project_id"] in ids][:10]}
|
|
return data
|
|
|
|
|
|
@router.get("/runs/{run_id}")
|
|
def run_detail(run_id: str, request: Request, domain: Literal["current", "lab"] = "current"):
|
|
service = selected_service(request, domain)
|
|
run = translate(lambda: service.resource("run", run_id))
|
|
check_project(request, run["project_id"])
|
|
project = service.resource("project", run["project_id"])
|
|
if not visible(project, request.state.identity):
|
|
raise HTTPException(403, "无权读取该数据域")
|
|
with service.connect() as db:
|
|
release = {r["id"]: r["version_id"] for r in listing(db, "release")}
|
|
components = run["components"]
|
|
applied = []
|
|
for component in [components["plan"]] + components["skills"] + components["rules"] + [components["template"]]:
|
|
applied.append({k: component.get(k) for k in ("id", "name", "kind", "version", "content_hash", "engine", "inputs", "conditions")})
|
|
applied[-1]["current_version"] = release.get(component["id"], "旧配置目录")
|
|
applied[-1]["changed"] = component["id"] in release and release[component["id"]] != component["id"] + "@" + component["version"]
|
|
snapshot = get(db, "input_snapshot", run["input_snapshot_id"])
|
|
events = [dict(seq=r[0], **json.loads(r[1])) for r in db.execute("SELECT seq,body FROM assistant_events WHERE project_id=? AND run_id=? ORDER BY seq", (run["project_id"], run_id))]
|
|
result = service.public_run(run)
|
|
result["steps"] = [{**{k: v for k, v in s.items() if k != "trace"}, "elapsed_seconds": seconds(s.get("started_at"), s.get("ended_at"))} for s in run["steps"]]
|
|
return {"run": result, "components": applied, "events": events, "source_revision": snapshot.get("lab_source_revision"),
|
|
"source_hash": snapshot.get("lab_source_hash"), "data_revision": snapshot["data_revision"], "values": snapshot["values"]}
|
|
|
|
|
|
@router.get("/lab/source")
|
|
def source(request: Request):
|
|
lab_guard(request)
|
|
lab = demo_service()
|
|
with lab.connect() as db:
|
|
release = get(db, "release", "lab.coverage")
|
|
rule = get(db, "component_version", release["version_id"])
|
|
project = get(db, "project", PROJECT)
|
|
return {**lab.source(), "release": release, "threshold": rule["conditions"][0]["threshold"],
|
|
"synced_revision": project.get("lab_source_revision"), "database": "operations_demo.sqlite3", "isolated": True}
|
|
|
|
|
|
class SourceInput(Input):
|
|
revision: int = Field(ge=1)
|
|
amount: str = Field(max_length=40)
|
|
liabilities: str = Field(max_length=40)
|
|
dd_liabilities: str = Field(max_length=40)
|
|
cashflow: str | None = Field(default=None, max_length=40)
|
|
duplicate: bool = True
|
|
enabled: bool = True
|
|
|
|
|
|
@router.post("/lab/source")
|
|
def edit_source(payload: SourceInput, request: Request):
|
|
lab_guard(request, True)
|
|
values = payload.model_dump()
|
|
revision = values.pop("revision")
|
|
return translate(lambda: demo_service().edit_source(request.state.identity["id"], revision, **values))
|
|
|
|
|
|
class SubmitInput(Input):
|
|
source_revision: int = Field(ge=1)
|
|
model: bool = False
|
|
requirement: str = Field(default="", max_length=2000)
|
|
|
|
|
|
@router.post("/lab/runs", status_code=202)
|
|
def submit(payload: SubmitInput, request: Request):
|
|
lab_guard(request, True)
|
|
key = request.headers.get("idempotency-key", "")
|
|
if not key or len(key) > 160:
|
|
raise HTTPException(422, "请提供不超过160字符的幂等键")
|
|
if payload.requirement.strip() and not payload.model:
|
|
raise HTTPException(422, "自然语言要求需要启用真实模型步骤，确定性模式不会解释提示词")
|
|
def perform():
|
|
with SUBMIT_LOCK:
|
|
lab = demo_service()
|
|
project = lab.ingest(request.state.identity["id"], payload.source_revision)
|
|
return lab.enqueue(PROJECT, request.state.identity, payload.requirement, "lab.model" if payload.model else "lab.deterministic", key, project["revision"])
|
|
return translate(perform)
|
|
|
|
|
|
class RuleInput(Input):
|
|
version: Literal["1.0.0", "2.0.0"]
|
|
revision: int = Field(ge=1)
|
|
|
|
|
|
@router.post("/lab/rule")
|
|
def switch_rule(payload: RuleInput, request: Request):
|
|
lab_guard(request, True)
|
|
return translate(lambda: demo_service().switch_rule(request.state.identity["id"], payload.version, payload.revision))
|
|
|
|
|
|
@router.post("/lab/runs/{run_id}/cancel")
|
|
def cancel(run_id: str, request: Request):
|
|
lab_guard(request, True)
|
|
return translate(lambda: demo_service().cancel(PROJECT, run_id, request.state.identity["id"]))
|
|
|
|
|
|
@router.get("/health")
|
|
def health(request: Request):
|
|
role(request, ("admin",))
|
|
service = current_service()
|
|
with service.connect() as db:
|
|
integrity = db.execute("PRAGMA quick_check").fetchone()[0]
|
|
try:
|
|
model = configuration()
|
|
except ValueError:
|
|
model = {"available": False}
|
|
return {"time": now(), "checks": [
|
|
{"name": "FastAPI / Next.js 代理", "status": "ok", "detail": "本请求已成功到达后端；不代表所有业务链路均可用"},
|
|
{"name": "项目 SQLite 完整性", "status": "ok" if integrity == "ok" else "error", "detail": "PRAGMA quick_check: " + integrity},
|
|
{"name": "业务模型", "status": "unverified" if model.get("available") else "unavailable", "detail": "已配置，尚未执行连通性及质量测试" if model.get("available") else "未配置可用内网模型；确定性演示可独立运行"},
|
|
{"name": "验证边界", "status": "unverified", "detail": "本页不验证 OCR、Word 版式或 ERP；不展示虚构 Token、成本或通过率"}]}
|
|
|
|
|
|
@router.post("/lab/evaluate")
|
|
def evaluate(request: Request):
|
|
lab_guard(request, True)
|
|
if not EVALUATION_LOCK.acquire(blocking=False):
|
|
raise HTTPException(409, "已有隔离验收在执行，请稍后查看")
|
|
try:
|
|
result = translate(lambda: regression(request.state.identity))
|
|
with demo_service().connect() as db:
|
|
put(db, "operations_evaluation", result, immutable=True)
|
|
event(db, PROJECT, "lab.evaluated", request.state.identity["id"], {"evaluation_id": result["id"], "passed": result["passed"]})
|
|
return result
|
|
finally:
|
|
EVALUATION_LOCK.release()
|