Files
AIPortPilot/backend/app/routers/dashboard.py
T

301 lines
11 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.
"""健康度评分 + 仪表盘路由。"""
from fastapi import APIRouter, Depends, Query
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import get_db
from app.core.dependencies import get_current_user
from app.models.company import Company
from app.models.health_score import HealthScore
from app.models.report import MonthlyReport
from app.models.risk import RiskEvent
from app.models.user import User
from app.schemas.common import ApiResponse, success
from app.schemas.health_score import DashboardSummary, HealthScoreResponse
from app.services.predictor import predict_trend, detect_anomalies
router = APIRouter(prefix="/dashboard", tags=["dashboard"])
# 14 维度 key 列表
_DIMENSION_KEYS = [
"financial_score", "operational_score", "ai_commercial_score", "ai_cost_score",
"org_talent_score", "product_tech_score", "market_compete_score", "governance_score",
"financing_score", "synergy_score", "ai_model_product_score",
"data_compliance_score", "team_tech_score", "customer_success_score",
]
@router.get("/summary", response_model=ApiResponse[DashboardSummary])
async def get_dashboard_summary(
company_id: str | None = Query(default=None, description="指定企业 ID,不传则汇总全租户"),
db: AsyncSession = Depends(get_db),
user: User = Depends(get_current_user),
):
"""获取仪表盘汇总数据。支持按企业过滤。"""
tenant_id = user.tenant_id
# 企业总数(单企业视角时为 1
if company_id:
total_companies = 1
else:
companies_result = await db.execute(
select(func.count()).select_from(Company).where(Company.tenant_id == tenant_id)
)
total_companies = companies_result.scalar_one()
# 平均健康度(按企业过滤时只算该企业)
avg_query = (
select(func.avg(HealthScore.total_score))
.join(Company, HealthScore.company_id == Company.id)
.where(Company.tenant_id == tenant_id)
)
if company_id:
avg_query = avg_query.where(HealthScore.company_id == company_id)
avg_result = await db.execute(avg_query)
avg_score = avg_result.scalar_one()
avg_health_score = float(avg_score) if avg_score else 0.0
# 高风险事件数
risk_query = (
select(func.count())
.select_from(RiskEvent)
.join(Company, RiskEvent.company_id == Company.id)
.where(Company.tenant_id == tenant_id, RiskEvent.status.in_(["open", "assigned", "in_progress"]))
)
if company_id:
risk_query = risk_query.where(RiskEvent.company_id == company_id)
risk_result = await db.execute(risk_query)
high_risk_count = risk_result.scalar_one()
# 待审阅月报数
pending_query = (
select(func.count())
.select_from(MonthlyReport)
.join(Company, MonthlyReport.company_id == Company.id)
.where(Company.tenant_id == tenant_id, MonthlyReport.status.in_(["submitted", "ai_parsed"]))
)
if company_id:
pending_query = pending_query.where(MonthlyReport.company_id == company_id)
pending_result = await db.execute(pending_query)
pending_reports = pending_result.scalar_one()
# 最近评分 — 每家企业只取最新一条(避免历史数据导致企业重复)
subq = (
select(
HealthScore.company_id,
func.max(HealthScore.calculated_at).label("max_at"),
)
.join(Company, HealthScore.company_id == Company.id)
.where(Company.tenant_id == tenant_id)
.group_by(HealthScore.company_id)
)
if company_id:
subq = subq.where(HealthScore.company_id == company_id)
subq = subq.subquery()
recent_query = (
select(HealthScore, Company.name.label("company_name"))
.join(Company, HealthScore.company_id == Company.id)
.join(subq, (HealthScore.company_id == subq.c.company_id) & (HealthScore.calculated_at == subq.c.max_at))
.order_by(HealthScore.calculated_at.desc())
.limit(10)
)
recent_result = await db.execute(recent_query)
recent_rows = recent_result.all()
recent_scores = []
for row in recent_rows:
score = row[0]
company_name = row[1]
resp = HealthScoreResponse.model_validate(score, from_attributes=True)
resp.company_name = company_name
recent_scores.append(resp)
return success(
data=DashboardSummary(
total_companies=total_companies,
avg_health_score=round(avg_health_score, 1),
high_risk_count=high_risk_count,
pending_reports=pending_reports,
recent_scores=recent_scores,
)
)
@router.get("/scores", response_model=ApiResponse[list[HealthScoreResponse]])
async def list_health_scores(
company_id: str | None = Query(default=None),
limit: int = Query(default=20, ge=1, le=100),
db: AsyncSession = Depends(get_db),
user: User = Depends(get_current_user),
):
"""获取健康度评分列表 — 每家企业只返回最新一条。"""
subq = (
select(
HealthScore.company_id,
func.max(HealthScore.calculated_at).label("max_at"),
)
.join(Company, HealthScore.company_id == Company.id)
.where(Company.tenant_id == user.tenant_id)
.group_by(HealthScore.company_id)
)
if company_id:
subq = subq.where(HealthScore.company_id == company_id)
subq = subq.subquery()
query = (
select(HealthScore, Company.name.label("company_name"))
.join(Company, HealthScore.company_id == Company.id)
.join(subq, (HealthScore.company_id == subq.c.company_id) & (HealthScore.calculated_at == subq.c.max_at))
.order_by(HealthScore.calculated_at.desc())
.limit(limit)
)
result = await db.execute(query)
scores = []
for row in result.all():
score = row[0]
company_name = row[1]
resp = HealthScoreResponse.model_validate(score, from_attributes=True)
resp.company_name = company_name
scores.append(resp)
return success(data=scores)
@router.get("/heatmap", response_model=ApiResponse[list[dict]])
async def get_health_heatmap(
db: AsyncSession = Depends(get_db),
user: User = Depends(get_current_user),
):
"""获取健康度热力图数据 — 企业 × 维度评分矩阵。
返回格式:[{ company_id, company_name, scores: { dimension: score } }]
"""
# 获取租户下所有企业
companies_result = await db.execute(
select(Company).where(Company.tenant_id == user.tenant_id).order_by(Company.name)
)
companies = companies_result.scalars().all()
# 获取每个企业最新评分
heatmap = []
for company in companies:
score_result = await db.execute(
select(HealthScore)
.where(HealthScore.company_id == company.id)
.order_by(HealthScore.calculated_at.desc())
.limit(1)
)
score = score_result.scalar_one_or_none()
scores_dict = {}
if score:
for dim_key in _DIMENSION_KEYS:
val = getattr(score, dim_key, None)
if val is not None:
scores_dict[dim_key] = val
heatmap.append({
"company_id": company.id,
"company_name": company.name,
"total_score": score.total_score if score else None,
"scores": scores_dict,
})
return success(data=heatmap)
@router.get("/trends", response_model=ApiResponse[list[dict]])
async def get_health_trends(
company_id: str | None = Query(default=None, description="指定企业 ID,不传则返回全部企业"),
months: int = Query(default=6, ge=1, le=24, description="趋势月数"),
db: AsyncSession = Depends(get_db),
user: User = Depends(get_current_user),
):
"""获取健康度趋势数据 — 按企业分组,每家企业一条折线。
返回格式:[{ company_id, company_name, data: [{ period, score }] }]
"""
query = (
select(HealthScore, Company.name.label("company_name"))
.join(Company, HealthScore.company_id == Company.id)
.where(Company.tenant_id == user.tenant_id)
)
if company_id:
query = query.where(HealthScore.company_id == company_id)
query = query.order_by(HealthScore.calculated_at.asc()).limit(months * 100)
result = await db.execute(query)
# 按企业分组
by_company: dict[str, dict[str, list]] = {}
for row in result.all():
score = row[0]
cname = row[1]
cid = str(score.company_id)
if cid not in by_company:
by_company[cid] = {"company_name": cname, "scores": []}
by_company[cid]["scores"].append(score)
trends = []
for cid, info in by_company.items():
# 每月取最新一条
monthly: dict[str, float] = {}
for s in info["scores"]:
period = s.calculated_at.strftime("%Y-%m")
monthly[period] = s.total_score
data_points = [
{"period": p, "score": round(v, 1)}
for p, v in sorted(monthly.items())
]
trends.append({
"company_id": cid,
"company_name": info["company_name"],
"data": data_points,
})
return success(data=trends)
@router.get("/forecasts", response_model=ApiResponse[dict])
async def get_health_forecasts(
company_id: str | None = Query(default=None, description="指定企业 ID,不传则汇总全租户"),
months_ahead: int = Query(default=3, ge=1, le=6, description="预测月数"),
db: AsyncSession = Depends(get_db),
user: User = Depends(get_current_user),
):
"""获取健康度预测数据 — 基于历史评分预测未来趋势 + 异常检测。
返回格式:{ predictions: [...], trend_direction, confidence, anomalies: [...] }
"""
query = (
select(HealthScore)
.join(Company, HealthScore.company_id == Company.id)
.where(Company.tenant_id == user.tenant_id)
)
if company_id:
query = query.where(HealthScore.company_id == company_id)
query = query.order_by(HealthScore.calculated_at.asc()).limit(24)
result = await db.execute(query)
scores = result.scalars().all()
historical = [s.total_score for s in scores]
forecast = predict_trend(historical, months_ahead)
# 异常检测 — 各维度
anomalies_by_dim: dict[str, list[int]] = {}
for dim_key in _DIMENSION_KEYS:
dim_values = [getattr(s, dim_key) for s in scores if getattr(s, dim_key) is not None]
if len(dim_values) >= 3:
dim_anomalies = detect_anomalies(dim_values)
if dim_anomalies:
anomalies_by_dim[dim_key] = dim_anomalies
return success(data={
"predictions": forecast.get("predicted", []),
"slope": forecast.get("slope", 0),
"confidence": forecast.get("confidence", 0),
"trend_direction": "up" if forecast.get("slope", 0) > 1 else "down" if forecast.get("slope", 0) < -1 else "stable",
"anomalies": anomalies_by_dim,
"historical_count": len(historical),
})