274 lines
8.0 KiB
Python
274 lines
8.0 KiB
Python
# Description: Structured logging system for EU-Utility
|
|
# Centralized logging with levels, rotation, and formatting
|
|
|
|
"""
|
|
EU-Utility Logging System
|
|
|
|
Structured logging with:
|
|
- Multiple log levels (DEBUG, INFO, WARNING, ERROR, CRITICAL)
|
|
- Log rotation by size and time
|
|
- Formatted output
|
|
- File and console handlers
|
|
- Context tracking
|
|
"""
|
|
|
|
import logging
|
|
import logging.handlers
|
|
import os
|
|
import sys
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
|
|
class Logger:
|
|
"""
|
|
Centralized logging system for EU-Utility.
|
|
|
|
Usage:
|
|
from core.logger import get_logger
|
|
|
|
logger = get_logger("plugin_name")
|
|
logger.info("Application started")
|
|
logger.error("Something went wrong", exc_info=True)
|
|
"""
|
|
|
|
_instance = None
|
|
_initialized = False
|
|
|
|
def __new__(cls):
|
|
if cls._instance is None:
|
|
cls._instance = super().__new__(cls)
|
|
return cls._instance
|
|
|
|
def __init__(self):
|
|
if Logger._initialized:
|
|
return
|
|
|
|
Logger._initialized = True
|
|
|
|
# Setup log directory
|
|
self.log_dir = Path.home() / '.eu-utility' / 'logs'
|
|
self.log_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Create main logger
|
|
self.logger = logging.getLogger('eu_utility')
|
|
self.logger.setLevel(logging.DEBUG)
|
|
|
|
# Prevent duplicate handlers
|
|
if self.logger.handlers:
|
|
return
|
|
|
|
# File handler with rotation (10MB per file, keep 5 backups)
|
|
file_handler = logging.handlers.RotatingFileHandler(
|
|
self.log_dir / 'eu_utility.log',
|
|
maxBytes=10*1024*1024, # 10MB
|
|
backupCount=5,
|
|
encoding='utf-8'
|
|
)
|
|
file_handler.setLevel(logging.DEBUG)
|
|
|
|
# Console handler
|
|
console_handler = logging.StreamHandler(sys.stdout)
|
|
console_handler.setLevel(logging.INFO)
|
|
|
|
# Formatters
|
|
file_formatter = logging.Formatter(
|
|
'%(asctime)s - %(name)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(message)s',
|
|
datefmt='%Y-%m-%d %H:%M:%S'
|
|
)
|
|
|
|
console_formatter = logging.Formatter(
|
|
'[%(levelname)s] %(message)s'
|
|
)
|
|
|
|
file_handler.setFormatter(file_formatter)
|
|
console_handler.setFormatter(console_formatter)
|
|
|
|
self.logger.addHandler(file_handler)
|
|
self.logger.addHandler(console_handler)
|
|
|
|
# Plugin loggers cache
|
|
self._plugin_loggers = {}
|
|
|
|
def get_logger(self, name):
|
|
"""Get a logger for a specific component."""
|
|
if name not in self._plugin_loggers:
|
|
logger = logging.getLogger(f'eu_utility.{name}')
|
|
self._plugin_loggers[name] = logger
|
|
return self._plugin_loggers[name]
|
|
|
|
def set_level(self, level):
|
|
"""Set the global log level."""
|
|
self.logger.setLevel(level)
|
|
|
|
def get_log_path(self):
|
|
"""Get the path to the current log file."""
|
|
return str(self.log_dir / 'eu_utility.log')
|
|
|
|
def get_recent_logs(self, lines=100):
|
|
"""Get recent log entries."""
|
|
try:
|
|
log_file = self.log_dir / 'eu_utility.log'
|
|
if not log_file.exists():
|
|
return []
|
|
|
|
with open(log_file, 'r', encoding='utf-8') as f:
|
|
all_lines = f.readlines()
|
|
return all_lines[-lines:]
|
|
except Exception:
|
|
return []
|
|
|
|
|
|
# Global instance
|
|
_logger = None
|
|
|
|
def get_logger(name=None):
|
|
"""
|
|
Get a logger instance.
|
|
|
|
Args:
|
|
name: Component name (e.g., 'loot_tracker', 'core.event_bus')
|
|
|
|
Returns:
|
|
logging.Logger instance
|
|
"""
|
|
global _logger
|
|
if _logger is None:
|
|
_logger = Logger()
|
|
|
|
if name:
|
|
return _logger.get_logger(name)
|
|
return _logger.logger
|
|
|
|
|
|
# Convenience functions
|
|
def debug(msg, *args, **kwargs):
|
|
"""Log debug message."""
|
|
get_logger().debug(msg, *args, **kwargs)
|
|
|
|
def info(msg, *args, **kwargs):
|
|
"""Log info message."""
|
|
get_logger().info(msg, *args, **kwargs)
|
|
|
|
def warning(msg, *args, **kwargs):
|
|
"""Log warning message."""
|
|
get_logger().warning(msg, *args, **kwargs)
|
|
|
|
def error(msg, *args, **kwargs):
|
|
"""Log error message."""
|
|
get_logger().error(msg, *args, **kwargs)
|
|
|
|
def critical(msg, *args, **kwargs):
|
|
"""Log critical message."""
|
|
get_logger().critical(msg, *args, **kwargs)
|
|
|
|
|
|
class LogViewer:
|
|
"""Utility class for viewing and filtering logs."""
|
|
|
|
def __init__(self):
|
|
self.log_dir = Path.home() / '.eu-utility' / 'logs'
|
|
|
|
def get_logs(self, level=None, component=None, since=None, until=None):
|
|
"""
|
|
Get filtered logs.
|
|
|
|
Args:
|
|
level: Filter by level (DEBUG, INFO, WARNING, ERROR, CRITICAL)
|
|
component: Filter by component name
|
|
since: Filter by start datetime
|
|
until: Filter by end datetime
|
|
|
|
Returns:
|
|
List of log entries
|
|
"""
|
|
log_file = self.log_dir / 'eu_utility.log'
|
|
if not log_file.exists():
|
|
return []
|
|
|
|
entries = []
|
|
|
|
try:
|
|
with open(log_file, 'r', encoding='utf-8') as f:
|
|
for line in f:
|
|
# Parse log line
|
|
# Format: 2025-01-15 10:30:45 - name - LEVEL - [file:line] - message
|
|
try:
|
|
parts = line.split(' - ', 4)
|
|
if len(parts) < 4:
|
|
continue
|
|
|
|
timestamp_str = parts[0]
|
|
name = parts[1]
|
|
log_level = parts[2]
|
|
message = parts[-1]
|
|
|
|
# Parse timestamp
|
|
timestamp = datetime.strptime(timestamp_str, '%Y-%m-%d %H:%M:%S')
|
|
|
|
# Apply filters
|
|
if level and log_level != level:
|
|
continue
|
|
|
|
if component and component not in name:
|
|
continue
|
|
|
|
if since and timestamp < since:
|
|
continue
|
|
|
|
if until and timestamp > until:
|
|
continue
|
|
|
|
entries.append({
|
|
'timestamp': timestamp,
|
|
'component': name,
|
|
'level': log_level,
|
|
'message': message.strip()
|
|
})
|
|
|
|
except Exception:
|
|
continue
|
|
|
|
except Exception as e:
|
|
print(f"Error reading logs: {e}")
|
|
|
|
return entries
|
|
|
|
def tail(self, lines=50):
|
|
"""Get last N log lines."""
|
|
return get_logger().get_recent_logs(lines)
|
|
|
|
def export_logs(self, output_path, **filters):
|
|
"""Export filtered logs to file."""
|
|
entries = self.get_logs(**filters)
|
|
|
|
with open(output_path, 'w', encoding='utf-8') as f:
|
|
for entry in entries:
|
|
f.write(f"{entry['timestamp']} - {entry['component']} - {entry['level']} - {entry['message']}\n")
|
|
|
|
|
|
# Plugin integration helper
|
|
class PluginLogger:
|
|
"""Helper class for plugins to use logging."""
|
|
|
|
def __init__(self, plugin_name):
|
|
self.logger = get_logger(f"plugins.{plugin_name}")
|
|
|
|
def debug(self, msg, **kwargs):
|
|
self.logger.debug(msg, **kwargs)
|
|
|
|
def info(self, msg, **kwargs):
|
|
self.logger.info(msg, **kwargs)
|
|
|
|
def warning(self, msg, **kwargs):
|
|
self.logger.warning(msg, **kwargs)
|
|
|
|
def error(self, msg, **kwargs):
|
|
self.logger.error(msg, **kwargs)
|
|
|
|
def critical(self, msg, **kwargs):
|
|
self.logger.critical(msg, **kwargs)
|
|
|
|
def exception(self, msg, **kwargs):
|
|
self.logger.exception(msg, **kwargs)
|