fix: 战略地图P0修复 — KPI code统一+放开同层连线+数据迁移

This commit is contained in:
Hermes CI Fix
2026-07-14 16:02:33 +08:00
parent c7d436ec7a
commit 6218a7d2db
8775 changed files with 4125 additions and 1758555 deletions
Binary file not shown.
Binary file not shown.
Binary file not shown.
+45
View File
@@ -0,0 +1,45 @@
#!/bin/bash
# 每日ERP数据同步 — 凌晨3点执行
# 由Hermes cron调度
# P1扩展: 增加CRM模块和产模块同步
source /root/cma-management/backend/venv/bin/activate
cd /root/cma-management/backend
log() {
echo "[$(date '+%Y-%m-%d %H:%M:%S')] $*"
}
log "=" 60
log "ERP每日同步开始"
log "=" 60
# 1. 总账模块同步(原有)
log "--- 总账模块同步 ---"
python3 -c "
from scripts.erp_sync import run_sync
import logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')
run_sync(dry_run=False, use_api=False)
" >> /root/cma-management/backend/logs/erp_sync_daily.log 2>&1
# 2. CRM模块同步(P1新增)
log "--- CRM模块同步 ---"
python3 -c "
from scripts.erp_sync import run_sync
import logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')
run_sync(dry_run=False, kpi_codes=['C_RETENTION_RATE', 'C_NEW_CLIENTS'], use_api=False)
" >> /root/cma-management/backend/logs/erp_sync_crm.log 2>&1
# 3. 生产模块同步(P1新增)
log "--- 生产模块同步 ---"
python3 -c "
from scripts.erp_sync import run_sync
import logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')
run_sync(dry_run=False, kpi_codes=['F_QUALITY_RATE', 'F_REWORK_RATE'], use_api=False)
" >> /root/cma-management/backend/logs/erp_sync_production.log 2>&1
log "=" 60
log "ERP每日同步完成"
log "=" 60
+127
View File
@@ -0,0 +1,127 @@
"""
ERP数据源配置首批 — 任务8
配置data_source_config + 同步端点 + 定时任务
"""
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()
# 首批8个ERP-KPI的数据源配置
ERP_SOURCES = [
{
"name": "ERP总账-营业收入",
"kpi_codes": ["F_REVENUE"],
"source_type": "erp",
"api_endpoint": "http://127.0.0.1:8300/api/v1/stats/monthly",
"query_sql": "SELECT COALESCE(SUM(SumMoney), 0) as value FROM MasterBill WHERE BillType=1 AND BillState>=3 AND Period=:period",
"sync_type": "daily",
},
{
"name": "ERP总账-毛利率",
"kpi_codes": ["F_GROSS_MARGIN", "F_PROFIT_RATE"],
"source_type": "erp",
"api_endpoint": "http://127.0.0.1:8300/api/v1/stats/gross-profit",
"query_sql": "SELECT CASE WHEN SUM(SumMoney) > 0 THEN ROUND((SUM(SumMoney)-COALESCE(SUM(SumCostMoney),0))/SUM(SumMoney)*100,2) ELSE 0 END as value FROM MasterBill WHERE BillType=1 AND BillState>=3 AND Period=:period",
"sync_type": "daily",
},
{
"name": "ERP总账-净利润",
"kpi_codes": ["F_NET_PROFIT", "F_NET_PROFIT_RATE"],
"source_type": "erp",
"api_endpoint": "http://127.0.0.1:8300/api/v1/stats/monthly",
"query_sql": "SELECT COALESCE(SUM(SumMoney),0)-COALESCE(SUM(SumCostMoney),0)-COALESCE(SUM(SumExpense),0) as value FROM MasterBill WHERE BillType IN (1,2,3) AND BillState>=3 AND Period=:period",
"sync_type": "daily",
},
{
"name": "ERP总账-费用率",
"kpi_codes": ["F_COST_RATIO"],
"source_type": "erp",
"api_endpoint": "http://127.0.0.1:8300/api/v1/stats/monthly",
"query_sql": "SELECT CASE WHEN SUM(CASE WHEN BillType=1 THEN SumMoney ELSE 0 END) > 0 THEN ROUND((COALESCE(SUM(CASE WHEN BillType=2 THEN SumMoney ELSE 0 END),0)+COALESCE(SUM(CASE WHEN BillType=3 THEN SumMoney ELSE 0 END),0))/SUM(CASE WHEN BillType=1 THEN SumMoney ELSE 0 END)*100,2) ELSE 0 END as value FROM MasterBill WHERE BillState>=3 AND Period=:period",
"sync_type": "daily",
},
{
"name": "ERP总账-经营性现金流",
"kpi_codes": ["F_OP_CFLOW", "F_CASH_FLOW"],
"source_type": "erp",
"api_endpoint": "http://127.0.0.1:8300/api/v1/cashflow",
"query_sql": "SELECT COALESCE(SUM(CashIn),0)-COALESCE(SUM(CashOut),0) as value FROM CashFlow WHERE Period=:period",
"sync_type": "daily",
},
{
"name": "ERP应收-周转天数",
"kpi_codes": ["F_AR_DAYS", "F_AR_TURNOVER"],
"source_type": "erp",
"api_endpoint": "http://127.0.0.1:8300/api/v1/ar/aging",
"query_sql": "SELECT CASE WHEN total_sales > 0 THEN ROUND(AVG_receivable/total_sales*365,0) ELSE 0 END as value FROM (SELECT SUM(CASE WHEN BillType=1 THEN SumMoney ELSE 0 END) as total_sales, AVG(CASE WHEN BillType=1 THEN SumMoney ELSE 0 END) as AVG_receivable FROM MasterBill WHERE BillState>=3 AND Period<=:period) t",
"sync_type": "daily",
},
{
"name": "ERP交付-及时率",
"kpi_codes": ["P_DELIVERY", "P_DELIVERY_ON_TIME"],
"source_type": "erp",
"api_endpoint": "http://127.0.0.1:8300/api/v1/delivery/rate",
"query_sql": "SELECT CASE WHEN total_orders > 0 THEN ROUND(on_time_orders/total_orders*100,2) ELSE 0 END as value FROM (SELECT COUNT(*) as total_orders, SUM(CASE WHEN ActualDelivery<=PlanDelivery THEN 1 ELSE 0 END) as on_time_orders FROM OrderDelivery WHERE Period=:period) t",
"sync_type": "daily",
},
{
"name": "ERP质量-缺陷率",
"kpi_codes": ["P_BUG_RATE", "P_DEFECT_RATE"],
"source_type": "erp",
"api_endpoint": "http://127.0.0.1:8300/api/v1/quality/defect",
"query_sql": "SELECT CASE WHEN total_qty > 0 THEN ROUND(defect_qty/total_qty*100,2) ELSE 0 END as value FROM (SELECT COUNT(*) as total_qty, SUM(CASE WHEN QualityStatus='NG' THEN 1 ELSE 0 END) as defect_qty FROM QualityInspection WHERE Period=:period) t",
"sync_type": "daily",
},
]
def log(msg):
print(f"[{datetime.now():%H:%M:%S}] {msg}")
with engine.connect() as conn:
print("=" * 70)
log("配置ERP数据源 (data_source_config)")
print("=" * 70)
existing = conn.execute(text("SELECT COUNT(*) FROM data_source_config")).scalar()
log(f"当前已有 {existing} 条数据源配置")
created = 0
for src in ERP_SOURCES:
exists = conn.execute(
text("SELECT id FROM data_source_config WHERE name=:name"),
{"name": src["name"]}
).fetchone()
if exists:
log(f" ⏭️ 已存在: {src['name']} (id={exists[0]})")
continue
conn.execute(text("""
INSERT INTO data_source_config
(name, source_type, api_endpoint, query_sql, sync_type, status, created_at)
VALUES (:name, :source_type, :api_endpoint, :query_sql, :sync_type, 'active', NOW())
"""), {
"name": src["name"],
"source_type": src["source_type"],
"api_endpoint": src["api_endpoint"],
"query_sql": src["query_sql"],
"sync_type": src["sync_type"],
})
created += 1
log(f" ✅ 新增: {src['name']} — 关联KPI: {', '.join(src['kpi_codes'])}")
conn.commit()
log(f"\n✅ 数据源配置完成: 新增 {created} 条, 当前共 {existing + created} 条")
# 打印所有数据源
rows = conn.execute(text("SELECT id, name, source_type, sync_type, status FROM data_source_config ORDER BY id")).fetchall()
print("\n数据源清单:")
for r in rows:
print(f" [{r[0]}] {r[1]:30s} type={r[2]:10s} sync={r[3]:10s} status={r[4]}")
+148
View File
@@ -0,0 +1,148 @@
""""
ERP数据源扩展 — 任务5: CRM+生产模块对接
执行: cd /root/cma-management/backend && python3 scripts/erp_p1_crm_prod.py
"""
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}")
# CRM + 生产模块数据源配置
NEW_SOURCES = [
# ─── CRM模块 ───
{
"name": "ERP-CRM-客户保留率",
"kpi_codes": ["C_RETENTION_RATE"],
"source_type": "erp",
"api_endpoint": "http://erp-api.sxbh.ltd/api/v1/crm/retention-rate",
"query_sql": """SELECT CASE WHEN total_customers > 0
THEN ROUND(renew_customers/total_customers*100, 2) ELSE 0 END as value
FROM (SELECT COUNT(*) as total_customers,
SUM(CASE WHEN DATEDIFF(day, last_order_date, GETDATE()) <= 365 THEN 1 ELSE 0 END) as renew_customers
FROM Units WHERE UnitType='Customer') t""",
"sync_type": "daily",
},
{
"name": "ERP-CRM-新客户数",
"kpi_codes": ["C_NEW_CLIENTS"],
"source_type": "erp",
"api_endpoint": "http://erp-api.sxbh.ltd/api/v1/crm/new-customers",
"query_sql": "SELECT COUNT(*) as value FROM Units WHERE UnitType='Customer' AND DATEDIFF(day, CreateDate, GETDATE()) <= 30",
"sync_type": "daily",
},
# ─── 生产模块 ───
{
"name": "ERP-生产-产品合格率",
"kpi_codes": ["F_QUALITY_RATE"],
"source_type": "erp",
"api_endpoint": "http://erp-api.sxbh.ltd/api/v1/production/quality-rate",
"query_sql": """SELECT CASE WHEN total_qty > 0
THEN ROUND(qualified_qty/total_qty*100, 2) ELSE 0 END as value
FROM (SELECT COUNT(*) as total_qty,
SUM(CASE WHEN QualityStatus='OK' THEN 1 ELSE 0 END) as qualified_qty
FROM QualityInspection WHERE Period=:period) t""",
"sync_type": "daily",
},
{
"name": "ERP-生产-返工率",
"kpi_codes": ["F_REWORK_RATE"],
"source_type": "erp",
"api_endpoint": "http://erp-api.sxbh.ltd/api/v1/production/rework-rate",
"query_sql": """SELECT CASE WHEN total_qty > 0
THEN ROUND(rework_qty/total_qty*100, 2) ELSE 0 END as value
FROM (SELECT COUNT(*) as total_qty,
SUM(CASE WHEN ReworkStatus='Rework' THEN 1 ELSE 0 END) as rework_qty
FROM ProductionOrder WHERE Period=:period) t""",
"sync_type": "daily",
},
]
with engine.connect() as conn:
print("=" * 70)
log("开始ERP数据源扩展 — 任务5: CRM+生产模块")
print("=" * 70)
created = 0
for src in NEW_SOURCES:
exists = conn.execute(
text("SELECT id FROM data_source_config WHERE name=:name"),
{"name": src["name"]}
).fetchone()
if exists:
log(f" ⏭️ 已存在: {src['name']} (id={exists[0]})")
continue
conn.execute(text("""
INSERT INTO data_source_config
(name, source_type, api_endpoint, query_sql, sync_type, status, created_at)
VALUES (:name, :source_type, :api_endpoint, :query_sql, :sync_type, 'active', NOW())
"""), {
"name": src["name"],
"source_type": src["source_type"],
"api_endpoint": src["api_endpoint"],
"query_sql": src["query_sql"],
"sync_type": src["sync_type"],
})
created += 1
log(f" ✅ 新增: {src['name']} — 关联KPI: {', '.join(src['kpi_codes'])}")
conn.commit()
# 打印所有数据源
rows = conn.execute(text(
"SELECT id, name, source_type, sync_type, status FROM data_source_config ORDER BY id"
)).fetchall()
print("\n数据源清单:")
for r in rows:
print(f" [{r[0]}] {r[1]:35s} type={r[2]:10s} sync={r[3]:10s} status={r[4]}")
log(f"\n✅ CRM+生产模块数据源配置完成: 新增 {created} 条, 共 {len(rows)} 条")
# 更新KPI的data_source_config字段(存储关联的数据源ID)
log("\n更新KPI的data_source_config字段...")
for src in NEW_SOURCES:
src_row = conn.execute(
text("SELECT id FROM data_source_config WHERE name=:name"),
{"name": src["name"]}
).fetchone()
if not src_row:
continue
for kpi_code in src["kpi_codes"]:
kpi = conn.execute(
text("SELECT id, data_source_config FROM kpi_definitions WHERE kpi_code=:code AND status='active'"),
{"code": kpi_code}
).fetchone()
if kpi:
existing_config = kpi[1] or {}
if isinstance(existing_config, str):
import json
try:
existing_config = json.loads(existing_config)
except:
existing_config = {}
# 确保是dict
if not isinstance(existing_config, dict):
existing_config = {"source_ids": []}
if "source_ids" not in existing_config:
existing_config["source_ids"] = []
if src_row[0] not in existing_config["source_ids"]:
existing_config["source_ids"].append(src_row[0])
import json
conn.execute(
text("UPDATE kpi_definitions SET data_source_config=:config WHERE id=:id"),
{"config": json.dumps(existing_config, ensure_ascii=False), "id": kpi[0]}
)
log(f" ✅ KPI {kpi_code}: data_source_config 已更新")
conn.commit()
log("\n✅ 任务5: CRM+生产模块对接完成")
print("=" * 70)
+22
View File
@@ -110,6 +110,12 @@ def fetch_via_api(kpi: KPIDefinition, parsed: dict, period: str) -> float:
"CUSTOMER_COUNT": f"{ERP_API_BASE}/stats/monthly?year={period_year}",
"SALES_PROFIT_RATE": f"{ERP_API_BASE}/stats/gross-profit?year={period_year}&month={period_month}",
"TOP5_CUSTOMER_RATIO": f"{ERP_API_BASE}/stats/customer-top?year={period_year}&limit=5",
# P1: CRM模块
"C_RETENTION_RATE": f"{ERP_API_BASE}/crm/retention-rate?year={period_year}&month={period_month}",
"C_NEW_CLIENTS": f"{ERP_API_BASE}/crm/new-customers?year={period_year}&month={period_month}",
# P1: 生产模块
"F_QUALITY_RATE": f"{ERP_API_BASE}/production/quality-rate?year={period_year}&month={period_month}",
"F_REWORK_RATE": f"{ERP_API_BASE}/production/rework-rate?year={period_year}&month={period_month}",
}
headers = {"X-API-Key": ERP_API_KEY, "User-Agent": "CMA-ERP-SYNC/1.0"}
@@ -159,6 +165,22 @@ def fetch_via_api(kpi: KPIDefinition, parsed: dict, period: str) -> float:
return round(top5_total / total_amount * 100, 2)
return 0
# P1: CRM模块 — 客户保留率
elif kpi_code == "C_RETENTION_RATE":
return float(data.get("retention_rate", data.get("value", 0)))
# P1: CRM模块 — 新客户数
elif kpi_code == "C_NEW_CLIENTS":
return float(data.get("new_customers", data.get("value", 0)))
# P1: 生产模块 — 产品合格率
elif kpi_code == "F_QUALITY_RATE":
return float(data.get("quality_rate", data.get("value", 0)))
# P1: 生产模块 — 返工率
elif kpi_code == "F_REWORK_RATE":
return float(data.get("rework_rate", data.get("value", 0)))
raise ValueError(f"未实现的API映射: {kpi_code}")
+354
View File
@@ -0,0 +1,354 @@
"""
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)
+256
View File
@@ -0,0 +1,256 @@
""""
KPI字典库优化P1阶段 — Tasks 1-4: 新增9个P1级KPI
执行: cd /root/cma-management/backend && python3 scripts/kpi_p1_optimization.py
"""
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):
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
def activate_template(conn, template_code, target_code, target_name, threshold_green, threshold_yellow, threshold_red, data_source_type="manual", frequency="monthly"):
"""从模板激活KPI"""
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=:code"
), {"code": template_code}).fetchone()
if not t:
log(f" ❌ 模板不存在: {template_code}")
return None, False
nid, is_new = insert_kpi(conn,
template_id=t[0],
is_system=1,
kpi_code=target_code,
kpi_name=target_name or t[2],
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=data_source_type,
frequency=frequency,
status="active",
epic="Epic2",
threshold_green=threshold_green,
threshold_yellow=threshold_yellow,
threshold_red=threshold_red,
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]})
return nid, is_new
with engine.connect() as conn:
print("=" * 70)
log("开始KPI字典库优化P1 — 任务1~4: 新增9个P1级KPI")
print("=" * 70)
# =========================================================
# 任务1: 新增财务P1级KPI(3个)
# =========================================================
log("\n【任务1】新增财务P1级KPI")
print("-" * 50)
finance_kpis = [
("F_DEBT_RATIO", "资产负债率", "finance", "cash_risk",
"总负债/总资产", "%", 50.0, "<50", "<70", ">=70", "erp"),
("F_INTEREST_COVER", "利息保障倍数", "finance", "profitability",
"EBIT/利息费用", "倍", 5.0, ">5", ">2", "<=2", "erp"),
("F_EVA", "经济增加值(EVA)", "finance", "profitability",
"税后净营业利润-资本成本", "元", 0.0, ">0", ">-100000", "<=-100000", "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(),
)
# =========================================================
# 任务2: 新增客户P1级KPI(2个)
# =========================================================
log("\n【任务2】新增客户P1级KPI")
print("-" * 50)
customer_kpis = [
("C_MARKET_SHARE", "市场份额", "customer", "customer_scale",
"公司收入/行业总收入", "%", 10.0, ">10", ">5", "<=5", "erp"),
("C_NPS", "净推荐值(NPS)", "customer", "customer_satisfaction",
"NPS评分(-100~100)", "分", 50.0, ">50", ">0", "<=0", "manual"),
]
for code, name, dim, cat, formula, unit, target, tg, ty, tr, src in customer_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: 新增流程P1级KPI(3个)
# =========================================================
log("\n【任务3】新增流程P1级KPI")
print("-" * 50)
# 3.1 激活P_SUPPLY_CYCLE模板
log(" --- 激活模板: P_SUPPLY_CYCLE (供应链响应周期)")
activate_template(conn, "P_SUPPLY_CYCLE", "P_SUPPLY_CYCLE", None,
"<=7", "<=14", ">14", "erp", "monthly")
# 3.2 产能利用率
insert_kpi(conn,
is_system=0,
kpi_code="P_CAPACITY_UTIL",
kpi_name="产能利用率",
dimension="process",
category="supply_chain",
formula="实际产出/理论产能",
formula_desc="CMA标准产能利用率指标,反映生产资源利用效率",
unit="%",
target_value=85.0,
threshold_green=">85",
threshold_yellow=">70",
threshold_red="<=70",
data_source_type="erp",
frequency="monthly",
status="active",
epic="Epic2",
created_at=datetime.now(),
updated_at=datetime.now(),
)
# 3.3 研发投入占比
insert_kpi(conn,
is_system=0,
kpi_code="P_RD_RATIO",
kpi_name="研发投入占比",
dimension="process",
category="innovation",
formula="研发费用/收入",
formula_desc="CMA标准研发投入占比指标,衡量创新投入力度",
unit="%",
target_value=5.0,
threshold_green=">5",
threshold_yellow=">2",
threshold_red="<=2",
data_source_type="erp",
frequency="monthly",
status="active",
epic="Epic2",
created_at=datetime.now(),
updated_at=datetime.now(),
)
# =========================================================
# 任务4: 新增学习P1级KPI(2个)
# =========================================================
log("\n【任务4】新增学习P1级KPI")
print("-" * 50)
# 4.1 激活L_INNOVATION_COUNT模板
log(" --- 激活模板: L_INNOVATION_COUNT (创新提案数量)")
activate_template(conn, "L_INNOVATION_COUNT", "L_INNOVATION_COUNT", None,
">=12", ">=6", "<6", "manual", "monthly")
# 4.2 战略认知度
insert_kpi(conn,
is_system=0,
kpi_code="L_STRATEGY_AWARE",
kpi_name="战略认知度",
dimension="learning",
category="employee_engagement",
formula="员工战略理解度评分",
formula_desc="CMA标准战略认知度指标,通过员工调研获取",
unit="分",
target_value=80.0,
threshold_green=">80",
threshold_yellow=">60",
threshold_red="<=60",
data_source_type="manual",
frequency="quarterly",
status="active",
epic="Epic2",
created_at=datetime.now(),
updated_at=datetime.now(),
)
# =========================================================
# 提交事务
# =========================================================
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~4完成!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)
+75
View File
@@ -0,0 +1,75 @@
"""迁移脚本:更新现有战略地图KPI codes + 关联所有活跃KPI到地图"""
from app.database import get_engine
from sqlalchemy import text
import json
def migrate():
e = get_engine()
with e.begin() as c:
# 1. 更新现有地图的dimensions → 使用真实KPI codes
maps = c.execute(text('SELECT id, dimensions FROM strategic_maps')).fetchall()
for mid, dims_json in maps:
if not dims_json:
continue
dims = json.loads(dims_json) if isinstance(dims_json, str) else dims_json
# 定义层→节点名→真实KPI code的映射
name_to_kpi = {
# 财务层
"营收目标": ["F_REVENUE"],
"净利润率": ["F_NET_PROFIT"],
"现金流": ["F_OP_CFLOW"],
# 客户层
"客户满意度": ["C_SATISFACTION"],
"市场份额": ["C_MARKET_SHARE"],
"客户保留率": ["C_RETENTION_RATE"],
# 流程层
"运营效率": ["P_DELIVERY"],
"质量合格率": ["P_PASS_RATE"],
# 学习层
"关键岗位胜任度": ["L_COMPETENCY"],
"培训完成率": ["L_TRAINING"],
}
updated = False
for dim in dims:
for obj in dim.get("objectives", []):
name = obj.get("name", "")
if name in name_to_kpi:
real_kpis = name_to_kpi[name]
if obj.get("kpis") != real_kpis:
obj["kpis"] = real_kpis
updated = True
if updated:
c.execute(
text("UPDATE strategic_maps SET dimensions = :dims WHERE id = :mid"),
{"dims": json.dumps(dims, ensure_ascii=False), "mid": mid},
)
print(f"✅ 地图 {mid}: dimensions KPI codes 已更新")
# 2. 将所有活跃KPI的map_id更新为1
result = c.execute(
text("UPDATE kpi_definitions SET map_id = 1 WHERE status = 'active' AND (map_id IS NULL OR map_id != 1)")
)
affected = result.rowcount
print(f"✅ 已更新 {affected} 个活跃KPI的 map_id → 1")
# 3. 验证
total = c.execute(text("SELECT COUNT(*) FROM kpi_definitions WHERE status='active' AND map_id = 1")).scalar()
print(f"📊 活跃KPI关联到地图1的数量: {total}")
kpis_with_data = c.execute(text("""
SELECT k.kpi_code, k.kpi_name
FROM kpi_definitions k
WHERE EXISTS (
SELECT 1 FROM kpi_values v WHERE v.kpi_id = k.id
) AND k.map_id = 1
ORDER BY k.kpi_code
""")).fetchall()
print(f"📊 有历史数据的KPI ({len(kpis_with_data)}个):")
for r in kpis_with_data:
print(f" {r.kpi_code} - {r.kpi_name}")
if __name__ == "__main__":
migrate()
+203
View File
@@ -0,0 +1,203 @@
"""P2阶段: 预设KPI因果链(基于CMA四层因果链模型)
学习成长→内部流程→客户→财务
"""
import sys, os
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
from app.database import get_engine
from sqlalchemy import text
# 基于CMA四层因果链模型的预设因果关系
PRESET_CAUSALITIES = [
# ===== 学习成长 → 内部流程 =====
# 培训完成率 → 过程合格率
{"source": "L_TRAINING", "target": "P_PASS_RATE", "strength": 0.6, "lag": 2, "direction": "positive", "formula": "培训提升技能→过程质量提升"},
# 人均培训时长 → 缺陷率(负相关)
{"source": "L_TRAINING_HOURS", "target": "P_BUG_RATE", "strength": 0.5, "lag": 2, "direction": "negative", "formula": "更多培训→更少缺陷"},
# 员工满意度 → 返工率(负相关)
{"source": "L_EMPLOYEE_SAT", "target": "P_REWORK_RATE", "strength": 0.4, "lag": 1, "direction": "negative", "formula": "员工满意→更少返工"},
# 关键技术掌握率 → 交付及时率
{"source": "L_TECH_COVERAGE", "target": "P_DELIVERY", "strength": 0.7, "lag": 1, "direction": "positive", "formula": "技术掌握→交付提升"},
# 关键岗位胜任度 → 过程合格率
{"source": "L_COMPETENCY", "target": "P_PASS_RATE", "strength": 0.6, "lag": 1, "direction": "positive", "formula": "胜任度→质量提升"},
# 创新提案数量 → 新产品收入占比
{"source": "L_INNOVATION_COUNT", "target": "P_NEW_PROD_RATIO", "strength": 0.5, "lag": 3, "direction": "positive", "formula": "创新提案→新产品上市"},
# 数据自动化率 → 交付及时率
{"source": "L_DATA_AUTO_RATE", "target": "P_DELIVERY", "strength": 0.3, "lag": 1, "direction": "positive", "formula": "自动化→效率提升"},
# 系统覆盖率 → 供应链响应周期
{"source": "L_SYS_COVERAGE", "target": "P_SUPPLY_CYCLE", "strength": 0.4, "lag": 3, "direction": "negative", "formula": "系统覆盖→周期缩短"},
# 战略认知度 → 新产品收入占比
{"source": "L_STRATEGY_AWARE", "target": "P_NEW_PROD_RATIO", "strength": 0.3, "lag": 2, "direction": "positive", "formula": "战略理解→创新聚焦"},
# ===== 内部流程 → 客户 =====
# 过程合格率 → 客户满意度
{"source": "P_PASS_RATE", "target": "C_SATISFACTION", "strength": 0.7, "lag": 1, "direction": "positive", "formula": "质量提升→客户满意"},
# 交付及时率 → 客户满意度
{"source": "P_DELIVERY", "target": "C_SATISFACTION", "strength": 0.6, "lag": 0, "direction": "positive", "formula": "及时交付→客户满意"},
# 缺陷率 → 客户满意度(负相关)
{"source": "P_BUG_RATE", "target": "C_SATISFACTION", "strength": 0.5, "lag": 1, "direction": "negative", "formula": "缺陷多→不满意"},
# 返工率 → 客户满意度(负相关)
{"source": "P_REWORK_RATE", "target": "C_SATISFACTION", "strength": 0.3, "lag": 1, "direction": "negative", "formula": "返工多→不满意"},
# 供应链响应周期 → 交付及时率(负相关)
{"source": "P_SUPPLY_CYCLE", "target": "P_DELIVERY", "strength": 0.5, "lag": 1, "direction": "negative", "formula": "周期长→交付慢"},
# 产能利用率 → 交付及时率
{"source": "P_CAPACITY_UTIL", "target": "P_DELIVERY", "strength": 0.4, "lag": 0, "direction": "positive", "formula": "产能足→交付快"},
# 新产品收入占比 → 市场份额
{"source": "P_NEW_PROD_RATIO", "target": "C_MARKET_SHARE", "strength": 0.5, "lag": 3, "direction": "positive", "formula": "创新产品→市场占有率提升"},
# ===== 客户 → 财务 =====
# 客户满意度 → 客户保留率
{"source": "C_SATISFACTION", "target": "C_RETENTION_RATE", "strength": 0.8, "lag": 1, "direction": "positive", "formula": "满意→留存"},
# 客户满意度 → 营业收入
{"source": "C_SATISFACTION", "target": "F_REVENUE", "strength": 0.5, "lag": 2, "direction": "positive", "formula": "满意客户→复购增加"},
# 客户保留率 → 营业收入
{"source": "C_RETENTION_RATE", "target": "F_REVENUE", "strength": 0.6, "lag": 1, "direction": "positive", "formula": "老客户留存→稳定收入"},
# 客户保留率 → 客户生命周期价值
{"source": "C_RETENTION_RATE", "target": "C_LTV", "strength": 0.7, "lag": 2, "direction": "positive", "formula": "高留存→生命周期延长"},
# 新客户数 → 营业收入
{"source": "C_NEW_CLIENTS", "target": "F_REVENUE", "strength": 0.4, "lag": 1, "direction": "positive", "formula": "新客户→收入增长"},
# 市场份额 → 营业收入
{"source": "C_MARKET_SHARE", "target": "F_REVENUE", "strength": 0.5, "lag": 1, "direction": "positive", "formula": "市场扩大→收入增加"},
# 获客成本 → 净利润(负相关)
{"source": "C_ACQUISITION_COST", "target": "F_NET_PROFIT", "strength": 0.3, "lag": 1, "direction": "negative", "formula": "获客成本高→利润减少"},
# 客户集中度 → 净利润(负相关)
{"source": "C_CUST_CONCENTRATION", "target": "F_NET_PROFIT", "strength": 0.3, "lag": 1, "direction": "negative", "formula": "集中度高→风险增大"},
# ===== 财务内部因果 =====
# 营业收入 → 净利润
{"source": "F_REVENUE", "target": "F_NET_PROFIT", "strength": 0.7, "lag": 0, "direction": "positive", "formula": "收入增长→利润增加"},
# 毛利率 → 净利润
{"source": "F_GROSS_MARGIN", "target": "F_NET_PROFIT", "strength": 0.6, "lag": 0, "direction": "positive", "formula": "毛利提升→利润增加"},
# 费用率 → 净利润(负相关)
{"source": "F_COST_RATIO", "target": "F_NET_PROFIT", "strength": 0.5, "lag": 0, "direction": "negative", "formula": "费用高→利润减少"},
# 收入增长率 → 营业收入
{"source": "F_REVENUE_GROWTH", "target": "F_REVENUE", "strength": 0.8, "lag": 1, "direction": "positive", "formula": "增长加速→收入提升"},
# 净利润 → 经济增加值
{"source": "F_NET_PROFIT", "target": "F_EVA", "strength": 0.9, "lag": 0, "direction": "positive", "formula": "净利润→经济增加值"},
# 经营性现金流 → 净利润(滞后反馈)
{"source": "F_OP_CFLOW", "target": "F_NET_PROFIT", "strength": 0.4, "lag": 1, "direction": "positive", "formula": "现金充裕→运营改善"},
# 产品合格率 → 返工率(负相关)
{"source": "F_QUALITY_RATE", "target": "F_REWORK_RATE", "strength": 0.6, "lag": 1, "direction": "negative", "formula": "合格率高→返工少"},
# 应收账款周转天数 → 经营性现金流(负相关)
{"source": "F_AR_DAYS", "target": "F_OP_CFLOW", "strength": 0.5, "lag": 1, "direction": "negative", "formula": "回款慢→现金流紧张"},
# 流动比率 → 资产负债率
{"source": "F_CURRENT_RATIO", "target": "F_DEBT_RATIO", "strength": 0.3, "lag": 1, "direction": "negative", "formula": "流动性强→负债率低"},
# ===== 客户生命周期价值 =====
# 净推荐值(NPS) → 客户保留率
{"source": "C_NPS", "target": "C_RETENTION_RATE", "strength": 0.6, "lag": 1, "direction": "positive", "formula": "NPS高→留存好"},
# 客户生命周期价值 → 净利润
{"source": "C_LTV", "target": "F_NET_PROFIT", "strength": 0.5, "lag": 1, "direction": "positive", "formula": "客户终身价值→长期利润"},
]
def seed_causalities():
engine = get_engine()
with engine.connect() as conn:
# 先构建KPI代码→ID映射
kpis = conn.execute(
text("SELECT id, kpi_code FROM kpi_definitions WHERE status='active'")
).fetchall()
code_to_id = {r.kpi_code: r.id for r in kpis}
inserted = 0
skipped = 0
for item in PRESET_CAUSALITIES:
src_id = code_to_id.get(item["source"])
tgt_id = code_to_id.get(item["target"])
if not src_id:
print(f" ⚠️ 源KPI不存在: {item['source']}")
skipped += 1
continue
if not tgt_id:
print(f" ⚠️ 目标KPI不存在: {item['target']}")
skipped += 1
continue
# 检查是否已存在
existing = conn.execute(
text("SELECT id FROM kpi_causality WHERE source_kpi_id=:src AND target_kpi_id=:tgt"),
{"src": src_id, "tgt": tgt_id},
).fetchone()
if existing:
skipped += 1
continue
conn.execute(
text("""INSERT INTO kpi_causality
(source_kpi_id, target_kpi_id, strength, lag_months, formula, direction)
VALUES (:src, :tgt, :strength, :lag, :formula, :dir)"""),
{
"src": src_id, "tgt": tgt_id,
"strength": item["strength"],
"lag": item["lag"],
"formula": item["formula"],
"dir": item["direction"],
},
)
inserted += 1
conn.commit()
print(f"因果链: 新增{inserted}, 跳过{skipped}")
return inserted
def seed_bi_templates():
engine = get_engine()
PRESET_TEMPLATES = [
{
"name": "四层指标总览",
"report_type": "overview",
"is_system": 1,
"config": '{"description":"展示财务/客户/流程/学习四层维度KPI概览","layout":"grid","dimensions":["finance","customer","process","learning"],"metrics":["count","avg_value","alert_count"],"chart_type":"gauge_card"}',
},
{
"name": "同比趋势分析",
"report_type": "trend",
"is_system": 1,
"config": '{"description":"各KPI近12个月趋势对比","period":"monthly","window_months":12,"chart_type":"line","show_compare":true}',
},
{
"name": "实际vs预算对比",
"report_type": "comparison",
"is_system": 1,
"config": '{"description":"KPI实际值vs目标值偏差分析","chart_type":"bar","show_deviation":true,"group_by":"dimension"}',
},
{
"name": "TOP N异常KPI",
"report_type": "topn",
"is_system": 1,
"config": '{"description":"排名前N的异常KPI红黄灯","top_n":10,"sort_by":"deviation","chart_type":"horizontal_bar","show_threshold":true}',
},
{
"name": "因果链推演",
"report_type": "causality",
"is_system": 1,
"config": '{"description":"基于KPI因果链推演分析","chart_type":"force_graph","max_depth":3,"min_strength":0.3}',
},
]
with engine.connect() as conn:
inserted = 0
for tpl in PRESET_TEMPLATES:
existing = conn.execute(
text("SELECT id FROM bi_report_templates WHERE name=:name AND is_system=1"),
{"name": tpl["name"]},
).fetchone()
if existing:
continue
conn.execute(
text("INSERT INTO bi_report_templates (name, report_type, config, is_system) VALUES (:name, :type, :cfg, 1)"),
{"name": tpl["name"], "type": tpl["report_type"], "cfg": tpl["config"]},
)
inserted += 1
conn.commit()
print(f"报表模板: 新增{inserted}")
return inserted
if __name__ == "__main__":
print("=== 种子数据初始化 ===\n")
seed_causalities()
seed_bi_templates()
print("\n完成!")
+120
View File
@@ -0,0 +1,120 @@
"""P2阶段: 新增4个高级KPI"""
import sys, os
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
from app.database import get_engine
from sqlalchemy import text
NEW_KPIS = [
{
"kpi_code": "C_LTV",
"kpi_name": "客户生命周期价值",
"dimension": "customer",
"category": "customer_scale",
"objective": "提升客户长期价值",
"formula": "Σ(每年贡献利润×折现率)",
"formula_desc": "客户全周期内贡献的净利润折现总和",
"data_source_type": "manual",
"frequency": "quarterly",
"unit": "万元",
"target_value": 15.0,
"threshold_green": ">=10",
"threshold_yellow": ">=5",
"threshold_red": "<=5",
"responsible_dept": "市场营销部",
"responsible_user": "市场总监",
"epic": "Epic2",
"status": "active",
},
{
"kpi_code": "P_NEW_PROD_RATIO",
"kpi_name": "新产品收入占比",
"dimension": "process",
"category": "innovation",
"objective": "促进产品创新",
"formula": "新产品收入/总收入×100%",
"formula_desc": "近12个月上市的新产品收入占总收入比例",
"data_source_type": "erp",
"frequency": "monthly",
"unit": "%",
"target_value": 25.0,
"threshold_green": ">20",
"threshold_yellow": ">10",
"threshold_red": "<=10",
"responsible_dept": "产品研发部",
"responsible_user": "产品总监",
"epic": "Epic2",
"status": "active",
},
{
"kpi_code": "L_DATA_AUTO_RATE",
"kpi_name": "数据自动化率",
"dimension": "learning",
"category": "innovation",
"objective": "提升数据管理效率",
"formula": "自动报表数/总报表数×100%",
"formula_desc": "已实现自动化生成的管理报表占总报表数的比例",
"data_source_type": "manual",
"frequency": "monthly",
"unit": "%",
"target_value": 75.0,
"threshold_green": ">70",
"threshold_yellow": ">50",
"threshold_red": "<=50",
"responsible_dept": "信息技术部",
"responsible_user": "CIO",
"epic": "Epic2",
"status": "active",
},
{
"kpi_code": "L_SYS_COVERAGE",
"kpi_name": "系统覆盖率",
"dimension": "learning",
"category": "innovation",
"objective": "推动管理信息化",
"formula": "信息化覆盖业务环节数/总业务环节数×100%",
"formula_desc": "已实现信息系统支持的核心业务环节占总环节数的比例",
"data_source_type": "manual",
"frequency": "quarterly",
"unit": "%",
"target_value": 85.0,
"threshold_green": ">80",
"threshold_yellow": ">60",
"threshold_red": "<=60",
"responsible_dept": "信息技术部",
"responsible_user": "CIO",
"epic": "Epic2",
"status": "active",
},
]
def seed_p2_kpis():
engine = get_engine()
with engine.connect() as conn:
inserted = 0
skipped = 0
for kpi in NEW_KPIS:
exists = conn.execute(
text("SELECT id FROM kpi_definitions WHERE kpi_code=:code"),
{"code": kpi["kpi_code"]},
).fetchone()
if exists:
print(f" ⏭️ 跳过已存在: {kpi['kpi_code']} (id={exists[0]})")
skipped += 1
continue
cols = ", ".join(kpi.keys())
vals = ", ".join(f":{k}" for k in kpi.keys())
conn.execute(text(f"INSERT INTO kpi_definitions ({cols}) VALUES ({vals})"), kpi)
new_id = conn.execute(text("SELECT LAST_INSERT_ID()")).scalar()
print(f" ✅ 新增: {kpi['kpi_code']} - {kpi['kpi_name']} (id={new_id})")
inserted += 1
conn.commit()
print(f"\n总计: 新增{inserted}, 跳过{skipped}")
return inserted
if __name__ == "__main__":
seed_p2_kpis()
+49
View File
@@ -0,0 +1,49 @@
"""验证预警规则系统"""
import sys, os, json
sys.path.insert(0, '/root/cma-management/backend')
os.chdir('/root/cma-management/backend')
from dotenv import load_dotenv
load_dotenv()
from app.database import get_session_local
from app.models import KPIDefinition
from sqlalchemy import text, func
db = get_session_local()()
# 检查 alert_rules 表
try:
total = db.execute(text("SELECT COUNT(*) FROM alert_rules")).scalar()
type_counts = db.execute(text("SELECT rule_type, COUNT(*) FROM alert_rules GROUP BY rule_type")).fetchall()
print(f"Total alert rules: {total}")
for t, c in type_counts:
print(f" {t}: {c}")
except Exception as e:
print(f"表不存在: {e}")
# 检查 dynamic_threshold_cache
try:
dc = db.execute(text("SELECT COUNT(*) FROM dynamic_threshold_cache")).scalar()
print(f"Dynamic threshold caches: {dc}")
except Exception as e:
print(f"dynamic_threshold_cache表不存在: {e}")
# 取样显示预警规则
try:
samples = db.execute(text("""
SELECT ar.id, ar.kpi_id, ar.rule_type, ar.enabled, ar.params, k.kpi_code, k.kpi_name
FROM alert_rules ar
LEFT JOIN kpi_definitions k ON k.id = ar.kpi_id
LIMIT 6
""")).fetchall()
print("\nSample rules:")
for s in samples:
print(f" [{s.id}] {s.kpi_code}({s.kpi_name}) type={s.rule_type} enabled={s.enabled}")
except Exception as e:
print(f"查询失败: {e}")
# 总KPI数验证
total_kpis = db.execute(text("SELECT COUNT(*) FROM kpi_definitions WHERE status='active'")).scalar()
print(f"\nTotal active KPIs: {total_kpis}")
db.close()
print("\n✅ 验证完成")