""" Custom Log Formatters for GEPA Optimizer. Provides formatters for: - Console output with colors and emoji - JSON structured logging for production - Plain text for file logging """ import json import logging from datetime import datetime from typing import Any, Dict, Optional # ANSI color codes for terminal output class Colors: """ANSI color codes for terminal coloring.""" RESET = "\033[0m" BOLD = "\033[1m" DIM = "\033[2m" # Log level colors DEBUG = "\033[36m" # Cyan INFO = "\033[32m" # Green WARNING = "\033[33m" # Yellow ERROR = "\033[31m" # Red CRITICAL = "\033[35m" # Magenta # Semantic colors TIMESTAMP = "\033[90m" # Gray MODULE = "\033[34m" # Blue MESSAGE = "\033[0m" # Default # Emoji prefixes for visual log scanning LEVEL_EMOJI = { logging.DEBUG: "🔍", logging.INFO: "ℹī¸ ", logging.WARNING: "⚠ī¸ ", logging.ERROR: "❌", logging.CRITICAL: "🚨", } # Level colors mapping LEVEL_COLORS = { logging.DEBUG: Colors.DEBUG, logging.INFO: Colors.INFO, logging.WARNING: Colors.WARNING, logging.ERROR: Colors.ERROR, logging.CRITICAL: Colors.CRITICAL, } class GepaFormatter(logging.Formatter): """ Custom formatter for GEPA Optimizer logs. Features: - Optional color output for console - Optional emoji prefixes for visual scanning - Structured extra fields support - Clean, readable format Example output: 2024-01-15 10:30:45 | INFO | ℹī¸ gepa_optimizer.core.optimizer | Starting optimization iteration=5 """ def __init__( self, fmt: Optional[str] = None, datefmt: Optional[str] = None, use_colors: bool = True, include_emoji: bool = True, ): """ Initialize the formatter. Args: fmt: Format string (uses default if not provided) datefmt: Date format string use_colors: Whether to use ANSI colors include_emoji: Whether to include emoji prefixes """ super().__init__(fmt=fmt, datefmt=datefmt) self.use_colors = use_colors self.include_emoji = include_emoji def format(self, record: logging.LogRecord) -> str: """Format a log record with colors and emoji.""" # Store original values original_msg = record.msg original_levelname = record.levelname try: # Add emoji prefix if enabled if self.include_emoji: emoji = LEVEL_EMOJI.get(record.levelno, "") record.levelname = f"{emoji} {record.levelname}" # Add colors if enabled if self.use_colors: color = LEVEL_COLORS.get(record.levelno, Colors.RESET) record.levelname = f"{color}{record.levelname}{Colors.RESET}" record.name = f"{Colors.MODULE}{record.name}{Colors.RESET}" # Format extra fields if present extra_str = self._format_extra(record) if extra_str: record.msg = f"{record.msg} | {extra_str}" # Call parent formatter formatted = super().format(record) return formatted finally: # Restore original values record.msg = original_msg record.levelname = original_levelname def _format_extra(self, record: logging.LogRecord) -> str: """ Format extra fields from the log record. Extra fields are passed via the 'extra' parameter to logging calls: logger.info("Message", extra={"key": "value"}) """ # Standard LogRecord attributes to exclude standard_attrs = { 'name', 'msg', 'args', 'created', 'filename', 'funcName', 'levelname', 'levelno', 'lineno', 'module', 'msecs', 'pathname', 'process', 'processName', 'relativeCreated', 'stack_info', 'exc_info', 'exc_text', 'thread', 'threadName', 'taskName', 'message' } # Collect extra fields extra_fields = { k: v for k, v in record.__dict__.items() if k not in standard_attrs and not k.startswith('_') } if not extra_fields: return "" # Format as key=value pairs parts = [] for key, value in extra_fields.items(): if isinstance(value, str): parts.append(f"{key}={value}") elif isinstance(value, (int, float)): parts.append(f"{key}={value}") elif isinstance(value, bool): parts.append(f"{key}={str(value).lower()}") else: parts.append(f"{key}={repr(value)}") return " ".join(parts) class JsonFormatter(logging.Formatter): """ JSON formatter for structured logging. Outputs each log record as a single JSON line, suitable for: - Log aggregation systems (ELK, Splunk) - Cloud logging (CloudWatch, Stackdriver) - Log parsing and analysis Example output: {"timestamp": "2024-01-15T10:30:45.123Z", "level": "INFO", "logger": "gepa_optimizer.core", "message": "Starting optimization", "iteration": 5} """ def __init__( self, include_timestamp: bool = True, include_location: bool = False, ): """ Initialize JSON formatter. Args: include_timestamp: Include ISO timestamp include_location: Include file/line information """ super().__init__() self.include_timestamp = include_timestamp self.include_location = include_location def format(self, record: logging.LogRecord) -> str: """Format record as JSON string.""" log_dict: Dict[str, Any] = {} # Timestamp if self.include_timestamp: log_dict["timestamp"] = datetime.utcfromtimestamp( record.created ).isoformat() + "Z" # Core fields log_dict["level"] = record.levelname log_dict["logger"] = record.name log_dict["message"] = record.getMessage() # Location info if self.include_location: log_dict["file"] = record.filename log_dict["line"] = record.lineno log_dict["function"] = record.funcName # Exception info if record.exc_info: log_dict["exception"] = self.formatException(record.exc_info) # Extra fields standard_attrs = { 'name', 'msg', 'args', 'created', 'filename', 'funcName', 'levelname', 'levelno', 'lineno', 'module', 'msecs', 'pathname', 'process', 'processName', 'relativeCreated', 'stack_info', 'exc_info', 'exc_text', 'thread', 'threadName', 'taskName', 'message' } for key, value in record.__dict__.items(): if key not in standard_attrs and not key.startswith('_'): try: # Ensure value is JSON serializable json.dumps(value) log_dict[key] = value except (TypeError, ValueError): log_dict[key] = str(value) return json.dumps(log_dict, default=str) class CompactFormatter(logging.Formatter): """ Compact formatter for minimal log output. Useful for: - CI/CD pipelines - Reduced log verbosity - Quick debugging Example output: 10:30:45 INFO optimizer: Starting optimization """ def format(self, record: logging.LogRecord) -> str: """Format record in compact form.""" # Short timestamp (time only) time_str = datetime.fromtimestamp(record.created).strftime("%H:%M:%S") # Short module name (last part only) short_name = record.name.split(".")[-1] return f"{time_str} {record.levelname:5s} {short_name}: {record.getMessage()}"