diff --git a/shared/logger.py b/shared/logger.py new file mode 100644 index 0000000..73db45d --- /dev/null +++ b/shared/logger.py @@ -0,0 +1,147 @@ +import os +import warnings +import logging +import time +from typing import Optional, Tuple + +logging_level = os.getenv("LOGGING_LEVEL", "INFO").upper() + +class RelativePathMultiLineFormatter(logging.Formatter): + def __init__( + self, + fmt: Optional[str] = None, + datefmt: Optional[str] = None, + remove_prefix: Optional[str] = None, + ) -> None: + super().__init__(fmt, datefmt) + self.remove_prefix = remove_prefix or os.getcwd() + # Ensure the prefix ends with a separator + if not self.remove_prefix.endswith(os.sep): + self.remove_prefix += os.sep + + def format(self, record: logging.LogRecord) -> str: + """Create a shallow copy of the record and rewrite the pathname + to be relative to the configured prefix, then handle multi-line formatting. + """ + # Make a copy of the record dict so we don't mutate the caller's record + record_dict = record.__dict__.copy() + new_record = logging.makeLogRecord(record_dict) + + # Remove the prefix from pathname if present + pathname = getattr(new_record, "pathname", "") + if pathname.startswith(self.remove_prefix): + new_record.pathname = pathname[len(self.remove_prefix) :] + + # Get the formatted message (with relative path) + formatted = super().format(new_record) + + # Handle multi-line formatting + lines = formatted.split('\n') + + # If there's only one line, return as-is + if len(lines) == 1: + return formatted + + # For multi-line messages, we need to add the prefix to each subsequent line + # Create a prefix by formatting an empty message to get the timestamp/level/location part + empty_record_dict = new_record.__dict__.copy() + empty_record_dict['msg'] = '' + empty_record_dict['args'] = () + empty_record = logging.makeLogRecord(empty_record_dict) + prefix = super().format(empty_record) + + # Apply prefix to each continuation line + formatted_lines = [lines[0]] # First line already has prefix + for line in lines[1:]: + formatted_lines.append(prefix + line) + + return '\n'.join(formatted_lines) + + +def _setup_logging(level: str=logging_level) -> logging.Logger: + os.environ["TORCH_CPP_LOG_LEVEL"] = "ERROR" + warnings.filterwarnings( + "ignore", message="Overriding a previously registered kernel" + ) + warnings.filterwarnings("ignore", message="Warning only once for all operators") + warnings.filterwarnings("ignore", message=".*Couldn't find ffmpeg or avconv.*") + warnings.filterwarnings("ignore", message="'force_all_finite' was renamed to") + warnings.filterwarnings("ignore", message="n_jobs value 1 overridden") + warnings.filterwarnings("ignore", message=".*websocket.*is deprecated") + + logging.getLogger("aiortc").setLevel(logging.WARNING) + logging.getLogger("aioice").setLevel(logging.WARNING) + logging.getLogger("asyncio").setLevel(logging.WARNING) + logging.getLogger("httpx").setLevel(logging.WARNING) + + numeric_level = getattr(logging, level.upper(), None) + if not isinstance(numeric_level, int): + raise ValueError(f"Invalid log level: {level}") + + # Create the combined formatter + formatter = RelativePathMultiLineFormatter( + fmt="%(asctime)s.%(msecs)03d %(levelname)s - %(pathname)s:%(lineno)d - %(message)s", + datefmt="%H:%M:%S", + ) + + # Create a handler (e.g., StreamHandler for console output) + handler = logging.StreamHandler() + formatter.converter = time.localtime + handler.setFormatter(formatter) + + # Simple repeat-suppression filter + class _RepeatFilter(logging.Filter): + def __init__(self, interval: float = 5.0) -> None: + super().__init__() + self._interval = interval + self._last: Optional[Tuple[int, str]] = None + self._last_time: float = 0.0 + + def filter(self, record: logging.LogRecord) -> bool: + try: + msg = record.getMessage() + except Exception: + # Fallback to a string representation if getMessage fails + msg = str(record) + + key: Tuple[int, str] = (getattr(record, "levelno", 0), msg) + now = time.time() + if self._last == key and (now - self._last_time) < self._interval: + return False + self._last = key + self._last_time = now + return True + + handler.addFilter(_RepeatFilter()) + + # Configure root logger + logging.basicConfig( + level=numeric_level, + handlers=[handler], # Use only your handler + force=True, + ) + + # Set levels for noisy loggers + for noisy_logger in ( + "uvicorn", + "uvicorn.error", + "uvicorn.access", + "fastapi", + "starlette", + ): + logger = logging.getLogger(noisy_logger) + logger.setLevel(logging.WARNING) + logger.handlers = [] # Remove default handlers + logger.addHandler(handler) # Add your custom handler + + logger = logging.getLogger(__name__) + return logger + + +logger = _setup_logging(level=logging_level) +logger.debug(f"Logging initialized with level: {logging_level}") + +# Test the multi-line formatting +if __name__ == "__main__": + logger.info("Single line message") + logger.info("Multi-line message:\nSecond line\nThird line") \ No newline at end of file