feat(action-plans): COSO 自检未通过条目一键生成改善行动(接通死按钮)
ci/woodpecker/push/main Pipeline failed

- 后端新增 POST /api/cma/action-plans/coso-generate-actions:未通过条目→action_plans,
  账套取 token 绑定(忽略 body entity_id)、幂等键 (entity_id,title)、不写 auto_verify_rule
- 前端 ActionPlanLibrary.vue 按钮接 @click(含未通过条数提示 + 二次确认 + 生成后刷新并跳列表)
- api 客户端补 cosoGenerateActions
- 新增 tests/test_coso_generate_actions_20260921.py(生成/幂等/隔离/未认证/归属校验/空集)
This commit is contained in:
fullstack
2026-09-23 01:03:19 +08:00
committed by Hermes CI Fix
parent 2c2dbc2b2d
commit bd4f58aae1
4 changed files with 470 additions and 3 deletions
+157 -2
View File
@@ -1,7 +1,8 @@
"""改善行动计划 API — 管理会计OS"""
from fastapi import APIRouter, Depends, HTTPException, Request
from fastapi import APIRouter, Body, Depends, HTTPException, Request
from sqlalchemy import or_
from sqlalchemy.orm import Session
from datetime import datetime, timezone
from datetime import datetime, timezone, timedelta
from typing import Optional
import re
import logging
@@ -520,3 +521,157 @@ def get_coso_checklist(
if not data:
raise HTTPException(404, f"实体 entity={slug} 无 COSO 自检数据")
return data
@router.post("/coso-generate-actions")
def generate_coso_actions(
request: Request,
data: Optional[dict] = Body(default=None),
db: Session = Depends(get_db),
current_user: User = Depends(require_auth),
entity_id: int = Depends(get_entity_id),
):
"""COSO内控自检 → 生成改善行动(把自检「未通过」条目一键转为行动计划)
口径说明:
- 账套(多租户铁律):写入 entity_id 只取 token 绑定账套(get_entity_id),
body 里若含 entity_id 字段一律忽略;
- 数据源:与 get_coso_checklist 同口径(resolve_session_slug 按会话账套取同一份
自检)——屏幕上看到哪份自检,就用哪份生成,不做跨账套内容搬运;
- 关联 KPI:action_plans.kpi_id 是 NOT NULL 且列表按 KPI 归属做账套 join,
故必须挂本账套 KPI;
- 幂等:判重键 (entity_id, title),已存在的计入 skipped 不再 INSERT;
- auto_verify_rule 显式不生成:COSO 条目没有可复算的 KPI 判据。
"""
data = data or {}
# ① 取自检数据(与只读端点同口径解析 slug)
slug = resolve_session_slug(request, "hanke")
coso = COSO_CHECKLIST_DATA.get(slug)
if not coso:
raise HTTPException(404, f"实体 entity={slug} 无 COSO 自检数据")
# ② 待生成条目:passed 为真的一律不生成
failed_items = []
for el in coso["elements"]:
for item in el.get("items") or []:
if not item.get("passed"):
failed_items.append((el, item))
# ③ 无未通过条目 → 返回空结果结构(不算错误)
if not failed_items:
return {
"entity_id": entity_id,
"entity_slug": slug,
"kpi_id": None,
"kpi_name": None,
"failed_total": 0,
"created": 0,
"skipped": 0,
"created_ids": [],
"source": "coso",
}
# ④ 关联 KPI(必须,kpi_id NOT NULL 且列表按 KPI 归属做账套 join)
if data.get("kpi_id"):
# 显式指定:必须属于当前账套(跨账套 KPI 一律 404,不得跨账套写)
kpi = db.query(KPIDefinition).filter(
KPIDefinition.id == data["kpi_id"],
KPIDefinition.entity_id == entity_id,
).first()
if not kpi:
raise HTTPException(404, "关联KPI不存在")
else:
# 自动选取(判据显式、可解释):
# a. 名称/编码含关键词(按序:内控/合规/COSO/control/compliance),id 最小者优先
# b. 都没有 → 本账套 id 最小的 KPI(确定性兜底)
# c. 本账套一个 KPI 都没有 → 400(无法满足 kpi_id NOT NULL)
kpi = None
for kw in ("内控", "合规", "COSO", "control", "compliance"):
kpi = (db.query(KPIDefinition)
.filter(KPIDefinition.entity_id == entity_id)
.filter(or_(
KPIDefinition.kpi_name.like(f"%{kw}%"),
KPIDefinition.kpi_code.like(f"%{kw}%"),
))
.order_by(KPIDefinition.id)
.first())
if kpi:
break
if not kpi:
kpi = (db.query(KPIDefinition)
.filter(KPIDefinition.entity_id == entity_id)
.order_by(KPIDefinition.id)
.first())
if not kpi:
raise HTTPException(400, "当前账套没有可关联的KPI,无法生成改善行动")
# ⑤ 截止日期:body 显式传则校验格式且不得早于今天;否则默认 +30 天
due_date = None
if data.get("due_date"):
try:
due_date = datetime.strptime(str(data["due_date"]), "%Y-%m-%d")
except (ValueError, TypeError):
raise HTTPException(400, f"due_date 格式非法: {data['due_date']}(应为 YYYY-MM-DD)")
if due_date.date() < datetime.now().date():
raise HTTPException(400, "due_date 不能早于今天")
else:
due_date = datetime.now() + timedelta(days=30)
# ⑥ 幂等判重键 (entity_id, title):带账套过滤 —— 对端账套同标题行不抑制本账套创建
existing_titles = {
p.title for p in db.query(ActionPlan).filter(ActionPlan.entity_id == entity_id).all()
}
created_plans = []
skipped_count = 0
for el, item in failed_items:
title = f"[COSO] {el['name']}·{item['text']}"[:200] # title 列宽 String(200)
if title in existing_titles:
skipped_count += 1
continue
existing_titles.add(title) # 本批次内自去重
score = el.get("score") or 0
if score < 40:
priority = "high"
elif score < 60:
priority = "medium"
else:
priority = "low"
plan = ActionPlan(
entity_id=entity_id,
kpi_id=kpi.id,
title=title,
# 来源标注便于回溯(自检哪条条目生成了这条行动)
description=(
f"来源: COSO内控自检 / {el['name']}({el['id']})/ 条目 {item['id']}: {item['text']}\n"
f"自检结论: {item.get('detail') or '未通过'}"
),
status="pending",
progress=0,
priority=priority,
due_date=due_date,
created_by=current_user.name or current_user.username,
# 不传 auto_verify_rule:该列是 MySQL json 列,显式传 None 会被落成 JSON null
# (JSON_TYPE()='NULL'、IS NOT NULL 为真)而不是 SQL NULL,会让 bot_iron_law 的
# auto_verify_rule.isnot(None) 把「没有规则」的行算成「有规则」。不赋值则 INSERT
# 省略该列 → 保持 SQL NULL。COSO 条目没有可复算判据,引擎对空规则是跳过(安全)。
)
db.add(plan)
created_plans.append(plan)
# ⑦ 单次 add 循环 + 一次 commit:任一异常向上抛,不留半批
db.commit()
created_ids = [p.id for p in created_plans]
return {
"entity_id": entity_id,
"entity_slug": slug,
"kpi_id": kpi.id,
"kpi_name": kpi.kpi_name,
"failed_total": len(failed_items), # 自检未通过条目总数
"created": len(created_ids), # 本次新建条数
"skipped": skipped_count, # 已存在(幂等命中)条数
"created_ids": created_ids, # 新建行 id 列表
"source": "coso",
}