src/common/logging.py

"""Structured logging: colored console renderer + the `step()` stage timer.

Logging standard for this project: every pipeline stage runs inside ``with step("<concern>.<action>") as
metrics:`` — it logs ``step.start`` then ``step.end`` with the elapsed seconds and whatever you attach via
``metrics["key"] = value``. No ``print()`` in the engine; no engine/library names in user-facing strings.

Configuration is read from the environment (so this module has no dependency on the settings layer yet):
``RERANKER_LOG_LEVEL`` (default ``INFO``), ``RERANKER_LOG`` (set ``0`` to silence), ``NO_COLOR`` to drop
ANSI. The stdlib ``import logging`` below is an ABSOLUTE import — Python resolves top-level imports against
``sys.path``, never this same-named sibling — so ``logging.INFO`` references the standard library correctly.
"""

from __future__ import annotations

import contextlib
import logging
import os
import sys
import time
from contextlib import contextmanager
from typing import Any

import structlog
from structlog.typing import EventDict

_log_configured = False
# this module's own functions to skip when locating the real caller (logging plumbing, not user code)
_CALLSITE_SKIP = {"step", "get_logger", "_callsite", "_configure_logging"}

_ANSI = {
    "reset": "\033[0m", "bold": "\033[1m", "gray": "\033[90m", "red": "\033[31m",
    "green": "\033[32m", "yellow": "\033[33m", "blue": "\033[94m", "magenta": "\033[35m", "cyan": "\033[36m",
}
_LEVEL_COLOR = {"debug": "blue", "info": "green", "warning": "yellow", "error": "red", "critical": "red"}


def _callsite(_logger: Any, _method_name: str, event_dict: EventDict) -> EventDict:
    """Add ``at`` (file:line) + ``func`` to each line, pointing at the code that opened the step."""
    frame: Any = sys._getframe()
    while frame is not None:
        filename = frame.f_code.co_filename
        func = frame.f_code.co_name
        skip = (
            "structlog" in filename
            or "contextlib" in filename
            or (os.path.basename(filename) == "logging.py" and func in _CALLSITE_SKIP)
        )
        if not skip:
            event_dict["at"] = f"{os.path.basename(filename)}:{frame.f_lineno}"
            event_dict["func"] = func
            break
        frame = frame.f_back
    return event_dict


def _make_console_renderer(colors: bool):
    """Build a renderer: ``time [level] [step] [file:line func] event  key=value …``."""
    def paint(text: str, *styles: str) -> str:
        if not colors:
            return text
        prefix = "".join(_ANSI[style] for style in styles)
        return f"{prefix}{text}{_ANSI['reset']}"

    def render(_logger: Any, _method: str, event_dict: EventDict) -> str:
        extra = dict(event_dict)
        timestamp = str(extra.pop("timestamp", ""))
        level = str(extra.pop("level", "info"))
        event = str(extra.pop("event", ""))
        name = extra.pop("step", None) or extra.pop("logger", None)
        extra.pop("logger", None)
        location = f"{extra.pop('at', '')} {extra.pop('func', '')}".strip()
        blocks = [paint(timestamp, "gray"), paint(f"[{level:<7}]", "bold", _LEVEL_COLOR.get(level, "green"))]
        if name:
            blocks.append(paint(f"[{name}]", "cyan"))
        if location:
            blocks.append(paint(f"[{location}]", "yellow"))
        blocks.append(paint(event, "bold", "blue"))
        header_line = " ".join(block for block in blocks if block)
        pairs = " ".join(
            f"{paint(key, 'magenta')}={paint(str(value), 'green')}" for key, value in extra.items()
        )
        return f"{header_line}  {pairs}".rstrip()

    return render


def _configure_logging(force: bool = False) -> None:
    """Configure structlog once from the environment (idempotent)."""
    global _log_configured
    if _log_configured and not force:
        return
    if os.environ.get("RERANKER_LOG", "1") == "0":
        # Silence: filter at CRITICAL so info/warning/error are dropped (the engine never logs CRITICAL).
        # NOTE: structlog only accepts standard levels here — CRITICAL+1 raises KeyError.
        structlog.configure(
            wrapper_class=structlog.make_filtering_bound_logger(logging.CRITICAL),
            cache_logger_on_first_use=True,
        )
        _log_configured = True
        return
    with contextlib.suppress(Exception):
        sys.stdout.reconfigure(encoding="utf-8", line_buffering=True)  # type: ignore[union-attr]
    colors = "NO_COLOR" not in os.environ
    if colors:
        with contextlib.suppress(Exception):  # wrap stdout so ANSI works on legacy Windows consoles
            import colorama

            colorama.init()
    level = getattr(logging, os.environ.get("RERANKER_LOG_LEVEL", "INFO").upper(), logging.INFO)
    structlog.configure(
        processors=[
            structlog.contextvars.merge_contextvars,
            structlog.processors.add_log_level,
            structlog.processors.TimeStamper(fmt="%H:%M:%S", utc=False),
            _callsite,
            _make_console_renderer(colors),
        ],
        wrapper_class=structlog.make_filtering_bound_logger(level),
        cache_logger_on_first_use=True,
    )
    _log_configured = True


def get_logger(name: str | None = None):
    """Return a configured structlog logger, with the concern ``name`` bound as ``logger=``."""
    _configure_logging()
    log = structlog.get_logger(name)
    return log.bind(logger=name) if name else log


def _phase_banner(text: str) -> None:
    """A plain full-width divider line bracketing every phase, for grep/scroll tracing in a long log stream
    (``grep -A20 "PHASE START: rank.explain"``). Bypasses the structured renderer on purpose — this is a visual
    marker, not a parseable event — but still respects ``RERANKER_LOG=0`` silencing."""
    if os.environ.get("RERANKER_LOG", "1") == "0":
        return
    print(f"--------------------------- {text} ---------------------------")


@contextmanager
def step(label: str, **context: Any):
    """Log ``step.start`` / ``step.end`` (+seconds, +metrics), bracketed by PHASE banners. Yields a mutable
    metrics dict."""
    _phase_banner(f"PHASE START: {label}")
    log = get_logger("step").bind(step=label, **context)
    log.info("step.start")
    metrics: dict[str, Any] = {}
    start = time.perf_counter()
    try:
        yield metrics
    except Exception as exc:
        log.error("step.fail", seconds=round(time.perf_counter() - start, 3), error=str(exc))
        _phase_banner(f"PHASE FAILED: {label} ({round(time.perf_counter() - start, 3)}s)")
        raise
    else:
        log.info("step.end", seconds=round(time.perf_counter() - start, 3), **metrics)
        _phase_banner(f"PHASE END: {label} ({round(time.perf_counter() - start, 3)}s)")