"""Central textual audit log for user-driven warehouse operations.""" from __future__ import annotations import json import logging from pathlib import Path from typing import Any from user_session import UserSession AUDIT_LOG_PATH = Path(__file__).with_name("warehouse_audit.log") _LOGGER = logging.getLogger("warehouse_audit") _LOGGER_CONFIGURED = False def _configure_logger() -> None: """Configure the append-only textual audit logger once.""" global _LOGGER_CONFIGURED if _LOGGER_CONFIGURED: return _LOGGER.setLevel(logging.INFO) _LOGGER.propagate = False handler = logging.FileHandler(AUDIT_LOG_PATH, encoding="utf-8") handler.setFormatter(logging.Formatter("%(asctime)s | %(message)s")) _LOGGER.addHandler(handler) _LOGGER_CONFIGURED = True def _user_label(session: UserSession | None) -> str: """Render the user identity consistently in the audit trail.""" if session is None: return "anonymous" return f"{session.login or 'anonymous'}#{session.operator_id}" def _payload( *, session: UserSession | None, module: str, action: str, outcome: str, target: str = "", details: dict[str, Any] | None = None, ) -> str: """Serialize one audit event as a compact text line.""" base = { "user": _user_label(session), "module": module, "action": action, "outcome": outcome, "target": target, "details": details or {}, } return json.dumps(base, ensure_ascii=False, default=str) def log_user_action( session: UserSession | None, *, module: str, action: str, outcome: str, target: str = "", details: dict[str, Any] | None = None, ) -> None: """Write one user action event to the textual audit file.""" _configure_logger() _LOGGER.info( _payload( session=session, module=module, action=action, outcome=outcome, target=target, details=details, ) ) def log_session_event( session: UserSession | None, *, action: str, outcome: str, details: dict[str, Any] | None = None, ) -> None: """Write one session lifecycle event to the textual audit file.""" log_user_action( session, module="session", action=action, outcome=outcome, details=details, )