550 lines
24 KiB
Python
550 lines
24 KiB
Python
"""增长质量诊断 API — 五维评分 + 诊断结论 + 跨期对比
|
||
|
||
数据来源：KPI字典 (kpi_definitions) + KPI实际值 (kpi_values)
|
||
五维度：营收增长 / 利润质量 / 现金质量 / 增长效率 / 组织健康
|
||
评分区间：0-100（>=80 好 / 60-79 中 / <60 差）
|
||
"""
|
||
from fastapi import APIRouter, Depends, HTTPException, Query
|
||
from sqlalchemy.orm import Session
|
||
from typing import Optional, List
|
||
from datetime import datetime
|
||
import re
|
||
|
||
from app.database import get_db
|
||
from app.auth_middleware import require_auth, require_role
|
||
from app.models import KPIDefinition, KPIValue, Entity
|
||
|
||
router = APIRouter(prefix="/api/cma/growth-quality", tags=["增长质量诊断"],
|
||
dependencies=[Depends(require_role("ceo", "finance", "business", "it"))],
|
||
)
|
||
|
||
# 五维度定义
|
||
DIMENSIONS = [
|
||
{"key": "revenueGrowth", "name": "营收增长", "weight": 20, "icon": "📈",
|
||
"desc": "营收增速、增长持续性"},
|
||
{"key": "profitQuality", "name": "利润质量", "weight": 20, "icon": "💰",
|
||
"desc": "毛利率/净利率趋势、利润与收入匹配"},
|
||
{"key": "cashQuality", "name": "现金质量", "weight": 20, "icon": "🏦",
|
||
"desc": "经营现金流与净利润匹配度（含金量）"},
|
||
{"key": "growthEfficiency", "name": "增长效率", "weight": 20, "icon": "🚀",
|
||
"desc": "获客成本、单位增长投入产出"},
|
||
{"key": "orgHealth", "name": "组织健康", "weight": 20, "icon": "⚡",
|
||
"desc": "人效、费用结构"},
|
||
]
|
||
|
||
# 关键KPI编码 → 维度用途
|
||
KPI_CODES = {
|
||
"F_REVENUE": "营业收入(万元)",
|
||
"F_NET_PROFIT": "净利润(万元)",
|
||
"F_OP_CFLOW": "经营性现金流(万元)",
|
||
"F_GROSS_MARGIN": "毛利率(%)",
|
||
"F_OP_PROFIT_MARGIN": "经营利润率(%)",
|
||
"F_COST_RATIO": "费用率(%)",
|
||
"F_REVENUE_GROWTH": "收入增长率(%)",
|
||
"F_FCF": "自由现金流(万元)",
|
||
"C_NEW_CLIENTS": "新客户数",
|
||
"C_REBATE_RATE": "渠补率(%)",
|
||
"C_SATISFACTION": "客户满意度",
|
||
"L_TRAINING": "培训完成率",
|
||
"P_DELIVERY": "交付及时率",
|
||
"F_AR_DAYS": "应收账款周转天数",
|
||
}
|
||
|
||
MONTH_RE = re.compile(r"^\d{4}-\d{2}$")
|
||
|
||
|
||
def _fetch_kpi_values(db: Session, entity_id: int, period: str) -> dict:
|
||
"""拉取某实体某期间的全部KPI值 {kpi_code: actual_value}"""
|
||
rows = (db.query(KPIDefinition.kpi_code, KPIValue.actual_value)
|
||
.join(KPIValue, KPIValue.kpi_id == KPIDefinition.id)
|
||
.filter(KPIDefinition.entity_id == entity_id,
|
||
KPIValue.period == period,
|
||
KPIValue.actual_value.isnot(None))
|
||
.all())
|
||
return {code: value for code, value in rows}
|
||
|
||
|
||
def _fetch_history(db: Session, entity_id: int, limit: int = 12) -> List[dict]:
|
||
"""拉取最近 N 个期间（按月，含数据）的 KPI 值，供趋势/持续性分析"""
|
||
periods = (db.query(KPIValue.period)
|
||
.join(KPIDefinition, KPIDefinition.id == KPIValue.kpi_id)
|
||
.filter(KPIDefinition.entity_id == entity_id,
|
||
KPIDefinition.kpi_code.in_(["F_REVENUE", "F_NET_PROFIT", "F_OP_CFLOW"]))
|
||
.distinct().all())
|
||
plist = sorted({p[0] for p in periods}, reverse=True)
|
||
# 只保留 YYYY-MM 格式，按时间排序（旧→新）
|
||
months = sorted([p for p in plist if MONTH_RE.match(p)])
|
||
hist = []
|
||
for p in months[-limit:]:
|
||
hist.append({"period": p, **{k: None for k in KPI_CODES}})
|
||
if not hist:
|
||
return []
|
||
# 批量取数
|
||
rows = (db.query(KPIDefinition.kpi_code, KPIValue.period, KPIValue.actual_value)
|
||
.join(KPIValue, KPIValue.kpi_id == KPIDefinition.id)
|
||
.filter(KPIDefinition.entity_id == entity_id,
|
||
KPIValue.period.in_([h["period"] for h in hist]))
|
||
.all())
|
||
idx = {h["period"]: h for h in hist}
|
||
for code, period, val in rows:
|
||
if period in idx and code in idx[period]:
|
||
idx[period][code] = val
|
||
return hist
|
||
|
||
|
||
def _prev_period(period: str) -> Optional[str]:
|
||
"""计算上期（YYYY-MM → 上一月；其他格式 → None）"""
|
||
m = MONTH_RE.match(period)
|
||
if not m:
|
||
return None
|
||
y, mo = int(period[:4]), int(period[5:7])
|
||
if mo == 1:
|
||
return f"{y-1:04d}-12"
|
||
return f"{y:04d}-{mo-1:02d}"
|
||
|
||
|
||
def _yoy_period(period: str) -> Optional[str]:
|
||
"""计算去年同期（YYYY-MM → 去年同月）"""
|
||
m = MONTH_RE.match(period)
|
||
if not m:
|
||
return None
|
||
return f"{int(period[:4])-1:04d}-{period[5:7]}"
|
||
|
||
|
||
# ═══════════════════════════════════════════════
|
||
# 五维度评分引擎（0-100）
|
||
# ═══════════════════════════════════════════════
|
||
|
||
def _clamp(v: float, lo: float = 0.0, hi: float = 100.0) -> float:
|
||
return max(lo, min(hi, v))
|
||
|
||
|
||
def _linear(value, points: List[tuple]):
|
||
"""分段线性插值评分: points = [(x, score), ...] 按 x 升序"""
|
||
if value is None:
|
||
return 50.0
|
||
if value <= points[0][0]:
|
||
return points[0][1]
|
||
if value >= points[-1][0]:
|
||
return points[-1][1]
|
||
for (x1, s1), (x2, s2) in zip(points, points[1:]):
|
||
if x1 <= value <= x2:
|
||
if x2 == x1:
|
||
return s1
|
||
return s1 + (s2 - s1) * (value - x1) / (x2 - x1)
|
||
return 50.0
|
||
|
||
|
||
# ── 1. 营收增长：增速 + 持续性 ──
|
||
def _score_revenue_growth(cur: dict, prev: dict, history: List[dict]) -> float:
|
||
rev_cur = cur.get("F_REVENUE")
|
||
rev_prev = prev.get("F_REVENUE") if prev else None
|
||
# 优先用 KPI 直接给的收入增长率
|
||
kpi_growth = cur.get("F_REVENUE_GROWTH")
|
||
growth = None
|
||
if kpi_growth is not None:
|
||
growth = float(kpi_growth)
|
||
elif rev_cur is not None and rev_prev:
|
||
growth = (rev_cur - rev_prev) / rev_prev * 100 if rev_prev else None
|
||
|
||
score = _linear(growth, [
|
||
(-30, 5), (-20, 15), (-10, 30), (0, 45), (5, 60), (10, 70),
|
||
(20, 82), (30, 90), (50, 96),
|
||
])
|
||
# 增长持续性：近6个月中收入增长月占比
|
||
if len(history) >= 2:
|
||
revs = [h.get("F_REVENUE") for h in history if h.get("F_REVENUE") is not None]
|
||
ups = 0
|
||
for i in range(1, len(revs)):
|
||
if revs[i] > revs[i - 1]:
|
||
ups += 1
|
||
persist = ups / (len(revs) - 1) if len(revs) > 1 else 0.5
|
||
score = score * 0.7 + persist * 100 * 0.3
|
||
return round(_clamp(score), 1)
|
||
|
||
|
||
# ── 2. 利润质量：毛利率/净利率水平 + 趋势 + 收入匹配 ──
|
||
def _score_profit_quality(cur: dict, prev: dict) -> float:
|
||
gm = cur.get("F_GROSS_MARGIN")
|
||
np_ = cur.get("F_NET_PROFIT")
|
||
rev = cur.get("F_REVENUE")
|
||
net_margin = (np_ / rev * 100) if (np_ is not None and rev) else None
|
||
opm = cur.get("F_OP_PROFIT_MARGIN")
|
||
|
||
gm_score = _linear(gm, [(-10, 5), (0, 10), (10, 25), (20, 45), (30, 62),
|
||
(40, 75), (55, 88), (70, 95)])
|
||
nm_score = _linear(net_margin, [(-50, 0), (-20, 10), (-10, 20), (0, 35),
|
||
(10, 60), (20, 78), (30, 90)])
|
||
opm_score = _linear(opm, [(-20, 10), (0, 30), (10, 55), (20, 75), (35, 90)])
|
||
|
||
# 毛利率 40% + 净利率 40% + 经营利润率 20%
|
||
base = gm_score * 0.4 + nm_score * 0.4 + opm_score * 0.2
|
||
|
||
# 利润与收入匹配：收入升但利润降 → 扣分
|
||
if prev and rev is not None and np_ is not None:
|
||
prev_rev = prev.get("F_REVENUE")
|
||
prev_np = prev.get("F_NET_PROFIT")
|
||
if prev_rev and prev_np is not None:
|
||
rev_up = rev > prev_rev
|
||
np_down = np_ < prev_np
|
||
if rev_up and np_down:
|
||
base -= 10
|
||
elif np_down:
|
||
base -= 5
|
||
return round(_clamp(base), 1)
|
||
|
||
|
||
# ── 3. 现金质量：含金量(OCF/净利润) + 现金流强度 ──
|
||
def _score_cash_quality(cur: dict) -> float:
|
||
ocf = cur.get("F_OP_CFLOW")
|
||
np_ = cur.get("F_NET_PROFIT")
|
||
rev = cur.get("F_REVENUE")
|
||
fcf = cur.get("F_FCF")
|
||
|
||
# 含金量 = OCF / 净利润（净利润>0时）
|
||
gold = None
|
||
if ocf is not None and np_ is not None and np_ > 0:
|
||
gold = ocf / np_
|
||
# 净利润<=0：利润为负，含金量指标失效 → 低分（除非现金流强）
|
||
gold_score = _linear(gold, [(0, 10), (0.5, 35), (0.8, 55), (1.0, 70),
|
||
(1.2, 85), (1.5, 95)])
|
||
if np_ is not None and np_ <= 0:
|
||
gold_score = 15 if (ocf is None or ocf <= 0) else 35
|
||
|
||
# 现金流强度 = OCF / 收入
|
||
ocf_ratio = (ocf / rev * 100) if (ocf is not None and rev) else None
|
||
ocf_score = _linear(ocf_ratio, [(-20, 5), (0, 20), (10, 50), (20, 75),
|
||
(30, 90), (50, 98)])
|
||
fcf_score = _linear(fcf, [(-100, 10), (-20, 30), (0, 50), (20, 70),
|
||
(100, 90)]) if fcf is not None else 50.0
|
||
|
||
score = gold_score * 0.5 + ocf_score * 0.35 + fcf_score * 0.15
|
||
return round(_clamp(score), 1)
|
||
|
||
|
||
# ── 4. 增长效率：费用率水平 + 单位增长投入产出 + 获客成本 ──
|
||
def _score_growth_efficiency(cur: dict, prev: dict) -> float:
|
||
cost_ratio = cur.get("F_COST_RATIO")
|
||
cost_score = _linear(cost_ratio, [(10, 95), (20, 82), (30, 68), (40, 55),
|
||
(55, 40), (70, 25), (90, 10)])
|
||
|
||
# 费用增速 vs 收入增速（用费用率变化近似）
|
||
eff_score = 60.0
|
||
if prev is not None and cost_ratio is not None:
|
||
prev_cr = prev.get("F_COST_RATIO")
|
||
if prev_cr:
|
||
cr_change = cost_ratio - prev_cr
|
||
eff_score = _linear(cr_change, [(-15, 95), (-5, 80), (0, 65),
|
||
(5, 45), (15, 25), (30, 10)])
|
||
|
||
# 获客成本代理：收入/新客户数（越高越高效）
|
||
rev = cur.get("F_REVENUE")
|
||
new_clients = cur.get("C_NEW_CLIENTS")
|
||
cac_score = 60.0
|
||
if rev is not None and new_clients:
|
||
per_client = rev / new_clients
|
||
cac_score = _linear(per_client, [(0, 40), (50, 50), (200, 65),
|
||
(500, 78), (1000, 88)])
|
||
|
||
score = cost_score * 0.45 + eff_score * 0.35 + cac_score * 0.2
|
||
return round(_clamp(score), 1)
|
||
|
||
|
||
# ── 5. 组织健康：费用结构 + 人效/运营质量 ──
|
||
def _score_org_health(cur: dict) -> float:
|
||
cost_ratio = cur.get("F_COST_RATIO")
|
||
# 费用结构（费用率越低越健康）
|
||
cost_score = _linear(cost_ratio, [(10, 95), (20, 82), (30, 68), (40, 55),
|
||
(55, 40), (70, 25), (90, 10)])
|
||
# 运营/人效质量代理：满意度、培训、交付、应收
|
||
sat = cur.get("C_SATISFACTION")
|
||
train = cur.get("L_TRAINING")
|
||
deliver = cur.get("P_DELIVERY")
|
||
ar_days = cur.get("F_AR_DAYS")
|
||
op_vals = [v for v in [sat, train, deliver] if v is not None]
|
||
op_score = (sum(op_vals) / len(op_vals)) if op_vals else 55.0
|
||
ar_score = _linear(ar_days, [(15, 95), (30, 80), (45, 65), (60, 50),
|
||
(90, 30), (120, 15)]) if ar_days is not None else 55.0
|
||
|
||
score = cost_score * 0.4 + op_score * 0.35 + ar_score * 0.25
|
||
return round(_clamp(score), 1)
|
||
|
||
|
||
_SCORERS = {
|
||
"revenueGrowth": _score_revenue_growth,
|
||
"profitQuality": _score_profit_quality,
|
||
"cashQuality": _score_cash_quality,
|
||
"growthEfficiency": _score_growth_efficiency,
|
||
"orgHealth": _score_org_health,
|
||
}
|
||
|
||
|
||
# ═══════════════════════════════════════════════
|
||
# 明细指标 + 改善建议
|
||
# ═══════════════════════════════════════════════
|
||
|
||
def _fmt(v, unit=""):
|
||
if v is None:
|
||
return "—"
|
||
if isinstance(v, float) and v == int(v):
|
||
return f"{int(v)}{unit}"
|
||
return f"{round(v, 2)}{unit}"
|
||
|
||
|
||
def _dim_indicators(dim_key: str, cur: dict, prev: dict) -> List[dict]:
|
||
"""维度明细指标（label/value/verdict/status）"""
|
||
inds = []
|
||
def add(label, value, verdict, status):
|
||
inds.append({"label": label, "value": value, "verdict": verdict, "status": status})
|
||
|
||
if dim_key == "revenueGrowth":
|
||
rev, prev_rev = cur.get("F_REVENUE"), (prev or {}).get("F_REVENUE")
|
||
growth = None
|
||
if rev is not None and prev_rev:
|
||
growth = (rev - prev_rev) / prev_rev * 100
|
||
add("营业收入", _fmt(rev, "万"), "环比" + (_fmt(growth, "%") if growth is not None else "无上期数据"),
|
||
"success" if (growth or 0) >= 0 else "danger")
|
||
add("收入增长率(KPI)", _fmt(cur.get("F_REVENUE_GROWTH"), "%"),
|
||
"KPI直接值" if cur.get("F_REVENUE_GROWTH") is not None else "未录入",
|
||
"success" if (cur.get("F_REVENUE_GROWTH") or 0) >= 10 else "warning")
|
||
elif dim_key == "profitQuality":
|
||
rev, np_ = cur.get("F_REVENUE"), cur.get("F_NET_PROFIT")
|
||
nm = (np_ / rev * 100) if (np_ is not None and rev) else None
|
||
add("毛利率", _fmt(cur.get("F_GROSS_MARGIN"), "%"),
|
||
"毛利健康" if (cur.get("F_GROSS_MARGIN") or 0) >= 30 else "毛利偏低",
|
||
"success" if (cur.get("F_GROSS_MARGIN") or 0) >= 30 else "danger")
|
||
add("净利率", _fmt(nm, "%"),
|
||
"盈利" if (nm or 0) > 0 else "亏损",
|
||
"success" if (nm or 0) > 10 else "danger")
|
||
add("经营利润率", _fmt(cur.get("F_OP_PROFIT_MARGIN"), "%"),
|
||
"正常" if (cur.get("F_OP_PROFIT_MARGIN") or 0) >= 15 else "偏低",
|
||
"success" if (cur.get("F_OP_PROFIT_MARGIN") or 0) >= 15 else "warning")
|
||
elif dim_key == "cashQuality":
|
||
ocf, np_ = cur.get("F_OP_CFLOW"), cur.get("F_NET_PROFIT")
|
||
gold = (ocf / np_) if (ocf is not None and np_ and np_ > 0) else None
|
||
add("经营现金流", _fmt(ocf, "万"),
|
||
"现金流入" if (ocf or 0) > 0 else "现金流出",
|
||
"success" if (ocf or 0) > 0 else "danger")
|
||
add("含金量(OCF/净利润)", _fmt(gold, "倍"),
|
||
"含金量高" if (gold or 0) >= 1 else ("利润为负" if (np_ or 0) <= 0 else "含金量低"),
|
||
"success" if (gold or 0) >= 1 else "danger")
|
||
add("自由现金流", _fmt(cur.get("F_FCF"), "万"),
|
||
"正常" if (cur.get("F_FCF") or 0) > 0 else "为负",
|
||
"success" if (cur.get("F_FCF") or 0) > 0 else "warning")
|
||
elif dim_key == "growthEfficiency":
|
||
rev, nc = cur.get("F_REVENUE"), cur.get("C_NEW_CLIENTS")
|
||
per = (rev / nc) if (rev is not None and nc) else None
|
||
add("费用率", _fmt(cur.get("F_COST_RATIO"), "%"),
|
||
"费用可控" if (cur.get("F_COST_RATIO") or 0) <= 30 else "费用偏高",
|
||
"success" if (cur.get("F_COST_RATIO") or 0) <= 30 else "warning")
|
||
add("单位客户营收(万/户)", _fmt(per),
|
||
"获客效率高" if (per or 0) >= 200 else "获客效率一般",
|
||
"success" if (per or 0) >= 500 else "warning")
|
||
add("渠补率", _fmt(cur.get("C_REBATE_RATE"), "%"),
|
||
"渠道依赖" if (cur.get("C_REBATE_RATE") or 0) > 50 else "渠道健康",
|
||
"danger" if (cur.get("C_REBATE_RATE") or 0) > 50 else "success")
|
||
elif dim_key == "orgHealth":
|
||
add("费用率(结构)", _fmt(cur.get("F_COST_RATIO"), "%"),
|
||
"结构健康" if (cur.get("F_COST_RATIO") or 0) <= 30 else "结构偏重",
|
||
"success" if (cur.get("F_COST_RATIO") or 0) <= 30 else "warning")
|
||
add("应收周转天数", _fmt(cur.get("F_AR_DAYS"), "天"),
|
||
"回款快" if (cur.get("F_AR_DAYS") or 0) <= 45 else "回款偏慢",
|
||
"success" if (cur.get("F_AR_DAYS") or 0) <= 45 else "warning")
|
||
add("客户满意度", _fmt(cur.get("C_SATISFACTION")),
|
||
"满意" if (cur.get("C_SATISFACTION") or 0) >= 80 else "待提升",
|
||
"success" if (cur.get("C_SATISFACTION") or 0) >= 80 else "warning")
|
||
add("培训完成率", _fmt(cur.get("L_TRAINING"), "%"),
|
||
"学习投入足" if (cur.get("L_TRAINING") or 0) >= 80 else "学习投入不足",
|
||
"success" if (cur.get("L_TRAINING") or 0) >= 80 else "warning")
|
||
return inds
|
||
|
||
|
||
def _dim_suggestions(dim_key: str, score: float, cur: dict) -> List[str]:
|
||
"""按维度评分生成改善建议"""
|
||
if score >= 80:
|
||
return ["该维度表现良好，建议保持并固化为标准流程"]
|
||
sug = []
|
||
if dim_key == "revenueGrowth":
|
||
sug = ["挖掘存量客户复购，稳定收入基本盘",
|
||
"拓展新渠道/新产品线，提升营收增速",
|
||
"跟踪F_REVENUE_GROWTH KPI按月更新，建立增长预警线"]
|
||
elif dim_key == "profitQuality":
|
||
sug = ["排查毛利率下滑原因（成本/价格/渠补），优先止血",
|
||
"控制费用增速不超过收入增速，改善净利率",
|
||
"对亏损产品线做盈亏平衡分析，必要时收缩"]
|
||
elif dim_key == "cashQuality":
|
||
sug = ["加强应收账款催收，缩短回款周期",
|
||
"压缩非必要开支，提升经营现金流净额",
|
||
"建立现金流月度滚动预测，防范断流风险"]
|
||
elif dim_key == "growthEfficiency":
|
||
sug = ["优化费用结构，降低费用率至30%以下",
|
||
"评估渠道返利政策，降低渠补率与渠道依赖",
|
||
"提高获客转化率，降低单位获客成本"]
|
||
elif dim_key == "orgHealth":
|
||
sug = ["精简组织与费用结构，提升人效",
|
||
"强化培训与人才梯队建设（盯L_TRAINING）",
|
||
"优化应收管理，缩短周转天数"]
|
||
return sug
|
||
|
||
|
||
def _level_of(overall: float) -> dict:
|
||
if overall >= 80:
|
||
return {"level": "好", "level_type": "success", "desc": "增长质量优秀，增长可持续"}
|
||
if overall >= 60:
|
||
return {"level": "中", "level_type": "warning", "desc": "增长质量中等，存在优化空间"}
|
||
return {"level": "差", "level_type": "danger", "desc": "增长质量堪忧，需立即干预"}
|
||
|
||
|
||
# ═══════════════════════════════════════════════
|
||
# 诊断主流程
|
||
# ═══════════════════════════════════════════════
|
||
|
||
def _diagnose(db: Session, entity_id: int, period: str, history: List[dict]):
|
||
"""对单个期间执行五维诊断，返回完整诊断对象"""
|
||
cur = _fetch_kpi_values(db, entity_id, period)
|
||
prev_period = _prev_period(period)
|
||
prev = _fetch_kpi_values(db, entity_id, prev_period) if prev_period else {}
|
||
|
||
scores = {}
|
||
dims_payload = {}
|
||
for dim in DIMENSIONS:
|
||
key = dim["key"]
|
||
scorer = _SCORERS[key]
|
||
if key == "revenueGrowth":
|
||
s = scorer(cur, prev, history)
|
||
elif key in ("cashQuality", "orgHealth"):
|
||
s = scorer(cur)
|
||
else:
|
||
s = scorer(cur, prev)
|
||
scores[key] = s
|
||
dims_payload[key] = {
|
||
"key": key, "name": dim["name"], "icon": dim["icon"],
|
||
"desc": dim["desc"], "weight": dim["weight"],
|
||
"score": s,
|
||
"indicators": _dim_indicators(key, cur, prev),
|
||
"suggestions": _dim_suggestions(key, s, cur),
|
||
}
|
||
|
||
overall = round(sum(scores.values()) / len(scores), 1)
|
||
level = _level_of(overall)
|
||
|
||
# 诊断结论文本
|
||
low_dims = [d for d in DIMENSIONS if scores[d["key"]] < 60]
|
||
mid_dims = [d for d in DIMENSIONS if 60 <= scores[d["key"]] < 80]
|
||
lines = [f"{period} 综合增长质量评分 {overall} 分（{level['level']}）：{level['desc']}。"]
|
||
if low_dims:
|
||
lines.append("需重点关注：" + "、".join(f"{d['name']}({scores[d['key']]}分)" for d in low_dims) + "。")
|
||
if mid_dims:
|
||
lines.append("可优化：" + "、".join(f"{d['name']}({scores[d['key']]}分)" for d in mid_dims) + "。")
|
||
if not low_dims:
|
||
lines.append("各维度均处于健康区间，增长质量扎实。")
|
||
diagnosis = "".join(lines)
|
||
|
||
return {
|
||
"entity_id": entity_id,
|
||
"period": period,
|
||
"overall": overall,
|
||
"level": level["level"],
|
||
"level_type": level["level_type"],
|
||
"diagnosis": diagnosis,
|
||
"dimensions": dims_payload,
|
||
"kpi_available": {k: cur.get(k) is not None for k in KPI_CODES},
|
||
}
|
||
|
||
|
||
@router.get("/periods")
|
||
def list_periods(entity_id: int = Query(1), db: Session = Depends(get_db)):
|
||
"""列出某实体有KPI数据的期间（按月，含数据覆盖度，用于前端默认期间选择）"""
|
||
rows = (db.query(KPIValue.period, KPIValue.kpi_id)
|
||
.join(KPIDefinition, KPIDefinition.id == KPIValue.kpi_id)
|
||
.filter(KPIDefinition.entity_id == entity_id)
|
||
.all())
|
||
counts: dict = {}
|
||
for period, kpi_id in rows:
|
||
if MONTH_RE.match(period or ""):
|
||
counts[period] = counts.get(period, 0) + 1
|
||
periods = sorted(counts.keys(), reverse=True)
|
||
return {"entity_id": entity_id, "periods": periods,
|
||
"coverage": {p: counts[p] for p in periods}}
|
||
|
||
|
||
@router.get("/diagnosis")
|
||
def growth_quality_diagnosis(
|
||
entity_id: int = Query(1, ge=1),
|
||
period: Optional[str] = Query(None, description="期间 YYYY-MM，默认最近有数据期间"),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
"""增长质量诊断 — 五维评分(0-100) + 诊断结论 + 跨期对比（本期/上期/去年同期）"""
|
||
ent = db.query(Entity).filter(Entity.id == entity_id).first()
|
||
if not ent:
|
||
raise HTTPException(404, f"实体 {entity_id} 不存在")
|
||
|
||
history = _fetch_history(db, entity_id, limit=12)
|
||
if not history:
|
||
raise HTTPException(400, "该实体暂无月度KPI数据，请先录入KPI实际值")
|
||
|
||
# 默认取最近且有足够数据覆盖的期间（>=5个KPI值，退化为最近一个）
|
||
if not period:
|
||
cov_rows = (db.query(KPIValue.period)
|
||
.join(KPIDefinition, KPIDefinition.id == KPIValue.kpi_id)
|
||
.filter(KPIDefinition.entity_id == entity_id)
|
||
.all())
|
||
cov: dict = {}
|
||
for (p,) in cov_rows:
|
||
if MONTH_RE.match(p or ""):
|
||
cov[p] = cov.get(p, 0) + 1
|
||
candidates = sorted([p for p in cov if cov[p] >= 5], reverse=True)
|
||
period = candidates[0] if candidates else history[-1]["period"]
|
||
|
||
current = _diagnose(db, entity_id, period, history)
|
||
|
||
# 跨期对比：上期 + 去年同期
|
||
prev_p = _prev_period(period)
|
||
yoy_p = _yoy_period(period)
|
||
prev_data = _fetch_kpi_values(db, entity_id, prev_p) if prev_p else {}
|
||
yoy_data = _fetch_kpi_values(db, entity_id, yoy_p) if yoy_p else {}
|
||
|
||
comparison = {
|
||
"current": {
|
||
"period": period, "overall": current["overall"],
|
||
"level": current["level"], "level_type": current["level_type"],
|
||
"dimensions": {k: v["score"] for k, v in current["dimensions"].items()},
|
||
},
|
||
}
|
||
if prev_p and prev_data:
|
||
pdiag = _diagnose(db, entity_id, prev_p, history)
|
||
comparison["previous"] = {
|
||
"period": prev_p, "overall": pdiag["overall"],
|
||
"level": pdiag["level"], "level_type": pdiag["level_type"],
|
||
"dimensions": {k: v["score"] for k, v in pdiag["dimensions"].items()},
|
||
}
|
||
if yoy_p and yoy_data:
|
||
ydiag = _diagnose(db, entity_id, yoy_p, history)
|
||
comparison["yoy"] = {
|
||
"period": yoy_p, "overall": ydiag["overall"],
|
||
"level": ydiag["level"], "level_type": ydiag["level_type"],
|
||
"dimensions": {k: v["score"] for k, v in ydiag["dimensions"].items()},
|
||
}
|
||
|
||
# 趋势：近12个月综合评分
|
||
trend = []
|
||
for h in history:
|
||
try:
|
||
d = _diagnose(db, entity_id, h["period"], history)
|
||
trend.append({"period": h["period"], "overall": d["overall"]})
|
||
except Exception:
|
||
continue
|
||
|
||
return {
|
||
"entity": {"id": ent.id, "name": ent.name, "short_name": ent.short_name},
|
||
"period": period,
|
||
"overall": current["overall"],
|
||
"level": current["level"],
|
||
"level_type": current["level_type"],
|
||
"diagnosis": current["diagnosis"],
|
||
"dimensions": current["dimensions"],
|
||
"comparison": comparison,
|
||
"trend": trend,
|
||
"kpi_available": current["kpi_available"],
|
||
}
|