i/Aether
1
0
Fork 0
mirror of https://github.com/fawney19/Aether.git synced 2026-09-03 01:40:21 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: 统计数据优化 - 支持细粒度时间范围和多维度分析

Browse Source 
- 新增 StatsHourly/StatsDaily 预聚合表，支持任意时区的精确统计
- 实现 UTC datetime 范围查询策略，边界数据实时聚合
- 新增统计 API：用户/API Key 维度、成本分析、性能百分位、错误分类
- 新增前端页面：成本分析、性能分析、用户统计
- 新增 TimeRangePicker 组件和统计可视化组件
- 优化 Dashboard 和 Usage 页面支持时间范围筛选

Close #135
This commit is contained in:
fawney19
2026-02-04 02:04:54 +08:00
parent 6f363a7703
commit f3e2f84b38
59 changed files with 6392 additions and 500 deletions
Download Patch File Download Diff File Expand all files Collapse all files

144
src/services/system/maintenance_scheduler.py
View File

@@ -123,14 +123,24 @@ class MaintenanceScheduler:
scheduler = get_scheduler()
# 注册定时任务（使用业务时区）
# 统计聚合任务 - 凌晨 1 点执行
# 注册定时任务
# 统计聚合任务 - UTC 00:05 执行
scheduler.add_cron_job(
self._scheduled_stats_aggregation,
hour=1,
minute=0,
hour=0,
minute=5,
job_id="stats_aggregation",
name="统计数据聚合",
timezone="UTC",
)
# 小时统计聚合任务 - 每小时 05 分执行（UTC）
scheduler.add_cron_job(
self._scheduled_hourly_stats_aggregation,
hour="*",
minute=5,
job_id="stats_hourly_aggregation",
name="统计小时数据聚合",
timezone="UTC",
)
# 统计聚合补偿任务 - 每 30 分钟检查缺失并回填
scheduler.add_interval_job(
@@ -231,6 +241,10 @@ class MaintenanceScheduler:
"""统计聚合任务（定时调用）"""
await self._perform_stats_aggregation(backfill=backfill)
async def _scheduled_hourly_stats_aggregation(self) -> None:
"""小时统计聚合任务（定时调用）"""
await self._perform_hourly_stats_aggregation()
async def _scheduled_cleanup(self) -> None:
"""清理任务（定时调用）"""
await self._perform_cleanup()
@@ -282,17 +296,12 @@ class MaintenanceScheduler:
logger.info("开始执行统计数据聚合...")
from zoneinfo import ZoneInfo
from src.models.database import StatsDaily
from src.models.database import User as DBUser
from src.services.system.scheduler import APP_TIMEZONE
# 使用业务时区计算日期，确保与定时任务触发时间一致
# 定时任务在 Asia/Shanghai 凌晨 1 点触发，此时应聚合 Asia/Shanghai 的"昨天"
app_tz = ZoneInfo(APP_TIMEZONE)
now_local = datetime.now(app_tz)
today_local = now_local.replace(hour=0, minute=0, second=0, microsecond=0)
# 使用 UTC 日期，定时任务在 UTC 00:05 触发，聚合 UTC 昨天
now_utc = datetime.now(timezone.utc)
today_utc = now_utc.replace(hour=0, minute=0, second=0, microsecond=0)
if backfill:
# 启动时检查并回填缺失的日期
@@ -314,14 +323,15 @@ class MaintenanceScheduler:
# 非首次运行，检查最近是否有缺失的日期需要回填
from src.models.database import StatsDailyModel, StatsDailyProvider
yesterday_business_date = today_local.date() - timedelta(days=1)
yesterday_utc_date = today_utc.date() - timedelta(days=1)
max_backfill_days: int = (
SystemConfigService.get_config(db, "max_stats_backfill_days", 30) or 30
)
# 计算回填检查的起始日期
check_start_date = yesterday_business_date - timedelta(
days=max_backfill_days - 1
check_start_date = yesterday_utc_date - timedelta(days=max_backfill_days - 1)
check_start_dt = datetime.combine(
check_start_date, datetime.min.time(), tzinfo=timezone.utc
)
# 获取 StatsDaily 和 StatsDailyModel 中已有数据的日期集合
@@ -330,41 +340,39 @@ class MaintenanceScheduler:
existing_provider_dates = set()
daily_stats = (
db.query(StatsDaily.date)
.filter(StatsDaily.date >= check_start_date.isoformat())
.all()
db.query(StatsDaily.date).filter(StatsDaily.date >= check_start_dt).all()
)
for (stat_date,) in daily_stats:
if stat_date.tzinfo is None:
stat_date = stat_date.replace(tzinfo=timezone.utc)
existing_daily_dates.add(stat_date.astimezone(app_tz).date())
existing_daily_dates.add(stat_date.date())
model_stats = (
db.query(StatsDailyModel.date)
.filter(StatsDailyModel.date >= check_start_date.isoformat())
.filter(StatsDailyModel.date >= check_start_dt)
.distinct()
.all()
)
for (stat_date,) in model_stats:
if stat_date.tzinfo is None:
stat_date = stat_date.replace(tzinfo=timezone.utc)
existing_model_dates.add(stat_date.astimezone(app_tz).date())
existing_model_dates.add(stat_date.date())
provider_stats = (
db.query(StatsDailyProvider.date)
.filter(StatsDailyProvider.date >= check_start_date.isoformat())
.filter(StatsDailyProvider.date >= check_start_dt)
.distinct()
.all()
)
for (stat_date,) in provider_stats:
if stat_date.tzinfo is None:
stat_date = stat_date.replace(tzinfo=timezone.utc)
existing_provider_dates.add(stat_date.astimezone(app_tz).date())
existing_provider_dates.add(stat_date.date())
# 找出需要回填的日期
all_dates = set()
current = check_start_date
while current <= yesterday_business_date:
while current <= yesterday_utc_date:
all_dates.add(current)
current += timedelta(days=1)
@@ -389,43 +397,17 @@ class MaintenanceScheduler:
)
users = db.query(DBUser.id).filter(DBUser.is_active.is_(True)).all()
user_ids = [user_id for (user_id,) in users]
failed_dates = 0
failed_users = 0
for current_date in sorted_dates:
try:
current_date_local = datetime.combine(
current_date, datetime.min.time(), tzinfo=app_tz
current_date_utc = datetime.combine(
current_date, datetime.min.time(), tzinfo=timezone.utc
)
StatsAggregatorService.aggregate_daily_stats_bundle(
db, current_date_utc, user_ids=user_ids
)
# 只在缺失时才聚合对应的表
if current_date in missing_daily_dates:
StatsAggregatorService.aggregate_daily_stats(
db, current_date_local
)
if current_date in missing_model_dates:
StatsAggregatorService.aggregate_daily_model_stats(
db, current_date_local
)
if current_date in missing_provider_dates:
StatsAggregatorService.aggregate_daily_provider_stats(
db, current_date_local
)
# 用户统计在任一缺失时都回填
for (user_id,) in users:
try:
StatsAggregatorService.aggregate_user_daily_stats(
db, user_id, current_date_local
)
except Exception as e:
failed_users += 1
logger.warning(
f"回填用户 {user_id} 日期 {current_date} 失败: {e}"
)
try:
db.rollback()
except Exception as rollback_err:
logger.error(f"回滚失败: {rollback_err}")
except Exception as e:
failed_dates += 1
logger.warning(f"回填日期 {current_date} 失败: {e}")
@@ -436,10 +418,10 @@ class MaintenanceScheduler:
StatsAggregatorService.update_summary(db)
if failed_dates > 0 or failed_users > 0:
if failed_dates > 0:
logger.warning(
f"回填完成，共处理 {len(dates_to_process)} 天，"
f"失败: {failed_dates} 天, {failed_users} 个用户记录"
f"失败: {failed_dates} 天"
)
else:
logger.info(f"缺失数据回填完成，共处理 {len(dates_to_process)} 天")
@@ -447,25 +429,14 @@ class MaintenanceScheduler:
logger.info("统计数据已是最新，无需回填")
return
# 定时任务：聚合昨天的数据
yesterday_local = today_local - timedelta(days=1)
StatsAggregatorService.aggregate_daily_stats(db, yesterday_local)
StatsAggregatorService.aggregate_daily_model_stats(db, yesterday_local)
StatsAggregatorService.aggregate_daily_provider_stats(db, yesterday_local)
# 定时任务：聚合昨天 (UTC) 的数据
yesterday_utc = today_utc - timedelta(days=1)
users = db.query(DBUser.id).filter(DBUser.is_active.is_(True)).all()
for (user_id,) in users:
try:
StatsAggregatorService.aggregate_user_daily_stats(
db, user_id, yesterday_local
)
except Exception as e:
logger.warning(f"聚合用户 {user_id} 统计数据失败: {e}")
try:
db.rollback()
except Exception:
pass
user_ids = [user_id for (user_id,) in users]
StatsAggregatorService.aggregate_daily_stats_bundle(
db, yesterday_utc, user_ids=user_ids
)
StatsAggregatorService.update_summary(db)
@@ -480,6 +451,27 @@ class MaintenanceScheduler:
finally:
db.close()
async def _perform_hourly_stats_aggregation(self) -> None:
"""执行小时统计聚合任务"""
db = create_session()
try:
if not SystemConfigService.get_config(db, "enable_stats_aggregation", True):
logger.info("统计聚合已禁用，跳过小时聚合任务")
return
now_utc = datetime.now(timezone.utc)
last_hour = now_utc.replace(minute=0, second=0, microsecond=0) - timedelta(hours=1)
StatsAggregatorService.aggregate_hourly_stats_bundle(db, last_hour)
logger.info(f"小时统计聚合完成: {last_hour.isoformat()}")
except Exception as e:
logger.exception(f"小时统计聚合任务执行失败: {e}")
try:
db.rollback()
except Exception:
pass
finally:
db.close()
async def _perform_pending_cleanup(self) -> None:
"""执行 pending 状态清理"""
db = create_session()

8
src/services/system/scheduler.py
View File

@@ -41,10 +41,11 @@ class TaskScheduler:
def add_cron_job(
self,
func: Callable[..., Any],
hour: int,
hour: int | str,
minute: int = 0,
job_id: str | None = None,
name: str | None = None,
timezone: str | None = None,
**kwargs: Any,
) -> Any:
"""
@@ -58,7 +59,8 @@ class TaskScheduler:
name: 任务名称（用于日志）
**kwargs: 传递给任务函数的参数
"""
trigger = CronTrigger(hour=hour, minute=minute, timezone=APP_TIMEZONE)
trigger_timezone = timezone or APP_TIMEZONE
trigger = CronTrigger(hour=hour, minute=minute, timezone=trigger_timezone)
job_id = job_id or func.__name__
display_name = name or job_id
@@ -74,7 +76,7 @@ class TaskScheduler:
logger.info(
f"已注册定时任务: {display_name}, "
f"执行时间: {hour:02d}:{minute:02d} ({APP_TIMEZONE})"
f"执行时间: {hour}:{minute:02d} ({trigger_timezone})"
)
def add_interval_job(

1268
src/services/system/stats_aggregator.py
View File

File diff suppressed because it is too large Load Diff

241
src/services/system/time_range.py Normal file
View File

@@ -0,0 +1,241 @@
"""Time range utilities for stats queries."""
from __future__ import annotations
from datetime import date, datetime, time, timedelta, timezone
from typing import Literal
from pydantic import BaseModel, model_validator
class TimeRangeParams(BaseModel):
"""
Time range parameters (local-date semantics).
Rules:
1) Inputs are user-local dates, backend converts to UTC datetime range.
2) Range is half-open: [start, end).
"""
start_date: date | None = None
end_date: date | None = None
preset: (
Literal[
"today",
"yesterday",
"last7days",
"last30days",
"last90days",
"this_week",
"last_week",
"this_month",
"last_month",
"this_year",
]
| None
) = None
granularity: Literal["hour", "day", "week", "month"] = "day"
timezone: str | None = None
tz_offset_minutes: int = 0
@model_validator(mode="after")
def validate_and_resolve(self) -> "TimeRangeParams":
"""Validate and resolve preset to concrete dates."""
if self.preset:
user_today = self._get_user_today()
match self.preset:
case "today":
self.start_date = self.end_date = user_today
case "yesterday":
self.start_date = self.end_date = user_today - timedelta(days=1)
case "last7days":
self.start_date = user_today - timedelta(days=6)
self.end_date = user_today
case "last30days":
self.start_date = user_today - timedelta(days=29)
self.end_date = user_today
case "last90days":
self.start_date = user_today - timedelta(days=89)
self.end_date = user_today
case "this_week":
self.start_date = user_today - timedelta(days=user_today.weekday())
self.end_date = user_today
case "last_week":
week_start = user_today - timedelta(days=user_today.weekday())
self.start_date = week_start - timedelta(days=7)
self.end_date = week_start - timedelta(days=1)
case "this_month":
self.start_date = user_today.replace(day=1)
self.end_date = user_today
case "last_month":
first_of_this_month = user_today.replace(day=1)
self.end_date = first_of_this_month - timedelta(days=1)
self.start_date = self.end_date.replace(day=1)
case "this_year":
self.start_date = user_today.replace(month=1, day=1)
self.end_date = user_today
if not self.preset and (self.start_date is None or self.end_date is None):
raise ValueError("Either preset or both start_date and end_date must be provided")
if self.start_date and self.end_date and self.start_date > self.end_date:
raise ValueError("start_date must be <= end_date")
if self.start_date and self.end_date:
max_days = 365
days = (self.end_date - self.start_date).days
if days > max_days:
raise ValueError(f"Query range cannot exceed {max_days} days")
if self.granularity == "hour":
if self.start_date != self.end_date:
raise ValueError("Hour granularity only supports single day query")
return self
def validate_for_time_series(self) -> "TimeRangeParams":
"""Extra validation for time series queries."""
if self.granularity == "hour" and self.start_date != self.end_date:
raise ValueError("Hour granularity only supports single day query")
if self.start_date and self.end_date:
days_inclusive = (self.end_date - self.start_date).days + 1
max_days_for_time_series = 90
if days_inclusive > max_days_for_time_series:
raise ValueError(
f"Time series query range cannot exceed {max_days_for_time_series} days "
f"(requested {days_inclusive} days). "
"For longer ranges, use aggregated statistics instead."
)
return self
def _get_user_today(self) -> date:
"""Get user-local 'today'."""
if self.timezone:
try:
from zoneinfo import ZoneInfo
user_tz = ZoneInfo(self.timezone)
return datetime.now(user_tz).date()
except Exception:
pass
user_now = datetime.now(timezone.utc) + timedelta(minutes=self.tz_offset_minutes)
return user_now.date()
def _get_tz_offset_for_date(self, local_date: date) -> timedelta:
"""Get timezone offset for a local date (DST-aware if timezone is provided)."""
if self.timezone:
try:
from zoneinfo import ZoneInfo
user_tz = ZoneInfo(self.timezone)
local_midnight = datetime.combine(local_date, time.min)
local_aware = local_midnight.replace(tzinfo=user_tz)
return local_aware.utcoffset() or timedelta(0)
except Exception:
pass
return timedelta(minutes=self.tz_offset_minutes)
def to_utc_datetime_range(self) -> tuple[datetime, datetime]:
"""Convert to UTC datetime range (half-open)."""
start_offset = self._get_tz_offset_for_date(self.start_date)
end_offset = self._get_tz_offset_for_date(self.end_date + timedelta(days=1))
local_start = datetime.combine(self.start_date, time.min)
local_end = datetime.combine(self.end_date + timedelta(days=1), time.min)
start_utc = (local_start - start_offset).replace(tzinfo=timezone.utc)
end_utc = (local_end - end_offset).replace(tzinfo=timezone.utc)
return start_utc, end_utc
def get_complete_utc_dates(
self,
) -> tuple[list[date], tuple[datetime, datetime] | None, tuple[datetime, datetime] | None]:
"""Split into complete UTC days + head/tail boundaries."""
start_utc, end_utc = self.to_utc_datetime_range()
if (
start_utc.hour == 0
and start_utc.minute == 0
and start_utc.second == 0
and start_utc.microsecond == 0
):
first_complete_date = start_utc.date()
head_boundary = None
else:
first_complete_date = start_utc.date() + timedelta(days=1)
head_boundary = (
start_utc,
datetime.combine(first_complete_date, time.min, tzinfo=timezone.utc),
)
if (
end_utc.hour == 0
and end_utc.minute == 0
and end_utc.second == 0
and end_utc.microsecond == 0
):
last_complete_date = end_utc.date() - timedelta(days=1)
tail_boundary = None
else:
last_complete_date = end_utc.date() - timedelta(days=1)
tail_start = datetime.combine(end_utc.date(), time.min, tzinfo=timezone.utc)
tail_boundary = (tail_start, end_utc)
complete_dates = []
if first_complete_date <= last_complete_date:
current = first_complete_date
while current <= last_complete_date:
complete_dates.append(current)
current += timedelta(days=1)
return complete_dates, head_boundary, tail_boundary
def get_local_day_hours(self) -> list[tuple[date, datetime, datetime]]:
"""Return local-day mapped UTC ranges for time-series."""
result = []
current_date = self.start_date
while current_date <= self.end_date:
offset = self._get_tz_offset_for_date(current_date)
local_start = datetime.combine(current_date, time.min)
local_end = datetime.combine(current_date + timedelta(days=1), time.min)
day_start_utc = (local_start - offset).replace(tzinfo=timezone.utc)
day_end_utc = (
local_end - self._get_tz_offset_for_date(current_date + timedelta(days=1))
).replace(tzinfo=timezone.utc)
result.append((current_date, day_start_utc, day_end_utc))
current_date += timedelta(days=1)
return result
def split_time_range_for_hourly(start_utc: datetime, end_utc: datetime) -> tuple[
tuple[datetime, datetime] | None,
list[datetime],
tuple[datetime, datetime] | None,
]:
"""Split into head fragment, complete hours, tail fragment."""
first_hour = start_utc.replace(minute=0, second=0, microsecond=0)
if start_utc > first_hour:
first_hour += timedelta(hours=1)
head_fragment = (start_utc, first_hour) if first_hour <= end_utc else None
else:
head_fragment = None
first_hour = start_utc
last_hour = end_utc.replace(minute=0, second=0, microsecond=0)
if end_utc > last_hour:
tail_fragment = (last_hour, end_utc) if last_hour >= first_hour else None
else:
tail_fragment = None
last_hour = end_utc
complete_hours = []
current = first_hour
while current < last_hour:
complete_hours.append(current)
current += timedelta(hours=1)
return head_fragment, complete_hours, tail_fragment

70
src/services/usage/error_classifier.py Normal file
View File

@@ -0,0 +1,70 @@
"""Error classification helpers for Usage records."""
from __future__ import annotations
from src.core.enums import ErrorCategory
_STATUS_CODE_MAP: dict[int, ErrorCategory] = {
400: ErrorCategory.INVALID_REQUEST,
401: ErrorCategory.AUTH,
403: ErrorCategory.AUTH,
404: ErrorCategory.NOT_FOUND,
408: ErrorCategory.TIMEOUT,
429: ErrorCategory.RATE_LIMIT,
500: ErrorCategory.SERVER_ERROR,
502: ErrorCategory.SERVER_ERROR,
503: ErrorCategory.SERVER_ERROR,
504: ErrorCategory.TIMEOUT,
}
_CONTEXT_LENGTH_PATTERNS = (
"context_length_exceeded",
"maximum context length",
"too many tokens",
"input is too long",
)
_CONTENT_FILTER_PATTERNS = (
"content_filter",
"content_policy",
"safety_block",
"blocked by content",
)
_NETWORK_PATTERNS = ("connection", "network", "dns", "socket")
def classify_error(
status_code: int | None,
error_message: str | None,
status: str | None = None,
) -> ErrorCategory:
"""Map provider errors to ErrorCategory."""
if status and status.lower() == "cancelled":
return ErrorCategory.CANCELLED
if status_code is not None:
mapped = _STATUS_CODE_MAP.get(status_code)
if mapped:
return mapped
if error_message:
msg_lower = error_message.lower()
if any(p in msg_lower for p in _CONTEXT_LENGTH_PATTERNS):
return ErrorCategory.CONTEXT_LENGTH
if any(p in msg_lower for p in _CONTENT_FILTER_PATTERNS):
return ErrorCategory.CONTENT_FILTER
if "rate limit" in msg_lower or "rate_limit" in msg_lower:
return ErrorCategory.RATE_LIMIT
if "timeout" in msg_lower or "timed out" in msg_lower:
return ErrorCategory.TIMEOUT
if any(p in msg_lower for p in _NETWORK_PATTERNS):
return ErrorCategory.NETWORK
if status_code is not None:
if status_code >= 500:
return ErrorCategory.SERVER_ERROR
if status_code >= 400:
return ErrorCategory.INVALID_REQUEST
return ErrorCategory.UNKNOWN

11
src/services/usage/service.py
View File

@@ -27,6 +27,7 @@ from src.models.database import (
)
from src.services.model.cost import ModelCostService
from src.services.system.config import SystemConfigService
from src.services.usage.error_classifier import classify_error
@dataclass
@@ -335,6 +336,10 @@ class UsageService:
actual_request_cost = request_cost * actual_rate_multiplier
actual_total_cost = total_cost * actual_rate_multiplier
error_category = None
if status_code >= 400 or error_message or status in {"failed", "cancelled"}:
error_category = classify_error(status_code, error_message, status).value
return {
"user_id": user.id if user else None,
"api_key_id": api_key.id if api_key else None,
@@ -376,6 +381,7 @@ class UsageService:
"is_stream": is_stream,
"status_code": status_code,
"error_message": error_message,
"error_category": error_category,
"response_time_ms": response_time_ms,
"first_byte_time_ms": first_byte_time_ms,
"status": status,
@@ -586,6 +592,7 @@ class UsageService:
existing_usage.status = usage_params["status"]
existing_usage.status_code = usage_params["status_code"]
existing_usage.error_message = usage_params["error_message"]
existing_usage.error_category = usage_params.get("error_category")
existing_usage.response_time_ms = usage_params["response_time_ms"]
existing_usage.first_byte_time_ms = usage_params["first_byte_time_ms"]
@@ -2068,7 +2075,7 @@ class UsageService:
if start_date:
query = query.filter(Usage.created_at >= start_date)
if end_date:
query = query.filter(Usage.created_at <= end_date)
query = query.filter(Usage.created_at < end_date)
# 使用跨数据库兼容的日期函数
from src.utils.database_helpers import date_trunc_portable
@@ -2108,7 +2115,7 @@ class UsageService:
if start_date:
summary = summary.filter(Usage.created_at >= start_date)
if end_date:
summary = summary.filter(Usage.created_at <= end_date)
summary = summary.filter(Usage.created_at < end_date)
summary = summary.group_by(date_func, Usage.provider_name, Usage.model).all()