diff --git a/backend/app/api/kpi_governance.py b/backend/app/api/kpi_governance.py new file mode 100644 index 00000000..f756f6d2 --- /dev/null +++ b/backend/app/api/kpi_governance.py @@ -0,0 +1,196 @@ +"""KPI数据治理4条规则：入库必检 + 元数据必填 + 编码规范 + 战略分级 + +规则1: 入库必检 — dimension/target_value/unit 必填（创建/更新强制拦截） +规则2: 元数据必填 — formula/data_source/data_owner 不能为空或占位符（待补充/待指定/-） +规则3: 编码规范 — kpi_code 必须以 F_/C_/P_/L_ 前缀开头且与 dimension 一致；禁止跨层同名 +规则4: 战略/运营分级 — kpi_level 必须是 strategic/operational + +API: +- POST /api/cma/kpi/validate 校验单个KPI数据 → {valid, errors} +- GET /api/cma/kpi/governance/audit 全量校验 → 按4条规则分组的不合规清单 +""" +from typing import List, Optional + +from fastapi import APIRouter, Depends +from sqlalchemy.orm import Session + +from app.database import get_db +from app.auth_middleware import require_auth, require_role +from app.models import KPIDefinition + +router = APIRouter( + prefix="/api/cma/kpi", + tags=["KPI数据治理"], + dependencies=[Depends(require_role("ceo", "finance", "business", "it"))], +) + +# 维度 → 编码前缀 +DIM_PREFIX = {"finance": "F", "customer": "C", "process": "P", "learning": "L"} +VALID_LEVELS = ("strategic", "operational") +# 视为"未完善"的占位符值 +PLACEHOLDERS = ("待补充", "待指定", "待完善", "待定", "暂无", "TBD", "tbd", "-", "--", "N/A", "n/a") + +# 规则2必填元数据字段 +META_FIELDS = [ + ("formula", "计算公式"), + ("data_source", "数据来源"), + ("data_owner", "数据责任人"), +] + + +def _clean_str(val) -> str: + if val is None: + return "" + if isinstance(val, str): + return val.strip() + return str(val).strip() + + +def _is_placeholder(val) -> bool: + """空值或占位符（待补充/待指定/- 等）视为未完善""" + s = _clean_str(val) + if not s: + return True + return s in PLACEHOLDERS + + +def validate_kpi_payload( + data: dict, + db: Session = None, + current_kpi_id: Optional[int] = None, + is_update: bool = False, +) -> List[dict]: + """校验单个KPI数据（4条规则）。 + + - data: 提交的KPI字段字典（创建或更新的载荷） + - db: SQLAlchemy Session（用于跨层同名/编码唯一性检查，可为None） + - current_kpi_id: 更新时传KPI自身id（避免自检误报） + - is_update: 更新模式 — 仅校验载荷中显式出现的字段 + + 返回 [{rule: int, field: str, message: str}, ...]，空列表=合规。 + """ + issues: List[dict] = [] + + def add(rule: int, field: str, message: str): + issues.append({"rule": rule, "field": field, "message": message}) + + code = _clean_str(data.get("kpi_code")) + dimension = _clean_str(data.get("dimension")) + has_code = bool(code) + + # ── 规则1: 入库必检 dimension/target_value/unit ── + if not is_update or "dimension" in data: + if not dimension: + add(1, "dimension", "缺少dimension(所属维度: finance/customer/process/learning)") + if not is_update or "target_value" in data: + tv = data.get("target_value") + if tv is None or (isinstance(tv, str) and _clean_str(tv) == ""): + add(1, "target_value", "缺少target_value(目标值)") + if not is_update or "unit" in data: + if _is_placeholder(data.get("unit")): + add(1, "unit", "缺少unit(单位)") + + # ── 规则2: 元数据必填（不能为空或占位符）── + for field, label in META_FIELDS: + if not is_update or field in data: + if _is_placeholder(data.get(field)): + add(2, field, f"元数据未完善: {label}({field})不能为空或占位符(待补充/待指定/-)") + + # ── 规则3: 编码规范 ── + if not is_update or "kpi_code" in data: + if not has_code: + add(3, "kpi_code", "缺少kpi_code(KPI编码)") + else: + prefix = code.split("_")[0] if "_" in code else code + if prefix not in ("F", "C", "P", "L"): + add(3, "kpi_code", f"编码前缀不符: {code} 应以F_/C_/P_/L_开头") + elif dimension and DIM_PREFIX.get(dimension) and prefix != DIM_PREFIX[dimension]: + expected = DIM_PREFIX[dimension] + add(3, "kpi_code", + f"编码前缀与维度不符: {code} 前缀{prefix}_ 与维度{dimension}(应为{expected}_)不一致") + + # 禁止跨层同名: 同一kpi_code不能用于不同dimension + if has_code and db is not None: + dup = db.query(KPIDefinition).filter(KPIDefinition.kpi_code == code).first() + if dup and dup.id != current_kpi_id: + if dimension and dup.dimension and dup.dimension != dimension: + add(3, "kpi_code", + f"跨层同名: {code} 已用于维度{dup.dimension}，不能用于维度{dimension}") + elif not dimension: + add(3, "kpi_code", f"编码已存在: {code} 已注册(维度{dup.dimension})，不能重复使用") + + # ── 规则4: 战略/运营分级 ── + if not is_update or "kpi_level" in data: + lv = data.get("kpi_level") + if lv is not None and lv not in VALID_LEVELS: + add(4, "kpi_level", f"kpi_level必须是strategic或operational，当前值: {lv}") + + return issues + + +def kpi_issues_message(issues: List[dict]) -> List[str]: + """issue dict列表 → 纯文本错误列表""" + return [i["message"] for i in issues] + + +@router.post("/validate") +def validate_kpi( + kpi_data: dict, + db: Session = Depends(get_db), + current_user=Depends(require_auth), +): + """校验单个KPI数据（不入库）。 + + 请求体: KPI字段字典，可选 kpi_id 标识正在编辑的KPI（避免跨层同名误报）。 + 返回: {"valid": bool, "errors": [str], "details": [{rule, field, message}]} + """ + kpi_id = kpi_data.get("kpi_id") if isinstance(kpi_data.get("kpi_id"), int) else None + issues = validate_kpi_payload(kpi_data, db=db, current_kpi_id=kpi_id) + return { + "valid": len(issues) == 0, + "errors": kpi_issues_message(issues), + "details": issues, + } + + +@router.get("/governance/audit") +def governance_audit( + db: Session = Depends(get_db), + current_user=Depends(require_auth), +): + """全量校验所有活跃KPI，输出按4条规则分组的不合规清单。""" + kpis = db.query(KPIDefinition).filter(KPIDefinition.status == "active").all() + + non_compliant = [] + by_rule: dict = {1: [], 2: [], 3: [], 4: []} + for k in kpis: + payload = {c.name: getattr(k, c.name) for c in k.__table__.columns} + issues = validate_kpi_payload(payload, db=db, current_kpi_id=k.id) + if issues: + entry = { + "kpi_id": k.id, + "kpi_code": k.kpi_code, + "kpi_name": k.kpi_name, + "dimension": k.dimension, + "kpi_level": k.kpi_level, + "issues": issues, + } + non_compliant.append(entry) + for i in issues: + by_rule.setdefault(i["rule"], []).append({ + "kpi_id": k.id, + "kpi_code": k.kpi_code, + "kpi_name": k.kpi_name, + "field": i["field"], + "message": i["message"], + }) + + rule_counts = {str(r): len(items) for r, items in by_rule.items()} + return { + "total": len(kpis), + "compliant": len(kpis) - len(non_compliant), + "non_compliant_count": len(non_compliant), + "rule_counts": rule_counts, + "by_rule": {str(r): items for r, items in by_rule.items()}, + "non_compliant": non_compliant, + } diff --git a/backend/app/api/kpis.py b/backend/app/api/kpis.py index 8faf5156..ace05c5d 100644 --- a/backend/app/api/kpis.py +++ b/backend/app/api/kpis.py @@ -9,6 +9,7 @@ import json from app.database import get_db from app.auth_middleware import require_auth, require_role, filter_kpis_by_role, kpi_visible_dims from app.models import StrategicMap, MapObjective, KPIDefinition, KPIValue, KPIAlert, OperationLog, Entity, KPICausality, KPIHierarchy +from app.api.kpi_governance import validate_kpi_payload, kpi_issues_message router = APIRouter(prefix="/api/cma/kpis", tags=["KPI字典"], dependencies=[Depends(require_role("ceo", "finance", "business", "it"))], @@ -27,6 +28,7 @@ def list_kpis( epic: Optional[str] = None, category: Optional[str] = None, entity_id: Optional[int] = None, + kpi_level: Optional[str] = None, db: Session = Depends(get_db), current_user = Depends(require_auth), ): @@ -45,6 +47,8 @@ def list_kpis( query = query.filter(KPIDefinition.category.in_(cats_list)) if entity_id is not None: query = query.filter(KPIDefinition.entity_id == entity_id) + if kpi_level: + query = query.filter(KPIDefinition.kpi_level == kpi_level) total = query.count() kpis = query.order_by(KPIDefinition.kpi_code).offset((page-1)*page_size).limit(page_size).all() result = {"total": total, "page": page, "page_size": page_size, "data": [kpi_to_dict(k) for k in kpis]} @@ -477,31 +481,10 @@ def get_kpi(kpi_id: int, db: Session = Depends(get_db)): return kpi_to_dict(kpi) -def _validate_kpi_data(data: dict, is_update: bool = False): - """数据治理：入库必检 + 元数据校验""" - errors = [] - - # 规则1: target_value 必填 - tv = data.get("target_value") - if tv is None or (isinstance(tv, (int, float)) and tv < 0 and not is_update): - if not is_update or "target_value" in data: - if tv is None: - errors.append("目标值(target_value)不能为空") - - # 规则1: unit 必填 - unit = data.get("unit") - if not unit or (isinstance(unit, str) and unit.strip() == ""): - if not is_update or "unit" in data: - errors.append("单位(unit)不能为空") - - # 规则2: 元数据必填 — formula/data_source/data_owner - for field, label in [("formula", "计算公式"), ("data_source", "数据来源"), ("data_owner", "数据责任人")]: - val = data.get(field) - if not val or (isinstance(val, str) and val.strip() == ""): - if not is_update or field in data: - errors.append(f"元数据字段'{label}'({field})不能为空") - - return errors +def _validate_kpi_data(data: dict, db: Session, current_kpi_id: Optional[int] = None, is_update: bool = False): + """数据治理4条规则校验（入库必检+元数据+编码规范+战略分级），返回错误信息列表""" + issues = validate_kpi_payload(data, db=db, current_kpi_id=current_kpi_id, is_update=is_update) + return kpi_issues_message(issues) @router.post("") @@ -510,8 +493,8 @@ def create_kpi(data: dict, db: Session = Depends(get_db), user=WRITE_ROLES): existing = db.query(KPIDefinition).filter(KPIDefinition.kpi_code == data.get("kpi_code", "")).first() if existing: raise HTTPException(400, f"KPI编码 {data['kpi_code']} 已存在") - # 数据治理校验 - errs = _validate_kpi_data(data, is_update=False) + # 数据治理校验（规则1强制拦截） + errs = _validate_kpi_data(data, db=db, is_update=False) if errs: raise HTTPException(422, detail={"message": "数据校验不通过", "errors": errs}) kpi = KPIDefinition(**data) @@ -528,7 +511,7 @@ def update_kpi(kpi_id: int, data: dict, db: Session = Depends(get_db), user=WRIT if not kpi: raise HTTPException(404, "KPI不存在") # 数据治理校验（更新时只检查传了但为空的字段） - errs = _validate_kpi_data(data, is_update=True) + errs = _validate_kpi_data(data, db=db, current_kpi_id=kpi_id, is_update=True) if errs: raise HTTPException(422, detail={"message": "数据校验不通过", "errors": errs}) for k, v in data.items(): diff --git a/backend/app/api/maps.py b/backend/app/api/maps.py index 2f62d784..20566514 100644 --- a/backend/app/api/maps.py +++ b/backend/app/api/maps.py @@ -327,8 +327,12 @@ def _merge_map_objectives(m, db): @router.get("/{map_id}/review") -def get_map_review(map_id: int, db: Session = Depends(get_db)): - """战略回顾会：返回目标状态、KPI值、改善行动""" +def get_map_review(map_id: int, level: Optional[str] = None, db: Session = Depends(get_db)): + """战略回顾会：返回目标状态、KPI值、改善行动 + + level: 可选 strategic/operational — 战略回顾(默认strategic)只显示战略级KPI； + 不传则返回全部KPI（向后兼容）。 + """ from app.models import KPIDefinition, KPIValue, ActionPlan m = db.query(StrategicMap).filter(StrategicMap.id == map_id).first() if not m: @@ -408,6 +412,9 @@ def get_map_review(map_id: int, db: Session = Depends(get_db)): kpi_def = kpi_map.get(code) if not kpi_def: continue + # 规则4: 战略地图默认只显示strategic级KPI（level过滤） + if level and kpi_def.kpi_level != level: + continue lv = latest_values.get(kpi_def.id, {}) actual = lv.get("actual_value") target = kpi_def.target_value diff --git a/backend/app/api/templates.py b/backend/app/api/templates.py index ceac6c69..16ef4a32 100644 --- a/backend/app/api/templates.py +++ b/backend/app/api/templates.py @@ -9,6 +9,7 @@ from datetime import datetime from app.database import get_db from app.auth_middleware import require_role from app.models import KPITemplate, KPIDefinition, OperationLog +from app.api.kpi_governance import validate_kpi_payload, kpi_issues_message router = APIRouter(prefix="/api/cma/templates", tags=["KPI模板库"], dependencies=[Depends(require_role("ceo", "finance", "business", "it"))], @@ -120,6 +121,20 @@ def instantiate_template(template_id: int, data: dict, db: Session = Depends(get if existing: raise HTTPException(400, f"KPI编码 {kpi_code} 已存在，请修改") + # 数据治理：入库必检(规则1) + 编码规范(规则3) + merged = { + "kpi_code": kpi_code, + "kpi_name": kpi_name, + "dimension": data.get("dimension", t.dimension), + "category": data.get("category", t.category), + "formula": data.get("formula", t.formula), + "unit": data.get("unit", t.unit or "%"), + "target_value": data.get("target_value", t.target_value), + } + gov_issues = [i for i in validate_kpi_payload(merged, db=db) if i["rule"] in (1, 3)] + if gov_issues: + raise HTTPException(422, detail={"message": "数据校验不通过", "errors": kpi_issues_message(gov_issues)}) + kpi = KPIDefinition( template_id=t.id, is_system=0, # 从模板实例化的KPI不是系统预置 diff --git a/backend/app/main.py b/backend/app/main.py index d13a33c3..b2995202 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -5,7 +5,7 @@ from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from dotenv import load_dotenv from app.database import init_db -from app.api import auth, kpis, templates, maps, dashboard, data, alerts, ai_analysis, alert_rules, users, thresholds, notifications, permissions, action_plans, alignment, org, objectives, versions, budget, cost, predict, reports, security, knowledge, bot_bridge, bot_bridge_v2, lead, tenant, customer_dashboard, deviation_push, budget_generate, knowledge_articles, kpi_causality, data_quality, bi_reports, entities, bsc_layers, okr, okr_templates, subjects, driver_budget, bot_kpis, bot_iron_law, analysis_results, expenses, cash, tax_compliance, verify +from app.api import auth, kpis, kpi_governance, templates, maps, dashboard, data, alerts, ai_analysis, alert_rules, users, thresholds, notifications, permissions, action_plans, alignment, org, objectives, versions, budget, cost, predict, reports, security, knowledge, bot_bridge, bot_bridge_v2, lead, tenant, customer_dashboard, deviation_push, budget_generate, knowledge_articles, kpi_causality, data_quality, bi_reports, entities, bsc_layers, okr, okr_templates, subjects, driver_budget, bot_kpis, bot_iron_law, analysis_results, expenses, cash, tax_compliance, verify from app.utils.cache import clear_all as clear_cache, delete as delete_cache from scripts.erp_sync import run_sync as run_erp_sync from app.auth_middleware import require_auth @@ -32,6 +32,7 @@ app.add_middleware( app.include_router(auth.router) app.include_router(kpis.router) +app.include_router(kpi_governance.router) app.include_router(templates.router) app.include_router(maps.router) app.include_router(dashboard.router)