620fa0ef2d
Complete rebuild of the Discord bot with modern architecture including: - Modular API client with proper error handling - Clean separation of models, services, and commands - Comprehensive test coverage with pytest - Structured logging and configuration management - Organized command structure for scalability 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
247 lines
8.0 KiB
Python
247 lines
8.0 KiB
Python
"""
|
|
Enhanced Logging Utilities
|
|
|
|
Provides structured logging with contextual information for Discord bot debugging.
|
|
Implements hybrid approach: human-readable console + structured JSON files.
|
|
"""
|
|
import contextvars
|
|
import json
|
|
import logging
|
|
import time
|
|
import uuid
|
|
from datetime import datetime
|
|
from typing import Dict, Any, Optional, Union
|
|
|
|
# Context variable for request tracking across async calls
|
|
log_context: contextvars.ContextVar[Dict[str, Any]] = contextvars.ContextVar('log_context', default={})
|
|
|
|
logger = logging.getLogger(f'{__name__}.logging_utils')
|
|
|
|
JSONValue = Union[
|
|
str,
|
|
int,
|
|
float,
|
|
bool,
|
|
None,
|
|
dict[str, Any], # nested object
|
|
list[Any] # arrays
|
|
]
|
|
|
|
|
|
class JSONFormatter(logging.Formatter):
|
|
"""Custom JSON formatter for structured file logging."""
|
|
|
|
def format(self, record) -> str:
|
|
"""Format log record as JSON with context information."""
|
|
# Base log object
|
|
log_obj: dict[str, JSONValue] = {
|
|
'timestamp': datetime.now().isoformat() + 'Z',
|
|
'level': record.levelname,
|
|
'logger': record.name,
|
|
'message': record.getMessage()
|
|
}
|
|
|
|
# Add function/line info if available
|
|
if hasattr(record, 'funcName') and record.funcName:
|
|
log_obj['function'] = record.funcName
|
|
if hasattr(record, 'lineno') and record.lineno:
|
|
log_obj['line'] = record.lineno
|
|
|
|
# Add exception info if present
|
|
if record.exc_info:
|
|
log_obj['exception'] = {
|
|
'type': record.exc_info[0].__name__ if record.exc_info[0] else 'Unknown',
|
|
'message': str(record.exc_info[1]) if record.exc_info[1] else 'No message',
|
|
'traceback': self.formatException(record.exc_info)
|
|
}
|
|
|
|
# Add context from contextvars
|
|
context = log_context.get({})
|
|
if context:
|
|
log_obj['context'] = context.copy()
|
|
|
|
# Add custom fields from extra parameter
|
|
excluded_keys = {
|
|
'name', 'msg', 'args', 'levelname', 'levelno', 'pathname',
|
|
'filename', 'module', 'lineno', 'funcName', 'created',
|
|
'msecs', 'relativeCreated', 'thread', 'threadName',
|
|
'processName', 'process', 'getMessage', 'exc_info',
|
|
'exc_text', 'stack_info'
|
|
}
|
|
|
|
extra_data = {}
|
|
for key, value in record.__dict__.items():
|
|
if key not in excluded_keys:
|
|
# Ensure JSON serializable
|
|
try:
|
|
json.dumps(value)
|
|
extra_data[key] = value
|
|
except (TypeError, ValueError):
|
|
extra_data[key] = str(value)
|
|
|
|
if extra_data:
|
|
log_obj['extra'] = extra_data
|
|
|
|
return json.dumps(log_obj, ensure_ascii=False)
|
|
|
|
|
|
class ContextualLogger:
|
|
"""
|
|
Logger wrapper that provides contextual information and structured logging.
|
|
|
|
Automatically includes Discord context (user, guild, command) in all log messages.
|
|
"""
|
|
|
|
def __init__(self, logger_name: str):
|
|
"""
|
|
Initialize contextual logger.
|
|
|
|
Args:
|
|
logger_name: Name for the underlying logger
|
|
"""
|
|
self.logger = logging.getLogger(logger_name)
|
|
self._start_time: Optional[float] = None
|
|
|
|
def start_operation(self, operation_name: Optional[str] = None) -> str:
|
|
"""
|
|
Start timing an operation and generate a trace ID.
|
|
|
|
Args:
|
|
operation_name: Optional name for the operation being tracked
|
|
|
|
Returns:
|
|
Generated trace ID for this operation
|
|
"""
|
|
self._start_time = time.time()
|
|
trace_id = str(uuid.uuid4())[:8]
|
|
|
|
# Add trace_id to context
|
|
current_context = log_context.get({})
|
|
current_context['trace_id'] = trace_id
|
|
if operation_name:
|
|
current_context['operation'] = operation_name
|
|
log_context.set(current_context)
|
|
|
|
return trace_id
|
|
|
|
def _get_duration_ms(self) -> Optional[int]:
|
|
"""Get operation duration in milliseconds if start_operation was called."""
|
|
if self._start_time:
|
|
return int((time.time() - self._start_time) * 1000)
|
|
return None
|
|
|
|
def debug(self, message: str, **kwargs):
|
|
"""Log debug message with context."""
|
|
duration = self._get_duration_ms()
|
|
if duration is not None:
|
|
kwargs['duration_ms'] = duration
|
|
self.logger.debug(message, extra=kwargs)
|
|
|
|
def info(self, message: str, **kwargs):
|
|
"""Log info message with context."""
|
|
duration = self._get_duration_ms()
|
|
if duration is not None:
|
|
kwargs['duration_ms'] = duration
|
|
self.logger.info(message, extra=kwargs)
|
|
|
|
def warning(self, message: str, **kwargs):
|
|
"""Log warning message with context."""
|
|
duration = self._get_duration_ms()
|
|
if duration is not None:
|
|
kwargs['duration_ms'] = duration
|
|
self.logger.warning(message, extra=kwargs)
|
|
|
|
def error(self, message: str, error: Optional[Exception] = None, **kwargs):
|
|
"""
|
|
Log error message with context and exception information.
|
|
|
|
Args:
|
|
message: Error message
|
|
error: Optional exception object
|
|
**kwargs: Additional context
|
|
"""
|
|
duration = self._get_duration_ms()
|
|
if duration is not None:
|
|
kwargs['duration_ms'] = duration
|
|
|
|
if error:
|
|
kwargs['error'] = {
|
|
'type': type(error).__name__,
|
|
'message': str(error)
|
|
}
|
|
self.logger.error(message, exc_info=True, extra=kwargs)
|
|
else:
|
|
self.logger.error(message, extra=kwargs)
|
|
|
|
def exception(self, message: str, **kwargs):
|
|
"""Log exception with full traceback and context."""
|
|
duration = self._get_duration_ms()
|
|
if duration is not None:
|
|
kwargs['duration_ms'] = duration
|
|
self.logger.exception(message, extra=kwargs)
|
|
|
|
|
|
def set_discord_context(
|
|
interaction: Optional[Any] = None,
|
|
user_id: Optional[Union[str, int]] = None,
|
|
guild_id: Optional[Union[str, int]] = None,
|
|
channel_id: Optional[Union[str, int]] = None,
|
|
command: Optional[str] = None,
|
|
**additional_context
|
|
):
|
|
"""
|
|
Set Discord-specific context for logging.
|
|
|
|
Args:
|
|
interaction: Discord interaction object (will extract user/guild/channel)
|
|
user_id: Discord user ID
|
|
guild_id: Discord guild ID
|
|
channel_id: Discord channel ID
|
|
command: Command name (e.g., '/player')
|
|
**additional_context: Any additional context to include
|
|
"""
|
|
context = log_context.get({}).copy()
|
|
|
|
# Extract from interaction if provided
|
|
if interaction:
|
|
context['user_id'] = str(interaction.user.id)
|
|
if interaction.guild:
|
|
context['guild_id'] = str(interaction.guild.id)
|
|
context['guild_name'] = interaction.guild.name
|
|
if interaction.channel:
|
|
context['channel_id'] = str(interaction.channel.id)
|
|
if hasattr(interaction, 'command') and interaction.command:
|
|
context['command'] = f"/{interaction.command.name}"
|
|
|
|
# Override with explicit parameters
|
|
if user_id:
|
|
context['user_id'] = str(user_id)
|
|
if guild_id:
|
|
context['guild_id'] = str(guild_id)
|
|
if channel_id:
|
|
context['channel_id'] = str(channel_id)
|
|
if command:
|
|
context['command'] = command
|
|
|
|
# Add any additional context
|
|
context.update(additional_context)
|
|
|
|
log_context.set(context)
|
|
|
|
|
|
def clear_context():
|
|
"""Clear the current logging context."""
|
|
log_context.set({})
|
|
|
|
|
|
def get_contextual_logger(logger_name: str) -> ContextualLogger:
|
|
"""
|
|
Get a contextual logger instance.
|
|
|
|
Args:
|
|
logger_name: Name for the logger (typically __name__)
|
|
|
|
Returns:
|
|
ContextualLogger instance
|
|
"""
|
|
return ContextualLogger(logger_name) |