feat: Bot API L1-L4风险分级标注 + 操作审计日志

- 新增 app/risk_levels.py: RISK_LEVELS定义 + @risk_level装饰器 + API_RISK_MAP
  (L1只读21 / L2业务写5 / L3批量写3 / L4=0安全底线)
- 新增 app/api/audit_log.py: Bot API审计中间件 → backend/logs/bot_audit.log
  (JSON行: timestamp/bot_name/endpoint/method/risk_level/entity_id/status,
   L3额外记rows行数, 不阻塞业务)
- bot_bridge/bot_bridge_v2/bot_kpis/bot_iron_law 全部29路由标注级别
- 新增 GET /api/cma/bot/risk-levels (X-BOT-KEY鉴权): API→级别→处理方式清单
- main.py 注册审计中间件
- tests/test_risk_levels.py: 覆盖路由标注/risk-levels端点/L4不存在/审计日志

backend/app/api/audit_log.py

@@ -0,0 +1,226 @@
+"""
+Bot API 操作审计中间件
+======================
+
+每次 Bot API 调用（/api/cma/bot*）记录一行 JSON 审计日志到
+backend/logs/bot_audit.log（可用环境变量 CMA_BOT_AUDIT_LOG 覆盖路径）。
+
+JSON 行字段: timestamp / bot_name / endpoint / method / risk_level /
+            entity_id / status
+L3 批量写额外记录 rows（导入行数 / 批量条数）。
+
+设计约束:
+  - 不阻塞业务: 所有日志写入失败仅静默跳过，不影响请求结果
+  - 只读 body（starlette 会缓存并回放给下游），仅在 application/json 时解析
+  - 不改动任何现有 API 行为与多租户隔离
+"""
+import json
+import logging
+import os
+from datetime import datetime
+
+from fastapi import Request
+
+_LOG_DIR = os.path.join(
+    os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))),
+    "logs",
+)
+_DEFAULT_LOG_PATH = os.path.join(_LOG_DIR, "bot_audit.log")
+
+_audit_logger = None
+
+
+def _get_logger():
+    """构造/复用审计 logger（路径取 CMA_BOT_AUDIT_LOG 覆盖值，便于测试隔离）"""
+    global _audit_logger
+    if _audit_logger is None:
+        log_path = os.getenv("CMA_BOT_AUDIT_LOG") or _DEFAULT_LOG_PATH
+        os.makedirs(os.path.dirname(log_path), exist_ok=True)
+        logger = logging.getLogger("cma.bot_audit")
+        logger.setLevel(logging.INFO)
+        logger.propagate = False
+        # 清掉旧 handler，避免日志名单例导致路径切换失效
+        for h in list(logger.handlers):
+            logger.removeHandler(h)
+            try:
+                h.close()
+            except Exception:
+                pass
+        handler = logging.FileHandler(log_path, encoding="utf-8")
+        handler.setFormatter(logging.Formatter("%(message)s"))
+        logger.addHandler(handler)
+        _audit_logger = logger
+    return _audit_logger
+
+
+def write_audit_line(record: dict):
+    """写一行 JSON 审计日志；任何异常都静默（不阻塞业务）"""
+    try:
+        _get_logger().info(json.dumps(record, ensure_ascii=False))
+    except Exception:
+        pass
+
+
+def _resolve_bot_name(request: Request) -> str:
+    """解析调用方标识: X-BOT-KEY → 配置名/原始key；bridge → bot-bridge；其余 → web-user/unknown"""
+    key = request.headers.get("X-BOT-KEY")
+    if key:
+        try:
+            from app.api.bot_bridge import _BOT_API_KEYS
+            info = _BOT_API_KEYS.get(key)
+            if isinstance(info, dict) and info.get("name"):
+                return info["name"]
+        except Exception:
+            pass
+        return key
+    if request.headers.get("X-BRIDGE-TOKEN"):
+        return "bot-bridge"
+    if (request.headers.get("Authorization") or "").startswith("Bearer "):
+        return "web-user"
+    return "unknown"
+
+
+def _resolve_risk_level(request: Request, route=None, endpoint=None):
+    """级别解析顺序: 路由函数装饰器标注 → API_RISK_MAP(route.path) → API_RISK_MAP(请求URL)"""
+    if endpoint is not None:
+        level = getattr(endpoint, "risk_level", None)
+        if level:
+            return level
+    if route is not None:
+        from app.risk_levels import get_risk_level
+        path = getattr(route, "path", None)
+        if path:
+            for m in getattr(route, "methods", set()) or set():
+                if m in ("GET", "POST", "PUT", "DELETE", "PATCH"):
+                    level = get_risk_level(m, path)
+                    if level:
+                        return level
+    try:
+        from app.risk_levels import get_risk_level
+        return get_risk_level(request.method, request.url.path)
+    except Exception:
+        return None
+
+
+async def _extract_entity_id(request: Request):
+    """多租户 entity_id 提取（尽力而为）: 查询参数 → JSON body（仅 application/json）"""
+    try:
+        q = request.query_params.get("entity_id")
+        if q is not None and str(q) != "":
+            return int(q) if str(q).isdigit() else q
+    except Exception:
+        pass
+    ctype = (request.headers.get("content-type") or "").lower()
+    if "application/json" in ctype:
+        try:
+            raw = await request.body()
+            if raw:
+                data = json.loads(raw)
+                eid = data.get("entity_id")
+                if eid is not None:
+                    return int(eid) if str(eid).isdigit() else eid
+        except Exception:
+            pass
+    return None
+
+
+def _extract_l3_rows(level, resp_body: bytes):
+    """L3 批量写: 从响应体提取行数/条数（imported / kpi_updated / causality_links / total）"""
+    if level != "L3" or not resp_body:
+        return None
+    try:
+        data = json.loads(resp_body)
+    except Exception:
+        return None
+    if not isinstance(data, dict):
+        return None
+    for key in ("imported", "kpi_updated"):
+        v = data.get(key)
+        if isinstance(v, (int, float)):
+            return int(v)
+    links = data.get("causality_links")
+    if isinstance(links, list):
+        return len(links)
+    total = data.get("total")
+    if isinstance(total, (int, float)):
+        return int(total)
+    return None
+
+
+async def bot_audit_middleware(request: Request, call_next):
+    """HTTP 中间件: 仅审计 /api/cma/bot* 路径；任何异常不影响业务"""
+    path = request.url.path
+    if not path.startswith("/api/cma/bot"):
+        return await call_next(request)
+
+    entity_id = None
+    try:
+        entity_id = await _extract_entity_id(request)
+    except Exception:
+        entity_id = None
+
+    try:
+        response = await call_next(request)
+    except Exception:
+        # 兜底记录（全局异常handler会返回500，此处保证审计不丢）
+        try:
+            write_audit_line({
+                "timestamp": datetime.now().isoformat(timespec="seconds"),
+                "bot_name": _resolve_bot_name(request),
+                "endpoint": path,
+                "method": request.method,
+                "risk_level": _resolve_risk_level(request) or "NA",
+                "entity_id": entity_id,
+                "status": 500,
+            })
+        except Exception:
+            pass
+        raise
+
+    # 级别解析（route 由路由器在 call_next 内写入 scope）
+    level = None
+    try:
+        route = request.scope.get("route")
+        endpoint = getattr(route, "endpoint", None) if route else None
+        level = _resolve_risk_level(request, route=route, endpoint=endpoint)
+    except Exception:
+        level = None
+
+    # 捕获响应体（L3 需要行数），并重放
+    resp_body = b""
+    try:
+        body_iterator = getattr(response, "body_iterator", None)
+        if body_iterator is not None:
+            chunks = [chunk async for chunk in body_iterator]
+            resp_body = b"".join(chunks)
+        rows = _extract_l3_rows(level, resp_body)
+    except Exception:
+        rows = None
+
+    record = {
+        "timestamp": datetime.now().isoformat(timespec="seconds"),
+        "bot_name": _resolve_bot_name(request),
+        "endpoint": path,
+        "method": request.method,
+        "risk_level": level or "NA",
+        "entity_id": entity_id,
+        "status": response.status_code,
+    }
+    if rows is not None:
+        record["rows"] = rows
+
+    try:
+        write_audit_line(record)
+    except Exception:
+        pass
+
+    # 响应体已被消费 → 重建响应（Bot接口均为小JSON，非流式）
+    if resp_body:
+        from fastapi.responses import Response
+        return Response(
+            content=resp_body,
+            status_code=response.status_code,
+            headers=dict(response.headers),
+            media_type=response.media_type,
+        )
+    return response

backend/app/api/bot_bridge.py

@@ -18,6 +18,7 @@ from app.models import (
 from app.models.budget_plan import BudgetPlan
 from app.models.cost_model import StandardCost, ActualCost, AbcActivity, AbcAllocation
 from app.models import KPICausality
+from app.risk_levels import risk_level
 import json
 
 logger = logging.getLogger("cma.bot_bridge")
@@ -79,6 +80,7 @@ def _model_dict(obj, fields: dict):
 # ═══════════════ 端点 ═══════════════
 
 @router.get("/ping")
+@risk_level("L1")
 def ping():
     return {"status": "ok", "version": "1.0", "timestamp": datetime.now().isoformat()}
 
@@ -86,6 +88,7 @@ def ping():
 # ── 总览 ──
 
 @router.get("/overview")
+@risk_level("L1")
 def bot_overview(
     bot: dict = Depends(verify_bot_key),
     db: Session = Depends(get_db),
@@ -110,6 +113,7 @@ def bot_overview(
 # ── KPI ──
 
 @router.get("/kpis")
+@risk_level("L1")
 def bot_kpis(
     dimension: Optional[str] = Query(None),
     status: str = Query("active"),
@@ -145,6 +149,7 @@ def bot_kpis(
 
 
 @router.get("/kpis/{kpi_id}/history")
+@risk_level("L1")
 def bot_kpi_history(
     kpi_id: int, limit: int = Query(12, le=60),
     bot: dict = Depends(verify_bot_key),
@@ -171,6 +176,7 @@ def bot_kpi_history(
 # ── 战略地图 ──
 
 @router.get("/strategic-maps")
+@risk_level("L1")
 def bot_maps(
     bot: dict = Depends(verify_bot_key),
     db: Session = Depends(get_db),
@@ -198,6 +204,7 @@ def bot_maps(
 # ── 预警 ──
 
 @router.get("/alerts")
+@risk_level("L1")
 def bot_alerts(
     status: str = Query("pending"),
     level: Optional[str] = Query(None),
@@ -228,6 +235,7 @@ def bot_alerts(
 # ── 预算 ──
 
 @router.get("/budget/plans")
+@risk_level("L1")
 def bot_budget_plans(
     year: Optional[int] = Query(None),
     bot: dict = Depends(verify_bot_key),
@@ -255,6 +263,7 @@ def bot_budget_plans(
 # ── 成本 ──
 
 @router.get("/cost/standard")
+@risk_level("L1")
 def bot_standard_costs(
     bot: dict = Depends(verify_bot_key),
     db: Session = Depends(get_db),
@@ -278,6 +287,7 @@ def bot_standard_costs(
 
 
 @router.get("/cost/actual")
+@risk_level("L1")
 def bot_actual_costs(
     period: Optional[str] = Query(None),
     bot: dict = Depends(verify_bot_key),
@@ -306,6 +316,7 @@ def bot_actual_costs(
 # ── 行动方案 ──
 
 @router.get("/actions")
+@risk_level("L1")
 def bot_actions(
     status: Optional[str] = Query(None),
     bot: dict = Depends(verify_bot_key),
@@ -333,6 +344,7 @@ def bot_actions(
 # ── 组织 ──
 
 @router.get("/organization")
+@risk_level("L1")
 def bot_org(
     bot: dict = Depends(verify_bot_key),
     db: Session = Depends(get_db),
@@ -354,6 +366,7 @@ def bot_org(
 # ── 数据源 ──
 
 @router.get("/data-sources")
+@risk_level("L1")
 def bot_data_sources(
     bot: dict = Depends(verify_bot_key),
     db: Session = Depends(get_db),
@@ -377,6 +390,7 @@ def bot_data_sources(
 # ── 用户 ──
 
 @router.get("/users")
+@risk_level("L1")
 def bot_users(
     bot: dict = Depends(verify_bot_key),
     db: Session = Depends(get_db),
@@ -395,6 +409,7 @@ def bot_users(
 # ── 统一查询（BOT首选） ──
 
 @router.get("/query")
+@risk_level("L1")
 def bot_query(
     q: str = Query("overview", description="overview/kpis/alerts/maps/budget/cost/actions/all"),
     bot: dict = Depends(verify_bot_key),
@@ -482,6 +497,7 @@ def bot_query(
 
 
 @router.post("/import")
+@risk_level("L3")
 def bot_import_excel(
     file: UploadFile = File(...),
     bot: dict = Depends(verify_bot_key),
@@ -546,6 +562,7 @@ def bot_import_excel(
 # ── 自然语言查询 ──
 
 @router.post("/okr/create")
+@risk_level("L2")
 def bot_okr_create(
     title: str = Query(...),
     quarter: str = Query(...),
@@ -563,6 +580,7 @@ def bot_okr_create(
 
 
 @router.get("/okr/list")
+@risk_level("L1")
 def bot_okr_list(
     quarter: Optional[str] = Query(None),
     bot: dict = Depends(verify_bot_key),
@@ -584,6 +602,7 @@ def bot_okr_list(
 
 
 @router.get("/nlp")
+@risk_level("L1")
 def bot_nlp(
     intent: str = Query("overview"),
     bot: dict = Depends(verify_bot_key),
@@ -613,6 +632,7 @@ def bot_nlp(
 # ════════════════════════════════════════════════════════════
 
 @router.post("/kpi-value-with-check")
+@risk_level("L2")
 def bot_kpi_value_with_check(data: dict, db: Session = Depends(get_db), bot: dict = Depends(verify_bot_key)):
     """聚合A: 写KPI值 + 自动跑该KPI预警检查（Agent一次调用，免自拼check-all）
     body: {kpi_id, actual_value, period?, entity_id?, run_check?}"""
@@ -681,6 +701,7 @@ def bot_kpi_value_with_check(data: dict, db: Session = Depends(get_db), bot: dic
 
 
 @router.post("/kpis/create-with-links")
+@risk_level("L3")
 def bot_kpi_create_with_links(data: dict, db: Session = Depends(get_db), bot: dict = Depends(verify_bot_key)):
     """聚合B: 创建KPI + 关联战略地图 + 批量因果链（Agent建KPI标准动作）
     body: {kpi_code, kpi_name, dimension, entity_id?, target_value?, unit?, link_map_id?, link_causality?}"""
@@ -736,3 +757,18 @@ def bot_kpi_create_with_links(data: dict, db: Session = Depends(get_db), bot: di
     db.commit()
     return {"kpi_id": kpi.id, "kpi_code": kpi.kpi_code, "map_id": link_map_id,
             "causality_links": links, "status": "ok"}
+
+
+@router.get("/risk-levels")
+@risk_level("L1")
+def bot_risk_levels(bot: dict = Depends(verify_bot_key)):
+    """Bot API风险分级清单（API→级别→处理方式）— 验收/巡检/授权决策用"""
+    from app.risk_levels import RISK_LEVELS, list_api_risk_map, risk_summary
+    return {
+        "bot": bot["name"],
+        "risk_levels": RISK_LEVELS,
+        "apis": list_api_risk_map(),
+        "summary": risk_summary(),
+        "note": "L4（危险）不向Bot API开放：Bot只能通过白名单API读写，"
+                "DROP/TRUNCATE/批量DELETE/生产结构修改物理不可能，由终端层+approval-gate拦截",
+    }

backend/app/api/bot_bridge_v2.py

@@ -23,6 +23,7 @@ from app.models import (
     KPIDefinition, KPIValue, KPIAlert,
     ActionPlan, Entity,
 )
+from app.risk_levels import risk_level
 
 logger = logging.getLogger("cma.bot_bridge_v2")
 
@@ -279,6 +280,7 @@ ALERT_THRESHOLDS = {
 # ═══════════════════════════════════════════════
 
 @router.post("/mpm-result")
+@risk_level("L3")
 def receive_mpm_result(
     data: dict,
     bridge_bot: str = Depends(verify_bridge_token),
@@ -404,6 +406,7 @@ def receive_mpm_result(
 
 
 @router.post("/kpi-result")
+@risk_level("L2")
 def push_kpi_result(
     data: dict,
     bridge_bot: str = Depends(verify_bridge_token),
@@ -521,6 +524,7 @@ def push_kpi_result(
 
 
 @router.post("/verify/{action_plan_id}")
+@risk_level("L2")
 def verify_action_plan(
     action_plan_id: int,
     bridge_bot: str = Depends(verify_bridge_token),
@@ -599,6 +603,7 @@ def verify_action_plan(
 
 
 @router.get("/verify/{action_plan_id}/history")
+@risk_level("L1")
 def verify_history(
     action_plan_id: int,
     bridge_bot: str = Depends(verify_bridge_token),

backend/app/api/bot_iron_law.py

@@ -9,6 +9,7 @@ from sqlalchemy.orm import Session
 from sqlalchemy import func
 from app.database import get_db
 from app.models import ActionPlan
+from app.risk_levels import risk_level
 
 logger = logging.getLogger("cma.iron_law")
 
@@ -150,6 +151,7 @@ def _query_action_plan_verify(db: Session):
 # ═══════════════ 端点 ═══════════════
 
 @router.get("/iron-law")
+@risk_level("L1")
 def get_iron_law_kpis(
     bot: dict = Depends(verify_bot_key),
     db: Session = Depends(get_db),
@@ -224,6 +226,7 @@ def get_iron_law_kpis(
 
 
 @router.get("/iron-law/bots")
+@risk_level("L1")
 def get_bot_iron_law_ranking(
     bot: dict = Depends(verify_bot_key),
     db: Session = Depends(get_db),

backend/app/api/bot_kpis.py

@@ -7,6 +7,7 @@ from datetime import datetime
 from app.database import get_db
 from app.auth_middleware import require_auth
 from app.models import KPIDefinition, KPIValue
+from app.risk_levels import risk_level
 
 router = APIRouter(prefix="/api/cma/bot-kpis", tags=["Bot KPI管理"],
     dependencies=[Depends(require_auth)],
@@ -46,6 +47,7 @@ def _calc_bot_kpi_score(current_value, target_value, is_reverse=False):
 
 
 @router.get("")
+@risk_level("L1")
 def list_bot_kpis(
     source: str = Query("finance-bot", description="Bot标识"),
     period: Optional[str] = None,
@@ -132,6 +134,7 @@ def list_bot_kpis(
 
 
 @router.post("/{kpi_id}/value")
+@risk_level("L2")
 def update_bot_kpi_value(
     kpi_id: int,
     data: dict,

backend/app/main.py

@@ -9,6 +9,7 @@ from app.api import auth, kpis, kpi_governance, templates, maps, dashboard, data
 from app.utils.cache import clear_all as clear_cache, delete as delete_cache
 from scripts.erp_sync import run_sync as run_erp_sync
 from app.auth_middleware import require_auth
+from app.api.audit_log import bot_audit_middleware
 
 load_dotenv()
 
@@ -30,6 +31,9 @@ app.add_middleware(
     allow_headers=["*"],
 )
 
+# Bot API 操作审计（L1-L4分级标注 + JSON行审计日志，不阻塞业务）
+app.middleware("http")(bot_audit_middleware)
+
 app.include_router(auth.router)
 app.include_router(kpis.router)
 app.include_router(kpi_governance.router)

backend/app/risk_levels.py

@@ -0,0 +1,120 @@
+"""
+CMA Bot API 风险分级（L1-L4）与路由标注装饰器
+=============================================
+
+分级定义（方案文档 nexa-permission-autonomy-plan-20260828 第二节）:
+    L1 只读查询       — X-BOT-KEY验证后直接放行
+    L2 业务写（单条） — 放行 + 写前校验（entity归属/字段校验）
+    L3 批量写/创建    — 放行 + 限制批量 + source_batch审计
+    L4 危险           — 不向Bot API开放（DROP/TRUNCATE/批量DELETE/生产结构修改）
+
+安全底线: Bot API面不存在L4端点（0项开放 = 天然隔离）。
+"""
+import functools
+
+RISK_LEVELS = {
+    "L1": "只读",
+    "L2": "业务写",
+    "L3": "批量写",
+    "L4": "危险",
+}
+
+RISK_HANDLING = {
+    "L1": "X-BOT-KEY验证后直接放行",
+    "L2": "放行 + 写前校验（entity归属/字段校验）",
+    "L3": "放行 + 限制批量 + source_batch审计",
+    "L4": "不向Bot API开放（终端层拦截 + 人工审批）",
+}
+
+# ────────────────────────────────────────────────
+# API → 风险级别 清单（Bot API面全量路由）
+# 键格式: "METHOD path"（path 与 FastAPI route.path 一致，含 {param} 占位符）
+# ────────────────────────────────────────────────
+API_RISK_MAP = {
+    # ── L1 只读（20项方案清单 + risk-levels查询端点） ──
+    "GET /api/cma/bot/ping": "L1",
+    "GET /api/cma/bot/overview": "L1",
+    "GET /api/cma/bot/kpis": "L1",
+    "GET /api/cma/bot/kpis/{kpi_id}/history": "L1",
+    "GET /api/cma/bot/strategic-maps": "L1",
+    "GET /api/cma/bot/alerts": "L1",
+    "GET /api/cma/bot/budget/plans": "L1",
+    "GET /api/cma/bot/cost/standard": "L1",
+    "GET /api/cma/bot/cost/actual": "L1",
+    "GET /api/cma/bot/actions": "L1",
+    "GET /api/cma/bot/organization": "L1",
+    "GET /api/cma/bot/data-sources": "L1",
+    "GET /api/cma/bot/users": "L1",
+    "GET /api/cma/bot/query": "L1",
+    "GET /api/cma/bot/okr/list": "L1",
+    "GET /api/cma/bot/nlp": "L1",
+    "GET /api/cma/bot/iron-law": "L1",
+    "GET /api/cma/bot/iron-law/bots": "L1",
+    "GET /api/cma/bot-bridge/verify/{action_plan_id}/history": "L1",
+    "GET /api/cma/bot-kpis": "L1",
+    # 本任务新增的只读端点
+    "GET /api/cma/bot/risk-levels": "L1",
+    # ── L2 业务写（4项方案清单 + okr/create单条业务写） ──
+    "POST /api/cma/bot/kpi-value-with-check": "L2",
+    "POST /api/cma/bot-bridge/kpi-result": "L2",
+    "POST /api/cma/bot-kpis/{kpi_id}/value": "L2",
+    "POST /api/cma/bot-bridge/verify/{action_plan_id}": "L2",
+    "POST /api/cma/bot/okr/create": "L2",  # 单条OKR创建（方案清单未列出，按单条业务写归类）
+    # ── L3 批量写/创建（3项） ──
+    "POST /api/cma/bot/import": "L3",
+    "POST /api/cma/bot/kpis/create-with-links": "L3",
+    "POST /api/cma/bot-bridge/mpm-result": "L3",
+    # ── L4 危险：Bot API面不存在（安全底线，不添加） ──
+}
+
+
+def risk_level(level: str):
+    """路由标注装饰器: @risk_level('L1') 挂在路由函数上（router.get/post 之下）。
+
+    同时把级别属性写到原函数与包装函数上，保证 route.endpoint 无论取到哪个
+    都能通过 getattr(endpoint, 'risk_level') 解析。
+    """
+    def decorator(func):
+        func.risk_level = level
+
+        @functools.wraps(func)
+        def wrapper(*args, **kwargs):
+            return func(*args, **kwargs)
+
+        wrapper.risk_level = level
+        return wrapper
+
+    return decorator
+
+
+def get_risk_level(method: str, path: str):
+    """按 METHOD + path（FastAPI模板路径）查级别，未标注返回 None"""
+    return API_RISK_MAP.get(f"{method.upper()} {path}")
+
+
+def get_handling(level: str) -> str:
+    """级别 → 处理方式说明"""
+    return RISK_HANDLING.get(level, "")
+
+
+def list_api_risk_map() -> list:
+    """返回 API→级别→处理方式 清单（供 GET /api/cma/bot/risk-levels 使用）"""
+    items = []
+    for key, level in API_RISK_MAP.items():
+        method, path = key.split(" ", 1)
+        items.append({
+            "method": method,
+            "path": path,
+            "risk_level": level,
+            "handling": get_handling(level),
+        })
+    items.sort(key=lambda x: (x["risk_level"], x["method"], x["path"]))
+    return items
+
+
+def risk_summary() -> dict:
+    """各级别端点数量统计"""
+    summary = {lv: 0 for lv in RISK_LEVELS}
+    for level in API_RISK_MAP.values():
+        summary[level] = summary.get(level, 0) + 1
+    return summary

backend/tests/test_risk_levels.py

@@ -0,0 +1,189 @@
+"""Bot API 风险分级（L1-L4）标注 + 操作审计日志 测试
+
+覆盖:
+  1. API_RISK_MAP 覆盖所有 /api/cma/bot* 路由（app.routes 遍历核对）
+  2. GET /api/cma/bot/risk-levels 返回200且含L1-L4定义
+  3. Bot API面不存在L4端点（无 drop/truncate/delete 批量端点，安全底线）
+  4. 审计日志在调用Bot API后写入（monkeypatch + 真实日志文件双验证）
+"""
+import io
+import json
+
+import pytest
+from openpyxl import Workbook
+
+from app.main import app
+from app.models import KPIDefinition
+from app.risk_levels import API_RISK_MAP, RISK_LEVELS
+import app.api.audit_log as audit_log_module
+
+BOT_KEY = {"X-BOT-KEY": "cma-bot-finance-2026"}
+_HTTP_METHODS = ("GET", "POST", "PUT", "DELETE", "PATCH")
+
+
+def _bot_routes():
+    """遍历 app.routes，返回所有 /api/cma/bot* 路由 (route, methods列表)"""
+    routes = []
+    for r in app.routes:
+        path = getattr(r, "path", "")
+        if path.startswith("/api/cma/bot"):
+            methods = sorted(m for m in (getattr(r, "methods", set()) or set())
+                             if m in _HTTP_METHODS)
+            routes.append((r, methods))
+    return routes
+
+
+class TestRiskMapCoverage:
+    def test_api_risk_map_covers_all_bot_routes(self):
+        """每个 /api/cma/bot* 路由都在 API_RISK_MAP 有级别标注，且函数有@risk_level装饰器"""
+        missing = []
+        unlabeled = []
+        for r, methods in _bot_routes():
+            for m in methods:
+                key = f"{m} {r.path}"
+                if key not in API_RISK_MAP:
+                    missing.append(key)
+            if not getattr(r.endpoint, "risk_level", None):
+                unlabeled.append(f"{sorted(methods)} {r.path}")
+        assert not missing, f"API_RISK_MAP 缺少以下路由标注: {missing}"
+        assert not unlabeled, f"以下路由函数缺少 @risk_level 装饰器: {unlabeled}"
+
+    def test_risk_level_counts(self):
+        """分级统计与方案一致：L1=21(20项清单+risk-levels端点)、L2=5(4项清单+okr/create)、L3=3、L4=0"""
+        from collections import Counter
+        counts = Counter(API_RISK_MAP.values())
+        assert counts["L1"] == 21, counts
+        assert counts["L2"] == 5, counts
+        assert counts["L3"] == 3, counts
+        assert counts["L4"] == 0, "Bot API面不得存在L4端点（安全底线）"
+
+    def test_no_l4_bot_endpoints(self):
+        """Bot API面不存在L4端点：无DELETE方法、无drop/truncate/delete危险路径"""
+        danger_keywords = ("drop", "truncate", "delete")
+        for r, methods in _bot_routes():
+            assert getattr(r.endpoint, "risk_level", None) != "L4", \
+                f"{r.path} 不应被标注为L4"
+            assert "DELETE" not in methods, f"Bot路由不应有DELETE方法: {r.path}"
+            low = r.path.lower()
+            for kw in danger_keywords:
+                assert kw not in low, f"Bot路由不应含危险路径片段: {r.path}"
+
+
+class TestRiskLevelsEndpoint:
+    def test_risk_levels_requires_bot_key(self, client):
+        """无X-BOT-KEY → 401"""
+        resp = client.get("/api/cma/bot/risk-levels")
+        assert resp.status_code == 401
+
+    def test_risk_levels_returns_200_with_definitions(self, client):
+        """X-BOT-KEY → 200，含L1-L4定义与API→级别→处理方式清单"""
+        resp = client.get("/api/cma/bot/risk-levels", headers=BOT_KEY)
+        assert resp.status_code == 200
+        data = resp.json()
+
+        # L1-L4 定义齐全
+        for level, label in RISK_LEVELS.items():
+            assert data["risk_levels"][level] == label, f"缺少 {level} 定义"
+
+        # API→级别→处理方式 清单
+        apis = {f"{a['method']} {a['path']}": a for a in data["apis"]}
+        assert apis["GET /api/cma/bot/ping"]["risk_level"] == "L1"
+        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 data["summary"]["L1"] == 21
+        assert data["summary"]["L2"] == 5
+        assert data["summary"]["L3"] == 3
+        assert data["summary"]["L4"] == 0
+
+
+class TestAuditLog:
+    def test_audit_record_after_bot_call(self, client, monkeypatch):
+        """调用Bot API后写出审计记录（monkeypatch捕获）"""
+        records = []
+        monkeypatch.setattr(audit_log_module, "write_audit_line",
+                            lambda rec: records.append(rec))
+        resp = client.get("/api/cma/bot/ping")
+        assert resp.status_code == 200
+        assert len(records) >= 1
+        rec = records[-1]
+        assert rec["method"] == "GET"
+        assert rec["endpoint"] == "/api/cma/bot/ping"
+        assert rec["risk_level"] == "L1"
+        assert rec["status"] == 200
+        # 字段齐全
+        for field in ("timestamp", "bot_name", "endpoint", "method",
+                      "risk_level", "entity_id", "status"):
+            assert field in rec, f"审计记录缺少字段: {field}"
+
+    def test_audit_file_written_json_lines(self, client, tmp_path, monkeypatch):
+        """真实日志文件：调用Bot API后 bot_audit.log 追加JSON行"""
+        log_file = tmp_path / "bot_audit.log"
+        monkeypatch.setenv("CMA_BOT_AUDIT_LOG", str(log_file))
+        monkeypatch.setattr(audit_log_module, "_audit_logger", None)
+
+        resp = client.get("/api/cma/bot/ping")
+        assert resp.status_code == 200
+        assert log_file.exists()
+
+        lines = log_file.read_text(encoding="utf-8").strip().splitlines()
+        assert lines, "审计日志文件为空"
+        rec = json.loads(lines[-1])
+        assert rec["endpoint"] == "/api/cma/bot/ping"
+        assert rec["risk_level"] == "L1"
+        assert rec["status"] == 200
+
+    def test_audit_entity_id_and_bot_name(self, client, db, monkeypatch):
+        """审计记录含 bot_name（X-BOT-KEY映射）与 entity_id"""
+        records = []
+        monkeypatch.setattr(audit_log_module, "write_audit_line",
+                            lambda rec: records.append(rec))
+        kpi = KPIDefinition(kpi_code="AUDIT_EID", kpi_name="审计实体", dimension="finance",
+                            status="active", target_value=1.0, entity_id=1)
+        db.add(kpi)
+        db.commit()
+
+        resp = client.post("/api/cma/bot/kpi-value-with-check",
+                           json={"kpi_id": kpi.id, "actual_value": 66.0,
+                                 "period": "2026-08", "entity_id": 1},
+                           headers=BOT_KEY)
+        assert resp.status_code == 200
+        assert records, "应有审计记录"
+        rec = records[-1]
+        assert rec["bot_name"] == "财务BOT"
+        assert rec["entity_id"] == 1
+        assert rec["risk_level"] == "L2"
+        assert rec["status"] == 200
+        # JSON body 读取未破坏业务
+        assert resp.json()["status"] == "ok"
+
+    def test_l3_batch_write_records_rows(self, client, db, monkeypatch):
+        """L3批量写（/import）：审计记录额外含 rows 行数"""
+        kpi = KPIDefinition(kpi_code="AUDIT_ROWS", kpi_name="审计行数", dimension="finance",
+                            status="active", target_value=1.0)
+        db.add(kpi)
+        db.commit()
+
+        wb = Workbook()
+        ws = wb.active
+        ws.append(["kpi_code", "period", "actual_value"])
+        ws.append(["AUDIT_ROWS", "2026-08", 88.0])
+        ws.append(["AUDIT_ROWS", "2026-07", 77.0])
+        buf = io.BytesIO()
+        wb.save(buf)
+
+        records = []
+        monkeypatch.setattr(audit_log_module, "write_audit_line",
+                            lambda rec: records.append(rec))
+        files = {"file": ("kpi.xlsx", buf.getvalue(),
+                          "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")}
+        resp = client.post("/api/cma/bot/import", headers=BOT_KEY, files=files)
+        assert resp.status_code == 200
+        assert resp.json()["imported"] == 2
+
+        assert records, "应有审计记录"
+        rec = records[-1]
+        assert rec["risk_level"] == "L3"
+        assert rec["rows"] == 2