"""Logging configuration for Aseprite MCP."""
import logging
import sys
from pathlib import Path
from typing import Optional
from logging.handlers import RotatingFileHandler
import json
from datetime import datetime
from .config import get_config
class StructuredFormatter(logging.Formatter):
"""Custom formatter that outputs structured logs."""
def format(self, record: logging.LogRecord) -> str:
"""Format log record as structured data."""
config = get_config()
# Basic log data
log_data = {
"timestamp": datetime.utcnow().isoformat(),
"level": record.levelname,
"logger": record.name,
"message": record.getMessage(),
"module": record.module,
"function": record.funcName,
"line": record.lineno
}
# Add extra fields if present
if hasattr(record, 'file_path'):
log_data['file_path'] = record.file_path
if hasattr(record, 'operation'):
log_data['operation'] = record.operation
if hasattr(record, 'duration'):
log_data['duration'] = record.duration
if hasattr(record, 'error_type'):
log_data['error_type'] = record.error_type
if hasattr(record, 'details'):
log_data['details'] = record.details
# Add exception info if present
if record.exc_info:
log_data['exception'] = self.formatException(record.exc_info)
# Format based on configuration
if config.log_format == "json":
return json.dumps(log_data, default=str)
else:
# Traditional format
base_msg = f"{log_data['timestamp']} - {log_data['logger']} - {log_data['level']} - {log_data['message']}"
if 'exception' in log_data:
base_msg += f"\n{log_data['exception']}"
return base_msg
class AsepriteLogger:
"""Centralized logging for Aseprite MCP."""
def __init__(self, name: str = "aseprite_mcp"):
"""Initialize logger with configuration."""
self.config = get_config()
self.logger = logging.getLogger(name)
self._setup_logger()
def _setup_logger(self):
"""Configure the logger based on settings."""
# Clear existing handlers
self.logger.handlers.clear()
# Set log level
self.logger.setLevel(getattr(logging, self.config.log_level))
# Console handler
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setFormatter(StructuredFormatter())
self.logger.addHandler(console_handler)
# File handler if configured
if self.config.log_file:
log_path = Path(self.config.log_file)
log_path.parent.mkdir(parents=True, exist_ok=True)
file_handler = RotatingFileHandler(
log_path,
maxBytes=10 * 1024 * 1024, # 10MB
backupCount=5
)
file_handler.setFormatter(StructuredFormatter())
self.logger.addHandler(file_handler)
def log_operation(self, operation: str, file_path: Optional[str] = None, **kwargs):
"""Log an operation with context."""
extra = {
'operation': operation,
'details': kwargs
}
if file_path:
extra['file_path'] = file_path
self.logger.info(f"Operation: {operation}", extra=extra)
def log_error(self, message: str, error: Exception, operation: Optional[str] = None, **kwargs):
"""Log an error with context."""
extra = {
'error_type': type(error).__name__,
'details': kwargs
}
if operation:
extra['operation'] = operation
self.logger.error(message, exc_info=error, extra=extra)
def log_performance(self, operation: str, duration: float, **kwargs):
"""Log performance metrics."""
extra = {
'operation': operation,
'duration': duration,
'details': kwargs
}
self.logger.info(f"Performance: {operation} took {duration:.3f}s", extra=extra)
def debug(self, message: str, **kwargs):
"""Log debug message with optional context."""
extra = {'details': kwargs} if kwargs else {}
self.logger.debug(message, extra=extra)
def info(self, message: str, **kwargs):
"""Log info message with optional context."""
extra = {'details': kwargs} if kwargs else {}
self.logger.info(message, extra=extra)
def warning(self, message: str, **kwargs):
"""Log warning message with optional context."""
extra = {'details': kwargs} if kwargs else {}
self.logger.warning(message, extra=extra)
def error(self, message: str, **kwargs):
"""Log error message with optional context."""
extra = {'details': kwargs} if kwargs else {}
self.logger.error(message, extra=extra)
# Global logger instance
_logger: Optional[AsepriteLogger] = None
def get_logger(name: Optional[str] = None) -> AsepriteLogger:
"""Get logger instance."""
global _logger
if _logger is None or name:
_logger = AsepriteLogger(name or "aseprite_mcp")
return _logger
# Convenience functions
def log_operation(operation: str, file_path: Optional[str] = None, **kwargs):
"""Log an operation."""
get_logger().log_operation(operation, file_path, **kwargs)
def log_error(message: str, error: Exception, operation: Optional[str] = None, **kwargs):
"""Log an error."""
get_logger().log_error(message, error, operation, **kwargs)
def log_performance(operation: str, duration: float, **kwargs):
"""Log performance metrics."""
get_logger().log_performance(operation, duration, **kwargs)
# Context manager for timing operations
class Timer:
"""Context manager for timing operations."""
def __init__(self, operation: str, log: bool = True, **log_kwargs):
self.operation = operation
self.log = log
self.log_kwargs = log_kwargs
self.start_time = None
self.duration = None
def __enter__(self):
import time
self.start_time = time.time()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
import time
self.duration = time.time() - self.start_time
if self.log:
if exc_type:
log_error(
f"Operation {self.operation} failed after {self.duration:.3f}s",
exc_val,
operation=self.operation,
duration=self.duration,
**self.log_kwargs
)
else:
log_performance(self.operation, self.duration, **self.log_kwargs)