245 lines
9.1 KiB
Python
245 lines
9.1 KiB
Python
"""客户维度KPI看板 API — P0-2"""
|
|
from fastapi import APIRouter, Depends, Query, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import func, desc
|
|
from typing import Optional
|
|
from datetime import datetime, timedelta
|
|
from app.database import get_db
|
|
from app.auth_middleware import require_auth, require_role
|
|
from app.models import KPIDefinition, KPIValue, KPIAlert, User
|
|
import logging
|
|
|
|
logger = logging.getLogger("cma.customer")
|
|
|
|
router = APIRouter(prefix="/api/cma/customer-dashboard", tags=["客户看板"],
|
|
dependencies=[Depends(require_role("ceo", "finance", "business", "it"))],
|
|
)
|
|
|
|
|
|
def parse_period(period_type: str, start_date: str = None, end_date: str = None):
|
|
"""解析时间区间"""
|
|
today = datetime.now()
|
|
if period_type == "month":
|
|
start = today.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
|
|
end = today
|
|
elif period_type == "quarter":
|
|
q = (today.month - 1) // 3
|
|
start = today.replace(month=q*3+1, day=1, hour=0, minute=0, second=0, microsecond=0)
|
|
end = today
|
|
elif period_type == "year":
|
|
start = today.replace(month=1, day=1, hour=0, minute=0, second=0, microsecond=0)
|
|
end = today
|
|
elif period_type == "custom" and start_date and end_date:
|
|
start = datetime.strptime(start_date, "%Y-%m-%d")
|
|
end = datetime.strptime(end_date, "%Y-%m-%d") + timedelta(days=1)
|
|
else:
|
|
start = today.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
|
|
end = today
|
|
return start, end
|
|
|
|
|
|
@router.get("")
|
|
def list_customer_kpis(
|
|
period: str = Query("month"),
|
|
start_date: str = Query(None),
|
|
end_date: str = Query(None),
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_auth),
|
|
):
|
|
"""获取客户维度KPI列表（含最新值、预警、趋势）"""
|
|
start, end = parse_period(period, start_date, end_date)
|
|
period_str = start.strftime("%Y-%m")
|
|
|
|
# 只查 customer 维度的 KPI
|
|
kpis = db.query(KPIDefinition).filter(
|
|
KPIDefinition.status == "active",
|
|
KPIDefinition.dimension == "customer",
|
|
).order_by(KPIDefinition.kpi_code).all()
|
|
|
|
result = []
|
|
for k in kpis:
|
|
base_query = db.query(KPIValue).filter(KPIValue.kpi_id == k.id)
|
|
|
|
if period == "month":
|
|
latest = base_query.filter(KPIValue.period == period_str).order_by(KPIValue.id.desc()).first()
|
|
elif period == "quarter":
|
|
q_month = (datetime.now().month - 1) // 3
|
|
months = [f"{datetime.now().year}-{m:02d}" for m in range(q_month*3+1, q_month*3+4)]
|
|
values = base_query.filter(KPIValue.period.in_(months)).all()
|
|
latest_val = sum(v.actual_value for v in values if v.actual_value) if values else None
|
|
latest = type('obj', (object,), {"actual_value": latest_val, "period": f"{months[0]}~{months[-1]}"})() if latest_val else None
|
|
elif period == "year":
|
|
values = base_query.filter(KPIValue.period.like(f"{period_str[:4]}%")).all()
|
|
latest_val = sum(v.actual_value for v in values if v.actual_value) if values else None
|
|
latest = type('obj', (object,), {"actual_value": latest_val, "period": period_str[:4]})() if latest_val else None
|
|
elif period == "custom" and start_date and end_date:
|
|
periods = []
|
|
d = start
|
|
while d <= end:
|
|
periods.append(d.strftime("%Y-%m"))
|
|
d += timedelta(days=32)
|
|
d = d.replace(day=1)
|
|
values = base_query.filter(KPIValue.period.in_(set(periods))).all()
|
|
latest_val = sum(v.actual_value for v in values if v.actual_value) if values else None
|
|
latest = type('obj', (object,), {"actual_value": latest_val, "period": f"{start_date}~{end_date}"})() if latest_val else None
|
|
else:
|
|
latest = base_query.order_by(KPIValue.period.desc()).first()
|
|
|
|
# 最新预警
|
|
alert = db.query(KPIAlert).filter(
|
|
KPIAlert.kpi_id == k.id,
|
|
KPIAlert.status == "pending",
|
|
).order_by(KPIAlert.id.desc()).first()
|
|
|
|
# 趋势（环比变化率）
|
|
trend = None
|
|
achievement_rate = None
|
|
period_values = []
|
|
|
|
if latest and latest.actual_value:
|
|
prev_period_str = None
|
|
if period == "month":
|
|
year_s, month_s = period_str.split("-")
|
|
y_s, m_s = int(year_s), int(month_s)
|
|
m_s -= 1
|
|
if m_s <= 0:
|
|
m_s += 12
|
|
y_s -= 1
|
|
prev_period_str = f"{y_s}-{m_s:02d}"
|
|
|
|
if prev_period_str:
|
|
prev_val = db.query(KPIValue).filter(
|
|
KPIValue.kpi_id == k.id,
|
|
KPIValue.period == prev_period_str,
|
|
).order_by(KPIValue.id.desc()).first()
|
|
if prev_val and prev_val.actual_value and prev_val.actual_value > 0:
|
|
trend = round((latest.actual_value - prev_val.actual_value) / prev_val.actual_value * 100, 2)
|
|
elif prev_val and prev_val.actual_value and prev_val.actual_value == 0:
|
|
trend = 100.0 if latest.actual_value > 0 else 0
|
|
|
|
# 达成率
|
|
if latest and latest.actual_value and k.target_value and k.target_value > 0:
|
|
achievement_rate = round(latest.actual_value / k.target_value * 100, 1)
|
|
|
|
# 最近6期趋势数据
|
|
period_q = db.query(KPIValue).filter(
|
|
KPIValue.kpi_id == k.id,
|
|
).order_by(KPIValue.period.desc()).limit(6).all()
|
|
period_values = [
|
|
{"period": v.period, "value": v.actual_value}
|
|
for v in reversed(period_q) if v.actual_value is not None
|
|
]
|
|
|
|
result.append({
|
|
"id": k.id,
|
|
"kpi_code": k.kpi_code,
|
|
"kpi_name": k.kpi_name,
|
|
"dimension": k.dimension,
|
|
"category": k.category,
|
|
"unit": k.unit,
|
|
"target_value": k.target_value,
|
|
"actual_value": latest.actual_value if latest else None,
|
|
"period": latest.period if latest else None,
|
|
"alert_level": alert.alert_level if alert else "none",
|
|
"alert_message": alert.alert_message if alert else None,
|
|
"frequency": k.frequency,
|
|
"responsible_dept": k.responsible_dept,
|
|
"responsible_user": k.responsible_user,
|
|
"trend": trend,
|
|
"achievement_rate": achievement_rate,
|
|
"period_values": period_values,
|
|
"kpi_name": k.kpi_name,
|
|
})
|
|
|
|
return {"data": result, "period": period, "total": len(result)}
|
|
|
|
|
|
@router.get("/trend/{kpi_id}")
|
|
def get_kpi_trend(
|
|
kpi_id: int,
|
|
months: int = Query(12, ge=3, le=24),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""获取单个KPI的历史趋势数据"""
|
|
kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first()
|
|
if not kpi:
|
|
raise HTTPException(404, "KPI不存在")
|
|
|
|
# 获取最近N期数据
|
|
values = db.query(KPIValue).filter(
|
|
KPIValue.kpi_id == kpi_id,
|
|
).order_by(KPIValue.period.desc()).limit(months).all()
|
|
|
|
trend_data = [
|
|
{"period": v.period, "value": v.actual_value}
|
|
for v in reversed(values) if v.actual_value is not None
|
|
]
|
|
|
|
# 计算预警水平和触发时间
|
|
alerts = db.query(KPIAlert).filter(
|
|
KPIAlert.kpi_id == kpi_id,
|
|
).order_by(KPIAlert.created_at.desc()).limit(10).all()
|
|
|
|
alert_logs = [
|
|
{
|
|
"level": a.alert_level,
|
|
"message": a.alert_message,
|
|
"time": a.created_at.isoformat() if a.created_at else None,
|
|
"status": a.status,
|
|
}
|
|
for a in alerts
|
|
]
|
|
|
|
return {
|
|
"kpi": {
|
|
"id": kpi.id,
|
|
"kpi_code": kpi.kpi_code,
|
|
"kpi_name": kpi.kpi_name,
|
|
"target_value": kpi.target_value,
|
|
"unit": kpi.unit,
|
|
"threshold_green": kpi.threshold_green,
|
|
"threshold_yellow": kpi.threshold_yellow,
|
|
"threshold_red": kpi.threshold_red,
|
|
},
|
|
"trend_data": trend_data,
|
|
"alerts": alert_logs,
|
|
}
|
|
|
|
|
|
@router.get("/summary")
|
|
def get_customer_summary(
|
|
period: str = Query("month"),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""客户维度概要统计"""
|
|
total = db.query(func.count(KPIDefinition.id)).filter(
|
|
KPIDefinition.status == "active",
|
|
KPIDefinition.dimension == "customer",
|
|
).scalar() or 0
|
|
|
|
# 预警统计
|
|
pending_alerts = db.query(func.count(KPIAlert.id)).filter(
|
|
KPIAlert.status == "pending",
|
|
KPIAlert.kpi_id.in_(
|
|
db.query(KPIDefinition.id).filter(
|
|
KPIDefinition.status == "active",
|
|
KPIDefinition.dimension == "customer",
|
|
)
|
|
),
|
|
).scalar() or 0
|
|
|
|
# 二级类别分布
|
|
cat_stats = db.query(
|
|
KPIDefinition.category,
|
|
func.count(KPIDefinition.id),
|
|
).filter(
|
|
KPIDefinition.status == "active",
|
|
KPIDefinition.dimension == "customer",
|
|
).group_by(KPIDefinition.category).all()
|
|
|
|
return {
|
|
"total": total,
|
|
"pending_alerts": pending_alerts,
|
|
"category_stats": [{"category": c[0], "count": c[1]} for c in cat_stats],
|
|
}
|