78 lines
2.3 KiB
Python
78 lines
2.3 KiB
Python
|
||
"""log_manager.py
|
||
|
||
简单的线程安全日志管理器：
|
||
- 追加日志（level/message/thread_id）
|
||
- 持久化到 data/logs.json
|
||
- 提供读取最近 N 条、清空
|
||
|
||
说明：采集器是异步运行，但日志写入使用 threading.Lock 足够。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import threading
|
||
from datetime import datetime
|
||
from typing import List, Dict, Optional
|
||
|
||
from app import config
|
||
|
||
|
||
class LogManager:
|
||
def __init__(self, log_file: Optional[str] = None, max_logs: int = 2000):
|
||
self.log_file = log_file or os.path.join(config.DATA_DIR, "logs.json")
|
||
self.max_logs = max_logs
|
||
self._lock = threading.Lock()
|
||
self.logs: List[Dict] = []
|
||
self._load()
|
||
|
||
def _load(self) -> None:
|
||
if not os.path.exists(self.log_file):
|
||
self.logs = []
|
||
return
|
||
try:
|
||
with open(self.log_file, "r", encoding="utf-8") as f:
|
||
data = json.load(f) or {}
|
||
self.logs = data.get("logs", []) or []
|
||
except Exception:
|
||
self.logs = []
|
||
|
||
def _save_locked(self) -> None:
|
||
os.makedirs(os.path.dirname(self.log_file), exist_ok=True)
|
||
tmp = self.log_file + ".tmp"
|
||
with open(tmp, "w", encoding="utf-8") as f:
|
||
json.dump(
|
||
{"logs": self.logs, "last_updated": datetime.now().isoformat(timespec="seconds")},
|
||
f,
|
||
ensure_ascii=False,
|
||
indent=2,
|
||
)
|
||
os.replace(tmp, self.log_file)
|
||
|
||
def append(self, level: str, message: str, thread_id: Optional[str] = None) -> Dict:
|
||
entry = {
|
||
"time": datetime.now().isoformat(timespec="seconds"),
|
||
"level": level.upper(),
|
||
"message": message,
|
||
}
|
||
if thread_id:
|
||
entry["thread_id"] = thread_id
|
||
|
||
with self._lock:
|
||
self.logs.append(entry)
|
||
if len(self.logs) > self.max_logs:
|
||
self.logs = self.logs[-self.max_logs :]
|
||
self._save_locked()
|
||
return entry
|
||
|
||
def clear(self) -> None:
|
||
with self._lock:
|
||
self.logs = []
|
||
self._save_locked()
|
||
|
||
def get_latest(self, limit: int = 200) -> List[Dict]:
|
||
with self._lock:
|
||
return list(self.logs[-max(1, int(limit)) :])
|