42 lines
2.6 KiB
Python
42 lines
2.6 KiB
Python
"""Read-only project index and cross-agent snapshot linkage."""
|
|
from fastapi import APIRouter, Depends, Request
|
|
from api import risk_agent_config as risk
|
|
from api.collateral_agent import decorated, existing_rows, storage
|
|
from api.risk_agent_access import authorize
|
|
|
|
router = APIRouter(prefix="/api/xingbang/risk-agent/workspace", dependencies=[Depends(authorize)])
|
|
|
|
|
|
@router.get("/projects")
|
|
def projects(request: Request):
|
|
with risk._db() as db:
|
|
storage(db)
|
|
objects = {}
|
|
for table, key in (("runs", "risk"), ("report_documents", "report"), ("coordination_projects", "coordination"), ("collateral_cases", "collateral")):
|
|
for obj in existing_rows(db, table):
|
|
project_id = obj["project_id"]
|
|
user = request.state.identity
|
|
if "*" not in user.get("projects", []) and project_id not in user.get("projects", []):
|
|
continue
|
|
entry = objects.setdefault(project_id, {"project_id": project_id, "risk": None, "report": None, "coordination": None, "collateral": None})
|
|
if entry[key] is not None:
|
|
continue
|
|
if key == "risk":
|
|
entry[key] = {"id": obj["id"], "time": obj["time"], "findings": len(obj["findings"]), "customer": obj.get("values", {}).get("customer.name", "")}
|
|
elif key == "report":
|
|
entry[key] = {"id": obj["id"], "status": obj["status"], "version": obj["version"], "source_id": obj["source_id"]}
|
|
elif key == "coordination":
|
|
entry[key] = {"id": obj["id"], "stage": obj["stage"], "owner": obj["owner_name"], "open_tasks": sum(t["status"] != "已解决" for t in obj["tasks"]), "plan_version": len(obj["plans"])}
|
|
else:
|
|
case = decorated(db, obj)
|
|
entry[key] = {"id": obj["id"], "stale": case["stale"], "open_checks": sum(c["status"] != "pass" for c in (obj.get("result") or {}).get("checks", [])), "has_result": bool(obj.get("result"))}
|
|
return {"projects": list(objects.values()), "identity": request.state.identity}
|
|
|
|
|
|
def report_links(db, project_id, source_id):
|
|
from api.collateral_agent import current_plan
|
|
coord, plan = current_plan(db, project_id)
|
|
case = next((c for c in existing_rows(db, "collateral_cases") if c["project_id"] == project_id and c["source_id"] == source_id and c.get("result") and not decorated(db, c)["stale"]), None)
|
|
return {"coordination_id": coord["id"] if coord else None, "plan": plan,
|
|
"verification": {"case_id": case["id"], "result": case["result"]} if case else None}
|