""" Centralized Logging System for Miku Discord Bot This module provides a robust, component-based logging system with: - Configurable log levels per component - Emoji-based log formatting - Multiple output handlers (console, separate log files per component) - Runtime configuration updates - API request filtering - Docker-compatible output Usage: from utils.logger import get_logger logger = get_logger('bot') logger.info("Bot started successfully") logger.error("Failed to connect", exc_info=True) """ import logging import sys import os from pathlib import Path from typing import Optional, Dict from logging.handlers import RotatingFileHandler import json # Log level emojis LEVEL_EMOJIS = { 'DEBUG': '🔍', 'INFO': 'ℹī¸', 'WARNING': '⚠ī¸', 'ERROR': '❌', 'CRITICAL': 'đŸ”Ĩ', 'API': '🌐', } # Custom API log level (between INFO and WARNING) API_LEVEL = 25 logging.addLevelName(API_LEVEL, 'API') # Component definitions COMPONENTS = { 'bot': 'Main bot lifecycle and events', 'api': 'FastAPI endpoints (non-HTTP)', 'api.requests': 'HTTP request/response logs', 'autonomous': 'Autonomous messaging system', 'persona': 'Bipolar/persona dialogue system', 'vision': 'Image and video processing', 'llm': 'LLM API calls and interactions', 'conversation': 'Conversation history management', 'mood': 'Mood system and state changes', 'dm': 'Direct message handling', 'scheduled': 'Scheduled tasks and cron jobs', 'gpu': 'GPU routing and model management', 'media': 'Media processing (audio, video, images)', 'server': 'Server management and configuration', 'commands': 'Command handling and routing', 'sentiment': 'Sentiment analysis', 'core': 'Core utilities and helpers', 'apscheduler': 'Job scheduler logs (APScheduler)', 'voice_manager': 'Voice channel session management', 'voice_commands': 'Voice channel commands', 'voice_audio': 'Voice audio streaming and TTS', 'error_handler': 'Error detection and webhook notifications', } # Global configuration _log_config: Optional[Dict] = None _loggers: Dict[str, logging.Logger] = {} _handlers_initialized = False # Log directory (in mounted volume so logs persist) LOG_DIR = Path(os.getenv('LOG_DIR', '/app/memory/logs')) class EmojiFormatter(logging.Formatter): """Custom formatter that adds emojis and colors to log messages.""" def __init__(self, use_emojis=True, use_colors=False, timestamp_format='datetime', *args, **kwargs): super().__init__(*args, **kwargs) self.use_emojis = use_emojis self.use_colors = use_colors self.timestamp_format = timestamp_format def format(self, record): # Add emoji prefix if self.use_emojis: emoji = LEVEL_EMOJIS.get(record.levelname, '') record.levelname_emoji = f"{emoji} {record.levelname}" else: record.levelname_emoji = record.levelname # Format timestamp based on settings if self.timestamp_format == 'off': record.timestamp_formatted = '' elif self.timestamp_format == 'time': record.timestamp_formatted = self.formatTime(record, '%H:%M:%S') + ' ' elif self.timestamp_format == 'date': record.timestamp_formatted = self.formatTime(record, '%Y-%m-%d') + ' ' elif self.timestamp_format == 'datetime': record.timestamp_formatted = self.formatTime(record, '%Y-%m-%d %H:%M:%S') + ' ' else: # Default to datetime if invalid option record.timestamp_formatted = self.formatTime(record, '%Y-%m-%d %H:%M:%S') + ' ' # Format the message return super().format(record) class ComponentFilter(logging.Filter): """Filter logs based on component configuration with individual level toggles.""" def __init__(self, component_name: str): super().__init__() self.component_name = component_name def filter(self, record): """Check if this log should be output based on enabled levels.""" config = get_log_config() if not config: return True component_config = config.get('components', {}).get(self.component_name, {}) # Check if component is enabled if not component_config.get('enabled', True): return False # Check if specific log level is enabled enabled_levels = component_config.get('enabled_levels', ['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL', 'API']) # Get the level name for this record level_name = logging.getLevelName(record.levelno) return level_name in enabled_levels def get_log_config() -> Optional[Dict]: """Get current log configuration.""" global _log_config if _log_config is None: # Try to load from file config_path = Path('/app/memory/log_settings.json') if config_path.exists(): try: with open(config_path, 'r') as f: _log_config = json.load(f) except Exception: _log_config = get_default_config() else: _log_config = get_default_config() return _log_config def get_default_config() -> Dict: """Get default logging configuration.""" # Read from environment variables # Enable api.requests by default (now that uvicorn access logs are disabled) enable_api_requests = os.getenv('LOG_ENABLE_API_REQUESTS', 'true').lower() == 'true' use_emojis = os.getenv('LOG_USE_EMOJIS', 'true').lower() == 'true' config = { 'version': '1.0', 'formatting': { 'use_emojis': use_emojis, 'use_colors': False, 'timestamp_format': 'datetime' # Options: 'off', 'time', 'date', 'datetime' }, 'components': {} } # Set defaults for each component for component in COMPONENTS.keys(): if component == 'api.requests': # API requests component defaults to only ERROR and CRITICAL default_levels = ['ERROR', 'CRITICAL'] if not enable_api_requests else ['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL', 'API'] config['components'][component] = { 'enabled': enable_api_requests, 'enabled_levels': default_levels, 'filters': { 'exclude_paths': ['/health', '/static/*'], 'exclude_status': [200, 304] if not enable_api_requests else [], 'include_slow_requests': True, 'slow_threshold_ms': 1000 } } elif component == 'apscheduler': # APScheduler defaults to WARNING and above (lots of INFO noise) config['components'][component] = { 'enabled': True, 'enabled_levels': ['WARNING', 'ERROR', 'CRITICAL'] } else: # All other components default to all levels enabled config['components'][component] = { 'enabled': True, 'enabled_levels': ['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] } return config def reload_config(): """Reload configuration from file.""" global _log_config _log_config = None get_log_config() # Update all existing loggers for component_name, logger in _loggers.items(): _configure_logger(logger, component_name) def save_config(config: Dict): """Save configuration to file.""" global _log_config _log_config = config config_path = Path('/app/memory/log_settings.json') config_path.parent.mkdir(parents=True, exist_ok=True) with open(config_path, 'w') as f: json.dump(config, f, indent=2) # Reload all loggers reload_config() def _setup_handlers(): """Set up log handlers (console and file).""" global _handlers_initialized if _handlers_initialized: return # Create log directory LOG_DIR.mkdir(parents=True, exist_ok=True) _handlers_initialized = True def _configure_logger(logger: logging.Logger, component_name: str): """Configure a logger with handlers and filters.""" config = get_log_config() formatting = config.get('formatting', {}) # Clear existing handlers logger.handlers.clear() # Set logger level to DEBUG so handlers can filter logger.setLevel(logging.DEBUG) logger.propagate = False # Create formatter timestamp_format = formatting.get('timestamp_format', 'datetime') # 'off', 'time', 'date', or 'datetime' use_emojis = formatting.get('use_emojis', True) use_colors = formatting.get('use_colors', False) # Console handler - goes to Docker logs console_handler = logging.StreamHandler(sys.stdout) console_formatter = EmojiFormatter( fmt='%(timestamp_formatted)s[%(levelname_emoji)s] [%(name)s] %(message)s', use_emojis=use_emojis, use_colors=use_colors, timestamp_format=timestamp_format ) console_handler.setFormatter(console_formatter) console_handler.addFilter(ComponentFilter(component_name)) logger.addHandler(console_handler) # File handler - separate file per component log_file = LOG_DIR / f'{component_name.replace(".", "_")}.log' file_handler = RotatingFileHandler( log_file, maxBytes=10 * 1024 * 1024, # 10MB backupCount=5, encoding='utf-8' ) file_formatter = EmojiFormatter( fmt='%(timestamp_formatted)s[%(levelname)s] [%(name)s] %(message)s', use_emojis=False, # No emojis in file logs use_colors=False, timestamp_format=timestamp_format ) file_handler.setFormatter(file_formatter) file_handler.addFilter(ComponentFilter(component_name)) logger.addHandler(file_handler) def get_logger(component: str) -> logging.Logger: """ Get a logger for a specific component. Args: component: Component name (e.g., 'bot', 'api', 'autonomous') Returns: Configured logger instance Example: logger = get_logger('bot') logger.info("Bot started") logger.error("Connection failed", exc_info=True) """ if component not in COMPONENTS: raise ValueError( f"Unknown component '{component}'. " f"Available: {', '.join(COMPONENTS.keys())}" ) if component in _loggers: return _loggers[component] # Setup handlers if not done _setup_handlers() # Create logger logger = logging.Logger(component) # Add custom API level method def api(self, message, *args, **kwargs): if self.isEnabledFor(API_LEVEL): self._log(API_LEVEL, message, args, **kwargs) logger.api = lambda msg, *args, **kwargs: api(logger, msg, *args, **kwargs) # Configure logger _configure_logger(logger, component) # Cache it _loggers[component] = logger return logger def list_components() -> Dict[str, str]: """Get list of all available components with descriptions.""" return COMPONENTS.copy() def get_component_stats() -> Dict[str, Dict]: """Get statistics about each component's logging.""" stats = {} for component in COMPONENTS.keys(): log_file = LOG_DIR / f'{component.replace(".", "_")}.log' stats[component] = { 'enabled': True, # Will be updated from config 'log_file': str(log_file), 'file_exists': log_file.exists(), 'file_size': log_file.stat().st_size if log_file.exists() else 0, } # Update from config config = get_log_config() component_config = config.get('components', {}).get(component, {}) stats[component]['enabled'] = component_config.get('enabled', True) stats[component]['level'] = component_config.get('level', 'INFO') stats[component]['enabled_levels'] = component_config.get('enabled_levels', ['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL']) return stats def intercept_external_loggers(): """ Intercept logs from external libraries (APScheduler, etc.) and route them through our system. Call this after initializing your application. """ # Intercept APScheduler loggers apscheduler_loggers = [ 'apscheduler', 'apscheduler.scheduler', 'apscheduler.executors', 'apscheduler.jobstores', ] our_logger = get_logger('apscheduler') for logger_name in apscheduler_loggers: ext_logger = logging.getLogger(logger_name) # Remove existing handlers ext_logger.handlers.clear() ext_logger.propagate = False # Add our handlers for handler in our_logger.handlers: ext_logger.addHandler(handler) # Set level ext_logger.setLevel(logging.DEBUG) # Initialize on import _setup_handlers()