""" ShortSmith v2 - Centralized Logging Module Provides consistent logging across all components with: - File and console handlers - Different log levels per module - Timing decorators for performance tracking - Structured log formatting """ import logging import sys import time import functools from pathlib import Path from typing import Optional, Callable, Any from datetime import datetime from contextlib import contextmanager # Custom log format LOG_FORMAT = "%(asctime)s | %(levelname)-8s | %(name)-20s | %(message)s" LOG_DATE_FORMAT = "%Y-%m-%d %H:%M:%S" # Module-specific log levels (can be overridden) MODULE_LOG_LEVELS = { "shortsmith": logging.INFO, "shortsmith.models": logging.INFO, "shortsmith.core": logging.INFO, "shortsmith.pipeline": logging.INFO, "shortsmith.scoring": logging.DEBUG, } # Track if logging has been set up _logging_initialized = False class ColoredFormatter(logging.Formatter): """Formatter that adds colors to log levels for console output.""" COLORS = { logging.DEBUG: "\033[36m", # Cyan logging.INFO: "\033[32m", # Green logging.WARNING: "\033[33m", # Yellow logging.ERROR: "\033[31m", # Red logging.CRITICAL: "\033[35m", # Magenta } RESET = "\033[0m" def format(self, record: logging.LogRecord) -> str: """Format log record with colors.""" # Add color to levelname color = self.COLORS.get(record.levelno, "") record.levelname = f"{color}{record.levelname}{self.RESET}" return super().format(record) def setup_logging( log_level: str = "INFO", log_file: Optional[str] = None, log_to_console: bool = True, use_colors: bool = True, ) -> None: """ Initialize the logging system. Args: log_level: Default logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL) log_file: Path to log file (None to disable file logging) log_to_console: Whether to log to console use_colors: Whether to use colored output in console Raises: ValueError: If invalid log level provided """ global _logging_initialized if _logging_initialized: return # Validate log level numeric_level = getattr(logging, log_level.upper(), None) if not isinstance(numeric_level, int): raise ValueError(f"Invalid log level: {log_level}") # Get root logger for shortsmith root_logger = logging.getLogger("shortsmith") root_logger.setLevel(logging.DEBUG) # Capture all, handlers will filter # Clear existing handlers root_logger.handlers.clear() # Console handler if log_to_console: console_handler = logging.StreamHandler(sys.stdout) console_handler.setLevel(numeric_level) if use_colors and sys.stdout.isatty(): console_formatter = ColoredFormatter(LOG_FORMAT, LOG_DATE_FORMAT) else: console_formatter = logging.Formatter(LOG_FORMAT, LOG_DATE_FORMAT) console_handler.setFormatter(console_formatter) root_logger.addHandler(console_handler) # File handler if log_file: try: log_path = Path(log_file) log_path.parent.mkdir(parents=True, exist_ok=True) file_handler = logging.FileHandler(log_file, encoding="utf-8") file_handler.setLevel(logging.DEBUG) # Log everything to file file_formatter = logging.Formatter(LOG_FORMAT, LOG_DATE_FORMAT) file_handler.setFormatter(file_formatter) root_logger.addHandler(file_handler) except (OSError, PermissionError) as e: # Log to console if file logging fails if log_to_console: root_logger.warning(f"Could not create log file {log_file}: {e}") # Apply module-specific levels for module, level in MODULE_LOG_LEVELS.items(): logging.getLogger(module).setLevel(level) _logging_initialized = True root_logger.info(f"Logging initialized at level {log_level}") def get_logger(name: str) -> logging.Logger: """ Get a logger instance for a specific module. Args: name: Module name (will be prefixed with 'shortsmith.') Returns: Configured logger instance """ # Ensure logging is initialized if not _logging_initialized: setup_logging() # Prefix with shortsmith if not already if not name.startswith("shortsmith"): name = f"shortsmith.{name}" return logging.getLogger(name) class LogTimer: """ Context manager and decorator for timing operations. Usage as context manager: with LogTimer(logger, "Processing video"): process_video() Usage as decorator: @LogTimer.decorator(logger, "Processing") def process_video(): ... """ def __init__( self, logger: logging.Logger, operation: str, level: int = logging.INFO, ): """ Initialize timer. Args: logger: Logger to use for output operation: Description of the operation being timed level: Log level for timing messages """ self.logger = logger self.operation = operation self.level = level self.start_time: Optional[float] = None self.end_time: Optional[float] = None def __enter__(self) -> "LogTimer": """Start timing.""" self.start_time = time.perf_counter() self.logger.log(self.level, f"Starting: {self.operation}") return self def __exit__(self, exc_type, exc_val, exc_tb) -> None: """Stop timing and log duration.""" self.end_time = time.perf_counter() duration = self.end_time - self.start_time if exc_type is not None: self.logger.error( f"Failed: {self.operation} after {duration:.2f}s - {exc_type.__name__}: {exc_val}" ) else: self.logger.log( self.level, f"Completed: {self.operation} in {duration:.2f}s" ) @property def elapsed(self) -> float: """Get elapsed time in seconds.""" if self.start_time is None: return 0.0 end = self.end_time if self.end_time else time.perf_counter() return end - self.start_time @staticmethod def decorator( logger: logging.Logger, operation: Optional[str] = None, level: int = logging.INFO, ) -> Callable: """ Create a timing decorator. Args: logger: Logger to use operation: Operation name (defaults to function name) level: Log level Returns: Decorator function """ def decorator_func(func: Callable) -> Callable: op_name = operation or func.__name__ @functools.wraps(func) def wrapper(*args, **kwargs) -> Any: with LogTimer(logger, op_name, level): return func(*args, **kwargs) return wrapper return decorator_func @contextmanager def log_context(logger: logging.Logger, context: str): """ Context manager that logs entry and exit of a code block. Args: logger: Logger instance context: Description of the context Yields: None """ logger.debug(f"Entering: {context}") try: yield except Exception as e: logger.error(f"Error in {context}: {type(e).__name__}: {e}") raise finally: logger.debug(f"Exiting: {context}") def log_exception(logger: logging.Logger, message: str = "An error occurred"): """ Decorator that logs exceptions with full context. Args: logger: Logger instance message: Custom error message prefix Returns: Decorator function """ def decorator(func: Callable) -> Callable: @functools.wraps(func) def wrapper(*args, **kwargs) -> Any: try: return func(*args, **kwargs) except Exception as e: logger.exception(f"{message} in {func.__name__}: {e}") raise return wrapper return decorator # Export public interface __all__ = [ "setup_logging", "get_logger", "LogTimer", "log_context", "log_exception", ]