diff --git a/backend/app/api/alert_rules.py b/backend/app/api/alert_rules.py index aeaf333e..02b7f3ce 100644 --- a/backend/app/api/alert_rules.py +++ b/backend/app/api/alert_rules.py @@ -260,7 +260,7 @@ def run_all_alert_checks(db: Session = Depends(get_db)): value = latest_value.actual_value period = latest_value.period - params = rule.params or {} + import json; params = json.loads(rule.params) if isinstance(rule.params, str) else (rule.params or {}) alert_level = None alert_message = None @@ -534,7 +534,7 @@ def _check_forecast_alerts(db: Session) -> int: continue # 检查预测值是否超限 - params = rule.params or {} + import json; params = json.loads(rule.params) if isinstance(rule.params, str) else (rule.params or {}) params["kpi"] = kpi for forecast in latest_forecasts: value = forecast.predicted_cash diff --git a/backend/app/api/data_quality.py b/backend/app/api/data_quality.py index 882f9870..642f6cbe 100644 --- a/backend/app/api/data_quality.py +++ b/backend/app/api/data_quality.py @@ -253,10 +253,72 @@ def quality_stats(db: Session = Depends(get_db)): if cnt: type_counts[t] = cnt + # ── 数据审计看板统计 ── + # KPI完整度评分 + all_kpis = db.query(KPIDefinition).filter(KPIDefinition.status == "active").all() + total = len(all_kpis) + complete_kpis = 0 + missing_metadata_count = 0 + missing_data_count = 0 + stale_data_count = 0 + + from datetime import datetime, timedelta + six_months_ago = datetime.now() - timedelta(days=180) + + for kpi in all_kpis: + # 元数据完整度检查 + has_meta = all([ + kpi.formula and kpi.formula.strip(), + kpi.data_source and kpi.data_source.strip(), + kpi.data_owner and kpi.data_owner.strip(), + kpi.unit and kpi.unit.strip(), + kpi.target_value is not None, + ]) + if has_meta: + complete_kpis += 1 + else: + missing_metadata_count += 1 + + # 数据缺失检查（是否有实际值） + val = db.query(KPIValue).filter( + KPIValue.kpi_id == kpi.id, + KPIValue.actual_value.isnot(None), + ).first() + if not val: + missing_data_count += 1 + + # 超30天未更新预警 + latest_val = db.query(KPIValue).filter( + KPIValue.kpi_id == kpi.id, + KPIValue.actual_value.isnot(None), + ).order_by(KPIValue.period.desc()).first() + if latest_val and latest_val.calculated_at: + if latest_val.calculated_at < six_months_ago: + stale_data_count += 1 + + completeness_score = round(complete_kpis / total * 100, 1) if total > 0 else 0 + missing_rate = round(missing_data_count / total * 100, 1) if total > 0 else 0 + return { "total_kpis": total_kpis, "total_logs": total_logs, "open_logs": open_logs, "severity_counts": severity_counts, "type_counts": type_counts, + # 数据审计看板 + "completeness": { + "score": completeness_score, + "complete": complete_kpis, + "total": total, + "missing_metadata": missing_metadata_count, + }, + "data_missing": { + "count": missing_data_count, + "rate": missing_rate, + "total": total, + }, + "stale_data": { + "count": stale_data_count, + "threshold_days": 180, + }, } diff --git a/backend/app/api/kpis.py b/backend/app/api/kpis.py index a67a5636..ed61d8c3 100644 --- a/backend/app/api/kpis.py +++ b/backend/app/api/kpis.py @@ -406,12 +406,43 @@ 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 + + @router.post("") 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) + if errs: + raise HTTPException(422, detail={"message": "数据校验不通过", "errors": errs}) kpi = KPIDefinition(**data) db.add(kpi) db.commit() @@ -425,6 +456,10 @@ def update_kpi(kpi_id: int, data: dict, db: Session = Depends(get_db), user=WRIT kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first() if not kpi: raise HTTPException(404, "KPI不存在") + # 数据治理校验（更新时只检查传了但为空的字段） + errs = _validate_kpi_data(data, is_update=True) + if errs: + raise HTTPException(422, detail={"message": "数据校验不通过", "errors": errs}) for k, v in data.items(): if hasattr(kpi, k) and v is not None: setattr(kpi, k, v) diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index 510f02e3..ba0deff9 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -59,6 +59,8 @@ class KPIDefinition(Base): formula_desc = Column(String(500), nullable=True, comment="公式说明") data_source_type = Column(String(20), default="manual", comment="erp/business/excel/manual") data_source_config = Column(JSON, nullable=True, comment="数据源配置") + data_source = Column(String(500), default="待补充", comment="数据来源") + data_owner = Column(String(100), default="待指定", comment="数据责任人") frequency = Column(String(20), default="monthly", comment="daily/weekly/monthly/quarterly/yearly") unit = Column(String(50), default="%", comment="单位") target_value = Column(Float, nullable=True, comment="目标值") diff --git a/backend/scripts/cleanup_kpi_encoding.py b/backend/scripts/cleanup_kpi_encoding.py new file mode 100644 index 00000000..80e9a404 --- /dev/null +++ b/backend/scripts/cleanup_kpi_encoding.py @@ -0,0 +1,126 @@ +"""数据治理：编码规范清洗 — 检查KPI编码前缀与维度一致性 + 修复误分类""" +import pymysql +import os +import logging + +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger("encoding-cleanup") + +DB_USER = os.getenv("CMA_DB_USER", "cma_user") +DB_PASS = os.getenv("CMA_DB_PASS", "cma_pass_2026") +DB_HOST = os.getenv("CMA_DB_HOST", "127.0.0.1") +DB_PORT = int(os.getenv("CMA_DB_PORT", "3306")) +DB_NAME = os.getenv("CMA_DB_NAME", "cma") + +conn = pymysql.connect( + host=DB_HOST, port=DB_PORT, user=DB_USER, password=DB_PASS, database=DB_NAME, + charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor, +) +cur = conn.cursor() + +# 编码前缀 → 正确维度映射 +PREFIX_DIM_MAP = { + "F_": "finance", + "C_": "customer", + "P_": "process", + "L_": "learning", +} + +# 已知误分类修复（编码 → 正确维度） +KNOWN_FIXES = { + "F_QUALITY_RATE": "process", # 产品合格率 → 流程层 + "F_REWORK_RATE": "process", # 返工率 → 流程层 + "P_COST_CUT": "process", # 招待费砍半 → 流程层 + "P_TRAIN_PASS": "process", # Model C考核通过 → 流程层 +} + + +def run(): + logger.info("=== 编码规范清洗 开始 ===") + + cur.execute("SELECT id, entity_id, kpi_code, kpi_name, dimension FROM kpi_definitions WHERE status = 'active'") + kpis = cur.fetchall() + + issues = [] + fixes_applied = 0 + + for kpi in kpis: + kpi_code = kpi["kpi_code"] + current_dim = kpi["dimension"] + entity_id = kpi["entity_id"] + + # 检查前缀 + prefix = kpi_code[:2] if len(kpi_code) >= 2 else "" + expected_dim = PREFIX_DIM_MAP.get(prefix) + + if expected_dim and current_dim != expected_dim: + # 先检查是否在已知修复列表 + correct_dim = KNOWN_FIXES.get(kpi_code, expected_dim) + issues.append({ + "kpi_code": kpi_code, + "kpi_name": kpi["kpi_name"], + "current_dim": current_dim, + "expected_dim": correct_dim, + "prefix": prefix, + "entity_id": entity_id, + }) + + if kpi_code in KNOWN_FIXES: + logger.info(f" 🔧 修复: {kpi_code} ({kpi['kpi_name']}) {current_dim} → {correct_dim} (entity={entity_id})") + cur.execute( + "UPDATE kpi_definitions SET dimension = %s WHERE id = %s", + (correct_dim, kpi["id"]), + ) + fixes_applied += 1 + + conn.commit() + + # 输出报告 + logger.info(f"\n=== 清洗报告 ===") + logger.info(f" 检查KPI总数: {len(kpis)}") + logger.info(f" 编码-维度不一致: {len(issues)}") + logger.info(f" 已自动修复: {fixes_applied}") + + if issues: + logger.info(f"\n 不一致详情:") + for i, iss in enumerate(issues, 1): + status = "✅ 已修复" if iss["kpi_code"] in KNOWN_FIXES else "⚠️ 需人工确认" + logger.info(f" {i}. {iss['kpi_code']} ({iss['kpi_name']}) " + f"当前维度={iss['current_dim']}, 期望维度={iss['expected_dim']} [{status}]") + + # 检查未命名规范问题 + logger.info(f"\n 编码前缀统计:") + for prefix, dim in PREFIX_DIM_MAP.items(): + cur.execute("SELECT COUNT(*) as cnt FROM kpi_definitions WHERE kpi_code LIKE %s AND status='active'", (f"{prefix}%",)) + row = cur.fetchone() + cnt = row["cnt"] if row else 0 + logger.info(f" {prefix} → {dim}: {cnt} 个KPI") + + # 检查前缀不匹配编码 + cur.execute(""" + SELECT kpi_code, dimension FROM kpi_definitions + WHERE status='active' + AND ( + (kpi_code LIKE 'F_%' AND dimension != 'finance') + OR (kpi_code LIKE 'C_%' AND dimension != 'customer') + OR (kpi_code LIKE 'P_%' AND dimension != 'process') + OR (kpi_code LIKE 'L_%' AND dimension != 'learning') + ) + """) + remaining = cur.fetchall() + if remaining: + logger.warning(f"\n ⚠️ 仍有 {len(remaining)} 个KPI编码前缀与维度不匹配:") + for r in remaining: + logger.warning(f" {r['kpi_code']} → {r['dimension']}") + else: + logger.info(f"\n ✅ 所有KPI编码前缀与维度一致！") + + logger.info("\n=== 编码规范清洗 完成 ===") + + +if __name__ == "__main__": + try: + run() + finally: + cur.close() + conn.close() diff --git a/backend/scripts/migrate_data_governance.py b/backend/scripts/migrate_data_governance.py new file mode 100644 index 00000000..5c9f8cb2 --- /dev/null +++ b/backend/scripts/migrate_data_governance.py @@ -0,0 +1,105 @@ +"""数据治理 migration: 入库必检约束 + 元数据字段补充""" +import pymysql +import os +import logging + +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger("data-governance") + +DB_USER = os.getenv("CMA_DB_USER", "cma_user") +DB_PASS = os.getenv("CMA_DB_PASS", "cma_pass_2026") +DB_HOST = os.getenv("CMA_DB_HOST", "127.0.0.1") +DB_PORT = int(os.getenv("CMA_DB_PORT", "3306")) +DB_NAME = os.getenv("CMA_DB_NAME", "cma") + +conn = pymysql.connect( + host=DB_HOST, port=DB_PORT, user=DB_USER, password=DB_PASS, database=DB_NAME, + charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor, +) +cur = conn.cursor() + + +def run(): + logger.info("=== 数据治理 Migration 开始 ===") + + # ── 0. 添加缺失列 ── + logger.info("[步骤0] 检查并补充缺失列...") + + for col_name, col_def in [ + ("data_source", "ALTER TABLE kpi_definitions ADD COLUMN data_source VARCHAR(500) DEFAULT NULL COMMENT '数据来源' AFTER data_source_type"), + ("data_owner", "ALTER TABLE kpi_definitions ADD COLUMN data_owner VARCHAR(100) DEFAULT NULL COMMENT '数据责任人' AFTER data_source"), + ]: + cur.execute("SHOW COLUMNS FROM kpi_definitions LIKE %s", (col_name,)) + if not cur.fetchone(): + logger.info(f" 添加 {col_name} 列...") + cur.execute(col_def) + conn.commit() + logger.info(f" {col_name} 列已添加") + else: + logger.info(f" {col_name} 列已存在") + + # ── 1. 修复空值 ── + logger.info("[步骤1] 修复空值...") + + for col, default, label in [ + ("target_value", "0", "NULL target_value"), + ("unit", "'-'", "NULL/empty unit"), + ("formula", "'待补充'", "NULL/empty formula"), + ("data_source", "'待补充'", "NULL/empty data_source"), + ("data_owner", "'待指定'", "NULL/empty data_owner"), + ]: + if col in ("target_value",): + r = cur.execute(f"SELECT COUNT(*) as cnt FROM kpi_definitions WHERE {col} IS NULL") + else: + r = cur.execute(f"SELECT COUNT(*) as cnt FROM kpi_definitions WHERE {col} IS NULL OR {col} = ''") + row = cur.fetchone() + cnt = row["cnt"] if row else 0 + logger.info(f" {label}: {cnt} 条") + if cnt > 0: + if col in ("target_value",): + cur.execute(f"UPDATE kpi_definitions SET {col} = {default} WHERE {col} IS NULL") + else: + cur.execute(f"UPDATE kpi_definitions SET {col} = {default} WHERE {col} IS NULL OR {col} = ''") + logger.info(f" 已修复 {cur.rowcount} 条") + + conn.commit() + + # ── 2. 修改列约束为 NOT NULL ── + logger.info("[步骤2] 修改列约束...") + + alters = [ + ("target_value", "DECIMAL(15,2) NOT NULL DEFAULT 0"), + ("unit", "VARCHAR(50) NOT NULL DEFAULT '-'"), + ("formula", "TEXT NOT NULL"), + ("data_source", "VARCHAR(500) NOT NULL DEFAULT '待补充'"), + ("data_owner", "VARCHAR(100) NOT NULL DEFAULT '待指定'"), + ] + for col, col_type in alters: + col_comment = { + "target_value": "目标值", "unit": "单位", "formula": "计算公式", + "data_source": "数据来源", "data_owner": "数据责任人", + }[col] + try: + cur.execute(f"ALTER TABLE kpi_definitions MODIFY {col} {col_type} COMMENT '{col_comment}'") + logger.info(f" {col} → {col_type}") + except Exception as e: + logger.warning(f" {col} 修改失败: {e}") + + conn.commit() + + # ── 3. 验证 ── + logger.info("[步骤3] 验证约束...") + cur.execute("DESCRIBE kpi_definitions") + for col in cur.fetchall(): + if col['Field'] in ('target_value', 'unit', 'formula', 'data_source', 'data_owner'): + logger.info(f" {col['Field']}: Null={col['Null']}, Default={col['Default']}, Type={col['Type']}") + + logger.info("=== 数据治理 Migration 完成 ===") + + +if __name__ == "__main__": + try: + run() + finally: + cur.close() + conn.close() diff --git a/frontend/src/views/DataQuality.vue b/frontend/src/views/DataQuality.vue index a113be7d..c56fae92 100644 --- a/frontend/src/views/DataQuality.vue +++ b/frontend/src/views/DataQuality.vue @@ -44,6 +44,73 @@ + + + + + +
+
+ {{ stats.completeness?.score ?? '-' }}% +
+
+ 完整 {{ stats.completeness?.complete ?? 0 }} / 总计 {{ stats.completeness?.total ?? 0 }} +
+
+ 缺失元数据: {{ stats.completeness?.missing_metadata ?? 0 }} 个KPI +
+ +
+ + + + + +
+
+ {{ stats.data_missing?.rate ?? '-' }}% +
+
+ 无数据值的KPI: {{ stats.data_missing?.count ?? 0 }} +
+
+ 总计 {{ stats.data_missing?.total ?? 0 }} 个KPI +
+ +
+ + + + + +
+
+ {{ stats.stale_data?.count ?? 0 }} +
+
+ 个KPI超过 {{ stats.stale_data?.threshold_days ?? 180 }} 天未更新 +
+ + 建议立即检查数据源 + + + 数据更新正常 + +
+ + + + @@ -147,6 +214,20 @@ const typeChartOption = computed(() => ({ }], })) +const completenessColor = computed(() => { + const s = stats.value.completeness?.score ?? 0 + if (s >= 80) return '#67c23a' + if (s >= 50) return '#e6a23c' + return '#f56c6c' +}) + +const staleDataColor = computed(() => { + const c = stats.value.stale_data?.count ?? 0 + if (c === 0) return '#67c23a' + if (c <= 5) return '#e6a23c' + return '#f56c6c' +}) + async function loadStats() { try { const r: any = await dataQualityApi.stats(); stats.value = r } catch (e) {} } diff --git a/frontend/src/views/KPIDetail.vue b/frontend/src/views/KPIDetail.vue index 4f887b08..79eb9368 100644 --- a/frontend/src/views/KPIDetail.vue +++ b/frontend/src/views/KPIDetail.vue @@ -79,6 +79,37 @@ 保存 + + + + + + + {{ kpi.formula }} + 缺失 + + + {{ kpi.data_source }} + {{ kpi.data_source || '缺失' }} + + + {{ kpi.data_owner }} + {{ kpi.data_owner || '缺失' }} + + {{ kpi.unit || '-' }} + {{ kpi.target_value ?? '-' }} + {{ kpi.data_source_type || '-' }} + {{ dimLabel(kpi.dimension) }} + {{ kpi.frequency || '-' }} + {{ catLabel(kpi.category) || '-' }} + + @@ -354,6 +385,14 @@ function dimLabel(d: string) { return ({ finance: '财务', customer: '客户', function catLabel(c: string) { return CAT_MAP[c] || c } function dimTagType(d: string) { return ({ finance: '', customer: 'success', process: 'warning', learning: 'info' } as any)[d] || '' } +const metadataComplete = computed(() => { + if (!kpi.value) return false + const k = kpi.value + return !!(k.formula && k.formula.trim() && k.data_source && k.data_source.trim() && k.data_source !== '待补充' + && k.data_owner && k.data_owner.trim() && k.data_owner !== '待指定' + && k.unit && k.unit.trim() && k.target_value !== null && k.target_value !== undefined) +}) + const chartOption = computed(() => ({ tooltip: { trigger: 'axis' }, xAxis: { type: 'category', data: values.value.map(v => v.period) },