Checkpoint before ghost pallet cleanup workflow

audit_log.py

@@ -0,0 +1,102 @@
+"""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,
+    )