|
|
""" |
|
|
Centralized Logger |
|
|
""" |
|
|
import json |
|
|
from datetime import datetime |
|
|
from pathlib import Path |
|
|
from typing import Optional, Dict, Any, List |
|
|
from core.config import LOGS_DIR |
|
|
|
|
|
class Logger: |
|
|
"""Thread-safe logger with file persistence""" |
|
|
|
|
|
def __init__(self, max_logs: int = 1000): |
|
|
self.log_file = LOGS_DIR / "app.log" |
|
|
self.max_logs = max_logs |
|
|
self._logs: List[Dict] = [] |
|
|
self._load_logs() |
|
|
|
|
|
def _load_logs(self): |
|
|
"""Load existing logs from file""" |
|
|
if self.log_file.exists(): |
|
|
try: |
|
|
self._logs = json.loads(self.log_file.read_text())[-self.max_logs:] |
|
|
except: |
|
|
self._logs = [] |
|
|
|
|
|
def _save_logs(self): |
|
|
"""Save logs to file""" |
|
|
try: |
|
|
self.log_file.write_text(json.dumps(self._logs[-self.max_logs:], indent=2)) |
|
|
except: |
|
|
pass |
|
|
|
|
|
def _log(self, level: str, module: str, message: str, data: Optional[Dict] = None): |
|
|
"""Internal log method""" |
|
|
entry = { |
|
|
"timestamp": datetime.now().isoformat(), |
|
|
"level": level, |
|
|
"module": module, |
|
|
"message": message, |
|
|
"data": data |
|
|
} |
|
|
self._logs.append(entry) |
|
|
|
|
|
|
|
|
icons = {"INFO": "ℹ️", "WARN": "⚠️", "ERROR": "❌", "EVENT": "📌"} |
|
|
icon = icons.get(level, "•") |
|
|
timestamp = datetime.now().strftime("%H:%M:%S") |
|
|
print(f"[{timestamp}] {icon} [{module}] {message}") |
|
|
|
|
|
|
|
|
if len(self._logs) % 10 == 0: |
|
|
self._save_logs() |
|
|
|
|
|
def info(self, module: str, message: str, data: Optional[Dict] = None): |
|
|
self._log("INFO", module, message, data) |
|
|
|
|
|
def warn(self, module: str, message: str, data: Optional[Dict] = None): |
|
|
self._log("WARN", module, message, data) |
|
|
|
|
|
def error(self, module: str, message: str, data: Optional[Dict] = None): |
|
|
self._log("ERROR", module, message, data) |
|
|
|
|
|
def event(self, module: str, message: str, data: Optional[Dict] = None): |
|
|
self._log("EVENT", module, message, data) |
|
|
|
|
|
def get_logs(self, level: Optional[str] = None, limit: int = 50) -> List[Dict]: |
|
|
"""Get recent logs, optionally filtered by level""" |
|
|
logs = self._logs |
|
|
if level: |
|
|
logs = [l for l in logs if l["level"] == level] |
|
|
return logs[-limit:] |
|
|
|
|
|
def clear(self, level: Optional[str] = None): |
|
|
"""Clear logs""" |
|
|
if level: |
|
|
self._logs = [l for l in self._logs if l["level"] != level] |
|
|
else: |
|
|
self._logs = [] |
|
|
self._save_logs() |
|
|
|
|
|
|
|
|
|
|
|
logger = Logger() |
|
|
|
