Magent/backend/app/logging_config.py

import contextvars
import json
import logging
import os
from logging.handlers import RotatingFileHandler
from typing import Any, Mapping, Optional
from urllib.parse import parse_qs

REQUEST_ID_CONTEXT: contextvars.ContextVar[str] = contextvars.ContextVar(
    "magent_request_id", default="-"
)

_SENSITIVE_KEYWORDS = (
    "api_key",
    "authorization",
    "cert",
    "cookie",
    "jwt",
    "key",
    "pass",
    "password",
    "pem",
    "private",
    "secret",
    "session",
    "signature",
    "token",
)
_MAX_BODY_BYTES = 4096


class RequestContextFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.request_id = REQUEST_ID_CONTEXT.get("-")
        return True


def bind_request_id(request_id: str) -> contextvars.Token[str]:
    return REQUEST_ID_CONTEXT.set(request_id or "-")


def reset_request_id(token: contextvars.Token[str]) -> None:
    REQUEST_ID_CONTEXT.reset(token)


def current_request_id() -> str:
    return REQUEST_ID_CONTEXT.get("-")


def _is_sensitive_key(key: str) -> bool:
    lowered = key.strip().lower()
    return any(marker in lowered for marker in _SENSITIVE_KEYWORDS)


def _redact_scalar(value: Any) -> Any:
    if value is None or isinstance(value, (int, float, bool)):
        return value
    text = str(value)
    if len(text) <= 4:
        return "***"
    return f"{text[:2]}***{text[-2:]}"


def sanitize_value(value: Any, *, key_hint: Optional[str] = None, depth: int = 0) -> Any:
    if key_hint and _is_sensitive_key(key_hint):
        return _redact_scalar(value)
    if value is None or isinstance(value, (bool, int, float)):
        return value
    if isinstance(value, bytes):
        return f"<bytes:{len(value)}>"
    if isinstance(value, str):
        return value if len(value) <= 512 else f"{value[:509]}..."
    if depth >= 3:
        return f"<{type(value).__name__}>"
    if isinstance(value, Mapping):
        return {
            str(key): sanitize_value(item, key_hint=str(key), depth=depth + 1)
            for key, item in value.items()
        }
    if isinstance(value, (list, tuple, set)):
        return [sanitize_value(item, depth=depth + 1) for item in list(value)[:20]]
    if hasattr(value, "model_dump"):
        try:
            return sanitize_value(value.model_dump(), depth=depth + 1)
        except Exception:
            return f"<{type(value).__name__}>"
    return str(value)


def sanitize_headers(headers: Mapping[str, Any]) -> dict[str, Any]:
    return {
        str(key).lower(): sanitize_value(value, key_hint=str(key))
        for key, value in headers.items()
    }


def summarize_http_body(body: bytes, content_type: Optional[str]) -> Any:
    if not body:
        return None
    normalized = (content_type or "").split(";")[0].strip().lower()
    if normalized == "application/json":
        preview = body[:_MAX_BODY_BYTES]
        try:
            payload = json.loads(preview.decode("utf-8"))
            summary = sanitize_value(payload)
            if len(body) > _MAX_BODY_BYTES:
                return {"truncated": True, "bytes": len(body), "payload": summary}
            return summary
        except Exception:
            pass
    if normalized == "application/x-www-form-urlencoded":
        try:
            parsed = parse_qs(body.decode("utf-8"), keep_blank_values=True)
            compact = {
                key: value[0] if len(value) == 1 else value
                for key, value in parsed.items()
            }
            return sanitize_value(compact)
        except Exception:
            pass
    if normalized.startswith("multipart/"):
        return {"content_type": normalized, "bytes": len(body)}
    preview = body[: min(len(body), 256)].decode("utf-8", errors="replace")
    return {
        "content_type": normalized or "unknown",
        "bytes": len(body),
        "preview": preview if len(body) <= 256 else f"{preview}...",
    }


def _coerce_level(level_name: Optional[str], fallback: int) -> int:
    if not level_name:
        return fallback
    return getattr(logging, str(level_name).upper(), fallback)


def configure_logging(
    log_level: Optional[str],
    log_file: Optional[str],
    *,
    log_file_max_bytes: int = 20_000_000,
    log_file_backup_count: int = 10,
    log_http_client_level: Optional[str] = "INFO",
    log_background_sync_level: Optional[str] = "INFO",
) -> None:
    level_name = (log_level or "INFO").upper()
    level = getattr(logging, level_name, logging.INFO)

    handlers: list[logging.Handler] = []
    stream_handler = logging.StreamHandler()
    handlers.append(stream_handler)

    if log_file:
        log_path = log_file
        if not os.path.isabs(log_path):
            log_path = os.path.join(os.getcwd(), log_path)
        os.makedirs(os.path.dirname(log_path), exist_ok=True)
        file_handler = RotatingFileHandler(
            log_path,
            maxBytes=max(1_000_000, int(log_file_max_bytes or 20_000_000)),
            backupCount=max(1, int(log_file_backup_count or 10)),
            encoding="utf-8",
        )
        handlers.append(file_handler)

    context_filter = RequestContextFilter()
    formatter = logging.Formatter(
        fmt="%(asctime)s | %(levelname)s | %(name)s | request_id=%(request_id)s | %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
    )
    for handler in handlers:
        handler.addFilter(context_filter)
        handler.setFormatter(formatter)

    root = logging.getLogger()
    for handler in list(root.handlers):
        root.removeHandler(handler)
    for handler in handlers:
        root.addHandler(handler)
    root.setLevel(level)

    logging.getLogger("uvicorn").setLevel(level)
    logging.getLogger("uvicorn.error").setLevel(level)
    logging.getLogger("uvicorn.access").setLevel(logging.WARNING)
    http_client_level = _coerce_level(log_http_client_level, logging.DEBUG)
    background_sync_level = _coerce_level(log_background_sync_level, logging.INFO)
    logging.getLogger("app.clients.base").setLevel(http_client_level)
    logging.getLogger("app.routers.requests").setLevel(background_sync_level)
    logging.getLogger("httpx").setLevel(logging.WARNING if level > logging.DEBUG else logging.INFO)
    logging.getLogger("httpcore").setLevel(logging.WARNING)