""" KPI字典库优化P0阶段 — 批量脚本 涵盖任务1-6: 激活模板KPI + 新增KPI + 修复员工满意度 """ import sys, os sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from dotenv import load_dotenv load_dotenv() from app.database import get_engine from sqlalchemy import text from datetime import datetime engine = get_engine() def log(msg): print(f"[{datetime.now():%H:%M:%S}] {msg}") def insert_kpi(conn, **kw): """安全插入KPI（检查唯一性）""" code = kw["kpi_code"] exists = conn.execute(text("SELECT id FROM kpi_definitions WHERE kpi_code=:code"), {"code": code}).fetchone() if exists: log(f" ⏭️ 已存在: {code} (id={exists[0]})") return exists[0], False cols = ", ".join(kw.keys()) vals = ", ".join(f":{k}" for k in kw.keys()) conn.execute(text(f"INSERT INTO kpi_definitions ({cols}) VALUES ({vals})"), kw) new_id = conn.execute(text("SELECT LAST_INSERT_ID()")).scalar() log(f" ✅ 新增: {code} (id={new_id})") return new_id, True with engine.connect() as conn: print("=" * 70) log("开始KPI字典库优化P0 — 任务1~6") print("=" * 70) # ========================================================= # 任务1: 激活4个系统模板KPI # ========================================================= log("\n【任务1】激活系统模板KPI") print("-" * 50) # 从kpi_templates读取模板数据 templates = conn.execute(text( "SELECT id, kpi_code, kpi_name, dimension, category, formula, formula_desc, " "unit, target_value, description FROM kpi_templates WHERE kpi_code IN " "('F_ASSET_TURNOVER', 'F_REVENUE_GROWTH', 'L_EMPLOYEE_TURNOVER', 'L_TECH_COVERAGE')" )).fetchall() for t in templates: tid, code, name, dim, cat, formula, fdesc, unit, target, desc = t nid, is_new = insert_kpi(conn, template_id=tid, is_system=1, kpi_code=code, kpi_name=name, dimension=dim, category=cat, formula=formula or "", formula_desc=fdesc or desc or "", unit=unit or "%", target_value=target, data_source_type="manual", frequency="monthly", status="active", epic="Epic2", created_at=datetime.now(), updated_at=datetime.now(), ) if is_new: # 设置阈值 if code == "F_ASSET_TURNOVER": conn.execute(text("UPDATE kpi_definitions SET threshold_green='>=1.0', threshold_yellow='>=0.8', threshold_red='<0.8' WHERE id=:id"), {"id": nid}) elif code == "F_REVENUE_GROWTH": conn.execute(text("UPDATE kpi_definitions SET threshold_green='>=20', threshold_yellow='>=15', threshold_red='<15' WHERE id=:id"), {"id": nid}) elif code == "L_EMPLOYEE_TURNOVER": conn.execute(text("UPDATE kpi_definitions SET threshold_green='<=5', threshold_yellow='<=10', threshold_red='>10' WHERE id=:id"), {"id": nid}) elif code == "L_TECH_COVERAGE": conn.execute(text("UPDATE kpi_definitions SET threshold_green='>=90', threshold_yellow='>=80', threshold_red='<80' WHERE id=:id"), {"id": nid}) # 更新模板使用计数 conn.execute(text("UPDATE kpi_templates SET usage_count = IFNULL(usage_count,0)+1 WHERE id=:id"), {"id": tid}) log(f" 阈值已配置") # ========================================================= # 任务2: 新增6个财务KPI # ========================================================= log("\n【任务2】新增6个财务KPI") print("-" * 50) finance_kpis = [ ("F_CURRENT_RATIO", "流动比率", "finance", "cash_risk", "流动资产/流动负债", "%", 200.0, ">=200", ">=150", "<150", "erp"), ("F_QUICK_RATIO", "速动比率", "finance", "cash_risk", "(流动资产-存货)/流动负债", "%", 100.0, ">=100", ">=80", "<80", "erp"), ("F_INV_DAYS", "存货周转天数", "finance", "asset_efficiency", "365/存货周转率", "天", 45.0, "<=30", "<=45", ">45", "erp"), ("F_ROI", "总资产报酬率(ROI)", "finance", "profitability", "净利润/平均总资产*100", "%", 8.0, ">=12", ">=8", "<8", "erp"), ("F_QUALITY_RATE", "产品合格率", "finance", "delivery_quality", "正品数/总产量*100", "%", 98.0, ">=99", ">=98", "<98", "erp"), ("F_REWORK_RATE", "返工率", "finance", "delivery_quality", "返工工时/总工时*100", "%", 5.0, "<=3", "<=5", ">5", "erp"), ] for code, name, dim, cat, formula, unit, target, tg, ty, tr, src in finance_kpis: nid, is_new = insert_kpi(conn, is_system=0, kpi_code=code, kpi_name=name, dimension=dim, category=cat, formula=formula, formula_desc=f"CMA标准{name}指标", unit=unit, target_value=target, threshold_green=tg, threshold_yellow=ty, threshold_red=tr, data_source_type=src, frequency="monthly", status="active", epic="Epic2", created_at=datetime.now(), updated_at=datetime.now(), ) # ========================================================= # 任务3: 新增3个客户KPI # ========================================================= log("\n【任务3】新增3个客户KPI") print("-" * 50) # 激活C_CUSTOMER_CONCENTRATION模板 t = conn.execute(text( "SELECT id, kpi_code, kpi_name, dimension, category, formula, formula_desc, " "unit, target_value, description FROM kpi_templates WHERE kpi_code='C_CUSTOMER_CONCENTRATION'" )).fetchone() if t: nid, is_new = insert_kpi(conn, template_id=t[0], is_system=1, kpi_code="C_CUST_CONCENTRATION", kpi_name="客户集中度", dimension=t[3], category=t[4], formula=t[5] or "", formula_desc=t[6] or t[9] or "", unit=t[7] or "%", target_value=t[8], data_source_type="erp", frequency="monthly", status="active", epic="Epic2", threshold_green="<=20", threshold_yellow="<=30", threshold_red=">30", created_at=datetime.now(), updated_at=datetime.now(), ) if is_new: conn.execute(text("UPDATE kpi_templates SET usage_count = IFNULL(usage_count,0)+1 WHERE id=:id"), {"id": t[0]}) # C_RETENTION_RATE insert_kpi(conn, is_system=0, kpi_code="C_RETENTION_RATE", kpi_name="客户保留率", dimension="customer", category="customer_scale", formula="续约客户数/总客户数*100", formula_desc="CMA标准客户保留率指标，反映客户粘性", unit="%", target_value=85.0, threshold_green=">=90", threshold_yellow=">=85", threshold_red="<85", data_source_type="erp", frequency="monthly", status="active", epic="Epic2", created_at=datetime.now(), updated_at=datetime.now(), ) # C_CAC — 注意C_CAC可能已存在（reset_kpi_dict中有定义），改用C_ACQUISITION_COST insert_kpi(conn, is_system=0, kpi_code="C_ACQUISITION_COST", kpi_name="获客成本(CAC)", dimension="customer", category="customer_scale", formula="营销费用/新客户数", formula_desc="CMA标准客户获取成本指标", unit="元", target_value=None, threshold_green="<=500", threshold_yellow="<=800", threshold_red=">800", data_source_type="erp", frequency="monthly", status="active", epic="Epic2", created_at=datetime.now(), updated_at=datetime.now(), ) # ========================================================= # 任务4: 新增2个流程KPI # ========================================================= log("\n【任务4】新增2个流程KPI") print("-" * 50) insert_kpi(conn, is_system=0, kpi_code="P_PASS_RATE", kpi_name="过程合格率", dimension="process", category="delivery_quality", formula="过程合格批次数/总检验批次数*100", formula_desc="流程层关注过程质量合格率，区别于财务层产品合格率", unit="%", target_value=95.0, threshold_green=">=98", threshold_yellow=">=95", threshold_red="<95", data_source_type="erp", frequency="monthly", status="active", epic="Epic2", created_at=datetime.now(), updated_at=datetime.now(), ) insert_kpi(conn, is_system=0, kpi_code="P_REWORK_RATE", kpi_name="流程返工率", dimension="process", category="delivery_quality", formula="返工批次/总生产批次*100", formula_desc="流程层返工率，体现过程质量控制水平", unit="%", target_value=5.0, threshold_green="<=3", threshold_yellow="<=5", threshold_red=">5", data_source_type="erp", frequency="monthly", status="active", epic="Epic2", created_at=datetime.now(), updated_at=datetime.now(), ) # ========================================================= # 任务5: 新增2个学习KPI # ========================================================= log("\n【任务5】新增2个学习KPI") print("-" * 50) # 激活L_TRAINING_HOURS模板 t = conn.execute(text( "SELECT id, kpi_code, kpi_name, dimension, category, formula, formula_desc, " "unit, target_value, description FROM kpi_templates WHERE kpi_code='L_TRAINING_HOURS'" )).fetchone() if t: nid, is_new = insert_kpi(conn, template_id=t[0], is_system=1, kpi_code="L_TRAINING_HOURS", kpi_name="人均培训时长", dimension=t[3], category=t[4], formula=t[5] or "", formula_desc=t[6] or t[9] or "", unit=t[7] or "小时", target_value=t[8], data_source_type="manual", frequency="monthly", status="active", epic="Epic2", threshold_green=">=40", threshold_yellow=">=20", threshold_red="<20", created_at=datetime.now(), updated_at=datetime.now(), ) if is_new: conn.execute(text("UPDATE kpi_templates SET usage_count = IFNULL(usage_count,0)+1 WHERE id=:id"), {"id": t[0]}) # L_COMPETENCY insert_kpi(conn, is_system=0, kpi_code="L_COMPETENCY", kpi_name="关键岗位胜任度", dimension="learning", category="talent_pipeline", formula="胜任评估得分≥80分人数/关键岗位总人数*100", formula_desc="CMA标准关键岗位胜任度指标，通过胜任力评估获取", unit="%", target_value=85.0, threshold_green=">=90", threshold_yellow=">=85", threshold_red="<85", data_source_type="manual", frequency="quarterly", status="active", epic="Epic2", created_at=datetime.now(), updated_at=datetime.now(), ) # ========================================================= # 任务6: 修复员工满意度阈值 # ========================================================= log("\n【任务6】修复员工满意度阈值") print("-" * 50) sat = conn.execute(text("SELECT id FROM kpi_definitions WHERE kpi_code='L_EMPLOYEE_SAT'")).fetchone() if sat: conn.execute(text( "UPDATE kpi_definitions SET threshold_green='>=80', threshold_yellow='>=60', threshold_red='<60' WHERE id=:id" ), {"id": sat[0]}) log(f" ✅ 员工满意度(id={sat[0]}) 阈值已设置: 绿>=80, 黄>=60, 红<60") else: log(" ⚠️ 员工满意度KPI不存在") # ========================================================= # 提交事务 # ========================================================= conn.commit() # 统计 total = conn.execute(text("SELECT COUNT(*) FROM kpi_definitions WHERE status='active'")).scalar() by_dim = conn.execute(text( "SELECT dimension, COUNT(*) FROM kpi_definitions WHERE status='active' GROUP BY dimension ORDER BY dimension" )).fetchall() no_threshold = conn.execute(text( "SELECT kpi_code, kpi_name FROM kpi_definitions WHERE status='active' AND (threshold_green IS NULL OR threshold_yellow IS NULL OR threshold_red IS NULL)" )).fetchall() print("\n" + "=" * 70) log(f"✅ 任务1~6完成！KPI字典共 {total} 条") for d, c in by_dim: print(f" {d}: {c} 条") if no_threshold: print(f"\n⚠️ 以下KPI仍缺阈值:") for code, name in no_threshold: print(f" {code}: {name}") else: print("\n✅ 所有KPI均有阈值配置") print("=" * 70)