app/core/logger.py

from __future__ import annotations

import contextvars
import json
import logging
import re
import sys
from datetime import datetime, timezone
from typing import Any, TextIO

from app.core.config import get_settings

request_id_context: contextvars.ContextVar[str] = contextvars.ContextVar("request_id", default="-")


class JsonFormatter(logging.Formatter):
    """One-line structured JSON logs suitable for container log collectors."""

    _standard = set(logging.makeLogRecord({}).__dict__) | {"message", "asctime"}
    _sensitive_keys = frozenset(
        {
            "access_token",
            "refresh_token",
            "id_token",
            "authorization",
            "client_secret",
            "secret",
            "password",
            "api_key",
            "credential",
            "cookie",
        }
    )
    _bearer = re.compile(r"(?i)\bbearer\s+[A-Za-z0-9._~+/=-]+")
    _assigned_secret = re.compile(
        r"(?i)(access[_-]?token|refresh[_-]?token|id[_-]?token|client[_-]?secret|"
        r"authorization|api[_-]?key|password|secret|credential)"
        r"([\"']?\s*[:=]\s*[\"']?)([^\"'\s,&}]+)"
    )

    def format(self, record: logging.LogRecord) -> str:
        payload: dict[str, Any] = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "message": self._redact_text(record.getMessage()),
            "request_id": getattr(record, "request_id", request_id_context.get()),
        }
        for key, value in record.__dict__.items():
            if key not in self._standard and not key.startswith("_"):
                payload[key] = self._json_safe(value, key=key)
        if record.exc_info:
            payload["exception"] = self._redact_text(
                self.formatException(record.exc_info)
            )
        return json.dumps(payload, ensure_ascii=False, separators=(",", ":"))

    @classmethod
    def _json_safe(cls, value: Any, *, key: str | None = None) -> Any:
        if key is not None and cls._is_sensitive_key(key):
            return "[REDACTED]"
        if isinstance(value, dict):
            return {
                str(item_key): cls._json_safe(item, key=str(item_key))
                for item_key, item in value.items()
            }
        if isinstance(value, (list, tuple, set)):
            return [cls._json_safe(item) for item in value]
        if isinstance(value, str):
            return cls._redact_text(value)
        try:
            json.dumps(value)
            return value
        except (TypeError, ValueError):
            return cls._redact_text(str(value))

    @classmethod
    def _is_sensitive_key(cls, key: str) -> bool:
        normalized = key.strip().lower().replace("-", "_")
        return any(part in normalized for part in cls._sensitive_keys)

    @classmethod
    def _redact_text(cls, value: str) -> str:
        redacted = cls._bearer.sub("Bearer [REDACTED]", value)
        return cls._assigned_secret.sub(
            lambda match: f"{match.group(1)}{match.group(2)}[REDACTED]",
            redacted,
        )


def configure_logging(stream: TextIO | None = None) -> None:
    """Configure structured logging, optionally targeting a stdio-safe stream."""
    settings = get_settings()
    handler = logging.StreamHandler(stream or sys.stdout)
    handler.setFormatter(JsonFormatter())
    root = logging.getLogger()
    root.handlers.clear()
    root.addHandler(handler)
    root.setLevel(settings.log_level)
    for name in ("uvicorn.access", "uvicorn.error"):
        logging.getLogger(name).handlers.clear()
        logging.getLogger(name).propagate = True


def get_logger(name: str) -> logging.Logger:
    return logging.getLogger(name)