48 lines
1.7 KiB
Python
48 lines
1.7 KiB
Python
"""Dashboard model for Search Gateway observability."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import asdict, dataclass, field
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
|
|
|
|
def utc_now() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SearchGatewayMetric:
|
|
key: str
|
|
title: str
|
|
value: Any
|
|
status: str = "ok"
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
|
|
class SearchGatewayDashboard:
|
|
def overview(self, requests: list[dict[str, Any]] | None = None) -> dict[str, Any]:
|
|
requests = requests or []
|
|
blocked = [r for r in requests if r.get("status") == "blocked"]
|
|
failed = [r for r in requests if r.get("status") in {"failed", "not_found"}]
|
|
fallback = [r for r in requests if r.get("fallback_triggered")]
|
|
return {
|
|
"module": "search-gateway",
|
|
"route": "/admin/ai-evolution/search",
|
|
"title": "AIOS 联网搜索治理中心",
|
|
"status": "attention_required" if failed else "healthy",
|
|
"generated_at": utc_now(),
|
|
"metrics": [
|
|
SearchGatewayMetric("total_requests", "搜索请求数", len(requests)).to_dict(),
|
|
SearchGatewayMetric("blocked_requests", "合规阻断数", len(blocked), "warn" if blocked else "ok").to_dict(),
|
|
SearchGatewayMetric("not_found_or_failed", "未找到/失败", len(failed), "warn" if failed else "ok").to_dict(),
|
|
SearchGatewayMetric("fallback_triggered", "降级次数", len(fallback)).to_dict(),
|
|
],
|
|
"recent_requests": requests[-20:],
|
|
}
|
|
|
|
|
|
__all__ = ["SearchGatewayMetric", "SearchGatewayDashboard"]
|