Added missing file

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")