src/ad_creative_env/logging_setup.py

"""Logging entrypoint for the ad-creative env.

Standard-library pattern: library code only does `logging.getLogger(__name__)` and
emits — it attaches NO handlers except a NullHandler on the package logger (so a
"no handler" warning never fires and records still PROPAGATE to whatever the
application configures on the ROOT logger). An application entrypoint (our scripts,
or a future FE) calls configure_logging() ONCE to install a root handler; because our
loggers propagate, a FE that configures its own root logging captures our logs without
calling us, and there is no double-logging because the library owns no real handler.
AD_LOG_FORMAT=json gives hosting-ready structured output.
"""

from __future__ import annotations

import json
import logging
import os
import sys
from pathlib import Path

_PKG = "ad_creative_env"
_PLAIN_FMT = "%(asctime)s %(levelname)s %(name)s: %(message)s"
_FORMATS = ("plain", "json")
# Dev default is DEBUG so a plain run surfaces every boundary log without extra env.
# Flip to "INFO" (or set AD_LOG_LEVEL) when quieting down for production.
_DEFAULT_LEVEL = "DEBUG"
# Chatty third-party loggers pinned to WARNING so a DEBUG run surfaces OUR records,
# not their per-chunk trace (e.g. PIL's PngImagePlugin STREAM lines, urllib3 retries).
_NOISY_LIBS = ("PIL", "urllib3")

# Library null handler: silences "no handler" and lets records propagate to whatever
# handler the application installs on root. Added once, at import.
logging.getLogger(_PKG).addHandler(logging.NullHandler())

# Standard LogRecord attributes + our JSON schema keys — everything else on a record
# is a structured `extra=` field we surface in JSON. (makeLogRecord baseline is
# version-robust: 3.10/3.11 lack `taskName`, 3.12 adds it; the explicit union covers both.)
_RESERVED = set(vars(logging.makeLogRecord({}))) | {"message", "asctime", "taskName"}
_SCHEMA_KEYS = ("ts", "level", "logger", "msg", "exc", "stack")


def _safe_str(value) -> str:
    """str(value) that never raises (a hostile __str__/__repr__ can't crash logging)."""
    try:
        return str(value)
    except Exception:
        return "<unstringable>"


def _json_safe(value):
    """Return value if JSON-serializable, else a best-effort safe string, never raising."""
    try:
        json.dumps(value)
        return value
    except (TypeError, ValueError):
        return _safe_str(value)


class JsonFormatter(logging.Formatter):
    """One-line JSON per record: {ts, level, logger, msg, +extra}. Canonical keys win.

    Every value path is guarded so a hostile __str__/__repr__, malformed %-args, or a
    non-serializable exception argument degrades to a safe string instead of raising —
    a logging call must never crash the caller.
    """

    def format(self, record: logging.LogRecord) -> str:
        try:
            msg = record.getMessage()
        except Exception:  # malformed %-args
            msg = "<unformattable msg " + _safe_str(record.msg) + ">"
        # extras FIRST so the canonical schema keys assigned below always win.
        payload = {
            key: _json_safe(val)
            for key, val in record.__dict__.items()
            if key not in _RESERVED and key not in _SCHEMA_KEYS and not key.startswith("_")
        }
        payload["ts"] = _safe_str(self.formatTime(record))
        payload["level"] = record.levelname
        payload["logger"] = record.name
        payload["msg"] = msg
        if record.exc_info:
            try:
                payload["exc"] = self.formatException(record.exc_info)
            except Exception:
                payload["exc"] = "<unformattable exception>"
        if record.stack_info:
            try:
                payload["stack"] = self.formatStack(record.stack_info)
            except Exception:
                payload["stack"] = "<unformattable stack>"
        try:
            return json.dumps(payload, default=_safe_str)
        except Exception:  # last-resort: never let logging raise
            return json.dumps({"level": record.levelname, "logger": record.name,
                               "msg": "<unserializable log record>"})


def _valid_level(name: str) -> bool:
    # getLevelName(name) -> int for a known name, else a "Level X" string (3.10-safe).
    return isinstance(logging.getLevelName(name), int)


def configure_logging(level=None, fmt=None, *, log_file=None, force=False) -> logging.Logger:
    """Install managed handlers on the ROOT logger, basicConfig-style. Idempotent.

    level: arg > AD_LOG_LEVEL env > "DEBUG" (dev default). fmt: arg > AD_LOG_FORMAT env
    > "plain". log_file: arg > AD_LOG_FILE env > None. Raises ValueError on an unknown
    level or format (fail-fast for both env vars).

    Always installs a stderr StreamHandler (terminal output). When a log_file is
    resolved, ALSO installs a FileHandler opened in mode "w", so each configure call
    OVERWRITES the file — the file holds only the latest run. Both handlers are tagged
    managed, so a reconfigure drops+closes them and re-installs cleanly.

    Like logging.basicConfig, this is a NO-OP when the root logger already has FOREIGN
    (non-managed) handlers — i.e. a host app configured logging — unless force=True. So
    calling it from our scripts installs output, but if a host already owns root we do
    not duplicate its output or override its level (our loggers just propagate into it).
    """
    level = (level or os.environ.get("AD_LOG_LEVEL") or _DEFAULT_LEVEL).upper()
    fmt = (fmt or os.environ.get("AD_LOG_FORMAT") or "plain").lower()
    log_file = log_file or os.environ.get("AD_LOG_FILE")
    if fmt not in _FORMATS:
        raise ValueError(f"AD_LOG_FORMAT must be one of {_FORMATS}, got {fmt!r}")
    if not _valid_level(level):
        raise ValueError(f"AD_LOG_LEVEL must be a valid level name, got {level!r}")

    root = logging.getLogger()
    foreign = [h for h in root.handlers if not getattr(h, "_ad_managed", False)]
    if foreign and not force:
        return root  # a host owns root logging; our loggers propagate into it

    root.setLevel(level)
    for handler in list(root.handlers):  # idempotent: drop AND close our prior handlers
        if getattr(handler, "_ad_managed", False):
            root.removeHandler(handler)
            handler.close()

    formatter = JsonFormatter() if fmt == "json" else logging.Formatter(_PLAIN_FMT)
    handlers = [logging.StreamHandler(sys.stderr)]  # terminal, always
    if log_file:
        path = Path(log_file)
        path.parent.mkdir(parents=True, exist_ok=True)
        handlers.append(logging.FileHandler(path, mode="w", encoding="utf-8"))  # overwrite
    for handler in handlers:
        setattr(handler, "_ad_managed", True)  # sentinel: marks OUR handler vs a host's
        handler.setFormatter(formatter)
        root.addHandler(handler)
    for name in _NOISY_LIBS:  # only reached when WE own root (host-owned returns early)
        logging.getLogger(name).setLevel(logging.WARNING)
    return root