feat: 安全加固检查与评分（加权安全评分+加固建议+详情页评分展示）

app/routers/monitor.py

@@ -7,10 +7,16 @@ from sqlmodel import Session, select
 
 from app.database import get_metrics_session
 from app.models.monitor import MetricPoint, SecurityCheck, ServerInfo
+from app.services import security_service
 
 router = APIRouter(prefix="/api/monitor", tags=["monitor"])
 
 
+@router.get("/security-overview", summary="所有服务器安全评分总览（风险高的在前）")
+def security_overview(session: Session = Depends(get_metrics_session)):
+    return security_service.get_security_overview(session)
+
+
 @router.get("/{asset_id}/metrics", summary="资源监控时序（倒序，最新在前）")
 def metrics(
     asset_id: int,
@@ -47,3 +53,8 @@ def security(asset_id: int, session: Session = Depends(get_metrics_session)):
         if check.check_item not in latest_by_item:
             latest_by_item[check.check_item] = check
     return list(latest_by_item.values())
+
+
+@router.get("/{asset_id}/security-score", summary="单个服务器安全评分与加固建议")
+def security_score(asset_id: int, session: Session = Depends(get_metrics_session)):
+    return security_service.get_asset_security(session, asset_id)

app/services/security_service.py

@@ -0,0 +1,77 @@
+"""安全评分服务
+
+基于 Agent 上报的安全检查项（SecurityCheck）计算安全评分与加固建议。
+评分按检查项加权：pass 满分、warn 半分、fail/unknown 零分。
+"""
+
+from sqlmodel import Session, select
+
+from app.models.monitor import SecurityCheck
+
+# 检查项权重（总和约 100）
+CHECK_WEIGHTS = {"ssh_config": 40, "firewall": 35, "listening_ports": 25}
+DEFAULT_WEIGHT = 20
+
+
+def latest_checks(session: Session, asset_id: int) -> list:
+    """取每个检查项的最新一条"""
+    stmt = (
+        select(SecurityCheck)
+        .where(SecurityCheck.asset_id == asset_id)
+        .order_by(SecurityCheck.ts.desc())
+        .limit(50)
+    )
+    checks = session.exec(stmt).all()
+    latest = {}
+    for check in checks:
+        if check.check_item not in latest:
+            latest[check.check_item] = check
+    return list(latest.values())
+
+
+def compute_security_score(checks: list):
+    """计算安全评分（0-100），无检查数据返回 None"""
+    if not checks:
+        return None
+    total = 0
+    earned = 0.0
+    for check in checks:
+        weight = CHECK_WEIGHTS.get(check.check_item, DEFAULT_WEIGHT)
+        total += weight
+        if check.status == "pass":
+            earned += weight
+        elif check.status == "warn":
+            earned += weight * 0.5
+    return round(earned / total * 100) if total else None
+
+
+def score_level(score):
+    """评分分级：good(>=80) / warning(>=50) / risk(<50) / unknown"""
+    if score is None:
+        return "unknown"
+    if score >= 80:
+        return "good"
+    if score >= 50:
+        return "warning"
+    return "risk"
+
+
+def get_asset_security(session: Session, asset_id: int) -> dict:
+    """单个服务器的安全评分与加固建议"""
+    checks = latest_checks(session, asset_id)
+    score = compute_security_score(checks)
+    return {
+        "asset_id": asset_id,
+        "score": score,
+        "level": score_level(score),
+        "checks_count": len(checks),
+        "suggestions": [c.suggestion for c in checks if c.suggestion],
+    }
+
+
+def get_security_overview(session: Session) -> list:
+    """所有有安全检查数据的服务器评分总览（按评分升序，风险高的在前）"""
+    asset_ids = session.exec(select(SecurityCheck.asset_id).distinct()).all()
+    result = [get_asset_security(session, aid) for aid in asset_ids]
+    result.sort(key=lambda x: (x["score"] is None, x["score"] if x["score"] is not None else 0))
+    return result