src/utils/logger.py

"""Shared structured logger for every CellTriage module.



WHAT: ``get_logger(name)`` returns a configured logger that writes to both the

console and ``outputs/logs/<name>.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/<name>.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}")