50 lines
1.4 KiB
Python
50 lines
1.4 KiB
Python
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)
|