diff --git a/backend/app/api/audit_log.py b/backend/app/api/audit_log.py index 9754abb9..dcc812e9 100644 --- a/backend/app/api/audit_log.py +++ b/backend/app/api/audit_log.py @@ -134,7 +134,7 @@ def _extract_l3_rows(level, resp_body: bytes): return None if not isinstance(data, dict): return None - for key in ("imported", "kpi_updated"): + for key in ("imported", "kpi_updated", "upserted"): v = data.get(key) if isinstance(v, (int, float)): return int(v) diff --git a/backend/app/api/bot_bridge.py b/backend/app/api/bot_bridge.py index 2282d50c..3ce0c292 100644 --- a/backend/app/api/bot_bridge.py +++ b/backend/app/api/bot_bridge.py @@ -3,7 +3,7 @@ CMA BOT API桥接层 — 供财务BOT/店研学BOT调用 无需用户登录，使用 BOT API Key 认证 """ import os, json, logging -from fastapi import APIRouter, Depends, HTTPException, Query, Header, UploadFile, File +from fastapi import APIRouter, Depends, HTTPException, Query, Header, UploadFile, File, Request from sqlalchemy.orm import Session from sqlalchemy import func, desc from datetime import datetime @@ -798,6 +798,181 @@ def bot_kpi_create_with_links(data: dict, db: Session = Depends(get_db), bot: di "causality_links": links, "status": "ok"} +# ═══════════════ 智享通科目字典写入（zxt_account_dict 幂等 UPSERT）═══════════════ +# 2026-09-12 新增（ops-cma-dict-write-endpoint-20260912）：补齐 bot_bridge 字典表写入能力， +# 解除 ops 侧 zxt_dict_load.py / zxt_load_account_dict.py 直连管道的 API 迁移阻塞。 +# 分级 L3（批量写，与 /import 同级）；留痕走 audit middle（logs/bot_audit.log）+ operation_logs。 + +_ZXT_DICT_ALLOWED = ("CwAccountID", "subject_code", "subject_name", "zxt_code", "remark") +_ZXT_DICT_REQUIRED = ("CwAccountID",) +_ZXT_DICT_MAXLEN = {"CwAccountID": 20, "subject_code": 20, "subject_name": 200, + "zxt_code": 50, "remark": 200} +_ZXT_DICT_CODE_FIELDS = ("CwAccountID", "subject_code", "zxt_code") +_ZXT_DICT_MAX_ROWS = 5000 + + +def _validate_zxt_dict_rows(rows): + """全量先校验（任一错误 → 400，不落库），返回标准化行 [{字段: 字符串}]""" + import re + if not isinstance(rows, list) or not rows: + raise HTTPException(400, "请求体必须是非空的行数组：JSON [{...}] 或 multipart CSV 文件") + if len(rows) > _ZXT_DICT_MAX_ROWS: + raise HTTPException(400, f"单次批量上限 {_ZXT_DICT_MAX_ROWS} 行，本次 {len(rows)} 行") + + code_re = re.compile(r"^[A-Za-z0-9_.\-]+$") + clean, errs = [], [] + 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 _ZXT_DICT_ALLOWED] + if unknown: + errs.append(f"第{i}行: 非白名单字段 {unknown}，允许字段 {list(_ZXT_DICT_ALLOWED)}") + continue + item, bad = {}, False + for f in _ZXT_DICT_ALLOWED: + v = raw.get(f) + if v is None: + v = "" + elif isinstance(v, bool): + errs.append(f"第{i}行: {f} 类型非法（bool）") + bad = True + continue + elif isinstance(v, (int, float)): + v = str(int(v)) if float(v).is_integer() else str(v) + if not isinstance(v, str): + errs.append(f"第{i}行: {f} 必须是字符串（实际 {type(v).__name__}）") + bad = True + continue + v = v.strip() + if len(v) > _ZXT_DICT_MAXLEN[f]: + errs.append(f"第{i}行: {f} 超长（>{_ZXT_DICT_MAXLEN[f]}）") + bad = True + continue + if f in _ZXT_DICT_REQUIRED and not v: + errs.append(f"第{i}行: {f} 必填且不能为空") + bad = True + continue + if v and f in _ZXT_DICT_CODE_FIELDS and not code_re.match(v): + errs.append(f"第{i}行: {f} 含非法字符（仅限字母/数字/_/./-）：{v}") + bad = True + continue + item[f] = v + if bad: + continue + for f in _ZXT_DICT_ALLOWED: + item.setdefault(f, "") + clean.append(item) + + if errs: + raise HTTPException(400, detail={ + "message": f"参数校验失败（{len(errs)} 处），未写入任何数据", + "errors": errs[:20], + }) + return clean + + +async def _read_zxt_dict_body(request: Request): + """兼容两种入参: application/json（数组或 {"rows":[...]}）/ multipart CSV 文件（字段 file）""" + import csv, io + ctype = (request.headers.get("content-type") or "").lower() + if "multipart/form-data" in ctype: + form = await request.form() + up = None + for key in ("file", "csv", "upload"): + v = form.get(key) + if v is not None and hasattr(v, "read"): + up = v + break + if up is None: + for v in form.values(): + if hasattr(v, "read"): + up = v # type: ignore[assignment] + break + if up is None: + raise HTTPException(400, "multipart 请求缺少 CSV 文件字段（file）") + raw = up.read() + raw = await raw if hasattr(raw, "__await__") else raw + text_data = raw.decode("utf-8-sig", errors="replace") if isinstance(raw, (bytes, bytearray)) else str(raw) + reader = csv.DictReader(io.StringIO(text_data)) + if not reader.fieldnames: + raise HTTPException(400, "CSV 缺少表头行") + headers = [(h or "").strip() for h in reader.fieldnames] + unknown = [h for h in headers if h not in _ZXT_DICT_ALLOWED] + if unknown: + raise HTTPException(400, f"CSV 含非白名单列 {unknown}，允许列 {list(_ZXT_DICT_ALLOWED)}") + rows = [] + for r in reader: + if not any((v or "").strip() for v in r.values() if v is not None): + continue + rows.append({(k or "").strip(): ("" if v is None else v) for k, v in r.items()}) + if not rows: + raise HTTPException(400, "CSV 无数据行") + return {"rows": rows, "source": "csv"} + + try: + raw = await request.body() + data = json.loads(raw) if raw else None + except Exception as e: + raise HTTPException(400, f"JSON 解析失败: {e}") + if isinstance(data, dict) and isinstance(data.get("rows"), list): + data = data["rows"] + if not isinstance(data, list): + raise HTTPException(400, 'JSON 必须是行数组 [{"CwAccountID":...}] 或 {"rows":[...]}') + return {"rows": data, "source": "json"} + + +@router.post("/dict/zxt-account") +@risk_level("L3") +def bot_dict_zxt_account( + bot: dict = Depends(verify_bot_key), + payload: dict = Depends(_read_zxt_dict_body), + db: Session = Depends(get_db), +): + """智享通科目字典写入（zxt_account_dict 幂等 UPSERT，单事务） + + body 二选一: + 1) application/json: [{"CwAccountID","subject_code","subject_name","zxt_code"}] + 或 {"rows":[...]} + 2) multipart/form-data: file=（表头同字段名，utf-8/utf-8-sig） + 返回: {"ok":true,"upserted":N,"before":x,"after":y}；参数错误 400 且不落库，未授权 401。 + 凭据走后端 app.database（无硬编码）。 + """ + from sqlalchemy import text + clean = _validate_zxt_dict_rows(payload.get("rows")) + source = payload.get("source", "json") + try: + stmt = text( + "INSERT INTO zxt_account_dict " + "(CwAccountID, subject_code, subject_name, zxt_code, remark) " + "VALUES (:CwAccountID, :subject_code, :subject_name, :zxt_code, :remark) " + "ON DUPLICATE KEY UPDATE " + "subject_code=VALUES(subject_code), subject_name=VALUES(subject_name), " + "zxt_code=VALUES(zxt_code), remark=VALUES(remark)" + ) + before = db.execute(text("SELECT COUNT(*) FROM zxt_account_dict")).scalar() or 0 + db.execute(stmt, clean) + after = db.execute(text("SELECT COUNT(*) FROM zxt_account_dict")).scalar() or 0 + db.add(OperationLog( + action="import", target_type="zxt_account_dict", target_id=None, + detail={"bot": bot.get("name"), "endpoint": "/api/cma/bot/dict/zxt-account", + "upserted": len(clean), "before": before, "after": after, "source": source}, + )) + db.commit() + except HTTPException: + raise + except Exception as e: + db.rollback() + logger.error(f"zxt_account_dict upsert 失败: {e}") + raise HTTPException(500, f"字典写入失败: {e}") + + logger.info(f"zxt_account_dict UPSERT: bot={bot.get('name')} rows={len(clean)} " + f"before={before} after={after}") + return {"ok": True, "upserted": len(clean), "before": before, "after": after, + "inserted": max(after - before, 0), "updated": len(clean) - max(after - before, 0), + "table": "zxt_account_dict", "source": source} + + @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 1ebc7fa4..03cd6137 100644 --- a/backend/app/risk_levels.py +++ b/backend/app/risk_levels.py @@ -29,6 +29,7 @@ low/high 与 L1-L4 是两套正交分级，勿混: POST /api/cma/bot-bridge/kpi-result — 写KPI实际值 (source_batch) (L2) 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-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待办) @@ -89,10 +90,11 @@ 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 批量写/创建（3项） ── + # ── L3 批量写/创建（4项） ── "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） # ── L4 危险：Bot API面不存在（安全底线，不添加） ── } diff --git a/backend/tests/test_bot_bridge.py b/backend/tests/test_bot_bridge.py index 8697d2bb..b9932981 100644 --- a/backend/tests/test_bot_bridge.py +++ b/backend/tests/test_bot_bridge.py @@ -368,3 +368,51 @@ class TestOkrAndNlp: resp2 = client.get("/api/cma/bot/nlp?intent=总览", headers=BOT_KEY) assert resp2.status_code == 200 assert "overview" in resp2.json() + + +class TestDictZxtAccountWrite: + """智享通科目字典写入端点（POST /api/cma/bot/dict/zxt-account）负例 + + 说明: 正向 UPSERT 走 MySQL 专有 ON DUPLICATE KEY UPDATE，测试库为 SQLite + 无法覆盖——正向/幂等/落库一致性由生产 curl 实测（见任务验收记录）， + 此处覆盖鉴权与参数校验（校验先于任何 DB 访问，故 SQLite 可测）。 + """ + URL = "/api/cma/bot/dict/zxt-account" + + def test_no_key_401(self, client: TestClient): + resp = client.post(self.URL, json=[{"CwAccountID": "999", "subject_code": "1"}]) + assert resp.status_code == 401 + + def test_invalid_key_401(self, client: TestClient): + resp = client.post(self.URL, json=[{"CwAccountID": "999"}], + headers={"X-BOT-KEY": "wrong-key"}) + assert resp.status_code == 401 + + def test_non_whitelist_field_400(self, client: TestClient): + resp = client.post(self.URL, + json=[{"CwAccountID": "999", "entity_id": 1}], headers=BOT_KEY) + assert resp.status_code == 400 + assert "非白名单字段" in str(resp.json()) + + def test_missing_required_field_400(self, client: TestClient): + resp = client.post(self.URL, json=[{"subject_code": "1122"}], headers=BOT_KEY) + assert resp.status_code == 400 + assert "CwAccountID" in str(resp.json()) + + def test_illegal_value_400(self, client: TestClient): + resp = client.post(self.URL, json=[{"CwAccountID": "9 99;DROP"}], headers=BOT_KEY) + assert resp.status_code == 400 + + def test_empty_array_400(self, client: TestClient): + resp = client.post(self.URL, json=[], headers=BOT_KEY) + assert resp.status_code == 400 + + def test_non_array_json_400(self, client: TestClient): + resp = client.post(self.URL, json={"CwAccountID": "999"}, headers=BOT_KEY) + assert resp.status_code == 400 + + def test_rows_wrapper_accepted_shape(self, client: TestClient): + """{"rows":[...]} 包装形态被解析（校验失败仍 400，证明已进入校验而非形态拒绝）""" + resp = client.post(self.URL, json={"rows": [{"bad_field": 1}]}, headers=BOT_KEY) + assert resp.status_code == 400 + assert "非白名单字段" in str(resp.json()) diff --git a/backend/tests/test_risk_levels.py b/backend/tests/test_risk_levels.py index badea720..733666e5 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=3、L4=0""" + """分级统计与方案一致：L1=21(20项清单+risk-levels端点)、L2=6(4项清单+okr/create+bot建议通道)、L3=4(+字典写入)、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"] == 3, counts + assert counts["L3"] == 4, counts assert counts["L4"] == 0, "Bot API面不得存在L4端点（安全底线）" def test_no_l4_bot_endpoints(self): @@ -91,11 +91,12 @@ class TestRiskLevelsEndpoint: assert apis["POST /api/cma/bot/kpi-value-with-check"]["risk_level"] == "L2" 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 data["summary"]["L1"] == 21 assert data["summary"]["L2"] == 6 - assert data["summary"]["L3"] == 3 + assert data["summary"]["L3"] == 4 assert data["summary"]["L4"] == 0