| """请求日志审计 - 记录近期请求""" |
|
|
| import time |
| import asyncio |
| import orjson |
| from typing import List, Dict, Deque |
| from collections import deque |
| from dataclasses import dataclass, asdict |
| from pathlib import Path |
|
|
| from app.core.logger import logger |
|
|
| @dataclass |
| class RequestLog: |
| id: str |
| time: str |
| timestamp: float |
| ip: str |
| model: str |
| duration: float |
| status: int |
| key_name: str |
| token_suffix: str |
| error: str = "" |
|
|
| class RequestLogger: |
| """请求日志记录器""" |
| |
| _instance = None |
| |
| def __new__(cls): |
| if cls._instance is None: |
| cls._instance = super().__new__(cls) |
| return cls._instance |
| |
| def __init__(self, max_len: int = 1000): |
| if hasattr(self, '_initialized'): |
| return |
| |
| self.file_path = Path(__file__).parents[2] / "data" / "logs.json" |
| self._logs: Deque[Dict] = deque(maxlen=max_len) |
| self._lock = asyncio.Lock() |
| self._loaded = False |
| |
| self._initialized = True |
|
|
| async def init(self): |
| """初始化加载数据""" |
| if not self._loaded: |
| await self._load_data() |
|
|
| async def _load_data(self): |
| """从磁盘加载日志数据""" |
| if self._loaded: |
| return |
|
|
| if not self.file_path.exists(): |
| self._loaded = True |
| return |
|
|
| try: |
| async with self._lock: |
| content = await asyncio.to_thread(self.file_path.read_bytes) |
| if content: |
| data = orjson.loads(content) |
| if isinstance(data, list): |
| self._logs.clear() |
| self._logs.extend(data) |
| self._loaded = True |
| logger.debug(f"[Logger] 加载日志成功: {len(self._logs)} 条") |
| except Exception as e: |
| logger.error(f"[Logger] 加载日志失败: {e}") |
| self._loaded = True |
|
|
| async def _save_data(self): |
| """保存日志数据到磁盘""" |
| if not self._loaded: |
| return |
|
|
| try: |
| |
| self.file_path.parent.mkdir(parents=True, exist_ok=True) |
| |
| async with self._lock: |
| |
| content = orjson.dumps(list(self._logs)) |
| await asyncio.to_thread(self.file_path.write_bytes, content) |
| except Exception as e: |
| logger.error(f"[Logger] 保存日志失败: {e}") |
|
|
| async def add_log(self, |
| ip: str, |
| model: str, |
| duration: float, |
| status: int, |
| key_name: str, |
| token_suffix: str = "", |
| error: str = ""): |
| """添加日志""" |
| if not self._loaded: |
| await self.init() |
| |
| try: |
| now = time.time() |
| |
| time_str = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(now)) |
| |
| log = { |
| "id": str(int(now * 1000)), |
| "time": time_str, |
| "timestamp": now, |
| "ip": ip, |
| "model": model, |
| "duration": round(duration, 2), |
| "status": status, |
| "key_name": key_name, |
| "token_suffix": token_suffix, |
| "error": error |
| } |
| |
| async with self._lock: |
| self._logs.appendleft(log) |
| |
| |
| asyncio.create_task(self._save_data()) |
| |
| except Exception as e: |
| logger.error(f"[Logger] 记录日志失败: {e}") |
|
|
| async def get_logs(self, limit: int = 1000) -> List[Dict]: |
| """获取日志""" |
| async with self._lock: |
| return list(self._logs)[:limit] |
| |
| async def clear_logs(self): |
| """清空日志""" |
| async with self._lock: |
| self._logs.clear() |
| await self._save_data() |
|
|
|
|
| |
| request_logger = RequestLogger() |
|
|
