Spaces:
Paused
Paused
| # logging_config.py - Production-ready logging configuration | |
| import logging | |
| import logging.handlers | |
| import sys | |
| import os | |
| from typing import Optional | |
| def setup_logging(log_level: str = "INFO", log_dir: str = "logs") -> None: | |
| """ | |
| Configures structured logging for the AI Lab with rotation and proper encoding. | |
| Args: | |
| log_level: Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL) | |
| log_dir: Directory for log files | |
| """ | |
| try: | |
| os.makedirs(log_dir, exist_ok=True) | |
| except Exception as e: | |
| print(f"Error creating log directory: {e}") | |
| sys.exit(1) | |
| # Create formatters | |
| detailed_formatter = logging.Formatter( | |
| fmt="%(asctime)s [%(levelname)-8s] [%(name)-20s] %(funcName)-15s:%(lineno)-4d | %(message)s", | |
| datefmt="%Y-%m-%d %H:%M:%S" | |
| ) | |
| simple_formatter = logging.Formatter( | |
| fmt="%(asctime)s [%(levelname)s] %(message)s", | |
| datefmt="%H:%M:%S" | |
| ) | |
| # Create handlers | |
| handlers = [] | |
| # Rotating file handler with UTF-8 encoding | |
| try: | |
| file_handler = logging.handlers.RotatingFileHandler( | |
| filename=os.path.join(log_dir, "ai_lab.log"), | |
| maxBytes=10 * 1024 * 1024, # 10MB per file | |
| backupCount=5, # Keep 5 backup files | |
| encoding='utf-8', | |
| mode='a' | |
| ) | |
| file_handler.setFormatter(detailed_formatter) | |
| file_handler.setLevel(getattr(logging, log_level)) | |
| handlers.append(file_handler) | |
| except Exception as e: | |
| print(f"Warning: Could not create file handler: {e}") | |
| # Console handler with UTF-8 support | |
| console_handler = logging.StreamHandler(sys.stdout) | |
| console_handler.setFormatter(simple_formatter) | |
| console_handler.setLevel(logging.INFO) # Console shows INFO and above | |
| # Ensure UTF-8 encoding on Windows | |
| if hasattr(console_handler.stream, 'reconfigure'): | |
| console_handler.stream.reconfigure(encoding='utf-8') | |
| handlers.append(console_handler) | |
| # Configure root logger | |
| root_logger = logging.getLogger() | |
| root_logger.setLevel(getattr(logging, log_level)) | |
| # Remove any existing handlers | |
| for handler in root_logger.handlers[:]: | |
| root_logger.removeHandler(handler) | |
| # Add new handlers | |
| for handler in handlers: | |
| root_logger.addHandler(handler) | |
| # Configure third-party library logging levels | |
| noisy_libraries = [ | |
| "httpx", | |
| "urllib3", | |
| "sentence_transformers", | |
| "transformers", | |
| "faiss", | |
| "openai", | |
| "httpcore", | |
| "langchain" | |
| ] | |
| for lib in noisy_libraries: | |
| logging.getLogger(lib).setLevel(logging.WARNING) | |
| # Log startup message | |
| logging.info("="*60) | |
| logging.info("AI Lab Logging System Initialized") | |
| logging.info(f"Log Level: {log_level}") | |
| logging.info(f"Log Directory: {os.path.abspath(log_dir)}") | |
| logging.info("="*60) | |
| def get_logger(name: str) -> logging.Logger: | |
| """ | |
| Get a logger instance for a specific module. | |
| Args: | |
| name: Name of the module/component | |
| Returns: | |
| Configured logger instance | |
| """ | |
| return logging.getLogger(name) | |
| def log_exception(logger: logging.Logger, e: Exception, context: str = "") -> None: | |
| """ | |
| Helper function to log exceptions with full traceback. | |
| Args: | |
| logger: Logger instance to use | |
| e: Exception to log | |
| context: Additional context about where the exception occurred | |
| """ | |
| import traceback | |
| error_msg = f"Exception in {context}: {str(e)}" if context else str(e) | |
| logger.error(error_msg) | |
| logger.debug(f"Traceback:\n{traceback.format_exc()}") | |
| def setup_performance_logging() -> Optional[logging.Logger]: | |
| """ | |
| Set up a separate performance logger for tracking execution times and metrics. | |
| Returns: | |
| Performance logger instance | |
| """ | |
| perf_logger = logging.getLogger("performance") | |
| perf_logger.setLevel(logging.DEBUG) | |
| # Create performance log file | |
| try: | |
| perf_handler = logging.handlers.RotatingFileHandler( | |
| filename=os.path.join("logs", "performance.log"), | |
| maxBytes=5 * 1024 * 1024, # 5MB | |
| backupCount=3, | |
| encoding='utf-8' | |
| ) | |
| perf_formatter = logging.Formatter( | |
| fmt="%(asctime)s.%(msecs)03d | %(message)s", | |
| datefmt="%Y-%m-%d %H:%M:%S" | |
| ) | |
| perf_handler.setFormatter(perf_formatter) | |
| perf_logger.addHandler(perf_handler) | |
| return perf_logger | |
| except Exception as e: | |
| print(f"Warning: Could not create performance logger: {e}") | |
| return None | |
| # Context manager for timed operations | |
| class LoggedTimer: | |
| """Context manager for logging operation execution time.""" | |
| def __init__(self, logger: logging.Logger, operation_name: str): | |
| self.logger = logger | |
| self.operation_name = operation_name | |
| self.start_time = None | |
| def __enter__(self): | |
| import time | |
| self.start_time = time.time() | |
| self.logger.debug(f"Starting: {self.operation_name}") | |
| return self | |
| def __exit__(self, exc_type, exc_val, exc_tb): | |
| import time | |
| elapsed = time.time() - self.start_time | |
| if exc_type: | |
| self.logger.error(f"Failed: {self.operation_name} after {elapsed:.2f}s - {exc_val}") | |
| else: | |
| self.logger.debug(f"Completed: {self.operation_name} in {elapsed:.2f}s") | |
| # Initialize logging when module is imported (can be overridden) | |
| if __name__ != "__main__": | |
| # Auto-initialize with defaults when imported | |
| try: | |
| setup_logging() | |
| except Exception as e: | |
| print(f"Warning: Auto-initialization of logging failed: {e}") |