planet/backend/app/core/logging.py

from __future__ import annotations

import json
import logging
import os
import re

from collections.abc import Mapping, Sequence
from typing import Any

from app.core.request_context import get_request_id

DEFAULT_SERVICE = "backend"
DEFAULT_EVENT = "app.log"
DEFAULT_LOG_LEVEL = os.getenv("PLANET_LOG_LEVEL", "INFO").upper()
REDACTED = "[REDACTED]"
SENSITIVE_FIELD_NAMES = {
    "access_token",
    "api_key",
    "authorization",
    "cookie",
    "password",
    "refresh_token",
    "secret",
    "token",
}
SENSITIVE_TEXT_PATTERNS = (
    re.compile(r"(?i)(authorization\s*[:=]\s*)(.+)"),
    re.compile(r"(?i)(bearer\s+)([A-Za-z0-9._\-]+)"),
    re.compile(r"(?i)(token\s*[:=]\s*)(.+)"),
    re.compile(r"(?i)(password\s*[:=]\s*)(.+)"),
    re.compile(r"(?i)(cookie\s*[:=]\s*)(.+)"),
)


def sanitize_log_value(value: Any) -> Any:
    if isinstance(value, Mapping):
        return {
            str(key): (REDACTED if str(key).lower() in SENSITIVE_FIELD_NAMES else sanitize_log_value(item))
            for key, item in value.items()
        }
    if isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)):
        return [sanitize_log_value(item) for item in value]
    if isinstance(value, str):
        sanitized = value
        for pattern in SENSITIVE_TEXT_PATTERNS:
            sanitized = pattern.sub(lambda match: f"{match.group(1)}{REDACTED}", sanitized)
        return sanitized
    return value


def _normalize_context(context: Any) -> dict[str, Any]:
    if context is None:
        return {}
    if isinstance(context, Mapping):
        sanitized = sanitize_log_value(context)
        return {str(key): value for key, value in sanitized.items()}
    return {"value": sanitize_log_value(context)}


class PlanetContextFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.request_id = getattr(record, "request_id", None) or get_request_id() or "-"
        record.service = getattr(record, "service", None) or DEFAULT_SERVICE
        record.event = getattr(record, "event", None) or DEFAULT_EVENT
        record.context = _normalize_context(getattr(record, "context", None))
        record.message = sanitize_log_value(record.getMessage())
        return True


class PlanetFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        timestamp = self.formatTime(record, self.datefmt)
        level = record.levelname
        service = getattr(record, "service", DEFAULT_SERVICE)
        module_name = record.name
        event = getattr(record, "event", DEFAULT_EVENT)
        request_id = getattr(record, "request_id", "-")
        message = sanitize_log_value(record.getMessage())
        context = _normalize_context(getattr(record, "context", None))
        context_suffix = ""
        if context:
            context_suffix = f" context={json.dumps(context, ensure_ascii=False, sort_keys=True)}"
        rendered = (
            f"{timestamp} {level} service={service} module={module_name} "
            f"event={event} request_id={request_id} message={message}{context_suffix}"
        )
        if record.exc_info:
            rendered = f"{rendered}\n{self.formatException(record.exc_info)}"
        return rendered


class PlanetLoggerAdapter(logging.LoggerAdapter):
    def process(self, msg: Any, kwargs: dict[str, Any]) -> tuple[Any, dict[str, Any]]:
        extra = dict(self.extra)
        extra.update(kwargs.get("extra", {}))
        if "context" in extra:
            extra["context"] = _normalize_context(extra.get("context"))
        kwargs["extra"] = extra
        return sanitize_log_value(msg), kwargs

    def log_event(
        self,
        level: int,
        message: str,
        *,
        event: str,
        context: Mapping[str, Any] | None = None,
        **extra: Any,
    ) -> None:
        self.log(level, message, extra={"event": event, "context": context or {}, **extra})

    def debug_event(self, message: str, *, event: str, context: Mapping[str, Any] | None = None, **extra: Any) -> None:
        self.log_event(logging.DEBUG, message, event=event, context=context, **extra)

    def info_event(self, message: str, *, event: str, context: Mapping[str, Any] | None = None, **extra: Any) -> None:
        self.log_event(logging.INFO, message, event=event, context=context, **extra)

    def warning_event(self, message: str, *, event: str, context: Mapping[str, Any] | None = None, **extra: Any) -> None:
        self.log_event(logging.WARNING, message, event=event, context=context, **extra)

    def error_event(self, message: str, *, event: str, context: Mapping[str, Any] | None = None, **extra: Any) -> None:
        self.log_event(logging.ERROR, message, event=event, context=context, **extra)

    def exception_event(
        self,
        message: str,
        *,
        event: str,
        context: Mapping[str, Any] | None = None,
        **extra: Any,
    ) -> None:
        self.error(message, exc_info=True, extra={"event": event, "context": context or {}, **extra})


def get_logger(name: str, *, service: str = DEFAULT_SERVICE) -> PlanetLoggerAdapter:
    return PlanetLoggerAdapter(logging.getLogger(name), {"service": service})


def configure_logging(level: str | None = None) -> None:
    root_logger = logging.getLogger()
    if getattr(configure_logging, "_configured", False):
        if level:
            root_logger.setLevel(level.upper())
        return

    handler = logging.StreamHandler()
    handler.setFormatter(PlanetFormatter(datefmt="%Y-%m-%d %H:%M:%S"))
    handler.addFilter(PlanetContextFilter())

    root_logger.handlers.clear()
    root_logger.addHandler(handler)
    root_logger.setLevel((level or DEFAULT_LOG_LEVEL).upper())

    for logger_name in ("uvicorn", "uvicorn.error", "uvicorn.access"):
        target_logger = logging.getLogger(logger_name)
        target_logger.handlers.clear()
        target_logger.propagate = True

    logging.captureWarnings(True)
    configure_logging._configured = True