File size: 1,671 Bytes
3f820d9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 | """Centralized logging configuration.
The pipeline uses ``logging`` exclusively (no ``print``). Every module obtains
its logger via :func:`get_logger`, which guarantees handlers are configured
exactly once per process, even under multiprocessing worker re-imports.
"""
from __future__ import annotations
import logging
import sys
from pathlib import Path
_CONFIGURED = False
LOG_FORMAT = "%(asctime)s | %(levelname)-8s | %(processName)s | %(name)s | %(message)s"
DATE_FORMAT = "%Y-%m-%d %H:%M:%S"
def configure_logging(level: int = logging.INFO, log_file: "Path | None" = None) -> None:
"""Configure the root logger once for the entire process.
Args:
level: Root logging level (e.g. ``logging.INFO``, ``logging.DEBUG``).
log_file: Optional file path to additionally write logs to.
"""
global _CONFIGURED
root = logging.getLogger()
if _CONFIGURED:
root.setLevel(level)
return
root.setLevel(level)
formatter = logging.Formatter(LOG_FORMAT, datefmt=DATE_FORMAT)
stream_handler = logging.StreamHandler(stream=sys.stdout)
stream_handler.setFormatter(formatter)
root.addHandler(stream_handler)
if log_file is not None:
log_file.parent.mkdir(parents=True, exist_ok=True)
file_handler = logging.FileHandler(log_file, encoding="utf-8")
file_handler.setFormatter(formatter)
root.addHandler(file_handler)
_CONFIGURED = True
def get_logger(name: str) -> logging.Logger:
"""Return a module-scoped logger, configuring root logging with sane defaults if needed."""
if not _CONFIGURED:
configure_logging()
return logging.getLogger(name)
|