147 lines
5.3 KiB
Python
147 lines
5.3 KiB
Python
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") |