cma-management/backend/app/api/bot_bridge.py

"""
CMA BOT API桥接层 — 供财务BOT/店研学BOT调用
无需用户登录，使用 BOT API Key 认证
"""
import os, json, logging
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
from typing import Optional
from app.database import get_db
from app.deps import get_entity_id
from app.models import (
    User, UserEntity, StrategicMap, KPIDefinition, KPITemplate, KPIValue,
    DataSourceConfig, KPIAlert, OperationLog, NotificationChannel,
    NotificationLog, RolePermission, ActionPlan, OrgNode,
    StrategicMapVersion, MapObjective, Objective, KR, Subject,
)
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")

router = APIRouter(prefix="/api/cma/bot", tags=["BOT桥接"])

# ── BOT API Key 配置 ──
_BOT_API_KEYS = {}

def _load_bot_keys():
    global _BOT_API_KEYS
    raw = os.getenv("CMA_BOT_API_KEYS", "")
    if not raw:
        _BOT_API_KEYS = {
            "cma-bot-finance-2026": {"role": "finance", "name": "财务BOT"},
            "cma-bot-shop-2026": {"role": "business", "name": "店研学BOT"},
            "cma-bot-admin-2026": {"role": "ceo", "name": "管理BOT"},
        }
    else:
        try:
            _BOT_API_KEYS = json.loads(raw)
        except:
            _BOT_API_KEYS = {}

_load_bot_keys()

def verify_bot_key(x_bot_key: str = Header(None, alias="X-BOT-KEY")):
    if not x_bot_key or x_bot_key not in _BOT_API_KEYS:
        raise HTTPException(401, "无效的BOT API Key")
    bot_info = _BOT_API_KEYS[x_bot_key]
    logger.info(f"BOT访问: {bot_info['name']} ({bot_info['role']})")
    return bot_info


# ═══════════════ 通用工具 ═══════════════

def _float(v):
    if v is None: return None
    try: return float(v)
    except: return None

def _safe_iso(dt):
    if dt is None: return None
    try: return dt.isoformat() if hasattr(dt, 'isoformat') else str(dt)
    except: return None

def _model_dict(obj, fields: dict):
    """安全地将模型字段转为dict"""
    result = {}
    for key, attr in fields.items():
        v = getattr(obj, attr, None)
        if isinstance(v, float):
            result[key] = _float(v)
        else:
            result[key] = v
    return result


# ═══════════════ 端点 ═══════════════

@router.get("/ping")
@risk_level("L1")
def ping():
    return {"status": "ok", "version": "1.0", "timestamp": datetime.now().isoformat()}


# ── 总览 ──

@router.get("/overview")
@risk_level("L1")
def bot_overview(
    bot: dict = Depends(verify_bot_key),
    db: Session = Depends(get_db),
    entity_id: int = Depends(get_entity_id),
):
    """系统总览 — BOT首选入口（账套隔离 2026-08-31: 仅统计当前企业）"""
    return {
        "bot": bot,
        "entity_id": entity_id,
        "timestamp": datetime.now().isoformat(),
        "stats": {
            "kpis_total": db.query(func.count(KPIDefinition.id)).filter(KPIDefinition.status == "active", KPIDefinition.entity_id == entity_id).scalar() or 0,
            "alerts_open": db.query(func.count(KPIAlert.id)).filter(KPIAlert.status == "pending", KPIAlert.entity_id == entity_id).scalar() or 0,
            "maps_total": db.query(func.count(StrategicMap.id)).filter(StrategicMap.entity_id == entity_id).scalar() or 0,
            "budget_plans": db.query(func.count(BudgetPlan.id)).filter(BudgetPlan.entity_id == entity_id).scalar() or 0,
            "action_plans_pending": db.query(func.count(ActionPlan.id)).filter(ActionPlan.entity_id == entity_id, ActionPlan.status.in_(["pending", "in_progress"])).scalar() or 0,
            "data_sources": db.query(func.count(DataSourceConfig.id)).filter(DataSourceConfig.entity_id == entity_id).scalar() or 0,
            "users": db.query(func.count(User.id)).join(UserEntity, UserEntity.user_id == User.id).filter(UserEntity.entity_id == entity_id).scalar() or 0,
            "org_nodes": db.query(func.count(OrgNode.id)).filter(OrgNode.entity_id == entity_id).scalar() or 0,
        }
    }


# ── KPI ──

@router.get("/kpis")
@risk_level("L1")
def bot_kpis(
    dimension: Optional[str] = Query(None),
    status: str = Query("active"),
    limit: int = Query(200, le=1000),
    bot: dict = Depends(verify_bot_key),
    db: Session = Depends(get_db),
    entity_id: int = Depends(get_entity_id),
):
    query = db.query(KPIDefinition).filter(KPIDefinition.status == status, KPIDefinition.entity_id == entity_id)
    if dimension:
        query = query.filter(KPIDefinition.dimension == dimension)
    kpis = query.order_by(KPIDefinition.dimension, KPIDefinition.kpi_code).limit(limit).all()

    results = []
    for k in kpis:
        latest = db.query(KPIValue).filter(KPIValue.kpi_id == k.id)\
            .order_by(KPIValue.period.desc()).first()
        results.append({
            "id": k.id, "name": k.kpi_name, "code": k.kpi_code,
            "dimension": k.dimension, "category": k.category,
            "unit": k.unit, "formula": k.formula,
            "frequency": k.frequency, "data_source_type": k.data_source_type,
            "target_value": _float(k.target_value),
            "threshold_green": k.threshold_green,
            "threshold_yellow": k.threshold_yellow,
            "threshold_red": k.threshold_red,
            "responsible_dept": k.responsible_dept, "owner": k.responsible_user,
            "objective": k.objective, "formula_desc": k.formula_desc,
            "latest_value": _float(latest.actual_value) if latest else None,
            "latest_period": latest.period if latest else None,
            "status": k.status,
        })
    return {"total": len(results), "items": results}


@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),
    db: Session = Depends(get_db),
    entity_id: int = Depends(get_entity_id),
):
    kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first()
    if not kpi or kpi.entity_id != entity_id:
        raise HTTPException(404, "KPI不存在")
    values = db.query(KPIValue).filter(KPIValue.kpi_id == kpi_id)\
        .order_by(KPIValue.period.desc()).limit(limit).all()
    return {
        "kpi": {"id": kpi.id, "name": kpi.kpi_name, "code": kpi.kpi_code, "unit": kpi.unit},
        "values": [
            {
                "period": v.period,
                "actual": _float(v.actual_value),
                "source_type": v.source_type,
                "data_status": v.data_status,
            } for v in values
        ],
    }


# ── 战略地图 ──

@router.get("/strategic-maps")
@risk_level("L1")
def bot_maps(
    bot: dict = Depends(verify_bot_key),
    db: Session = Depends(get_db),
    entity_id: int = Depends(get_entity_id),
):
    maps = db.query(StrategicMap).filter(StrategicMap.entity_id == entity_id).order_by(StrategicMap.id.desc()).all()
    result = []
    for m in maps:
        objectives = db.query(MapObjective).filter(MapObjective.map_id == m.id).all()
        dims = {}
        for obj in objectives:
            dk = obj.dimension_key
            if dk not in dims:
                dims[dk] = []
            dims[dk].append({"id": obj.id, "name": obj.name, "description": obj.description})
        result.append({
            "id": m.id, "title": m.title, "version": m.version,
            "status": m.status, "dimensions": m.dimensions,
            "objectives": dims,
            "created_at": _safe_iso(m.created_at),
            "updated_at": _safe_iso(m.updated_at),
        })
    return {"total": len(result), "items": result}


# ── 预警 ──

@router.get("/alerts")
@risk_level("L1")
def bot_alerts(
    status: str = Query("pending"),
    level: Optional[str] = Query(None),
    limit: int = Query(50, le=200),
    bot: dict = Depends(verify_bot_key),
    db: Session = Depends(get_db),
    entity_id: int = Depends(get_entity_id),
):
    query = db.query(KPIAlert).filter(KPIAlert.entity_id == entity_id)
    query = query.filter(KPIAlert.status == status)
    if level:
        query = query.filter(KPIAlert.alert_level == level)
    alerts = query.order_by(KPIAlert.created_at.desc()).limit(limit).all()
    return {
        "total": len(alerts),
        "items": [
            {
                "id": a.id, "kpi_id": a.kpi_id,
                "level": a.alert_level, "message": a.alert_message,
                "status": a.status, "assignee": a.assignee,
                "resolution": a.resolution,
                "created_at": _safe_iso(a.created_at),
                "resolved_at": _safe_iso(a.resolved_at),
            } for a in alerts
        ],
    }


# ── 预算 ──

@router.get("/budget/plans")
@risk_level("L1")
def bot_budget_plans(
    year: Optional[int] = Query(None),
    bot: dict = Depends(verify_bot_key),
    db: Session = Depends(get_db),
    entity_id: int = Depends(get_entity_id),
):
    query = db.query(BudgetPlan).filter(BudgetPlan.entity_id == entity_id)
    if year:
        query = query.filter(BudgetPlan.budget_year == year)
    plans = query.order_by(BudgetPlan.period.desc()).limit(200).all()
    return {
        "total": len(plans),
        "items": [
            {
                "id": p.id, "kpi_id": p.kpi_id,
                "period": p.period,
                "budget_value": _float(p.budget_value),
                "year": p.budget_year, "month": p.budget_month,
                "version": p.version, "status": p.status,
                "remark": p.remark,
            } for p in plans
        ],
    }


# ── 成本 ──

@router.get("/cost/standard")
@risk_level("L1")
def bot_standard_costs(
    bot: dict = Depends(verify_bot_key),
    db: Session = Depends(get_db),
    entity_id: int = Depends(get_entity_id),
):
    costs = db.query(StandardCost).filter(StandardCost.status == "active", StandardCost.entity_id == entity_id).limit(200).all()
    return {
        "total": len(costs),
        "items": [
            {
                "id": c.id, "product_code": c.product_code,
                "product_name": c.product_name, "cost_type": c.cost_type,
                "item_name": c.item_name,
                "standard_quantity": _float(c.standard_quantity),
                "unit": c.unit,
                "standard_price": _float(c.standard_price),
                "standard_cost": _float(c.standard_cost),
                "version": c.version, "remark": c.remark,
            } for c in costs
        ],
    }


@router.get("/cost/actual")
@risk_level("L1")
def bot_actual_costs(
    period: Optional[str] = Query(None),
    bot: dict = Depends(verify_bot_key),
    db: Session = Depends(get_db),
    entity_id: int = Depends(get_entity_id),
):
    query = db.query(ActualCost).filter(ActualCost.entity_id == entity_id)
    if period:
        query = query.filter(ActualCost.period == period)
    costs = query.order_by(ActualCost.period.desc()).limit(200).all()
    return {
        "total": len(costs),
        "items": [
            {
                "id": c.id, "period": c.period,
                "product_code": c.product_code,
                "product_name": c.product_name,
                "cost_type": c.cost_type, "item_name": c.item_name,
                "actual_quantity": _float(c.actual_quantity),
                "actual_price": _float(c.actual_price),
                "actual_cost": _float(c.actual_cost),
            } for c in costs
        ],
    }


# ── 行动方案 ──

@router.get("/actions")
@risk_level("L1")
def bot_actions(
    status: Optional[str] = Query(None),
    bot: dict = Depends(verify_bot_key),
    db: Session = Depends(get_db),
    entity_id: int = Depends(get_entity_id),
):
    query = db.query(ActionPlan).filter(ActionPlan.entity_id == entity_id)
    if status:
        query = query.filter(ActionPlan.status == status)
    plans = query.order_by(ActionPlan.priority, ActionPlan.id.desc()).limit(100).all()
    return {
        "total": len(plans),
        "items": [
            {
                "id": p.id, "title": p.title,
                "description": p.description, "kpi_id": p.kpi_id,
                "assignee": p.assignee, "priority": p.priority,
                "status": p.status, "progress": p.progress,
                "due_date": _safe_iso(p.due_date),
                "created_at": _safe_iso(p.created_at),
            } for p in plans
        ],
    }


# ── 组织 ──

@router.get("/organization")
@risk_level("L1")
def bot_org(
    bot: dict = Depends(verify_bot_key),
    db: Session = Depends(get_db),
    entity_id: int = Depends(get_entity_id),
):
    nodes = db.query(OrgNode).filter(OrgNode.entity_id == entity_id).order_by(OrgNode.level, OrgNode.sort_order).all()
    return {
        "total": len(nodes),
        "items": [
            {
                "id": n.id, "name": n.name,
                "parent_id": n.parent_id, "level": n.level,
                "code": n.code, "sort_order": n.sort_order,
                "enabled": n.enabled,
            } for n in nodes
        ],
    }


# ── 数据源 ──

@router.get("/data-sources")
@risk_level("L1")
def bot_data_sources(
    bot: dict = Depends(verify_bot_key),
    db: Session = Depends(get_db),
    entity_id: int = Depends(get_entity_id),
):
    sources = db.query(DataSourceConfig).filter(DataSourceConfig.entity_id == entity_id).all()
    return {
        "total": len(sources),
        "items": [
            {
                "id": s.id, "name": s.name,
                "source_type": s.source_type,
                "api_endpoint": s.api_endpoint,
                "sync_type": s.sync_type,
                "status": s.status,
                "last_sync_at": _safe_iso(s.last_sync_at),
            } for s in sources
        ],
    }


# ── 用户 ──

@router.get("/users")
@risk_level("L1")
def bot_users(
    bot: dict = Depends(verify_bot_key),
    db: Session = Depends(get_db),
    entity_id: int = Depends(get_entity_id),
):
    # 用户是全局实体，通过 user_entities 授权表按企业过滤（安全修复 2026-08-31）
    users = db.query(User).join(UserEntity, UserEntity.user_id == User.id)\
        .filter(UserEntity.entity_id == entity_id).all()
    return {
        "total": len(users),
        "items": [
            {"id": u.id, "username": u.username, "name": u.name,
             "role": u.role, "phone": u.phone}
            for u in 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),
    db: Session = Depends(get_db),
    entity_id: int = Depends(get_entity_id),
):
    """统一查询入口 — BOT用这个一次拿完需要的数据（账套隔离 2026-08-31）"""
    result = {"bot": bot["name"], "role": bot["role"], "entity_id": entity_id, "timestamp": datetime.now().isoformat()}

    if q in ("overview", "all"):
        result["overview"] = {
            "kpis": db.query(func.count(KPIDefinition.id)).filter(KPIDefinition.status == "active", KPIDefinition.entity_id == entity_id).scalar() or 0,
            "alerts_open": db.query(func.count(KPIAlert.id)).filter(KPIAlert.status == "pending", KPIAlert.entity_id == entity_id).scalar() or 0,
            "maps": db.query(func.count(StrategicMap.id)).filter(StrategicMap.entity_id == entity_id).scalar() or 0,
            "budget_plans": db.query(func.count(BudgetPlan.id)).filter(BudgetPlan.entity_id == entity_id).scalar() or 0,
        }

    if q in ("kpis", "all"):
        kpis = db.query(KPIDefinition).filter(KPIDefinition.status == "active", KPIDefinition.entity_id == entity_id).limit(100).all()
        result["kpis"] = [
            {"id": k.id, "name": k.kpi_name, "code": k.kpi_code,
             "dimension": k.dimension, "target": _float(k.target_value), "unit": k.unit}
            for k in kpis
        ]

    if q in ("alerts", "all"):
        alerts = db.query(KPIAlert).filter(KPIAlert.status == "pending", KPIAlert.entity_id == entity_id)\
            .order_by(KPIAlert.created_at.desc()).limit(20).all()
        result["alerts"] = [
            {"id": a.id, "level": a.alert_level, "message": a.alert_message,
             "kpi_id": a.kpi_id, "created_at": _safe_iso(a.created_at)}
            for a in alerts
        ]

    if q in ("maps", "all"):
        maps = db.query(StrategicMap).filter(StrategicMap.entity_id == entity_id).limit(10).all()
        result["maps"] = [
            {"id": m.id, "title": m.title, "status": m.status,
             "version": m.version, "created_at": _safe_iso(m.created_at)}
            for m in maps
        ]

    if q in ("budget", "all"):
        plans = db.query(BudgetPlan).filter(BudgetPlan.entity_id == entity_id).limit(50).all()
        result["budget"] = [
            {"id": p.id, "period": p.period, "budget_value": _float(p.budget_value),
             "year": p.budget_year, "month": p.budget_month, "status": p.status,
             "kpi_id": p.kpi_id}
            for p in plans
        ]

    if q in ("cost", "all"):
        sc = db.query(StandardCost).filter(StandardCost.entity_id == entity_id).limit(50).all()
        result["costs"] = [
            {"id": c.id, "product": c.product_name, "type": c.cost_type,
             "standard": _float(c.standard_cost), "unit": c.unit}
            for c in sc
        ]

    if q in ("okr", "all"):
        objs = db.query(Objective).filter(Objective.status == "active", Objective.entity_id == entity_id).all()
        result["okr"] = []
        for o in objs:
            # KR完整修复(2026-08-27): 从krs表读取
            krs = db.query(KR).filter(KR.objective_id == o.id).all()
            result["okr"].append({
                "id": o.id, "title": o.title, "quarter": o.quarter,
                "dimension": o.dimension, "progress": o.progress,
                "confidence": o.confidence,
                "key_results": [
                    {"title": kr.title, "status": kr.status, "progress": kr.progress,
                     "operator": kr.operator, "target_value": _float(kr.target_value)}
                    for kr in krs
                ]
            })

    if q in ("actions", "all"):
        acts = db.query(ActionPlan).filter(ActionPlan.entity_id == entity_id).limit(30).all()
        result["actions"] = [
            {"id": a.id, "title": a.title, "status": a.status,
             "progress": a.progress, "assignee": a.assignee}
            for a in acts
        ]

    return result


@router.post("/import")
@risk_level("L3")
def bot_import_excel(
    file: UploadFile = File(...),
    bot: dict = Depends(verify_bot_key),
    db: Session = Depends(get_db),
    entity_id: int = Depends(get_entity_id),
):
    """Bot上传Excel导入KPI数据到CMA（账套隔离 2026-08-31: 仅导入当前企业KPI）"""
    import pandas as pd, io, hashlib
    from app.models import KPIValue
    try:
        content = file.file.read()
        df = pd.read_excel(io.BytesIO(content))
    except Exception as e:
        raise HTTPException(400, f"Excel解析失败: {e}")
    finally:
        file.file.close()

    # 智能识别列名
    col_map = {"kpi_code": ["kpi_code", "KPI编码", "指标编码", "code"],
               "period": ["period", "期间", "月份", "month", "日期"],
               "actual_value": ["actual_value", "实际值", "值", "金额", "value", "amount"]}
    
    mapped = {}
    for field, aliases in col_map.items():
        for col in df.columns:
            if str(col).strip() in aliases or str(col).strip().lower() in aliases:
                mapped[field] = str(col).strip()
                break
    
    if "actual_value" not in mapped:
        raise HTTPException(400, f"无法识别数值列，支持的列名: {col_map['actual_value']}")
    if "kpi_code" not in mapped:
        raise HTTPException(400, f"无法识别KPI编码列，支持的列名: {col_map['kpi_code']}")
    
    kpi_col = mapped["kpi_code"]
    val_col = mapped["actual_value"]
    period_col = mapped.get("period")
    
    count = 0
    errors = []
    for idx, row in df.iterrows():
        try:
            kpi_code = str(row[kpi_col]).strip()
            val = float(row[val_col])
            period = str(row[period_col]).strip() if period_col else datetime.now().strftime("%Y-%m")
            
            kpi = db.query(KPIDefinition).filter(KPIDefinition.kpi_code == kpi_code).first()
            if not kpi:
                errors.append(f"第{idx+2}行: KPI编码 '{kpi_code}' 不存在，跳过")
                continue
            if kpi.entity_id != entity_id:
                errors.append(f"第{idx+2}行: KPI编码 '{kpi_code}' 不属于当前企业(entity={kpi.entity_id})，跳过")
                continue
            
            kv = KPIValue(kpi_id=kpi.id, entity_id=kpi.entity_id, period=period, actual_value=val,
                          source_batch=hashlib.md5(f"{datetime.now()}".encode()).hexdigest()[:12])
            db.add(kv)
            count += 1
        except Exception as e:
            errors.append(f"第{idx+2}行: {e}")
    
    db.commit()
    return {"ok": True, "imported": count, "errors": len(errors), "detail": errors[:5]}


# ── 自然语言查询 ──

@router.post("/okr/create")
@risk_level("L2")
def bot_okr_create(
    title: str = Query(...),
    quarter: str = Query(...),
    dimension: Optional[str] = Query(None),
    bot: dict = Depends(verify_bot_key),
    db: Session = Depends(get_db),
    entity_id: int = Depends(get_entity_id),
):
    """Bot创建OKR目标（账套隔离 2026-08-31: 写入token企业）"""
    from app.models import Objective
    obj = Objective(entity_id=entity_id, title=title, quarter=quarter, dimension=dimension, owner=bot["name"])
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return {"ok": True, "id": obj.id, "title": obj.title, "confidence": obj.confidence}


@router.get("/okr/list")
@risk_level("L1")
def bot_okr_list(
    quarter: Optional[str] = Query(None),
    bot: dict = Depends(verify_bot_key),
    db: Session = Depends(get_db),
    entity_id: int = Depends(get_entity_id),
):
    """Bot列出OKR（含KR进度）（账套隔离 2026-08-31）"""
    from app.models import Objective
    q = db.query(Objective).filter(Objective.entity_id == entity_id)
    if quarter:
        q = q.filter(Objective.quarter == quarter)
    objs = q.order_by(Objective.quarter.desc()).all()
    return {"total": len(objs), "items": [
        {"id": o.id, "title": o.title, "quarter": o.quarter,
         "dimension": o.dimension, "progress": o.progress,
         "confidence": o.confidence, "status": o.status,
         "kr_count": db.query(func.count(ActionPlan.id)).filter(ActionPlan.objective_id == o.id).scalar() or 0}
        for o in objs
    ]}


@router.get("/nlp")
@risk_level("L1")
def bot_nlp(
    intent: str = Query("overview"),
    bot: dict = Depends(verify_bot_key),
    db: Session = Depends(get_db),
    entity_id: int = Depends(get_entity_id),
):
    """
    自然语言意图映射:
    overview/总览/finance/财务/alerts/预警/budget/预算/cost/成本/maps/战略/actions/行动
    """
    m = {
        "总览": "overview", "驾驶舱": "overview",
        "财务": "finance", "财务状况": "finance",
        "预警": "alerts", "风险": "alerts",
        "预算": "budget", "预算执行": "budget",
        "成本": "cost", "成本分析": "cost",
        "战略": "maps", "战略地图": "maps",
        "行动": "actions", "改善": "actions",
    "okr": "okr", "目标": "okr", "季度目标": "okr",
    }
    resolved = m.get(intent, intent)
    return bot_query(q=resolved, bot=bot, db=db, entity_id=entity_id)


# ════════════════════════════════════════════════════════════
# 聚合接口（Agent化生产链路 · 行动1, 2026-08-25）
# 建议A: KPI值更新联动预警检查 | 建议B: KPI创建联动关联
# ════════════════════════════════════════════════════════════

@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?}"""
    kpi_id = data.get("kpi_id")
    actual_value = data.get("actual_value")
    period = data.get("period")
    entity_id = int(data.get("entity_id") or 1)
    run_check = bool(data.get("run_check", True))
    if not kpi_id or actual_value is None:
        raise HTTPException(400, "kpi_id 和 actual_value 必填")
    # 校验 KPI 归属（多租户）
    kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first()
    if not kpi or kpi.entity_id != entity_id:
        raise HTTPException(404, "KPI不存在")
    period = period or datetime.now().strftime("%Y-%m")

    # ① 写值（upsert）
    existing = db.query(KPIValue).filter(
        KPIValue.kpi_id == kpi_id, KPIValue.period == period).first()
    if existing:
        existing.actual_value = float(actual_value)
        existing.source_type = "bot"
        kv = existing
    else:
        kv = KPIValue(kpi_id=kpi_id, entity_id=entity_id, period=period,
                      actual_value=float(actual_value), source_type="bot", data_status="verified")
        db.add(kv)
    db.commit()
    db.refresh(kv)

    # ② 跑该KPI关联的预警规则（复用 alert_rules 检查函数，非全量）
    alerts = []
    if run_check:
        from app.api.alert_rules import AlertRule, _check_static, _check_dynamic, _check_trend
        rules = db.query(AlertRule).filter(
            AlertRule.kpi_id == kpi_id, AlertRule.entity_id == entity_id,
            AlertRule.enabled == 1).all()
        for rule in rules:
            try:
                params = json.loads(rule.params) if isinstance(rule.params, str) else (rule.params or {})
                value = float(actual_value)
                if rule.rule_type == "static":
                    level, msg = _check_static(value, params, kpi)
                elif rule.rule_type == "dynamic":
                    level, msg = _check_dynamic(kpi_id, value, params, db)
                elif rule.rule_type == "trend_up":
                    level, msg = _check_trend(kpi_id, value, "up", params, db)
                elif rule.rule_type == "trend_down":
                    level, msg = _check_trend(kpi_id, value, "down", params, db)
                else:
                    continue
                if level and level != "green":
                    dup = db.query(KPIAlert).filter(
                        KPIAlert.kpi_id == kpi_id, KPIAlert.kpi_value_id == kv.id,
                        KPIAlert.alert_level == level, KPIAlert.status == "pending").first()
                    if not dup:
                        db.add(KPIAlert(entity_id=entity_id, kpi_id=kpi_id, kpi_value_id=kv.id, alert_level=level,
                                        alert_message=msg, status="pending", alert_type="bot"))
                        alerts.append({"rule_id": rule.id, "rule_type": rule.rule_type,
                                       "level": level, "message": msg})
            except Exception as e:
                logger.warning(f"聚合检查失败 rule={rule.id}: {e}")
        db.commit()
    return {"kpi_id": kpi_id, "kpi_code": kpi.kpi_code, "value": float(actual_value),
            "period": period, "alerts": alerts, "status": "ok"}


@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?, unit?, link_map_id?, link_causality?}

    分层授权收口（2026-09-03 P0-②）: Bot 创建KPI不得携带自定目标值——
    考核基准类（目标值）= 人拍板；target_value 一律显式置 0（满足治理必填，
    后续由人工设定或走 POST /api/cma/bot/ai/suggestions 建议通道待审批）。
    """
    entity_id = int(data.get("entity_id") or 1)
    from app.api.kpis import _validate_kpi_data, apply_calc_type_inference
    kpi_data = {k: v for k, v in data.items() if k not in ("entity_id", "link_map_id", "link_causality")}

    # 收口: 非零 target_value 直接拒绝（不静默丢弃，让调用方明确知晓规则）
    tv_raw = data.get("target_value")
    if tv_raw not in (None, 0, "0", 0.0):
        raise HTTPException(
            400,
            "Bot创建KPI不允许直写目标值（考核基准类=人拍板，2026-09-03分层授权）。"
            "请省略 target_value（自动置0，人工后续设定）或改走 POST /api/cma/bot/ai/suggestions 建议通道待审批。",
        )
    kpi_data["target_value"] = 0  # 显式置0（覆盖 0/"0"/None 三种传入）

    # ① 创建KPI（编码唯一 + 治理校验 + 强制企业）
    code = kpi_data.get("kpi_code", "")
    if not code:
        raise HTTPException(400, "kpi_code 必填")
    if db.query(KPIDefinition).filter(
            KPIDefinition.kpi_code == code, KPIDefinition.entity_id == entity_id).first():
        raise HTTPException(400, f"KPI编码 {code} 已存在")
    errs = _validate_kpi_data(kpi_data, db=db, is_update=False)
    if errs:
        raise HTTPException(422, detail={"message": "数据校验不通过", "errors": errs})
    kpi_data["entity_id"] = entity_id
    kpi_data = apply_calc_type_inference(kpi_data)
    kpi = KPIDefinition(**kpi_data)
    db.add(kpi)
    db.commit()
    db.refresh(kpi)

    # ② 关联战略地图
    link_map_id = data.get("link_map_id")
    if link_map_id:
        m = db.query(StrategicMap).filter(
            StrategicMap.id == link_map_id, StrategicMap.entity_id == entity_id).first()
        if m:
            kpi.map_id = link_map_id
            db.commit()

    # ③ 批量因果链（源=新KPI → 目标列表）
    links = []
    for c in data.get("link_causality") or []:
        tgt = c.get("target_kpi_id")
        if not tgt or int(tgt) == kpi.id:
            continue
        tgt_kpi = db.query(KPIDefinition).filter(
            KPIDefinition.id == int(tgt), KPIDefinition.entity_id == entity_id).first()
        if not tgt_kpi:
            continue
        if db.query(KPICausality).filter(
                KPICausality.source_kpi_id == kpi.id,
                KPICausality.target_kpi_id == int(tgt)).first():
            continue
        db.add(KPICausality(source_kpi_id=kpi.id, target_kpi_id=int(tgt),
                            strength=c.get("strength", 0.5), lag_months=c.get("lag_months", 1),
                            direction=c.get("direction", "positive")))
        links.append({"source": kpi.kpi_code, "target": tgt_kpi.kpi_code,
                      "strength": c.get("strength", 0.5)})
    db.commit()
    return {"kpi_id": kpi.id, "kpi_code": kpi.kpi_code, "map_id": link_map_id,
            "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=<csv>（表头同字段名，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}


# ═══════════════ 会计科目写入（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")
# category 白名单：以新30号准则科目字典的**库内实况**为准，不臆造枚举。
# 2026-09-12 修复（dama-cma-subjects-category-whitelist-20260912）：
#   原白名单 8 值漏了库内真实存在的 revenue(5 行: 6001/6011/6051/6111/611101) 与
#   profit_loss(1 行: 6901)，导致「对库内既有科目的任何合法更新」「按库内既有约定回填的
#   批次导入」整批 400 —— DAMA 独立复核实证（3 行收入类科目被阻断）。
#   证据：SELECT DISTINCT category FROM subjects → 9 值（2026-09-12 只读核验，207 行）：
#   asset/cost/detail/equity/expense/income/liability/profit_loss/revenue。
# 取值策略（采用任务方案 A「补齐白名单」，DAMA 反对 B 静默归一化——同一字典内同一概念
#   出现两种取值应由 DAMA 出标准化方案后统一改值，端点不得替用户改写数据）：
#   9 个库内真实值 + 'other'（既有契约的写入口袋值，缩小白名单会引入回归，故保留）。
# 防漂移：新增取值由 tests/test_bot_bridge.py::TestSubjectsWrite 的语料断言测试 +
#   scripts/check_subjects_category_drift.py（生产库 SELECT DISTINCT ⊄ 白名单 → 失败）兜底。
_SUBJ_CATEGORIES = ("asset", "liability", "equity", "cost", "income", "expense", "detail", "other",
                    "revenue", "profit_loss")
_SUBJ_LEVELS = (1, 2, 3)
_SUBJ_MAX_ROWS = 2000
# subject_code / parent_code 长度上限：任务规格为 ≤32，但 DB 列 subjects.subject_code /
# parent_code 实为 varchar(20) —— 取交集 20。若按 32 放行，21~32 字符的编码会穿过校验直达
# MySQL，报 (1406, Data too long) → 整批 500（2026-09-12 生产实测），故此处收敛到 DB 列宽。
_SUBJ_CODE_MAXLEN = 20
_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 非空且 ≤_SUBJ_CODE_MAXLEN（=20，DB varchar(20) 列宽）+ 批次内去重；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)):
    """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拦截",
    }