from __future__ import annotations import hashlib import json import logging import os import uuid from datetime import datetime, timezone from typing import Any _LOGGER_NAME = "sofiia" _JSON_ENABLED = os.getenv("SOFIIA_LOG_JSON", "1").strip().lower() not in {"0", "false", "no", "off"} _LOG_LEVEL_NAME = os.getenv("SOFIIA_LOG_LEVEL", "INFO").strip().upper() _LOG_LEVEL = getattr(logging, _LOG_LEVEL_NAME, logging.INFO) def configure_sofiia_logger() -> logging.Logger: logger = logging.getLogger(_LOGGER_NAME) logger.setLevel(_LOG_LEVEL) return logger def hash_idempotency_key(key: str) -> str: cleaned = (key or "").strip() if not cleaned: return "" return hashlib.sha256(cleaned.encode("utf-8")).hexdigest()[:12] def get_request_id(request: Any) -> str: headers = getattr(request, "headers", None) req_id = "" if headers is not None: req_id = (headers.get("X-Request-Id") or "").strip() return req_id[:64] if req_id else uuid.uuid4().hex def log_event(event: str, **fields: Any) -> None: logger = logging.getLogger(_LOGGER_NAME) payload = { "event": event, "ts": datetime.now(timezone.utc).isoformat(), } payload.update({k: v for k, v in fields.items() if v is not None}) if _JSON_ENABLED: logger.info(json.dumps(payload, ensure_ascii=True, separators=(",", ":"), default=str)) else: logger.info("%s %s", event, payload)