File size: 6,733 Bytes
e0b68d2 | 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 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 | """管理后台统计聚合辅助函数。"""
from __future__ import annotations
import os
import time
from typing import Any, Dict, Optional
try:
import psutil
except ImportError:
psutil = None
from app.services.request_log_dao import RequestLogDAO, get_request_log_dao
from app.services.token_dao import TokenDAO, get_token_dao
from app.utils.format import format_compact_number
from app.utils.token_pool import TokenPool, get_token_pool
_TOKEN_POOL_SENTINEL = object()
DEFAULT_TREND_WINDOW = "7d"
TREND_WINDOW_OPTIONS = (
{"key": "24h", "label": "24 小时"},
{"key": "7d", "label": "7 天"},
{"key": "30d", "label": "30 天"},
)
def _coerce_int(value: Any) -> int:
"""将数据库聚合结果安全转换为整数。"""
return int(value or 0)
def calculate_success_rate(
successful_requests: int,
total_requests: int,
) -> float:
"""计算成功率百分比。"""
if total_requests <= 0:
return 0.0
return round(successful_requests / total_requests * 100, 1)
def normalize_trend_window(value: Any) -> str:
"""规范化趋势窗口参数，非法值回退到默认值。"""
normalized = str(value or "").strip().lower()
if normalized in {"24h", "7d", "30d"}:
return normalized
if normalized == "1d":
return "24h"
return DEFAULT_TREND_WINDOW
def format_uptime(total_seconds: int) -> str:
"""格式化运行时长。"""
total_seconds = max(0, int(total_seconds))
days, remainder = divmod(total_seconds, 86400)
hours, remainder = divmod(remainder, 3600)
minutes, seconds = divmod(remainder, 60)
parts = []
if days:
parts.append(f"{days}天")
if days or hours:
parts.append(f"{hours}小时")
if days or hours or minutes:
parts.append(f"{minutes}分钟")
parts.append(f"{seconds}秒")
return " ".join(parts)
def get_process_uptime() -> str:
"""获取当前进程运行时长。"""
if psutil is None:
return "未知"
created_at = psutil.Process(os.getpid()).create_time()
return format_uptime(int(time.time() - created_at))
async def collect_admin_stats(
provider: str,
*,
token_dao: Optional[TokenDAO] = None,
request_log_dao: Optional[RequestLogDAO] = None,
token_pool: Any = _TOKEN_POOL_SENTINEL,
trend_window: str = DEFAULT_TREND_WINDOW,
) -> Dict[str, Any]:
"""聚合管理后台所需的 Token 与请求统计。"""
token_dao = token_dao or get_token_dao()
request_log_dao = request_log_dao or get_request_log_dao()
if token_pool is _TOKEN_POOL_SENTINEL:
token_pool = get_token_pool()
trend_window = normalize_trend_window(trend_window)
token_counts = await token_dao.get_provider_token_counts(provider)
request_stats = await request_log_dao.get_provider_request_stats(provider)
usage_trend = await request_log_dao.get_provider_usage_trend(
provider,
window=trend_window,
)
pool_status: Dict[str, Any] = {}
if isinstance(token_pool, TokenPool) or hasattr(token_pool, "get_pool_status"):
pool_status = await token_pool.get_pool_status() if token_pool else {}
total_tokens = _coerce_int(token_counts.get("total_tokens"))
enabled_tokens = _coerce_int(token_counts.get("enabled_tokens"))
user_tokens = _coerce_int(token_counts.get("user_tokens"))
guest_tokens = _coerce_int(token_counts.get("guest_tokens"))
unknown_tokens = _coerce_int(token_counts.get("unknown_tokens"))
pool_total_tokens = _coerce_int(pool_status.get("total_tokens"))
if pool_total_tokens == 0 and token_pool is None:
pool_total_tokens = max(0, enabled_tokens - guest_tokens)
available_tokens = _coerce_int(pool_status.get("available_tokens"))
healthy_tokens = _coerce_int(pool_status.get("healthy_tokens"))
unhealthy_tokens = _coerce_int(pool_status.get("unhealthy_tokens"))
total_requests = _coerce_int(request_stats.get("total_requests"))
successful_requests = _coerce_int(request_stats.get("successful_requests"))
failed_requests = _coerce_int(request_stats.get("failed_requests"))
input_tokens = _coerce_int(request_stats.get("input_tokens"))
output_tokens = _coerce_int(request_stats.get("output_tokens"))
total_consumed_tokens = _coerce_int(request_stats.get("total_tokens"))
cache_creation_tokens = _coerce_int(
request_stats.get("cache_creation_tokens")
)
cache_read_tokens = _coerce_int(request_stats.get("cache_read_tokens"))
cache_creation_requests = _coerce_int(
request_stats.get("cache_creation_requests")
)
cache_hit_requests = _coerce_int(request_stats.get("cache_hit_requests"))
average_latency = round(float(request_stats.get("avg_duration") or 0.0), 2)
average_first_token_latency = round(
float(request_stats.get("avg_first_token_time") or 0.0),
2,
)
total_cache_tokens = cache_creation_tokens + cache_read_tokens
return {
"total_tokens": total_tokens,
"enabled_tokens": enabled_tokens,
"user_tokens": user_tokens,
"guest_tokens": guest_tokens,
"unknown_tokens": unknown_tokens,
"pool_total_tokens": pool_total_tokens,
"available_tokens": available_tokens,
"healthy_tokens": healthy_tokens,
"unhealthy_tokens": unhealthy_tokens,
"total_requests": total_requests,
"successful_requests": successful_requests,
"failed_requests": failed_requests,
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"total_consumed_tokens": total_consumed_tokens,
"cache_creation_tokens": cache_creation_tokens,
"cache_read_tokens": cache_read_tokens,
"total_cache_tokens": total_cache_tokens,
"cache_creation_requests": cache_creation_requests,
"cache_hit_requests": cache_hit_requests,
"average_latency": average_latency,
"average_first_token_latency": average_first_token_latency,
"trend_window": trend_window,
"usage_trend": usage_trend,
"total_consumed_tokens_display": format_compact_number(
total_consumed_tokens
),
"total_cache_tokens_display": format_compact_number(
total_cache_tokens
),
"input_tokens_display": format_compact_number(input_tokens),
"output_tokens_display": format_compact_number(output_tokens),
"success_rate": calculate_success_rate(
successful_requests,
total_requests,
),
}
|