spb/ai-atlas

1"""Structured logging (JSON in production, human-readable in development)."""2from __future__ import annotations34import json5import logging6import sys7from datetime import UTC, datetime89from aiatlas.config import settings1011_RESERVED = {"name", "msg", "args", "levelname", "levelno", "pathname", "filename", "module", "exc_info", "exc_text", "stack_info",12             "lineno", "funcName", "created", "msecs", "relativeCreated", "thread", "threadName", "processName", "process",13             "message", "taskName"}141516class JsonFormatter(logging.Formatter):17    def __init__(self, service: str):18        super().__init__()19        self.service = service2021    def format(self, record: logging.LogRecord) -> str:22        payload = {23            "ts": datetime.now(UTC).isoformat(timespec="milliseconds"),24            "level": record.levelname,25            "logger": record.name,26            "service": self.service,27            "msg": record.getMessage(),28        }29        for key, value in record.__dict__.items():30            if key not in _RESERVED and not key.startswith("_"):31                payload[key] = value32        if record.exc_info:33            payload["exc"] = self.formatException(record.exc_info)34        return json.dumps(payload, default=str)353637class PlainFormatter(logging.Formatter):38    def format(self, record: logging.LogRecord) -> str:39        extras = {k: v for k, v in record.__dict__.items() if k not in _RESERVED and not k.startswith("_")}40        base = f"{datetime.now().strftime('%H:%M:%S')} {record.levelname:<7} {record.name}: {record.getMessage()}"41        if extras:42            base += " " + " ".join(f"{k}={v}" for k, v in extras.items())43        if record.exc_info:44            base += "\n" + self.formatException(record.exc_info)45        return base464748def setup_logging(level: int = logging.INFO, service: str = "aiatlas") -> None:49    root = logging.getLogger()50    root.handlers.clear()51    handler = logging.StreamHandler(sys.stderr)52    handler.setFormatter(JsonFormatter(service) if settings.log_json else PlainFormatter())53    root.addHandler(handler)54    root.setLevel(level)55    for noisy in ("httpx", "httpcore", "apscheduler", "asyncio", "urllib3"):56        logging.getLogger(noisy).setLevel(logging.WARNING)57