|
|
""" |
|
|
Logging Service Module |
|
|
Centralized logging with file persistence. |
|
|
""" |
|
|
import json |
|
|
import threading |
|
|
import uuid |
|
|
from pathlib import Path |
|
|
from datetime import datetime |
|
|
from typing import List, Dict, Optional |
|
|
from dataclasses import dataclass, asdict |
|
|
from enum import Enum |
|
|
|
|
|
from .config import LOGS_DIR |
|
|
|
|
|
|
|
|
class LogLevel(Enum): |
|
|
DEBUG = "DEBUG" |
|
|
INFO = "INFO" |
|
|
WARNING = "WARNING" |
|
|
ERROR = "ERROR" |
|
|
EVENT = "EVENT" |
|
|
|
|
|
|
|
|
@dataclass |
|
|
class LogEntry: |
|
|
timestamp: str |
|
|
level: str |
|
|
category: str |
|
|
message: str |
|
|
details: Optional[Dict] = None |
|
|
log_id: str = "" |
|
|
|
|
|
def __post_init__(self): |
|
|
if not self.log_id: |
|
|
self.log_id = str(uuid.uuid4())[:8] |
|
|
|
|
|
|
|
|
class LoggingService: |
|
|
"""Centralized logging service with file persistence""" |
|
|
|
|
|
def __init__(self, log_dir: Path): |
|
|
self.log_dir = log_dir |
|
|
self.log_file = log_dir / "app.log" |
|
|
self.event_log = log_dir / "events.log" |
|
|
self.error_log = log_dir / "errors.log" |
|
|
self.max_entries = 1000 |
|
|
self._lock = threading.Lock() |
|
|
|
|
|
def _write_log(self, entry: LogEntry, file_path: Path): |
|
|
with self._lock: |
|
|
try: |
|
|
logs = [] |
|
|
if file_path.exists(): |
|
|
try: |
|
|
logs = json.loads(file_path.read_text()) |
|
|
except: |
|
|
logs = [] |
|
|
|
|
|
logs.append(asdict(entry)) |
|
|
|
|
|
if len(logs) > self.max_entries: |
|
|
logs = logs[-self.max_entries:] |
|
|
|
|
|
file_path.write_text(json.dumps(logs, indent=2)) |
|
|
except Exception as e: |
|
|
print(f"Log write error: {e}") |
|
|
|
|
|
def log(self, level: LogLevel, category: str, message: str, details: Dict = None): |
|
|
entry = LogEntry( |
|
|
timestamp=datetime.now().isoformat(), |
|
|
level=level.value, |
|
|
category=category, |
|
|
message=message, |
|
|
details=details |
|
|
) |
|
|
|
|
|
self._write_log(entry, self.log_file) |
|
|
|
|
|
if level == LogLevel.ERROR: |
|
|
self._write_log(entry, self.error_log) |
|
|
elif level == LogLevel.EVENT: |
|
|
self._write_log(entry, self.event_log) |
|
|
|
|
|
return entry.log_id |
|
|
|
|
|
def info(self, category: str, message: str, details: Dict = None): |
|
|
return self.log(LogLevel.INFO, category, message, details) |
|
|
|
|
|
def error(self, category: str, message: str, details: Dict = None): |
|
|
return self.log(LogLevel.ERROR, category, message, details) |
|
|
|
|
|
def event(self, category: str, message: str, details: Dict = None): |
|
|
return self.log(LogLevel.EVENT, category, message, details) |
|
|
|
|
|
def warning(self, category: str, message: str, details: Dict = None): |
|
|
return self.log(LogLevel.WARNING, category, message, details) |
|
|
|
|
|
def get_logs(self, log_type: str = "all", limit: int = 100, level: str = None, category: str = None) -> List[Dict]: |
|
|
file_map = { |
|
|
"all": self.log_file, |
|
|
"events": self.event_log, |
|
|
"errors": self.error_log |
|
|
} |
|
|
|
|
|
file_path = file_map.get(log_type, self.log_file) |
|
|
|
|
|
if not file_path.exists(): |
|
|
return [] |
|
|
|
|
|
try: |
|
|
logs = json.loads(file_path.read_text()) |
|
|
|
|
|
if level: |
|
|
logs = [l for l in logs if l.get("level") == level] |
|
|
if category: |
|
|
logs = [l for l in logs if l.get("category") == category] |
|
|
|
|
|
return logs[-limit:][::-1] |
|
|
except: |
|
|
return [] |
|
|
|
|
|
def clear_logs(self, log_type: str = "all") -> bool: |
|
|
try: |
|
|
if log_type == "all": |
|
|
for f in [self.log_file, self.event_log, self.error_log]: |
|
|
if f.exists(): |
|
|
f.write_text("[]") |
|
|
else: |
|
|
file_map = {"events": self.event_log, "errors": self.error_log, "app": self.log_file} |
|
|
if log_type in file_map and file_map[log_type].exists(): |
|
|
file_map[log_type].write_text("[]") |
|
|
return True |
|
|
except: |
|
|
return False |
|
|
|
|
|
def export_logs(self, log_type: str = "all") -> str: |
|
|
logs = self.get_logs(log_type, limit=10000) |
|
|
return json.dumps(logs, indent=2) |
|
|
|
|
|
|
|
|
|
|
|
logger = LoggingService(LOGS_DIR) |
|
|
|
