""" Кастомная система логирования для PyServe Управляет логгерами всех пакетов и модулей, включая uvicorn и starlette """ import logging import logging.handlers import sys import time from pathlib import Path from typing import Dict, Any, List from . import __version__ class UvicornLogFilter(logging.Filter): def filter(self, record): if hasattr(record, 'name') and 'uvicorn.access' in record.name: if hasattr(record, 'getMessage'): msg = record.getMessage() if ' - "' in msg and '" ' in msg: parts = msg.split(' - "') if len(parts) >= 2: client_info = parts[0] request_part = parts[1].split('" ') if len(request_part) >= 2: method_path = request_part[0] status_part = request_part[1] record.msg = f"Access: {client_info} - {method_path} - {status_part}" return True class PyServeFormatter(logging.Formatter): COLORS = { 'DEBUG': '\033[36m', # Cyan 'INFO': '\033[32m', # Green 'WARNING': '\033[33m', # Yellow 'ERROR': '\033[31m', # Red 'CRITICAL': '\033[35m', # Magenta 'RESET': '\033[0m' # Reset } def __init__(self, use_colors: bool = True, show_module: bool = True, *args, **kwargs): super().__init__(*args, **kwargs) self.use_colors = use_colors and hasattr(sys.stderr, 'isatty') and sys.stderr.isatty() self.show_module = show_module def format(self, record): if self.use_colors: levelname = record.levelname if levelname in self.COLORS: record.levelname = f"{self.COLORS[levelname]}{levelname}{self.COLORS['RESET']}" if self.show_module and hasattr(record, 'name'): name = record.name if name.startswith('uvicorn'): record.name = 'uvicorn' elif name.startswith('pyserve'): pass elif name.startswith('starlette'): record.name = 'starlette' return super().format(record) class AccessLogHandler(logging.Handler): def __init__(self, logger_name: str = 'pyserve.access'): super().__init__() self.access_logger = logging.getLogger(logger_name) def emit(self, record): self.access_logger.handle(record) class PyServeLogManager: def __init__(self): self.configured = False self.handlers: Dict[str, logging.Handler] = {} self.loggers: Dict[str, logging.Logger] = {} self.original_handlers: Dict[str, List[logging.Handler]] = {} def setup_logging(self, config: Dict[str, Any]) -> None: if self.configured: return level = config.get('level', 'INFO').upper() console_output = config.get('console_output', True) log_file = config.get('log_file', './logs/pyserve.log') self._save_original_handlers() self._clear_all_handlers() root_logger = logging.getLogger() root_logger.setLevel(logging.DEBUG) detailed_formatter = PyServeFormatter( use_colors=False, show_module=True, fmt='%(asctime)s - %(name)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(message)s' ) console_formatter = PyServeFormatter( use_colors=True, show_module=True, fmt='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) if console_output: console_handler = logging.StreamHandler(sys.stdout) console_handler.setLevel(getattr(logging, level)) console_handler.setFormatter(console_formatter) console_handler.addFilter(UvicornLogFilter()) root_logger.addHandler(console_handler) self.handlers['console'] = console_handler if log_file: self._ensure_log_directory(log_file) file_handler = logging.handlers.RotatingFileHandler( log_file, maxBytes=10 * 1024 * 1024, # 10MB backupCount=5, encoding='utf-8' ) file_handler.setLevel(logging.DEBUG) file_handler.setFormatter(detailed_formatter) file_handler.addFilter(UvicornLogFilter()) root_logger.addHandler(file_handler) self.handlers['file'] = file_handler self._configure_library_loggers(level) self._intercept_uvicorn_logging() pyserve_logger = logging.getLogger('pyserve') pyserve_logger.setLevel(getattr(logging, level)) self.loggers['pyserve'] = pyserve_logger pyserve_logger.info(f"PyServe v{__version__} - Система логирования инициализирована") pyserve_logger.info(f"Уровень логирования: {level}") pyserve_logger.info(f"Консольный вывод: {'включен' if console_output else 'отключен'}") pyserve_logger.info(f"Файл логов: {log_file if log_file else 'отключен'}") self.configured = True def _save_original_handlers(self) -> None: logger_names = ['', 'uvicorn', 'uvicorn.access', 'uvicorn.error', 'starlette'] for name in logger_names: logger = logging.getLogger(name) self.original_handlers[name] = logger.handlers.copy() def _clear_all_handlers(self) -> None: root_logger = logging.getLogger() for handler in root_logger.handlers[:]: root_logger.removeHandler(handler) handler.close() logger_names = ['uvicorn', 'uvicorn.access', 'uvicorn.error', 'starlette'] for name in logger_names: logger = logging.getLogger(name) for handler in logger.handlers[:]: logger.removeHandler(handler) handler.close() self.handlers.clear() def _ensure_log_directory(self, log_file: str) -> None: log_dir = Path(log_file).parent log_dir.mkdir(parents=True, exist_ok=True) def _configure_library_loggers(self, main_level: str) -> None: library_configs = { # Uvicorn и связанные - только в DEBUG режиме 'uvicorn': 'DEBUG' if main_level == 'DEBUG' else 'WARNING', 'uvicorn.access': 'DEBUG' if main_level == 'DEBUG' else 'WARNING', 'uvicorn.error': 'DEBUG' if main_level == 'DEBUG' else 'ERROR', 'uvicorn.asgi': 'DEBUG' if main_level == 'DEBUG' else 'WARNING', # Starlette - только в DEBUG режиме 'starlette': 'DEBUG' if main_level == 'DEBUG' else 'WARNING', 'asyncio': 'WARNING', 'concurrent.futures': 'WARNING', 'multiprocessing': 'WARNING', 'pyserve': main_level, 'pyserve.server': main_level, 'pyserve.routing': main_level, 'pyserve.extensions': main_level, 'pyserve.config': main_level, } for logger_name, level in library_configs.items(): logger = logging.getLogger(logger_name) logger.setLevel(getattr(logging, level)) if logger_name.startswith('uvicorn') and logger_name != 'uvicorn': logger.propagate = False self.loggers[logger_name] = logger def _intercept_uvicorn_logging(self) -> None: uvicorn_logger = logging.getLogger('uvicorn') uvicorn_access_logger = logging.getLogger('uvicorn.access') for handler in uvicorn_logger.handlers[:]: uvicorn_logger.removeHandler(handler) for handler in uvicorn_access_logger.handlers[:]: uvicorn_access_logger.removeHandler(handler) uvicorn_logger.propagate = True uvicorn_access_logger.propagate = True def get_logger(self, name: str) -> logging.Logger: if name not in self.loggers: logger = logging.getLogger(name) self.loggers[name] = logger return self.loggers[name] def set_level(self, logger_name: str, level: str) -> None: if logger_name in self.loggers: self.loggers[logger_name].setLevel(getattr(logging, level.upper())) def add_handler(self, name: str, handler: logging.Handler) -> None: if name not in self.handlers: root_logger = logging.getLogger() root_logger.addHandler(handler) self.handlers[name] = handler def remove_handler(self, name: str) -> None: if name in self.handlers: root_logger = logging.getLogger() root_logger.removeHandler(self.handlers[name]) self.handlers[name].close() del self.handlers[name] def create_access_log(self, method: str, path: str, status_code: int, response_time: float, client_ip: str, user_agent: str = "") -> None: access_logger = self.get_logger('pyserve.access') log_message = f'{client_ip} - - [{time.strftime("%d/%b/%Y:%H:%M:%S %z")}] ' \ f'"{method} {path} HTTP/1.1" {status_code} - ' \ f'"{user_agent}" {response_time:.3f}s' access_logger.info(log_message) def shutdown(self) -> None: for handler in self.handlers.values(): handler.close() self.handlers.clear() for logger_name, handlers in self.original_handlers.items(): logger = logging.getLogger(logger_name) for handler in handlers: logger.addHandler(handler) self.loggers.clear() self.configured = False log_manager = PyServeLogManager() def setup_logging(config: Dict[str, Any]) -> None: log_manager.setup_logging(config) def get_logger(name: str) -> logging.Logger: return log_manager.get_logger(name) def create_access_log(method: str, path: str, status_code: int, response_time: float, client_ip: str, user_agent: str = "") -> None: log_manager.create_access_log(method, path, status_code, response_time, client_ip, user_agent) def shutdown_logging() -> None: log_manager.shutdown()