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