"""Shared structured logger for every CellTriage module. WHAT: ``get_logger(name)`` returns a configured logger that writes to both the console and ``outputs/logs/.log``. WHY: Two project rules depend on durable logs rather than console scrollback. Every excluded cell must be recorded with its ID and the reason it was dropped, and every reported number must be traceable to a generated artifact. A run whose diagnostics vanished with the terminal session cannot support either claim months later during review. """ from __future__ import annotations import logging import sys from pathlib import Path from src.utils.paths import LOG_DIR, relative_to_root #: Timestamped, level-tagged, module-tagged. The line number is included #: because a QC audit trail is only useful if a surprising log line can be #: traced back to the exact code that emitted it. _LOG_FORMAT = "%(asctime)s | %(levelname)-8s | %(name)s:%(lineno)d | %(message)s" _DATE_FORMAT = "%Y-%m-%d %H:%M:%S" DEFAULT_FILE_LEVEL = logging.DEBUG DEFAULT_CONSOLE_LEVEL = logging.INFO def _log_path_for(name: str) -> Path: """Map a logger name to its log file, flattening dotted module paths. ``src.data.mat_parser`` -> ``outputs/logs/mat_parser.log`` so that the log directory stays readable and one file corresponds to one module. """ stem = name.split(".")[-1] if name else "celltriage" return LOG_DIR / f"{stem}.log" def get_logger( name: str, *, file_level: int = DEFAULT_FILE_LEVEL, console_level: int = DEFAULT_CONSOLE_LEVEL, log_dir: Path | None = None, ) -> logging.Logger: """Return a logger writing to the console and to ``outputs/logs/.log``. Idempotent: calling this repeatedly with the same ``name`` returns the same logger without stacking duplicate handlers. WHY that matters: modules import each other freely, and duplicated handlers produce duplicated log lines, which makes a count of "cells excluded" read wrong by an integer factor. Args: name: Logger name, conventionally ``__name__`` of the calling module. file_level: Threshold for the file handler. DEBUG by default, because the file is the audit trail and disk is cheap. console_level: Threshold for the console handler. INFO by default to keep interactive runs readable. log_dir: Override for the log directory. Intended for tests. Returns: A configured :class:`logging.Logger`. """ logger = logging.getLogger(name) # The logger's own level must be the more permissive of the two handler # levels, otherwise records are dropped before any handler sees them. logger.setLevel(min(file_level, console_level)) # Do not propagate to the root logger: a library or notebook that has # configured root logging would otherwise duplicate every record. logger.propagate = False if getattr(logger, "_celltriage_configured", False): return logger target_dir = Path(log_dir) if log_dir is not None else LOG_DIR target_dir.mkdir(parents=True, exist_ok=True) log_file = target_dir / _log_path_for(name).name formatter = logging.Formatter(fmt=_LOG_FORMAT, datefmt=_DATE_FORMAT) # Append rather than truncate: a re-run must not destroy the record of the # run that preceded it. file_handler = logging.FileHandler(log_file, mode="a", encoding="utf-8") file_handler.setLevel(file_level) file_handler.setFormatter(formatter) logger.addHandler(file_handler) console_handler = logging.StreamHandler(stream=sys.stdout) console_handler.setLevel(console_level) console_handler.setFormatter(formatter) logger.addHandler(console_handler) logger._celltriage_configured = True # type: ignore[attr-defined] logger.debug("Logger initialised; writing to %s", relative_to_root(log_file)) return logger def log_section(logger: logging.Logger, title: str, width: int = 78) -> None: """Emit a visually distinct section banner. WHY: Phase summaries and reconciliation tables must be findable by eye in a long log file. """ logger.info("=" * width) logger.info(title) logger.info("=" * width) def run_logger_smoke_test() -> Path: """Write one record at every level and return the resulting log file path. WHY this exists as a callable entry point: "logging works" is a claim best demonstrated by an executed artifact on disk rather than asserted, and a misconfigured handler is otherwise only discovered when a real run needs it. """ logger = get_logger("logger_smoke_test") log_section(logger, "CellTriage logger smoke test") logger.debug("DEBUG record - visible in the file, suppressed on console.") logger.info("INFO record - the default console level.") logger.warning("WARNING record.") logger.error("ERROR record.") log_file = LOG_DIR / "logger_smoke_test.log" logger.info("Smoke test complete; log file at %s", relative_to_root(log_file)) return log_file if __name__ == "__main__": path = run_logger_smoke_test() print(f"Logger smoke test wrote: {path}")