"""用量统计 + 审计日志 API(admin):/admin/api/usage、/admin/api/audit。""" from __future__ import annotations from typing import Optional from fastapi import APIRouter, Depends, Query from ..services import usage_store from .admin import _require_admin router = APIRouter(prefix="/admin/api", tags=["usage-audit"]) @router.get("/usage") async def usage( hours: int = Query(default=24, ge=1, le=720), _admin: str = Depends(_require_admin), ): summary = usage_store.usage_summary(hours=hours) return _ok(summary) @router.get("/usage/timeline") async def usage_timeline_api( hours: int = Query(default=24, ge=1, le=720), granularity: str = Query(default="hour", pattern="^(hour|day)$"), _admin: str = Depends(_require_admin), ): timeline = usage_store.usage_timeline(hours=hours, granularity=granularity) return _ok({"items": timeline, "granularity": granularity}) @router.get("/audit") async def audit( hours: int = Query(default=24, ge=1, le=720), limit: int = Query(default=100, ge=1, le=1000), offset: int = Query(default=0, ge=0), action: Optional[str] = Query(default=None), _admin: str = Depends(_require_admin), ): items = usage_store.list_audit(hours=hours, limit=limit, offset=offset, action=action) return _ok({"items": items, "count": len(items)}) def _ok(payload: dict): from ._common import ok_with_cors return ok_with_cors(payload)