diff --git a/backend/app/api/action_plans.py b/backend/app/api/action_plans.py index 46393f2b..0269587e 100644 --- a/backend/app/api/action_plans.py +++ b/backend/app/api/action_plans.py @@ -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", + } diff --git a/backend/tests/test_coso_generate_actions_20260921.py b/backend/tests/test_coso_generate_actions_20260921.py new file mode 100644 index 00000000..93ac96f5 --- /dev/null +++ b/backend/tests/test_coso_generate_actions_20260921.py @@ -0,0 +1,278 @@ +"""COSO内控自检「生成改善行动」端点测试 + +任务:project-20260921-coso-generate-actions +(自检「未通过」条目 → action_plans 写路径,前端「生成改善行动」按钮的后端实现) + +本文件覆盖 + ① 生成:e1(hanke)会话点击 → 200;created == 未通过条目数(从 COSO_CHECKLIST_DATA + 现算,不写字面量);新建行 entity_id=1 / status=pending / kpi 属于 e1 / + title 前缀 [COSO] / description 含「来源: COSO内控自检」/ auto_verify_rule is None; + passed=True 的条目没有对应行 + ② 幂等:同一会话再调一次 → created=0、skipped=上次 created、DB 行数不变 + ③ 账套隔离:e2(bohai)会话生成的行 entity_id=2;e1/e2 列表互不可见 + ④ 未认证:无 token → 401 + ⑤ 显式 kpi_id 归属校验:e1 会话传 e2 的 kpi_id → 404(不得跨账套写) + ⑥ 无未通过条目:monkeypatch 全 passed=True → 200 / created=0 / failed_total=0 / 不建行 +""" +import copy + +from fastapi.testclient import TestClient +from sqlalchemy.orm import Session + +from app.api import action_plans as ap +from app.models import ActionPlan, Entity, KPIDefinition +from tests.conftest import ( + TEST_USER_PASSWORD, + auth_header, + create_test_kpi, + create_test_user, + get_token_for_user, +) + +E1_KPI_CODE = "ZZ_COSO_E1_KPI_001" +E2_KPI_CODE = "ZZ_COSO_E2_KPI_001" + + +# ────────────────────────────────────────────── +# 夹具 +# ────────────────────────────────────────────── + +def _ensure_entity(db: Session, eid: int, name: str) -> Entity: + ent = db.query(Entity).filter(Entity.id == eid).first() + if not ent: + ent = Entity(id=eid, name=name, short_name=name, status="active") + db.add(ent) + db.commit() + db.refresh(ent) + return ent + + +def _login(client: TestClient, entity_id: int, username: str = "testadmin") -> str: + """按账套登录并返回 token(e2 会话用;e1 用 conftest 的 get_token_for_user)""" + resp = client.post("/api/cma/auth/login", json={ + "username": username, + "password": TEST_USER_PASSWORD, + "entity_id": entity_id, + }) + assert resp.status_code == 200, f"登录 entity{entity_id} 失败: {resp.text[:300]}" + return resp.json()["token"] + + +def _failed_items(coso: dict) -> list: + """自检未通过条目 [(element, item)](与端点同一口径:not item.get('passed'))""" + out = [] + for el in coso["elements"]: + for item in el.get("items") or []: + if not item.get("passed"): + out.append((el, item)) + return out + + +def _failed_titles(slug: str) -> set: + """按端点生成规则计算的标题集合([COSO] {el}·{item},截断 200)""" + coso = ap.COSO_CHECKLIST_DATA[slug] + return { + f"[COSO] {el['name']}·{item['text']}"[:200] + for el, item in _failed_items(coso) + } + + +def _seed_e1(db: Session) -> KPIDefinition: + """e1(酣客)基础夹具:实体 + ceo 用户 + 1 个本账套 KPI""" + _ensure_entity(db, 1, "酣客") + create_test_user(db) + return create_test_kpi(db, kpi_code=E1_KPI_CODE, kpi_name=f"{E1_KPI_CODE}-名称", entity_id=1) + + +# ────────────────────────────────────────────── +# ① 生成 +# ────────────────────────────────────────────── + +def test_generate_creates_rows_for_failed_items(client: TestClient, db: Session): + """生成:未通过条目各建一行、字段口径全部达标;passed=True 条目不建行 + + 断言依据:hanke 自检共 13 条(未通过 10 条)——条数用代码从 COSO_CHECKLIST_DATA + 现算而非写字面量,防自检文案/条目数将来调整时用例失真。 + """ + kpi = _seed_e1(db) + token = get_token_for_user(client) + h = auth_header(token) + + resp = client.post("/api/cma/action-plans/coso-generate-actions", headers=h, json={}) + assert resp.status_code == 200, resp.text[:300] + body = resp.json() + + coso = ap.COSO_CHECKLIST_DATA["hanke"] + failed = _failed_items(coso) + expected_titles = {f"[COSO] {el['name']}·{it['text']}"[:200] for el, it in failed} + passed_titles = { + f"[COSO] {el['name']}·{it['text']}"[:200] + for el in coso["elements"] + for it in (el.get("items") or []) + if it.get("passed") + } + + assert len(failed) > 0, "夹具失配:hanke 自检应存在未通过条目" + assert body["failed_total"] == len(failed), body + assert body["created"] == len(failed), body # 首轮无已存在行,created == 未通过总数 + assert body["skipped"] == 0, body + assert body["entity_id"] == 1, body + assert body["entity_slug"] == "hanke", body + assert body["kpi_id"] == kpi.id, body # 自动选取落在本账套唯一 KPI + assert body["source"] == "coso", body + + rows = db.query(ActionPlan).filter(ActionPlan.title.like("[COSO]%")).all() + assert len(rows) == body["created"] == len(body["created_ids"]), body + actual_titles = {r.title for r in rows} + assert actual_titles == expected_titles, "新建标题集合与未通过条目集合不一致" + assert not (actual_titles & passed_titles), "passed=True 条目不应生成对应行" + for r in rows: + assert r.entity_id == 1, r.title # 写库账套只来自 token(get_entity_id) + assert r.status == "pending", r.title + assert r.progress == 0, r.title + assert r.kpi_id == kpi.id, r.title + # kpi 归属校验:挂的 KPI 必须属于本账套(列表按 KPI 归属做账套 join 的口径) + kpi_ent = db.query(KPIDefinition).filter(KPIDefinition.id == r.kpi_id).first() + assert kpi_ent is not None and kpi_ent.entity_id == 1, r.title + assert r.title.startswith("[COSO]"), r.title + assert "来源: COSO内控自检" in (r.description or ""), r.title + assert r.auto_verify_rule is None, r.title # 显式不生成验证规则 + assert r.due_date is not None, r.title # 默认 +30 天 + + +# ────────────────────────────────────────────── +# ② 幂等 +# ────────────────────────────────────────────── + +def test_idempotent_second_call_skips_existing(client: TestClient, db: Session): + """幂等:同一会话第二次调用 created=0、skipped=首次 created、DB 行数不变 + + 判重键 (entity_id, title):重复点击「生成改善行动」不得重复造行。 + """ + _seed_e1(db) + h = auth_header(get_token_for_user(client)) + + first = client.post("/api/cma/action-plans/coso-generate-actions", headers=h, json={}) + assert first.status_code == 200 + body1 = first.json() + count_after_first = db.query(ActionPlan).count() + assert count_after_first == body1["created"] > 0 + + second = client.post("/api/cma/action-plans/coso-generate-actions", headers=h, json={}) + assert second.status_code == 200 + body2 = second.json() + assert body2["created"] == 0, body2 + assert body2["skipped"] == body1["created"], body2 + assert body2["failed_total"] == body1["failed_total"], body2 + assert db.query(ActionPlan).count() == count_after_first, "幂等调用不得新增行" + + +# ────────────────────────────────────────────── +# ③ 账套隔离 +# ────────────────────────────────────────────── + +def test_entity_isolation_e1_e2(client: TestClient, db: Session): + """账套隔离:e2 生成的行 entity_id=2 且挂 e2 KPI;e1/e2 列表互不可见 + + 两个账套存在同名标题(如「控制环境·授权审批制度」两个账套都未通过)—— + 这正好验证判重键带账套过滤:对端账套同标题行不得抑制本账套创建, + 列表也不能把对端行算进来。 + """ + _ensure_entity(db, 1, "酣客") + _ensure_entity(db, 2, "博海") + create_test_user(db) + k1 = create_test_kpi(db, kpi_code=E1_KPI_CODE, kpi_name="e1内控KPI", entity_id=1) + k2 = create_test_kpi(db, kpi_code=E2_KPI_CODE, kpi_name="e2内控KPI", entity_id=2) + + h1 = auth_header(_login(client, 1)) + h2 = auth_header(_login(client, 2)) + + r1 = client.post("/api/cma/action-plans/coso-generate-actions", headers=h1, json={}) + r2 = client.post("/api/cma/action-plans/coso-generate-actions", headers=h2, json={}) + assert r1.status_code == 200, r1.text[:300] + assert r2.status_code == 200, r2.text[:300] + + # e2 落库行:entity_id=2 且挂 e2 自己的 KPI + e2_rows = db.query(ActionPlan).filter(ActionPlan.entity_id == 2).all() + assert len(e2_rows) == r2.json()["created"] == len(_failed_items(ap.COSO_CHECKLIST_DATA["bohai"])) + for row in e2_rows: + assert row.entity_id == 2, row.title + assert row.kpi_id == k2.id, row.title + + # 列表互不可见:各自列表 == 各自自检未通过标题全集(含同名标题,也绝不含对端专属行) + e1_list = {row["title"] for row in client.get("/api/cma/action-plans", headers=h1).json()["data"]} + e2_list = {row["title"] for row in client.get("/api/cma/action-plans", headers=h2).json()["data"]} + assert e1_list == _failed_titles("hanke"), e1_list + assert e2_list == _failed_titles("bohai"), e2_list + e1_only = _failed_titles("hanke") - _failed_titles("bohai") + e2_only = _failed_titles("bohai") - _failed_titles("hanke") + assert not (e1_list & e2_only), "e1 列表不应出现 e2 专属条目" + assert not (e2_list & e1_only), "e2 列表不应出现 e1 专属条目" + # 差异集非空才说明上面的断言有鉴别力(两账套未通过集合确有重叠与差异) + assert e1_only and e2_only + + +# ────────────────────────────────────────────── +# ④ 未认证 +# ────────────────────────────────────────────── + +def test_unauthenticated_returns_401(client: TestClient, db: Session): + """未认证:无 token 调生成端点 → 401(写路径必须鉴权)""" + _seed_e1(db) + resp = client.post("/api/cma/action-plans/coso-generate-actions", json={}) + assert resp.status_code == 401, resp.text[:300] + + +# ────────────────────────────────────────────── +# ⑤ 显式 kpi_id 归属校验 +# ────────────────────────────────────────────── + +def test_explicit_kpi_id_must_belong_to_session_entity(client: TestClient, db: Session): + """显式 kpi_id 归属校验:e1 会话传 e2 的 kpi_id → 404 且不落任何行 + + kpi_id 是本账套隔离的唯一 join 锚点——若允许挂对端 KPI, + 本账套列表会按 KPI 归属 join 而丢失该行,形成「孤儿行」。 + """ + _ensure_entity(db, 1, "酣客") + _ensure_entity(db, 2, "博海") + create_test_user(db) + create_test_kpi(db, kpi_code=E1_KPI_CODE, kpi_name="e1 KPI", entity_id=1) + k2 = create_test_kpi(db, kpi_code=E2_KPI_CODE, kpi_name="e2 KPI", entity_id=2) + + h = auth_header(_login(client, 1)) + resp = client.post("/api/cma/action-plans/coso-generate-actions", + headers=h, json={"kpi_id": k2.id}) + assert resp.status_code == 404, resp.text[:300] + assert db.query(ActionPlan).count() == 0, "跨账套 kpi_id 被拒后不得残留任何行" + + +# ────────────────────────────────────────────── +# ⑥ 无未通过条目 +# ────────────────────────────────────────────── + +def test_no_failed_items_returns_zero_not_error(client: TestClient, db: Session, monkeypatch): + """无未通过条目:全 passed=True → 200 / created=0 / failed_total=0 / 不建行 + + monkeypatch 模块级 COSO_CHECKLIST_DATA(深拷贝,测后自动还原)—— + 端点必须把它当「空结果」而不是报错。 + """ + _seed_e1(db) + original = ap.COSO_CHECKLIST_DATA # patch 前持有原始引用,用于测后还原校验 + patched = copy.deepcopy(ap.COSO_CHECKLIST_DATA) + for entity_data in patched.values(): + for el in entity_data["elements"]: + for it in el["items"]: + it["passed"] = True + monkeypatch.setattr(ap, "COSO_CHECKLIST_DATA", patched) + + h = auth_header(get_token_for_user(client)) + resp = client.post("/api/cma/action-plans/coso-generate-actions", headers=h, json={}) + assert resp.status_code == 200, resp.text[:300] + body = resp.json() + assert body["failed_total"] == 0, body + assert body["created"] == 0 and body["skipped"] == 0, body + assert body["created_ids"] == [], body + assert db.query(ActionPlan).count() == 0, "无未通过条目时不得新建行" + + # 还原校验:patch 基于深拷贝,原始数据未被改坏(仍含未通过条目) + assert len(_failed_items(original["hanke"])) > 0 diff --git a/frontend/src/api/index.ts b/frontend/src/api/index.ts index a05991ec..60b6a1dc 100644 --- a/frontend/src/api/index.ts +++ b/frontend/src/api/index.ts @@ -202,6 +202,7 @@ export const actionPlanApi = { update: (id: number, data: any) => api.put(`/action-plans/${id}`, data), delete: (id: number) => api.delete(`/action-plans/${id}`), cosoChecklist: (params?: any) => api.get('/action-plans/coso-checklist', { params }), + cosoGenerateActions: (data?: any) => api.post('/action-plans/coso-generate-actions', data || {}), stats: () => api.get('/action-plans/stats'), } diff --git a/frontend/src/views/ActionPlanLibrary.vue b/frontend/src/views/ActionPlanLibrary.vue index 94af51fb..4f1530b5 100644 --- a/frontend/src/views/ActionPlanLibrary.vue +++ b/frontend/src/views/ActionPlanLibrary.vue @@ -294,7 +294,8 @@ {{ entityLabel }} 刷新 - 生成改善行动 + 生成改善行动({{ cosoFailedCount }})
@@ -667,6 +668,38 @@ async function loadCosoChecklist() { cosoLoading.value = false } +const cosoGenerating = ref(false) +// 自检未通过(passed=false)条目数——即「生成改善行动」将覆盖的范围 +const cosoFailedCount = computed(() => { + const els = cosoData.value?.elements || [] + return els.reduce((n: number, el: any) => n + (el.items || []).filter((it: any) => !it.passed).length, 0) +}) + +async function generateCosoActions() { + if (!cosoData.value) return + const n = cosoFailedCount.value + if (n === 0) { + ElMessage.info('当前自检没有未通过条目,无需生成改善行动') + return + } + try { + await ElMessageBox.confirm( + `将把自检中 ${n} 条未通过条目生成改善行动(已存在的不会重复生成),是否继续?`, + '生成改善行动' + ) + } catch { return } + cosoGenerating.value = true + try { + const r: any = await actionPlanApi.cosoGenerateActions({}) + const created = r?.created ?? 0 + const skipped = r?.skipped ?? 0 + ElMessage.success(`已生成 ${created} 条改善行动${skipped ? `,跳过 ${skipped} 条已存在` : ''}`) + await Promise.all([loadData(), loadStats()]) + if (created > 0) activeTab.value = 'plans' // 切到行动方案列表即可确认 + } catch { ElMessage.error('生成改善行动失败') } + cosoGenerating.value = false +} + // ── 初始化 ── onMounted(async () => { loadStats()