xhs_crawler_claude/app/log_manager.py

"""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)) :])