Files
cma-management/backend/app/api/dashboard.py
T
Hermes CI Fix 50c15ddbf6 fix: 多租户数据隔离P1a+P1b — 工作台/预警/行动计划/因果链按企业过滤
P1a(有entity表查询补齐):
- kpis get/update/delete/restore 跨企业404校验
- kpis create 强制entity=token企业, update禁止改归属
P1b(无entity表join隔离):
- alerts list/resolve join kpi_definitions 按企业过滤
- action_plans list join过滤 + create校验关联KPI归属
- kpi_causality full-network/list join过滤
- dashboard my_dashboard(用户发现) assigned/preset均按企业隔离
测试: 测试KPI种子entity对齐(2→1), pytest 451 passed
实证: 酣客token 6KPI(无博海id) vs 博海token 1KPI(414) 切换隔离正确
2026-08-23 17:43:39 +08:00

548 lines
22 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""驾驶舱 API v2 — 支持时间区间"""
from fastapi import APIRouter, Depends, Query, Request, HTTPException
from sqlalchemy.orm import Session
from sqlalchemy import func, or_
from datetime import datetime, timedelta
from typing import Optional
from app.database import get_db
from app.auth_middleware import require_auth, require_role
from app.deps import get_entity_id
from app.models import KPIDefinition, KPIValue, KPIAlert, User
from app.utils.cache import get as cache_get, set as cache_set
import json
import logging
logger = logging.getLogger("cma.dashboard")
router = APIRouter(prefix="/api/cma/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
def period_prefix(period_type: str):
"""生成SQL期间前缀匹配"""
if period_type == "month":
return datetime.now().strftime("%Y-%m")
elif period_type == "quarter":
now = datetime.now()
q = (now.month - 1) // 3
months = [f"{now.year}-{m:02d}" for m in range(q*3+1, q*3+4)]
return months
elif period_type == "year":
return str(datetime.now().year)
return None
def kpi_target_by_frequency(k):
"""按考核频率返回对应周期的目标值(多粒度改造)
monthly/weekly -> target_monthly; quarterly/half_year -> target_quarterly; yearly -> target_yearly
兼容: 对应列无值时回退 target_value
"""
freq = (k.frequency or "monthly").lower()
if freq in ("monthly", "weekly"):
val = getattr(k, "target_monthly", None)
elif freq in ("quarterly", "half_year"):
val = getattr(k, "target_quarterly", None)
elif freq == "yearly":
val = getattr(k, "target_yearly", None)
else:
val = None
if val is None:
val = k.target_value
return val
@router.get("/summary")
def get_dashboard_summary(role: str = Query("ceo"), period: str = Query("month"),
db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)):
cache_key = f"summary:{role}:{period}:{entity_id}"
cached = cache_get("dashboard", cache_key)
if cached:
return cached
kpi_total = db.query(func.count(KPIDefinition.id)).filter(
KPIDefinition.status == "active", KPIDefinition.entity_id == entity_id).scalar()
alert_count = db.query(func.count(KPIAlert.id)).filter(KPIAlert.status == "pending").scalar()
dims = db.query(KPIDefinition.dimension, func.count(KPIDefinition.id)).filter(
KPIDefinition.status == "active", KPIDefinition.entity_id == entity_id
).group_by(KPIDefinition.dimension).all()
# 读取最近一次同步状态(从日志文件最后一行)
sync_status = {"last_sync": None, "status": "unknown", "detail": ""}
try:
with open("/var/log/cma-daily-sync.log", "r") as f:
lines = f.readlines()
# 从最后往前找包含 "完成" 或 "失败" 的行
for line in reversed(lines[-50:]):
if "全部完成" in line:
sync_status["status"] = "success"
sync_status["last_sync"] = line.strip()
break
elif "失败" in line or "ERROR" in line:
sync_status["status"] = "failed"
sync_status["last_sync"] = line.strip()
break
else:
# 没找到完成/失败标记,取最后一行
sync_status["last_sync"] = lines[-1].strip() if lines else None
except Exception as e:
sync_status["detail"] = str(e)
result = {
"kpi_total": kpi_total or 0, "alert_count": alert_count or 0,
"dimension_stats": [{"dimension": d[0], "count": d[1]} for d in dims],
"sync_status": sync_status,
}
cache_set("dashboard", cache_key, result, ttl_seconds=30)
return result
@router.get("/kpis")
def get_dashboard_kpis(role: str = Query("ceo"), period: str = Query("month"),
start_date: str = Query(None), end_date: str = Query(None),
db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)):
start, end = parse_period(period, start_date, end_date)
period_str = start.strftime("%Y-%m")
kpis = db.query(KPIDefinition).filter(
KPIDefinition.status == "active",
KPIDefinition.entity_id == entity_id
).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":
months = period_prefix("quarter")
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()
result.append({
"id": k.id, "kpi_code": k.kpi_code, "kpi_name": k.kpi_name,
"dimension": k.dimension, "unit": k.unit, "target_value": k.target_value,
"target_monthly": k.target_monthly, "target_quarterly": k.target_quarterly, "target_yearly": k.target_yearly,
"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,
})
return {"data": result, "period": period, "range": {"start": start.strftime("%Y-%m-%d"), "end": end.strftime("%Y-%m-%d")}}
@router.get("/my-kpis")
def get_my_kpis(
current_user: User = Depends(require_auth),
period: str = Query("month"),
db: Session = Depends(get_db),
):
"""获取当前用户负责的KPI
- business角色:只看自己负责的KPI
- 其他角色:看所有有预警的KPI
"""
role = current_user.role
username = current_user.username
name = current_user.name
period_str = datetime.now().strftime("%Y-%m")
kpis = db.query(KPIDefinition).filter(KPIDefinition.status == "active").all()
result = []
for k in kpis:
# business角色筛选
if role == "business":
responsible = (k.responsible_user or "").strip()
if responsible and responsible != username and responsible != name:
continue
latest = db.query(KPIValue).filter(
KPIValue.kpi_id == k.id,
KPIValue.period == period_str,
).order_by(KPIValue.id.desc()).first()
alert = db.query(KPIAlert).filter(
KPIAlert.kpi_id == k.id,
KPIAlert.status == "pending",
).order_by(KPIAlert.id.desc()).first()
trend_values = db.query(KPIValue).filter(
KPIValue.kpi_id == k.id,
).order_by(KPIValue.period.desc()).limit(6).all()
trend = [{"period": v.period, "value": v.actual_value} for v in reversed(trend_values)]
result.append({
"id": k.id, "kpi_code": k.kpi_code, "kpi_name": k.kpi_name,
"dimension": k.dimension, "unit": k.unit,
"target_value": k.target_value,
"target_monthly": k.target_monthly, "target_quarterly": k.target_quarterly, "target_yearly": k.target_yearly,
"actual_value": latest.actual_value if latest else None,
"period": latest.period if latest else period_str,
"alert_level": alert.alert_level if alert else "none",
"alert_message": alert.alert_message if alert else None,
"alert_id": alert.id if alert else None,
"frequency": k.frequency,
"responsible_dept": k.responsible_dept,
"responsible_user": k.responsible_user,
"trend": trend,
"threshold_green": k.threshold_green,
"threshold_yellow": k.threshold_yellow,
"threshold_red": k.threshold_red,
})
return {"data": result, "user_role": role, "user_name": name, "period": period_str}
@router.get("/finance-analysis")
def get_finance_analysis(
current_user: User = Depends(require_auth),
period: str = Query("month"),
db: Session = Depends(get_db),
entity_id: int = Depends(get_entity_id),
):
"""财务工作台分析数据"""
period_str = datetime.now().strftime("%Y-%m")
finance_kpis = db.query(KPIDefinition).filter(
KPIDefinition.status == "active",
KPIDefinition.dimension == "finance",
KPIDefinition.entity_id == entity_id,
).all()
kpi_data = []
for k in finance_kpis:
latest = db.query(KPIValue).filter(
KPIValue.kpi_id == k.id,
KPIValue.period == period_str,
).order_by(KPIValue.id.desc()).first()
trend_values = db.query(KPIValue).filter(
KPIValue.kpi_id == k.id,
).order_by(KPIValue.period.desc()).limit(6).all()
trend = [{"period": v.period, "value": v.actual_value} for v in reversed(trend_values)]
alert = db.query(KPIAlert).filter(
KPIAlert.kpi_id == k.id,
KPIAlert.status == "pending",
).order_by(KPIAlert.id.desc()).first()
kpi_data.append({
"id": k.id, "kpi_code": k.kpi_code, "kpi_name": k.kpi_name,
"unit": k.unit, "target_value": k.target_value,
"target_monthly": k.target_monthly, "target_quarterly": k.target_quarterly, "target_yearly": k.target_yearly,
"actual_value": latest.actual_value if latest else None,
"threshold_green": k.threshold_green,
"threshold_yellow": k.threshold_yellow,
"threshold_red": k.threshold_red,
"trend": trend,
"alert_level": alert.alert_level if alert else "none",
"frequency": k.frequency,
})
total_sales = next((k for k in kpi_data if k["kpi_code"] == "SALES_TOTAL"), None)
gross_profit = next((k for k in kpi_data if k["kpi_code"] == "SALES_PROFIT_RATE"), None)
cost_control = next((k for k in kpi_data if k["kpi_code"] == "COST_CONTROL_RATE"), None)
receivable = next((k for k in kpi_data if k["kpi_code"] == "RECEIVABLE_TURNOVER"), None)
return {
"period": period_str,
"kpis": kpi_data,
"summary": {
"total_sales": total_sales["actual_value"] if total_sales else None,
"gross_profit_rate": gross_profit["actual_value"] if gross_profit else None,
"cost_control_rate": cost_control["actual_value"] if cost_control else None,
"receivable_turnover": receivable["actual_value"] if receivable else None,
}
}
@router.get("/predict")
def predict_kpis(db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)):
"""基于历史趋势预测下月KPI值(简单线性回归)"""
from datetime import datetime, timedelta
period_str = datetime.now().strftime("%Y-%m")
next_month = int(period_str[5:7]) + 1
next_year = int(period_str[:4])
if next_month > 12:
next_month = 1
next_year += 1
next_period = f"{next_year}-{next_month:02d}"
kpis = db.query(KPIDefinition).filter(
KPIDefinition.status == "active",
KPIDefinition.entity_id == entity_id
).all()
predictions = []
for k in kpis:
values = db.query(KPIValue).filter(
KPIValue.kpi_id == k.id,
).order_by(KPIValue.period.asc()).all()
# 需要至少3个数据点才能做预测
if len(values) < 3:
continue
# 简单线性回归: y = a + bx
points = [(i, v.actual_value) for i, v in enumerate(values) if v.actual_value is not None]
if len(points) < 3:
continue
n = len(points)
sum_x = sum(p[0] for p in points)
sum_y = sum(p[1] for p in points)
sum_xy = sum(p[0] * p[1] for p in points)
sum_xx = sum(p[0] ** 2 for p in points)
# 斜率 b = (n*sum_xy - sum_x*sum_y) / (n*sum_xx - sum_x*sum_x)
denom = n * sum_xx - sum_x * sum_x
if denom == 0:
continue
b = (n * sum_xy - sum_x * sum_y) / denom
a = (sum_y - b * sum_x) / n
# 预测下个月(x = n,因为最后一个索引是 n-1)
predicted_value = a + b * n
# 检查预测值是否触发阈值
alert_level = "none"
if k.threshold_red:
op = k.threshold_red[:2] if k.threshold_red[1] in "=<>" else k.threshold_red[0]
val = float(k.threshold_red.replace(op, "").strip())
if (op in (">=", ">") and predicted_value >= val) or (op in ("<=", "<") and predicted_value <= val):
alert_level = "red"
if alert_level == "none" and k.threshold_yellow:
op = k.threshold_yellow[:2] if k.threshold_yellow[1] in "=<>" else k.threshold_yellow[0]
val = float(k.threshold_yellow.replace(op, "").strip())
if (op in (">=", ">") and predicted_value >= val) or (op in ("<=", "<") and predicted_value <= val):
alert_level = "yellow"
predictions.append({
"kpi_id": k.id,
"kpi_code": k.kpi_code,
"kpi_name": k.kpi_name,
"target_value": k.target_value,
"last_value": points[-1][1] if points else None,
"predicted_value": round(predicted_value, 2),
"predicted_period": next_period,
"alert_level": alert_level,
"trend": "up" if b > 0 else ("down" if b < 0 else "stable"),
"confidence": "high" if len(points) >= 6 else ("medium" if len(points) >= 4 else "low"),
"data_points": len(points),
})
return {
"current_period": period_str,
"next_period": next_period,
"predictions": predictions,
"kpi_count": len(kpis),
"predictable_count": len(predictions),
}
# ── 个人工作台 ──────────────────────────────
@router.get("/my-dashboard")
def my_dashboard(
current_user: User = Depends(require_auth),
db: Session = Depends(get_db),
entity_id: int = Depends(get_entity_id),
):
"""个人工作台:返回我的KPI、改善行动、待办提醒(账套隔离: 按token企业过滤)"""
username = current_user.username
name = current_user.name
role = current_user.role
# 角色预设KPI编码
ROLE_PRESET_KPIS = {
"ceo": ["F_REVENUE", "F_NET_PROFIT", "F_COST_RATIO", "C_REBATE_RATE", "P_DELIVERY", "F_FCF"],
"finance": ["F_REVENUE", "F_NET_PROFIT", "F_COST_RATIO", "F_OP_CFLOW", "F_ROE"],
"business": ["C_REBATE_RATE", "C_NEW_CLIENTS", "F_REVENUE"],
"it": [],
}
preset_codes = ROLE_PRESET_KPIS.get(role, [])
# 1. 我的KPIresponsible_user匹配用户名或姓名)+ 角色预设(均按企业隔离)
assigned_kpis = db.query(KPIDefinition).filter(
or_(
KPIDefinition.responsible_user == username,
KPIDefinition.responsible_user == name,
),
KPIDefinition.status == "active",
KPIDefinition.entity_id == entity_id,
).all()
assigned_ids = {k.id for k in assigned_kpis}
# 补充角色预设KPI(去重,按企业隔离)
preset_kpis = []
if preset_codes:
q = db.query(KPIDefinition).filter(
KPIDefinition.kpi_code.in_(preset_codes),
KPIDefinition.status == "active",
KPIDefinition.entity_id == entity_id,
)
if assigned_ids:
q = q.filter(~KPIDefinition.id.in_(assigned_ids))
preset_kpis = q.all()
all_kpis = assigned_kpis + preset_kpis
kpi_list = []
for k in all_kpis:
latest_v = db.query(KPIValue).filter(
KPIValue.kpi_id == k.id
).order_by(KPIValue.calculated_at.desc()).first()
actual = latest_v.actual_value if latest_v else None
target = kpi_target_by_frequency(k)
level = "gray"
if actual is not None and target:
# 反向指标(越低越好):费用率/渠补率/应收天数/返利率/成本率
REVERSE_INDICATORS = {
"F_COST_RATIO", "C_REBATE_RATE", "F_AR_DAYS",
"F_REBATE_RATE", "F_FACTORY_REBATE_RATE", "F_COST_CONTROL_RATE",
}
if k.kpi_code in REVERSE_INDICATORS:
# 反向:实际≤目标=绿;实际≤目标*1.1=黄;否则红
level = "green" if actual <= target else (
"yellow" if actual <= target * 1.1 else "red")
else:
ratio = actual / target
level = "green" if ratio >= 0.9 else ("yellow" if ratio >= 0.7 else "red")
kpi_list.append({
"id": k.id,
"kpi_code": k.kpi_code,
"kpi_name": k.kpi_name,
"dimension": k.dimension,
"category": k.category,
"target_value": target,
"target_monthly": k.target_monthly,
"target_quarterly": k.target_quarterly,
"target_yearly": k.target_yearly,
"frequency": k.frequency,
"actual_value": actual,
"unit": k.unit,
"level": level,
"period": latest_v.period if latest_v else None,
})
# 2. 我的改善行动(assignee匹配)
from app.models import ActionPlan
my_plans = db.query(ActionPlan).filter(
or_(
ActionPlan.assignee == username,
ActionPlan.assignee == name,
)
).order_by(ActionPlan.updated_at.desc()).all()
plan_list = []
for p in my_plans:
overdue = False
if p.due_date and p.status not in ("completed", "cancelled"):
overdue = p.due_date < datetime.now()
kpi_name = ""
kpi = db.query(KPIDefinition).filter(KPIDefinition.id == p.kpi_id).first()
if kpi:
kpi_name = kpi.kpi_name
plan_list.append({
"id": p.id,
"kpi_id": p.kpi_id,
"kpi_name": kpi_name,
"title": p.title,
"assignee": p.assignee,
"priority": p.priority,
"status": p.status,
"progress": p.progress or 0,
"due_date": p.due_date.isoformat() if p.due_date else None,
"overdue": overdue,
"created_at": p.created_at.isoformat() if p.created_at else None,
})
# 3. 待办提醒
reminders = []
# 逾期行动
for p in plan_list:
if p["overdue"]:
reminders.append({
"type": "overdue_plan",
"severity": "danger",
"message": f"你负责的「{p['title']}」已逾期",
"related_id": p["id"],
"related_type": "action_plan",
})
# 红色预警KPI
for k in kpi_list:
if k["level"] == "red":
reminders.append({
"type": "red_kpi",
"severity": "danger",
"message": f"你负责的KPI「{k['kpi_name']}」处于红色预警",
"related_id": k["id"],
"related_type": "kpi",
})
# 黄色预警KPI
for k in kpi_list:
if k["level"] == "yellow":
reminders.append({
"type": "yellow_kpi",
"severity": "warning",
"message": f"你负责的KPI「{k['kpi_name']}」处于黄色预警",
"related_id": k["id"],
"related_type": "kpi",
})
return {
"kpis": kpi_list,
"action_plans": plan_list,
"reminders": reminders,
}