Spaces:
Sleeping
Sleeping
| import logging | |
| import logging.handlers | |
| import sys | |
| from pathlib import Path | |
| from app.core.config import settings | |
| # Create logs directory if it doesn't exist | |
| logs_dir = Path("logs") | |
| logs_dir.mkdir(exist_ok=True) | |
| # Define formatters | |
| DETAILED_FORMATTER = logging.Formatter( | |
| "%(asctime)s - %(name)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(message)s", | |
| datefmt="%Y-%m-%d %H:%M:%S" | |
| ) | |
| CONSOLE_FORMATTER = logging.Formatter( | |
| "%(asctime)s - %(levelname)s - %(message)s", | |
| datefmt="%H:%M:%S" | |
| ) | |
| def setup_logging(): | |
| """Configure logging for the application""" | |
| # Get root logger | |
| root_logger = logging.getLogger() | |
| # Clear any existing handlers | |
| root_logger.handlers.clear() | |
| # Set log level from settings | |
| log_level = getattr(logging, settings.LOG_LEVEL.upper(), logging.INFO) | |
| root_logger.setLevel(log_level) | |
| # Console Handler | |
| console_handler = logging.StreamHandler(sys.stdout) | |
| console_handler.setFormatter(CONSOLE_FORMATTER) | |
| console_handler.setLevel(log_level) | |
| root_logger.addHandler(console_handler) | |
| # File Handler with rotation | |
| file_handler = logging.handlers.RotatingFileHandler( | |
| filename=logs_dir / "talk2sql.log", | |
| maxBytes=10 * 1024 * 1024, # 10MB | |
| backupCount=5, | |
| encoding="utf-8" | |
| ) | |
| file_handler.setFormatter(DETAILED_FORMATTER) | |
| file_handler.setLevel(log_level) | |
| root_logger.addHandler(file_handler) | |
| # Create separate error log file for ERROR and CRITICAL | |
| error_handler = logging.handlers.RotatingFileHandler( | |
| filename=logs_dir / "error.log", | |
| maxBytes=10 * 1024 * 1024, # 10MB | |
| backupCount=5, | |
| encoding="utf-8" | |
| ) | |
| error_handler.setFormatter(DETAILED_FORMATTER) | |
| error_handler.setLevel(logging.ERROR) | |
| root_logger.addHandler(error_handler) | |
| # Capture unhandled exceptions | |
| def handle_exception(exc_type, exc_value, exc_traceback): | |
| if issubclass(exc_type, KeyboardInterrupt): | |
| # Call the default handler for KeyboardInterrupt | |
| sys.__excepthook__(exc_type, exc_value, exc_traceback) | |
| return | |
| root_logger.error("Uncaught exception", exc_info=(exc_type, exc_value, exc_traceback)) | |
| sys.excepthook = handle_exception | |
| # Log initial configuration | |
| root_logger.info(f"Logging configured with level: {settings.LOG_LEVEL}") | |
| return root_logger | |
| def get_logger(name: str) -> logging.Logger: | |
| """Get a logger instance for a specific module""" | |
| return logging.getLogger(name) |