"""驾驶舱 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. 我的KPI(responsible_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, }