1257 lines
53 KiB
Python
1257 lines
53 KiB
Python
"""
|
||
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(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拦截",
|
||
}
|