"""Evennia -> Nexus event normalization — v2 with full audit event types.""" from __future__ import annotations from datetime import datetime, timezone def _ts(value: str | None = None) -> str: return value or datetime.now(timezone.utc).isoformat() # ── Session Events ────────────────────────────────────────── def player_join(account: str, character: str = "", ip_address: str = "", timestamp: str | None = None) -> dict: return { "type": "evennia.player_join", "account": account, "character": character, "ip_address": ip_address, "timestamp": _ts(timestamp), } def player_leave(account: str, character: str = "", reason: str = "quit", session_duration: float = 0, timestamp: str | None = None) -> dict: return { "type": "evennia.player_leave", "account": account, "character": character, "reason": reason, "session_duration_seconds": session_duration, "timestamp": _ts(timestamp), } def session_bound(hermes_session_id: str, evennia_account: str = "Timmy", evennia_character: str = "Timmy", timestamp: str | None = None) -> dict: return { "type": "evennia.session_bound", "hermes_session_id": hermes_session_id, "evennia_account": evennia_account, "evennia_character": evennia_character, "timestamp": _ts(timestamp), } # ── Movement Events ───────────────────────────────────────── def player_move(character: str, from_room: str, to_room: str, timestamp: str | None = None) -> dict: return { "type": "evennia.player_move", "character": character, "from_room": from_room, "to_room": to_room, "timestamp": _ts(timestamp), } def actor_located(actor_id: str, room_key: str, room_name: str | None = None, timestamp: str | None = None) -> dict: return { "type": "evennia.actor_located", "actor_id": actor_id, "room_id": room_key, "room_key": room_key, "room_name": room_name or room_key, "timestamp": _ts(timestamp), } def room_snapshot(room_key: str, title: str, desc: str, exits: list[dict] | None = None, objects: list[dict] | None = None, occupants: list[dict] | None = None, timestamp: str | None = None) -> dict: return { "type": "evennia.room_snapshot", "room_id": room_key, "room_key": room_key, "title": title, "desc": desc, "exits": exits or [], "objects": objects or [], "occupants": occupants or [], "timestamp": _ts(timestamp), } # ── Command Events ────────────────────────────────────────── def command_executed(character: str, command: str, args: str = "", success: bool = True, timestamp: str | None = None) -> dict: return { "type": "evennia.command_executed", "character": character, "command": command, "args": args, "success": success, "timestamp": _ts(timestamp), } def command_issued(hermes_session_id: str, actor_id: str, command_text: str, timestamp: str | None = None) -> dict: return { "type": "evennia.command_issued", "hermes_session_id": hermes_session_id, "actor_id": actor_id, "command_text": command_text, "timestamp": _ts(timestamp), } def command_result(hermes_session_id: str, actor_id: str, command_text: str, output_text: str, success: bool = True, timestamp: str | None = None) -> dict: return { "type": "evennia.command_result", "hermes_session_id": hermes_session_id, "actor_id": actor_id, "command_text": command_text, "output_text": output_text, "success": success, "timestamp": _ts(timestamp), } # ── Audit Summary ─────────────────────────────────────────── def audit_heartbeat(characters: list[dict], online_count: int, total_commands: int, total_movements: int, timestamp: str | None = None) -> dict: return { "type": "evennia.audit_heartbeat", "characters": characters, "online_count": online_count, "total_commands": total_commands, "total_movements": total_movements, "timestamp": _ts(timestamp), }