Spaces:
Running
Running
File size: 3,826 Bytes
f3220aa 6339a53 f3220aa | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 | """Admin dashboard for session monitoring."""
import time
from typing import Any
from fastapi import APIRouter, Depends, Request
from starlette.responses import HTMLResponse
from starlette.templating import Jinja2Templates
from core.session_tracker import SessionTracker
from .dependencies import require_api_key
router = APIRouter()
templates = Jinja2Templates(directory="templates")
def _get_blocked_providers() -> set[str]:
"""Get currently blocked provider IDs. Placeholder - returns empty set."""
return set()
def _format_time(seconds: float) -> str:
"""Format seconds as human-readable string."""
if seconds < 60:
return f"{int(seconds)}s ago"
elif seconds < 3600:
return f"{int(seconds // 60)}m ago"
else:
return f"{int(seconds // 3600)}h ago"
def _get_admin_data() -> dict[str, Any]:
"""Gather all data for the admin dashboard."""
tracker = SessionTracker.get_instance()
blocked = _get_blocked_providers()
session_loads = tracker.get_all_session_loads()
provider_loads = tracker.get_all_provider_loads(blocked)
now = time.monotonic()
total_requests = 0
all_providers = set()
sessions_list = []
for session_id, load in session_loads.items():
total_requests += load.total_requests
all_providers.update(load.providers.keys())
state = tracker._sessions.get(session_id)
last_activity_seconds = 0.0
if state:
last_activity_seconds = now - state.last_request_time
req_per_min = 0.0
if state and state.requests_in_window > 0:
req_per_min = (state.requests_in_window / tracker._window_seconds) * 60
sessions_list.append(
{
"session_id": session_id,
"total_requests": load.total_requests,
"providers": list(load.providers.keys()),
"last_activity_seconds": last_activity_seconds,
"last_activity_display": _format_time(last_activity_seconds),
"requests_per_minute": int(req_per_min),
"status": "Active" if last_activity_seconds < 300 else "Idle",
"status_class": "status-active"
if last_activity_seconds < 300
else "status-idle",
}
)
providers_list = []
for provider_id, load in provider_loads.items():
remaining = 0
if provider_id in blocked:
remaining = 30 # placeholder, real impl would check rate limiter
status = "Rate Limited" if not load.is_healthy else "Healthy"
if not load.is_healthy and remaining > 0:
status += f" ({remaining}s remaining)"
providers_list.append(
{
"provider_id": provider_id,
"active_requests": load.active_requests,
"sessions_count": load.session_count,
"requests_per_minute": int(load.requests_per_minute),
"is_healthy": load.is_healthy,
"status": status,
}
)
return {
"summary": {
"active_sessions": len(session_loads),
"total_requests": total_requests,
"providers": len(all_providers),
},
"sessions": sessions_list,
"providers": providers_list,
}
@router.get("/admin", response_class=HTMLResponse)
async def admin_dashboard(
request: Request,
_auth=Depends(require_api_key),
):
"""Admin dashboard HTML page."""
data = _get_admin_data()
return templates.TemplateResponse("admin.html", {"request": request, **data})
@router.get("/api/admin/sessions")
async def admin_sessions_json(
_auth=Depends(require_api_key),
):
"""JSON API for session data (for external tools)."""
return _get_admin_data()
|