# Trouve-KA — logging structuré JSON # Author: Simon-Pierre Boucher # Contact: contact@spboucher.ai """Logs structurés (JSON une ligne) avec contexte crawl_id/url_id/domain_id/worker_id.""" import json import logging import sys import time from typing import Any class JsonFormatter(logging.Formatter): def format(self, record: logging.LogRecord) -> str: payload: dict[str, Any] = { "ts": round(time.time(), 3), "level": record.levelname.lower(), "logger": record.name, "msg": record.getMessage(), } extra = getattr(record, "ctx", None) if isinstance(extra, dict): payload.update(extra) if record.exc_info and record.exc_info[0] is not None: payload["exc"] = self.formatException(record.exc_info) return json.dumps(payload, ensure_ascii=False, default=str) def get_logger(name: str) -> logging.Logger: """Retourne un logger JSON prêt à l'emploi. Utiliser `log.info("msg", extra={"ctx": {...}})`.""" logger = logging.getLogger(name) if not logger.handlers: handler = logging.StreamHandler(sys.stdout) handler.setFormatter(JsonFormatter()) logger.addHandler(handler) logger.setLevel(logging.INFO) logger.propagate = False return logger