From 840d98765d0d1756fa55a1ef2a9f779e4c3ce169 Mon Sep 17 00:00:00 2001 From: Hermes CI Fix Date: Sat, 12 Sep 2026 10:42:46 +0800 Subject: [PATCH] =?UTF-8?q?feat(bot-bridge):=20=E6=96=B0=E5=A2=9E=E4=BC=9A?= =?UTF-8?q?=E8=AE=A1=E7=A7=91=E7=9B=AE=E5=86=99=E5=85=A5=E7=AB=AF=E7=82=B9?= =?UTF-8?q?=20POST=20/api/cma/bot/subjects=20=E2=80=94=20=E8=A1=8C?= =?UTF-8?q?=E7=BA=A7=E5=85=A8=E9=87=8F=E6=A0=A1=E9=AA=8C+=E6=8C=89subject?= =?UTF-8?q?=5Fcode=E5=B9=82=E7=AD=89UPSERT(dry=5Frun/update=5Fexisting?= =?UTF-8?q?=E5=BC=80=E5=85=B3)+L3=E7=95=99=E7=97=95,=20=E7=BB=9D=E4=B8=8DD?= =?UTF-8?q?ELETE=20(dama-cma-subjects-write-endpoint-20260912)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/api/bot_bridge.py | 254 +++++++++++++++++++++++++++++- backend/app/risk_levels.py | 4 +- backend/tests/test_bot_bridge.py | 141 ++++++++++++++++- backend/tests/test_risk_levels.py | 7 +- 4 files changed, 400 insertions(+), 6 deletions(-) diff --git a/backend/app/api/bot_bridge.py b/backend/app/api/bot_bridge.py index 3ce0c292..fb97b9a9 100644 --- a/backend/app/api/bot_bridge.py +++ b/backend/app/api/bot_bridge.py @@ -14,7 +14,7 @@ from app.models import ( User, UserEntity, StrategicMap, KPIDefinition, KPITemplate, KPIValue, DataSourceConfig, KPIAlert, OperationLog, NotificationChannel, NotificationLog, RolePermission, ActionPlan, OrgNode, - StrategicMapVersion, MapObjective, Objective, KR, + StrategicMapVersion, MapObjective, Objective, KR, Subject, ) from app.models.budget_plan import BudgetPlan from app.models.cost_model import StandardCost, ActualCost, AbcActivity, AbcAllocation @@ -973,6 +973,258 @@ def bot_dict_zxt_account( "table": "zxt_account_dict", "source": source} +# ═══════════════ 会计科目写入（subjects 幂等 UPSERT）═══════════════ +# 2026-09-12 新增（dama-cma-subjects-write-endpoint-20260912）：补齐 bot_bridge 科目表写入能力， +# 使 DAMA 管道 gen_inject_sql.py 的「科目新增」DML 走 API（原「财务侧 cma_user 直连写库」路径 +# 已随 2026-09-12 库访问治理失效：ERROR 1142 INSERT command denied）。 +# 语义铁律：按 subject_code 幂等 UPSERT，**绝不 DELETE / 绝不清空重建** +# （v1 ops 直连版为「清空重建」，已记录为语义差异，禁止继承）。 +# 分级 L3（批量写）；留痕走 audit middle（logs/bot_audit.log）+ operation_logs。 +# 表设计说明：subjects = 新30号准则会计科目字典，全局共享、无 entity_id 列 +# （见 cma-fullstack-ops「账套隔离」豁免说明），故本端点不做 entity 归属校验。 + +_SUBJ_FIELDS = ("subject_code", "subject_name", "parent_code", "level", "category", "is_active") +_SUBJ_CATEGORIES = ("asset", "liability", "equity", "cost", "income", "expense", "detail", "other") +_SUBJ_LEVELS = (1, 2, 3) +_SUBJ_MAX_ROWS = 2000 +_SUBJ_CODE_MAXLEN = 32 +_SUBJ_NAME_MAXLEN = 200 + + +def _norm_subj_active(v): + """is_active → (0/1, None) 或 (None, 错误信息)""" + if isinstance(v, bool): + return int(v), None + if isinstance(v, int): + return (v, None) if v in (0, 1) else (None, f"is_active 只允许 0/1（实际 {v}）") + if isinstance(v, str) and v.strip() in ("0", "1"): + return int(v.strip()), None + return None, f"is_active 只允许 0/1（实际 {v!r}）" + + +def _validate_subject_rows(rows, db): + """行级全量校验（任一错误 → 400 且零落库），返回标准化行 [{字段}] + + 规则（任务 dama-cma-subjects-write-endpoint-20260912）： + subject_code 非空且 ≤32 字符 + 批次内去重；subject_name 非空（DB NOT NULL）且 ≤200； + level ∈ {1,2,3}（缺省 1）；parent_code 为空或存在于本表（或本批次内，支持层级同批导入）； + category 限枚举；is_active ∈ {0,1}（缺省 1）。 + """ + if not isinstance(rows, list) or not rows: + raise HTTPException(400, '请求体必须是非空行数组: {"subjects": [{...}]}') + if len(rows) > _SUBJ_MAX_ROWS: + raise HTTPException(400, f"单次批量上限 {_SUBJ_MAX_ROWS} 行，本次 {len(rows)} 行") + + errs, clean, seen, parents = [], [], {}, set() + for i, raw in enumerate(rows, start=1): + if not isinstance(raw, dict): + errs.append(f"第{i}行: 行必须是对象（dict）") + continue + unknown = [k for k in raw.keys() if k not in _SUBJ_FIELDS] + if unknown: + errs.append(f"第{i}行: 非白名单字段 {unknown}，允许字段 {list(_SUBJ_FIELDS)}") + continue + item, bad = {}, False + + # subject_code + code = raw.get("subject_code") + code = "" if code is None else str(code).strip() + if not code: + errs.append(f"第{i}行: subject_code 必填且不能为空") + bad = True + elif len(code) > _SUBJ_CODE_MAXLEN: + errs.append(f"第{i}行: subject_code 超长（>{_SUBJ_CODE_MAXLEN} 字符）：{code}") + bad = True + elif code in seen: + errs.append(f"第{i}行: 批次内 subject_code 重复（第{seen[code]}行已出现）：{code}") + bad = True + else: + seen[code] = i + item["subject_code"] = code + + # subject_name + name = raw.get("subject_name") + name = "" if name is None else str(name).strip() + if not name: + errs.append(f"第{i}行: subject_name 必填且不能为空") + bad = True + elif len(name) > _SUBJ_NAME_MAXLEN: + errs.append(f"第{i}行: subject_name 超长（>{_SUBJ_NAME_MAXLEN} 字符）") + bad = True + item["subject_name"] = name + + # level + lvl = raw.get("level", 1) + if lvl is None: + lvl = 1 + if isinstance(lvl, str) and lvl.strip().lstrip("-").isdigit(): + lvl = int(lvl.strip()) + if isinstance(lvl, bool) or not isinstance(lvl, int): + errs.append(f"第{i}行: level 必须为整数 {list(_SUBJ_LEVELS)}（实际 {raw.get('level')!r}）") + bad = True + elif lvl not in _SUBJ_LEVELS: + errs.append(f"第{i}行: level 只允许 {list(_SUBJ_LEVELS)}（实际 {lvl}）") + bad = True + item["level"] = lvl if isinstance(lvl, int) and not isinstance(lvl, bool) else 1 + + # parent_code + pc = raw.get("parent_code") + pc = None if pc is None or str(pc).strip() == "" else str(pc).strip() + if pc is not None: + if len(pc) > _SUBJ_CODE_MAXLEN: + errs.append(f"第{i}行: parent_code 超长（>{_SUBJ_CODE_MAXLEN} 字符）：{pc}") + bad = True + elif pc == code: + errs.append(f"第{i}行: parent_code 不能等于自身 subject_code：{pc}") + bad = True + else: + parents.add(pc) + item["parent_code"] = pc + + # category + cat = raw.get("category") + cat = None if cat is None or str(cat).strip() == "" else str(cat).strip() + if cat is not None and cat not in _SUBJ_CATEGORIES: + errs.append(f"第{i}行: category 只允许 {list(_SUBJ_CATEGORIES)}（实际 {cat}）") + bad = True + item["category"] = cat + + # is_active（null/缺省 → 1） + act_raw = raw.get("is_active") + act, aerr = _norm_subj_active(1 if act_raw is None else act_raw) + if aerr: + errs.append(f"第{i}行: {aerr}") + bad = True + item["is_active"] = act if act is not None else 1 + + if not bad: + clean.append(item) + + # parent_code 存在性：库内（本表）∪ 本批次 + unknown_parents = sorted(parents - set(seen.keys())) + if unknown_parents: + found = {c for (c,) in db.query(Subject.subject_code) + .filter(Subject.subject_code.in_(unknown_parents)).all()} + missing = [p for p in unknown_parents if p not in found] + if missing: + errs.append(f"parent_code 不存在于 subjects 表且不在本批次内: {missing}") + + if errs: + raise HTTPException(400, detail={ + "message": f"参数校验失败（{len(errs)} 处），未写入任何数据", + "errors": errs[:20], + }) + return clean + + +async def _read_subjects_body(request: Request): + """解析请求体: {"subjects": [...], "dry_run": bool, "update_existing": bool} + 兼容裸数组 [...]（等价于 subjects 列表）。""" + try: + raw = await request.body() + data = json.loads(raw) if raw else None + except Exception as e: + raise HTTPException(400, f"JSON 解析失败: {e}") + dry_run, update_existing = False, False + if isinstance(data, dict): + if "dry_run" in data and not isinstance(data["dry_run"], bool): + raise HTTPException(400, "dry_run 必须是布尔值") + if "update_existing" in data and not isinstance(data["update_existing"], bool): + raise HTTPException(400, "update_existing 必须是布尔值") + dry_run = bool(data.get("dry_run")) + update_existing = bool(data.get("update_existing")) + rows = data.get("subjects") + else: + rows = data + if not isinstance(rows, list): + raise HTTPException(400, 'JSON 必须是 {"subjects": [...]} 或裸数组 [...]') + return {"rows": rows, "dry_run": dry_run, "update_existing": update_existing} + + +@router.post("/subjects") +@risk_level("L3") +def bot_subjects_upsert( + bot: dict = Depends(verify_bot_key), + payload: dict = Depends(_read_subjects_body), + db: Session = Depends(get_db), +): + """会计科目写入（subjects 幂等 UPSERT，单事务；绝不 DELETE） + + body: {"subjects": [{"subject_code","subject_name","parent_code","level","category","is_active"}], + "dry_run": false, "update_existing": false} + - 全量校验通过才落库；任一非法行 → 400 且库内零变化 + - 已存在行：update_existing=false → skip（幂等默认）；true → 更新 name/parent/level/category + - dry_run=true → 只校验与回显，零写入 + 返回: {"ok":true,"inserted":n,"updated":m,"skipped":k,"errors":[]}；未授权 401。 + """ + clean = _validate_subject_rows(payload.get("rows"), db) + dry_run = payload.get("dry_run", False) + update_existing = payload.get("update_existing", False) + codes = [r["subject_code"] for r in clean] + + existing = set() + if codes: + existing = {c for (c,) in db.query(Subject.subject_code) + .filter(Subject.subject_code.in_(codes)).all()} + to_insert = [r for r in clean if r["subject_code"] not in existing] + existed = [r for r in clean if r["subject_code"] in existing] + to_update = existed if update_existing else [] + to_skip = [] if update_existing else existed + + before = db.query(func.count(Subject.id)).scalar() or 0 + + if dry_run: + logger.info(f"subjects UPSERT dry_run: bot={bot.get('name')} rows={len(clean)} " + f"would_insert={len(to_insert)} would_update={len(to_update)} " + f"would_skip={len(to_skip)}") + return {"ok": True, "dry_run": True, "applied": False, "table": "subjects", + "before": before, "after": before, + "inserted": 0, "updated": 0, "skipped": 0, "upserted": 0, + "would_insert": len(to_insert), "would_update": len(to_update), + "would_skip": len(to_skip), "validated": len(clean), + "errors": [], "update_existing": update_existing} + + try: + for r in to_insert: + db.add(Subject(subject_code=r["subject_code"], subject_name=r["subject_name"], + parent_code=r["parent_code"], level=r["level"], + category=r["category"], is_active=r["is_active"])) + for r in to_update: + db.query(Subject).filter(Subject.subject_code == r["subject_code"]).update({ + Subject.subject_name: r["subject_name"], + Subject.parent_code: r["parent_code"], + Subject.level: r["level"], + Subject.category: r["category"], + }, synchronize_session=False) + db.flush() + after = db.query(func.count(Subject.id)).scalar() or 0 + db.add(OperationLog( + action="import", target_type="subjects", target_id=None, + detail={"bot": bot.get("name"), "endpoint": "/api/cma/bot/subjects", + "risk_level": "L3", "inserted": len(to_insert), "updated": len(to_update), + "skipped": len(to_skip), "before": before, "after": after, + "update_existing": update_existing, "source": "json"}, + )) + db.commit() + except HTTPException: + raise + except Exception as e: + db.rollback() + logger.error(f"subjects upsert 失败: {e}") + raise HTTPException(500, f"科目写入失败: {e}") + + if after != before + len(to_insert): + logger.warning(f"subjects UPSERT 行数口径差异: before={before} after={after} " + f"inserted={len(to_insert)}") + logger.info(f"subjects UPSERT: bot={bot.get('name')} inserted={len(to_insert)} " + f"updated={len(to_update)} skipped={len(to_skip)} before={before} after={after}") + return {"ok": True, "dry_run": False, "applied": True, "table": "subjects", + "before": before, "after": after, + "inserted": after - before, "updated": len(to_update), "skipped": len(to_skip), + "upserted": (after - before) + len(to_update), "errors": [], + "update_existing": update_existing} + + @router.get("/risk-levels") @risk_level("L1") def bot_risk_levels(bot: dict = Depends(verify_bot_key)): diff --git a/backend/app/risk_levels.py b/backend/app/risk_levels.py index 03cd6137..45fbe19d 100644 --- a/backend/app/risk_levels.py +++ b/backend/app/risk_levels.py @@ -30,6 +30,7 @@ low/high 与 L1-L4 是两套正交分级，勿混: POST /api/cma/bot-kpis/{kpi_id}/value — 写KPI当前值 (L2) POST /api/cma/bot/import — Excel批量导入KPI实际值 (L3) POST /api/cma/bot/dict/zxt-account — 智享通科目字典 UPSERT（zxt_account_dict） (L3, 2026-09-12 新增) + POST /api/cma/bot/subjects — 会计科目字典 UPSERT（subjects，幂等/绝不DELETE） (L3, 2026-09-12 新增) POST /api/cma/bot-bridge/mpm-result — MPM分析结果→KPI实际值 (L3) POST /api/cma/bot-bridge/verify/{action_plan_id} — 验证ActionPlan执行 (L2, 轻量verify_log) POST /api/cma/bot/okr/create — 创建OKR Objective (L2, 管理类业务写，保留评审: 2026-09-03 P1待办) @@ -90,11 +91,12 @@ API_RISK_MAP = { "POST /api/cma/bot-bridge/verify/{action_plan_id}": "L2", "POST /api/cma/bot/okr/create": "L2", # 单条OKR创建（方案清单未列出，按单条业务写归类） "POST /api/cma/bot/ai/suggestions": "L2", # Bot高危建议创建通道（写操作语义 high，见文件头注释；本身只建建议不直写考核基准） - # ── L3 批量写/创建（4项） ── + # ── L3 批量写/创建（5项） ── "POST /api/cma/bot/import": "L3", "POST /api/cma/bot/kpis/create-with-links": "L3", "POST /api/cma/bot-bridge/mpm-result": "L3", "POST /api/cma/bot/dict/zxt-account": "L3", # 智享通科目字典 UPSERT（2026-09-12 ops-cma-dict-write-endpoint） + "POST /api/cma/bot/subjects": "L3", # 会计科目字典 UPSERT（2026-09-12 dama-cma-subjects-write-endpoint） # ── L4 危险：Bot API面不存在（安全底线，不添加） ── } diff --git a/backend/tests/test_bot_bridge.py b/backend/tests/test_bot_bridge.py index b9932981..ced2d6bc 100644 --- a/backend/tests/test_bot_bridge.py +++ b/backend/tests/test_bot_bridge.py @@ -16,7 +16,7 @@ from sqlalchemy.orm import Session from app.models import ( User, KPIDefinition, KPIValue, KPIAlert, StrategicMap, MapObjective, - ActionPlan, OrgNode, DataSourceConfig, Objective, + ActionPlan, OrgNode, DataSourceConfig, Objective, Subject, ) from app.models.budget_plan import BudgetPlan from app.models.cost_model import StandardCost, ActualCost @@ -416,3 +416,142 @@ class TestDictZxtAccountWrite: resp = client.post(self.URL, json={"rows": [{"bad_field": 1}]}, headers=BOT_KEY) assert resp.status_code == 400 assert "非白名单字段" in str(resp.json()) + + +class TestSubjectsWrite: + """会计科目写入端点（POST /api/cma/bot/subjects）— 鉴权/校验/dry_run/幂等/更新 + + 实现为纯 ORM（无 MySQL 专有 SQL），故正向/幂等路径可在 SQLite 测试库完整覆盖。 + """ + + URL = "/api/cma/bot/subjects" + + @staticmethod + def _row(code, name, **kw): + row = {"subject_code": code, "subject_name": name, "parent_code": None, + "level": 1, "category": "equity", "is_active": 1} + row.update(kw) + return row + + # ── 鉴权 ── + def test_no_key_401(self, client: TestClient): + resp = client.post(self.URL, json={"subjects": [self._row("9001", "验证科目")]}) + assert resp.status_code == 401 + + def test_invalid_key_401(self, client: TestClient): + resp = client.post(self.URL, json={"subjects": [self._row("9001", "验证科目")]}, + headers={"X-BOT-KEY": "wrong-key"}) + assert resp.status_code == 401 + + # ── 参数校验（全批拒绝、零落库）── + def test_level_out_of_range_400(self, client: TestClient, db: Session): + resp = client.post(self.URL, json={"subjects": [self._row("9001", "验证科目", level=9)]}, + headers=BOT_KEY) + assert resp.status_code == 400 + assert "level" in str(resp.json()) + assert db.query(Subject).filter(Subject.subject_code == "9001").count() == 0 + + def test_batch_duplicate_code_400(self, client: TestClient, db: Session): + resp = client.post(self.URL, json={"subjects": [ + self._row("9002", "验证科目A"), self._row("9002", "验证科目B")]}, headers=BOT_KEY) + assert resp.status_code == 400 + assert "重复" in str(resp.json()) + assert db.query(Subject).filter(Subject.subject_code == "9002").count() == 0 + + def test_parent_not_exists_400(self, client: TestClient, db: Session): + resp = client.post(self.URL, json={"subjects": [ + self._row("900301", "验证子科目", parent_code="999999", level=2)]}, headers=BOT_KEY) + assert resp.status_code == 400 + assert "parent_code" in str(resp.json()) + assert db.query(Subject).filter(Subject.subject_code == "900301").count() == 0 + + def test_illegal_category_400(self, client: TestClient): + resp = client.post(self.URL, json={"subjects": [ + self._row("9003", "验证科目", category="not_a_category")]}, headers=BOT_KEY) + assert resp.status_code == 400 + assert "category" in str(resp.json()) + + def test_illegal_is_active_400(self, client: TestClient): + resp = client.post(self.URL, json={"subjects": [ + self._row("9004", "验证科目", is_active=7)]}, headers=BOT_KEY) + assert resp.status_code == 400 + assert "is_active" in str(resp.json()) + + def test_empty_subjects_400(self, client: TestClient): + resp = client.post(self.URL, json={"subjects": []}, headers=BOT_KEY) + assert resp.status_code == 400 + + def test_non_whitelist_field_400(self, client: TestClient): + resp = client.post(self.URL, json={"subjects": [ + self._row("9005", "验证科目", entity_id=1)]}, headers=BOT_KEY) + assert resp.status_code == 400 + assert "非白名单字段" in str(resp.json()) + + def test_missing_name_400(self, client: TestClient): + resp = client.post(self.URL, json={"subjects": [ + {"subject_code": "9006", "level": 1}]}, headers=BOT_KEY) + assert resp.status_code == 400 + assert "subject_name" in str(resp.json()) + + # ── 落库语义 ── + def test_dry_run_zero_write(self, client: TestClient, db: Session): + before = db.query(Subject).count() + resp = client.post(self.URL, json={"subjects": [self._row("9101", "验证-干跑")], + "dry_run": True}, headers=BOT_KEY) + assert resp.status_code == 200 + data = resp.json() + assert data["dry_run"] is True and data["applied"] is False + assert data["inserted"] == 0 and data["upserted"] == 0 + assert data["would_insert"] == 1 + db.expire_all() + assert db.query(Subject).count() == before + assert db.query(Subject).filter(Subject.subject_code == "9101").count() == 0 + + def test_insert_skip_counts_and_idempotent(self, client: TestClient, db: Session): + db.add(Subject(subject_code="9201", subject_name="既有科目", level=1, + category="equity", is_active=1)) + db.commit() + before = db.query(Subject).count() + + payload = {"subjects": [self._row("9201", "既有科目"), self._row("9202", "新增科目A"), + self._row("9203", "新增科目B")]} + resp = client.post(self.URL, json=payload, headers=BOT_KEY) + assert resp.status_code == 200 + d1 = resp.json() + assert (d1["inserted"], d1["updated"], d1["skipped"]) == (2, 0, 1), d1 + assert d1["after"] == before + 2 + + db.expire_all() + assert db.query(Subject).filter(Subject.subject_code.in_(["9202", "9203"])).count() == 2 + + # 幂等：同 payload 二次调用 → 全部 skip，行数不变、无重复 + resp2 = client.post(self.URL, json=payload, headers=BOT_KEY) + assert resp2.status_code == 200 + d2 = resp2.json() + assert (d2["inserted"], d2["updated"], d2["skipped"]) == (0, 0, 3), d2 + assert d2["before"] == d2["after"] == before + 2 + db.expire_all() + assert db.query(Subject).count() == before + 2 + assert db.query(Subject).filter(Subject.subject_code == "9202").count() == 1 + + def test_update_existing_switch(self, client: TestClient, db: Session): + db.add(Subject(subject_code="9301", subject_name="旧名称", level=1, + category="equity", is_active=1)) + db.commit() + resp = client.post(self.URL, json={ + "subjects": [self._row("9301", "新名称", category="cost", level=2)], + "update_existing": True}, headers=BOT_KEY) + assert resp.status_code == 200 + d = resp.json() + assert (d["inserted"], d["updated"], d["skipped"]) == (0, 1, 0), d + db.expire_all() + row = db.query(Subject).filter(Subject.subject_code == "9301").first() + assert row.subject_name == "新名称" and row.category == "cost" and row.level == 2 + + def test_batch_parent_within_batch_ok(self, client: TestClient, db: Session): + """父科目在本批次内（层级同批导入）允许；父科目不存在则拒绝""" + resp = client.post(self.URL, json={"subjects": [ + self._row("9401", "父科目"), self._row("940101", "子科目", parent_code="9401", level=2)], + }, headers=BOT_KEY) + assert resp.status_code == 200, resp.json() + assert resp.json()["inserted"] == 2 diff --git a/backend/tests/test_risk_levels.py b/backend/tests/test_risk_levels.py index 733666e5..abb25d26 100644 --- a/backend/tests/test_risk_levels.py +++ b/backend/tests/test_risk_levels.py @@ -49,12 +49,12 @@ class TestRiskMapCoverage: assert not unlabeled, f"以下路由函数缺少 @risk_level 装饰器: {unlabeled}" def test_risk_level_counts(self): - """分级统计与方案一致：L1=21(20项清单+risk-levels端点)、L2=6(4项清单+okr/create+bot建议通道)、L3=4(+字典写入)、L4=0""" + """分级统计与方案一致：L1=21(20项清单+risk-levels端点)、L2=6(4项清单+okr/create+bot建议通道)、L3=5(+字典写入+科目写入)、L4=0""" from collections import Counter counts = Counter(API_RISK_MAP.values()) assert counts["L1"] == 21, counts assert counts["L2"] == 6, counts - assert counts["L3"] == 4, counts + assert counts["L3"] == 5, counts assert counts["L4"] == 0, "Bot API面不得存在L4端点（安全底线）" def test_no_l4_bot_endpoints(self): @@ -92,11 +92,12 @@ class TestRiskLevelsEndpoint: assert apis["POST /api/cma/bot/import"]["risk_level"] == "L3" assert apis["POST /api/cma/bot/import"]["handling"] # 处理方式非空 assert apis["POST /api/cma/bot/dict/zxt-account"]["risk_level"] == "L3" + assert apis["POST /api/cma/bot/subjects"]["risk_level"] == "L3" # 分级统计 assert data["summary"]["L1"] == 21 assert data["summary"]["L2"] == 6 - assert data["summary"]["L3"] == 4 + assert data["summary"]["L3"] == 5 assert data["summary"]["L4"] == 0