hive-manage-v3/backend/app/api/health.py

"""Health API — 健康分查询、对比、手动刷新。"""
from fastapi import APIRouter, Depends, HTTPException, Query

from app.middleware.auth import require_auth, require_role
from app.models.user import User
from app.services.health_scorer import (
    PLACEHOLDER_PROFILES,
    calculate_all_scores,
    calculate_health_score,
    invalidate_cache,
)

router = APIRouter(prefix="/health", tags=["health"])


def _score_to_row(hs, rank: int) -> dict:
    return {
        "rank": rank,
        "name": hs.profile_name,
        "total_score": hs.total_score,
        "grade": hs.grade,
        "gateway_score": hs.gateway.score,
        "config_score": hs.config.score,
        "skill_score": hs.skill.score,
        "cron_score": hs.cron.score,
        "deduct_summary": hs.deduct_summary,
    }


@router.get("/scores")
async def list_scores(
    grade: str | None = Query(None, pattern="^(A|B|C|D)$"),
    sort: str = Query("asc", pattern="^(asc|desc)$"),
    _user: User = Depends(require_auth),
):
    """返回全部 Profile 健康分排行列表。

    - ?grade=A|B|C|D  按等级筛选
    - ?sort=asc|desc   排序，默认从低到高（问题优先）
    """
    scores = calculate_all_scores()

    # 筛选
    if grade:
        scores = [s for s in scores if s.grade == grade]

    # 排序
    scores.sort(key=lambda s: s.total_score, reverse=(sort == "desc"))

    rows = [_score_to_row(hs, i + 1) for i, hs in enumerate(scores)]
    return {
        "total": len(rows),
        "distribution": {
            "A": sum(1 for s in scores if s.grade == "A"),
            "B": sum(1 for s in scores if s.grade == "B"),
            "C": sum(1 for s in scores if s.grade == "C"),
            "D": sum(1 for s in scores if s.grade == "D"),
        },
        "scores": rows,
    }


@router.get("/scores/{profile}")
async def get_profile_score(
    profile: str,
    _user: User = Depends(require_auth),
):
    """返回单个 Profile 的健康分详情，含四维度扣分项和修复建议。"""
    if profile not in PLACEHOLDER_PROFILES:
        raise HTTPException(status_code=404, detail=f"Profile '{profile}' 不存在")

    hs = calculate_health_score(profile)
    return {
        "profile_name": hs.profile_name,
        "total_score": hs.total_score,
        "grade": hs.grade,
        "calculated_at": hs.calculated_at,
        "dimensions": [
            {
                "name": "Gateway 稳定性",
                "weight": "40%",
                "score": hs.gateway.score,
                "max_score": hs.gateway.max_score,
                "deductions": hs.gateway.deductions,
                "suggestions": hs.gateway.suggestions,
            },
            {
                "name": "配置完整性",
                "weight": "25%",
                "score": hs.config.score,
                "max_score": hs.config.max_score,
                "deductions": hs.config.deductions,
                "suggestions": hs.config.suggestions,
            },
            {
                "name": "Skill 一致性",
                "weight": "20%",
                "score": hs.skill.score,
                "max_score": hs.skill.max_score,
                "deductions": hs.skill.deductions,
                "suggestions": hs.skill.suggestions,
            },
            {
                "name": "Cron 成功率",
                "weight": "15%",
                "score": hs.cron.score,
                "max_score": hs.cron.max_score,
                "deductions": hs.cron.deductions,
                "suggestions": hs.cron.suggestions,
            },
        ],
    }


@router.post("/compare")
async def compare_profiles(
    body: dict,
    _user: User = Depends(require_auth),
):
    """返回多个 Profile 的配置/Skill 横向对比。

    body: {"profile_names": ["hebo", "ops-engineer"]}
    """
    names = body.get("profile_names", [])
    if len(names) < 2:
        raise HTTPException(status_code=400, detail="至少选择 2 个 Profile")
    if len(names) > 20:
        raise HTTPException(status_code=400, detail="最多对比 20 个 Profile")

    invalid = [n for n in names if n not in PLACEHOLDER_PROFILES]
    if invalid:
        raise HTTPException(status_code=404, detail=f"Profile 不存在: {', '.join(invalid)}")

    scores = {hs.profile_name: hs for hs in calculate_all_scores(names)}

    # 收集所有配置字段和 Skill
    from app.services.health_scorer import _REQUIRED_CONFIG_FIELDS, _BASELINE_SKILLS
    from app.services.health_scorer import _placeholder_config_data, _placeholder_skill_data

    config_fields = list(_REQUIRED_CONFIG_FIELDS)
    skill_fields = sorted(_BASELINE_SKILLS)

    rows: list[dict] = []
    for field in config_fields:
        row: dict = {"field": field, "category": "config"}
        for name in names:
            s = scores.get(name)
            if s is None:
                row[name] = "⚠"
            else:
                data = _placeholder_config_data(name)
                row[name] = "✅" if field in data["present_fields"] else "❌"
        rows.append(row)

    for field in skill_fields:
        row = {"field": field, "category": "skill"}
        for name in names:
            s = scores.get(name)
            if s is None:
                row[name] = "⚠"
            else:
                data = _placeholder_skill_data(name)
                row[name] = "✅" if field in data["profile_skills"] else "❌"
        rows.append(row)

    # 检测差异行
    for row in rows:
        vals = {str(row[n]) for n in names}
        row["has_diff"] = len(vals) > 1

    return {
        "profiles": names,
        "rows": rows,
    }


@router.post("/calculate")
async def recalculate_scores(
    _user: User = Depends(require_role("owner")),
):
    """手动触发全量评分重算（仅 owner）。"""
    invalidate_cache()
    scores = calculate_all_scores()
    return {
        "message": f"已重新计算 {len(scores)} 个 Profile 的健康分",
        "count": len(scores),
    }