""" system_logs/services.py ──────────────────────── log_event() is the one way modules record integration traffic and failures: log_event( SystemLog.MODULE_PAYMENTS, "mpesa.callback.unmatched", "Callback matched no order or POS payment", level=SystemLog.LEVEL_ERROR, checkout_request_id=..., mpesa_receipt=..., response_data=payload, alert=True, ) Every entry is also written to the rotating log file (logger "system_logs"). Logging never raises — a broken log must not break a payment. alert=True on an error emails SYSTEM_ALERT_EMAILS in the background, throttled per event + reference so a burst of failures sends one email. """ import logging import threading import traceback as traceback_mod from datetime import timedelta from typing import Any from django.conf import settings from django.core.mail import send_mail from django.urls import reverse from django.utils import timezone from system_logs.models import SystemLog logger = logging.getLogger("system_logs") _SECRET_KEYS = { "password", "passkey", "authorization", "access_token", "token", "consumer_secret", "consumer_key", "secret", } _LOG_LEVELS = { SystemLog.LEVEL_INFO: logging.INFO, SystemLog.LEVEL_WARNING: logging.WARNING, SystemLog.LEVEL_ERROR: logging.ERROR, } ALERT_THROTTLE = timedelta(minutes=10) ALERTS_PER_EVENT_PER_HOUR = 10 def scrub(data: Any) -> Any: """Drop secrets (Daraja password, tokens, keys) from a payload before storing it.""" if isinstance(data, dict): return { k: ("[removed]" if str(k).lower() in _SECRET_KEYS else scrub(v)) for k, v in data.items() } if isinstance(data, (list, tuple)): return [scrub(v) for v in data] return data def log_event( module: str, event: str, message: str, *, level: str = SystemLog.LEVEL_INFO, reference: str = "", checkout_request_id: str = "", phone_number: str = "", mpesa_receipt: str = "", related: Any = None, http_status: int | None = None, request_data: Any = None, response_data: Any = None, exc: BaseException | None = None, user: Any = None, alert: bool = False, ) -> SystemLog | None: tb = "" if exc is not None: tb = "".join(traceback_mod.format_exception(type(exc), exc, exc.__traceback__)) refs = " ".join( f"{k}={v}" for k, v in ( ("ref", reference), ("checkout", checkout_request_id), ("phone", phone_number), ("receipt", mpesa_receipt), ) if v ) logger.log(_LOG_LEVELS.get(level, logging.INFO), "[%s] %s: %s %s", module, event, message, refs, exc_info=exc if exc is not None else None) if user is not None and not getattr(user, "is_authenticated", False): user = None try: entry = SystemLog.objects.create( level=level, module=module, event=event[:80], message=message, reference=str(reference or "")[:100], checkout_request_id=str(checkout_request_id or "")[:100], phone_number=str(phone_number or "")[:20], mpesa_receipt=str(mpesa_receipt or "")[:50], related_model=related._meta.label_lower if related is not None else "", related_id=str(related.pk) if related is not None and related.pk is not None else "", http_status=http_status, request_data=scrub(request_data), response_data=scrub(response_data), traceback=tb, user=user, ) except Exception: # noqa: BLE001 — logging must never break the caller logger.exception("Could not write SystemLog entry for %s", event) return None if alert and level == SystemLog.LEVEL_ERROR: _queue_alert(entry) return entry # --------------------------------------------------------------------------- # Alert emails # --------------------------------------------------------------------------- def alert_recipients() -> list[str]: recipients = list(getattr(settings, "SYSTEM_ALERT_EMAILS", []) or []) if recipients: return recipients try: from store.models import PaybillSettings admin_email = PaybillSettings.get().admin_notification_email if admin_email: return [admin_email] except Exception: # noqa: BLE001 pass return [settings.EMAIL_HOST_USER] if getattr(settings, "EMAIL_HOST_USER", "") else [] def _should_alert(entry: SystemLog) -> bool: now = timezone.now() sent = SystemLog.objects.filter(event=entry.event, alert_sent_at__isnull=False).exclude(pk=entry.pk) if sent.filter(alert_sent_at__gte=now - timedelta(hours=1)).count() >= ALERTS_PER_EVENT_PER_HOUR: return False key_refs = { f: getattr(entry, f) for f in ("reference", "checkout_request_id", "mpesa_receipt") if getattr(entry, f) } if not key_refs: return not sent.filter(alert_sent_at__gte=now - ALERT_THROTTLE).exists() return not sent.filter(alert_sent_at__gte=now - ALERT_THROTTLE, **key_refs).exists() def _queue_alert(entry: SystemLog) -> None: if getattr(settings, "SYSTEM_LOG_ASYNC_ALERTS", True): threading.Thread(target=send_alert, args=(entry.pk,), daemon=True).start() else: send_alert(entry.pk) def send_alert(entry_id: int) -> bool: try: entry = SystemLog.objects.get(pk=entry_id) recipients = alert_recipients() if not recipients or not _should_alert(entry): return False admin_url = getattr(settings, "ADMIN_BASE_URL", "").rstrip("/") + reverse( "admin:system_logs_systemlog_change", args=[entry.pk], ) details = "\n".join( f"{label}: {value}" for label, value in ( ("Module", entry.get_module_display()), ("Event", entry.event), ("Time", entry.created_at.strftime("%Y-%m-%d %H:%M:%S")), ("Reference", entry.reference), ("M-Pesa request", entry.checkout_request_id), ("Phone", entry.phone_number), ("M-Pesa receipt", entry.mpesa_receipt), ("HTTP status", entry.http_status), ) if value ) send_mail( subject=f"[TAI Alert] {entry.get_module_display()}: {entry.message[:80]}", message=( f"{entry.message}\n\n{details}\n\n" f"Full details (request, response, error):\n{admin_url}\n\n" "Mark the entry resolved in the System Log once it's handled." ), from_email=settings.DEFAULT_FROM_EMAIL, recipient_list=recipients, ) SystemLog.objects.filter(pk=entry.pk).update(alert_sent_at=timezone.now()) return True except Exception: # noqa: BLE001 logger.exception("Could not send alert email for SystemLog %s", entry_id) return False