AI-Talent-Force/dev_caio

1"""2ShortSmith v2 - Centralized Logging Module3 4Provides consistent logging across all components with:5- File and console handlers6- Different log levels per module7- Timing decorators for performance tracking8- Structured log formatting9"""10 11import logging12import sys13import time14import functools15from pathlib import Path16from typing import Optional, Callable, Any17from datetime import datetime18from contextlib import contextmanager19 20 21# Custom log format22LOG_FORMAT = "%(asctime)s | %(levelname)-8s | %(name)-20s | %(message)s"23LOG_DATE_FORMAT = "%Y-%m-%d %H:%M:%S"24 25# Module-specific log levels (can be overridden)26MODULE_LOG_LEVELS = {27    "shortsmith": logging.INFO,28    "shortsmith.models": logging.INFO,29    "shortsmith.core": logging.INFO,30    "shortsmith.pipeline": logging.INFO,31    "shortsmith.scoring": logging.DEBUG,32}33 34# Track if logging has been set up35_logging_initialized = False36 37 38class ColoredFormatter(logging.Formatter):39    """Formatter that adds colors to log levels for console output."""40 41    COLORS = {42        logging.DEBUG: "\033[36m",     # Cyan43        logging.INFO: "\033[32m",      # Green44        logging.WARNING: "\033[33m",   # Yellow45        logging.ERROR: "\033[31m",     # Red46        logging.CRITICAL: "\033[35m",  # Magenta47    }48    RESET = "\033[0m"49 50    def format(self, record: logging.LogRecord) -> str:51        """Format log record with colors."""52        # Add color to levelname53        color = self.COLORS.get(record.levelno, "")54        record.levelname = f"{color}{record.levelname}{self.RESET}"55        return super().format(record)56 57 58def setup_logging(59    log_level: str = "INFO",60    log_file: Optional[str] = None,61    log_to_console: bool = True,62    use_colors: bool = True,63) -> None:64    """65    Initialize the logging system.66 67    Args:68        log_level: Default logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL)69        log_file: Path to log file (None to disable file logging)70        log_to_console: Whether to log to console71        use_colors: Whether to use colored output in console72 73    Raises:74        ValueError: If invalid log level provided75    """76    global _logging_initialized77 78    if _logging_initialized:79        return80 81    # Validate log level82    numeric_level = getattr(logging, log_level.upper(), None)83    if not isinstance(numeric_level, int):84        raise ValueError(f"Invalid log level: {log_level}")85 86    # Get root logger for shortsmith87    root_logger = logging.getLogger("shortsmith")88    root_logger.setLevel(logging.DEBUG)  # Capture all, handlers will filter89 90    # Clear existing handlers91    root_logger.handlers.clear()92 93    # Console handler94    if log_to_console:95        console_handler = logging.StreamHandler(sys.stdout)96        console_handler.setLevel(numeric_level)97 98        if use_colors and sys.stdout.isatty():99            console_formatter = ColoredFormatter(LOG_FORMAT, LOG_DATE_FORMAT)100        else:101            console_formatter = logging.Formatter(LOG_FORMAT, LOG_DATE_FORMAT)102 103        console_handler.setFormatter(console_formatter)104        root_logger.addHandler(console_handler)105 106    # File handler107    if log_file:108        try:109            log_path = Path(log_file)110            log_path.parent.mkdir(parents=True, exist_ok=True)111 112            file_handler = logging.FileHandler(log_file, encoding="utf-8")113            file_handler.setLevel(logging.DEBUG)  # Log everything to file114            file_formatter = logging.Formatter(LOG_FORMAT, LOG_DATE_FORMAT)115            file_handler.setFormatter(file_formatter)116            root_logger.addHandler(file_handler)117        except (OSError, PermissionError) as e:118            # Log to console if file logging fails119            if log_to_console:120                root_logger.warning(f"Could not create log file {log_file}: {e}")121 122    # Apply module-specific levels123    for module, level in MODULE_LOG_LEVELS.items():124        logging.getLogger(module).setLevel(level)125 126    _logging_initialized = True127    root_logger.info(f"Logging initialized at level {log_level}")128 129 130def get_logger(name: str) -> logging.Logger:131    """132    Get a logger instance for a specific module.133 134    Args:135        name: Module name (will be prefixed with 'shortsmith.')136 137    Returns:138        Configured logger instance139    """140    # Ensure logging is initialized141    if not _logging_initialized:142        setup_logging()143 144    # Prefix with shortsmith if not already145    if not name.startswith("shortsmith"):146        name = f"shortsmith.{name}"147 148    return logging.getLogger(name)149 150 151class LogTimer:152    """153    Context manager and decorator for timing operations.154 155    Usage as context manager:156        with LogTimer(logger, "Processing video"):157            process_video()158 159    Usage as decorator:160        @LogTimer.decorator(logger, "Processing")161        def process_video():162            ...163    """164 165    def __init__(166        self,167        logger: logging.Logger,168        operation: str,169        level: int = logging.INFO,170    ):171        """172        Initialize timer.173 174        Args:175            logger: Logger to use for output176            operation: Description of the operation being timed177            level: Log level for timing messages178        """179        self.logger = logger180        self.operation = operation181        self.level = level182        self.start_time: Optional[float] = None183        self.end_time: Optional[float] = None184 185    def __enter__(self) -> "LogTimer":186        """Start timing."""187        self.start_time = time.perf_counter()188        self.logger.log(self.level, f"Starting: {self.operation}")189        return self190 191    def __exit__(self, exc_type, exc_val, exc_tb) -> None:192        """Stop timing and log duration."""193        self.end_time = time.perf_counter()194        duration = self.end_time - self.start_time195 196        if exc_type is not None:197            self.logger.error(198                f"Failed: {self.operation} after {duration:.2f}s - {exc_type.__name__}: {exc_val}"199            )200        else:201            self.logger.log(202                self.level,203                f"Completed: {self.operation} in {duration:.2f}s"204            )205 206    @property207    def elapsed(self) -> float:208        """Get elapsed time in seconds."""209        if self.start_time is None:210            return 0.0211        end = self.end_time if self.end_time else time.perf_counter()212        return end - self.start_time213 214    @staticmethod215    def decorator(216        logger: logging.Logger,217        operation: Optional[str] = None,218        level: int = logging.INFO,219    ) -> Callable:220        """221        Create a timing decorator.222 223        Args:224            logger: Logger to use225            operation: Operation name (defaults to function name)226            level: Log level227 228        Returns:229            Decorator function230        """231        def decorator_func(func: Callable) -> Callable:232            op_name = operation or func.__name__233 234            @functools.wraps(func)235            def wrapper(*args, **kwargs) -> Any:236                with LogTimer(logger, op_name, level):237                    return func(*args, **kwargs)238 239            return wrapper240        return decorator_func241 242 243@contextmanager244def log_context(logger: logging.Logger, context: str):245    """246    Context manager that logs entry and exit of a code block.247 248    Args:249        logger: Logger instance250        context: Description of the context251 252    Yields:253        None254    """255    logger.debug(f"Entering: {context}")256    try:257        yield258    except Exception as e:259        logger.error(f"Error in {context}: {type(e).__name__}: {e}")260        raise261    finally:262        logger.debug(f"Exiting: {context}")263 264 265def log_exception(logger: logging.Logger, message: str = "An error occurred"):266    """267    Decorator that logs exceptions with full context.268 269    Args:270        logger: Logger instance271        message: Custom error message prefix272 273    Returns:274        Decorator function275    """276    def decorator(func: Callable) -> Callable:277        @functools.wraps(func)278        def wrapper(*args, **kwargs) -> Any:279            try:280                return func(*args, **kwargs)281            except Exception as e:282                logger.exception(f"{message} in {func.__name__}: {e}")283                raise284 285        return wrapper286    return decorator287 288 289# Export public interface290__all__ = [291    "setup_logging",292    "get_logger",293    "LogTimer",294    "log_context",295    "log_exception",296]297