270 lines
10 KiB
Python
270 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Live Evennia -> Nexus WebSocket bridge.
|
|
|
|
Two modes:
|
|
1. Live tail: watches Evennia log files and streams parsed events to Nexus WS
|
|
2. Playback: replays a telemetry JSONL file (legacy mode)
|
|
|
|
The bridge auto-reconnects on both ends and survives Evennia restarts.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import time
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
try:
|
|
import websockets
|
|
except ImportError:
|
|
websockets = None
|
|
|
|
from nexus.evennia_event_adapter import (
|
|
audit_heartbeat,
|
|
command_executed,
|
|
player_join,
|
|
player_leave,
|
|
player_move,
|
|
)
|
|
|
|
ANSI_RE = re.compile(r"\x1b\[[0-9;]*[A-Za-z]")
|
|
# Regex patterns for log parsing
|
|
MOVE_RE = re.compile(r"AUDIT MOVE: (\w+) arrived at (.+?) from (.+)")
|
|
CMD_RE = re.compile(r"AUDIT CMD: (\w+) executed '(\w+)'(?: args: '(.*?)')?")
|
|
SESSION_START_RE = re.compile(r"AUDIT SESSION: (\w+) puppeted by (\w+)")
|
|
SESSION_END_RE = re.compile(r"AUDIT SESSION: (\w+) unpuppeted.*session (\d+)s")
|
|
LOGIN_RE = re.compile(r"Logged in: (\w+)\(account \d+\) ([\d.]+)")
|
|
LOGOUT_RE = re.compile(r"Logged out: (\w+)\(account \d+\) ([\d.]+)")
|
|
|
|
|
|
def strip_ansi(text: str) -> str:
|
|
return ANSI_RE.sub("", text or "")
|
|
|
|
|
|
class LogTailer:
|
|
"""Async file tailer that yields new lines as they appear."""
|
|
|
|
def __init__(self, path: str, poll_interval: float = 0.5):
|
|
self.path = path
|
|
self.poll_interval = poll_interval
|
|
self._offset = 0
|
|
|
|
async def tail(self):
|
|
"""Yield new lines from the file, starting from end."""
|
|
# Start at end of file
|
|
if os.path.exists(self.path):
|
|
self._offset = os.path.getsize(self.path)
|
|
|
|
while True:
|
|
try:
|
|
if not os.path.exists(self.path):
|
|
await asyncio.sleep(self.poll_interval)
|
|
continue
|
|
|
|
size = os.path.getsize(self.path)
|
|
if size < self._offset:
|
|
# File was truncated/rotated
|
|
self._offset = 0
|
|
|
|
if size > self._offset:
|
|
with open(self.path, "r") as f:
|
|
f.seek(self._offset)
|
|
for line in f:
|
|
line = line.strip()
|
|
if line:
|
|
yield line
|
|
self._offset = f.tell()
|
|
|
|
await asyncio.sleep(self.poll_interval)
|
|
except Exception as e:
|
|
print(f"[tailer] Error reading {self.path}: {e}", flush=True)
|
|
await asyncio.sleep(2)
|
|
|
|
|
|
def parse_log_line(line: str) -> Optional[dict]:
|
|
"""Parse a log line into a Nexus event, or None if not parseable."""
|
|
|
|
# Movement events
|
|
m = MOVE_RE.search(line)
|
|
if m:
|
|
return player_move(m.group(1), m.group(3), m.group(2))
|
|
|
|
# Command events
|
|
m = CMD_RE.search(line)
|
|
if m:
|
|
return command_executed(m.group(1), m.group(2), m.group(3) or "")
|
|
|
|
# Session start
|
|
m = SESSION_START_RE.search(line)
|
|
if m:
|
|
return player_join(m.group(2), m.group(1))
|
|
|
|
# Session end
|
|
m = SESSION_END_RE.search(line)
|
|
if m:
|
|
return player_leave("", m.group(1), session_duration=float(m.group(2)))
|
|
|
|
# Server login
|
|
m = LOGIN_RE.search(line)
|
|
if m:
|
|
return player_join(m.group(1), ip_address=m.group(2))
|
|
|
|
# Server logout
|
|
m = LOGOUT_RE.search(line)
|
|
if m:
|
|
return player_leave(m.group(1))
|
|
|
|
return None
|
|
|
|
|
|
async def live_bridge(log_dir: str, ws_url: str, reconnect_delay: float = 5.0):
|
|
"""
|
|
Main live bridge loop.
|
|
|
|
Tails all Evennia log files and streams parsed events to Nexus WebSocket.
|
|
Auto-reconnects on failure.
|
|
"""
|
|
log_files = [
|
|
os.path.join(log_dir, "command_audit.log"),
|
|
os.path.join(log_dir, "movement_audit.log"),
|
|
os.path.join(log_dir, "player_activity.log"),
|
|
os.path.join(log_dir, "server.log"),
|
|
]
|
|
|
|
event_queue: asyncio.Queue = asyncio.Queue(maxsize=10000)
|
|
|
|
async def tail_file(path: str):
|
|
"""Tail a single file and put events on queue."""
|
|
tailer = LogTailer(path)
|
|
async for line in tailer.tail():
|
|
event = parse_log_line(line)
|
|
if event:
|
|
try:
|
|
event_queue.put_nowait(event)
|
|
except asyncio.QueueFull:
|
|
pass # Drop oldest if queue full
|
|
|
|
async def ws_sender():
|
|
"""Send events from queue to WebSocket, with auto-reconnect."""
|
|
while True:
|
|
try:
|
|
if websockets is None:
|
|
print("[bridge] websockets not installed, logging events locally", flush=True)
|
|
while True:
|
|
event = await event_queue.get()
|
|
ts = event.get("timestamp", "")[:19]
|
|
print(f"[{ts}] {event['type']}: {json.dumps({k: v for k, v in event.items() if k not in ('type', 'timestamp')})}", flush=True)
|
|
|
|
print(f"[bridge] Connecting to {ws_url}...", flush=True)
|
|
async with websockets.connect(ws_url) as ws:
|
|
print(f"[bridge] Connected to Nexus at {ws_url}", flush=True)
|
|
while True:
|
|
event = await event_queue.get()
|
|
await ws.send(json.dumps(event))
|
|
except Exception as e:
|
|
print(f"[bridge] WebSocket error: {e}. Reconnecting in {reconnect_delay}s...", flush=True)
|
|
await asyncio.sleep(reconnect_delay)
|
|
|
|
# Start all tailers + sender
|
|
tasks = [asyncio.create_task(tail_file(f)) for f in log_files]
|
|
tasks.append(asyncio.create_task(ws_sender()))
|
|
|
|
print(f"[bridge] Live bridge started. Watching {len(log_files)} log files.", flush=True)
|
|
await asyncio.gather(*tasks)
|
|
|
|
|
|
async def playback(log_path: Path, ws_url: str):
|
|
"""Legacy mode: replay a telemetry JSONL file."""
|
|
from nexus.evennia_event_adapter import (
|
|
actor_located, command_issued, command_result,
|
|
room_snapshot, session_bound,
|
|
)
|
|
|
|
def clean_lines(text: str) -> list[str]:
|
|
text = strip_ansi(text).replace("\r", "")
|
|
return [line.strip() for line in text.split("\n") if line.strip()]
|
|
|
|
def parse_room_output(text: str):
|
|
lines = clean_lines(text)
|
|
if len(lines) < 2:
|
|
return None
|
|
title = lines[0]
|
|
desc = lines[1]
|
|
exits = []
|
|
objects = []
|
|
for line in lines[2:]:
|
|
if line.startswith("Exits:"):
|
|
raw = line.split(":", 1)[1].strip().replace(" and ", ", ")
|
|
exits = [{"key": t.strip(), "destination_id": t.strip().title(), "destination_key": t.strip().title()} for t in raw.split(",") if t.strip()]
|
|
elif line.startswith("You see:"):
|
|
raw = line.split(":", 1)[1].strip().replace(" and ", ", ")
|
|
parts = [t.strip() for t in raw.split(",") if t.strip()]
|
|
objects = [{"id": p.removeprefix("a ").removeprefix("an "), "key": p.removeprefix("a ").removeprefix("an "), "short_desc": p} for p in parts]
|
|
return {"title": title, "desc": desc, "exits": exits, "objects": objects}
|
|
|
|
def normalize_event(raw: dict, hermes_session_id: str) -> list[dict]:
|
|
out = []
|
|
event = raw.get("event")
|
|
actor = raw.get("actor", "Timmy")
|
|
timestamp = raw.get("timestamp")
|
|
if event == "connect":
|
|
out.append(session_bound(hermes_session_id, evennia_account=actor, evennia_character=actor, timestamp=timestamp))
|
|
parsed = parse_room_output(raw.get("output", ""))
|
|
if parsed:
|
|
out.append(actor_located(actor, parsed["title"], parsed["title"], timestamp=timestamp))
|
|
out.append(room_snapshot(parsed["title"], parsed["title"], parsed["desc"], exits=parsed["exits"], objects=parsed["objects"], timestamp=timestamp))
|
|
elif event == "command":
|
|
cmd = raw.get("command", "")
|
|
output = raw.get("output", "")
|
|
out.append(command_issued(hermes_session_id, actor, cmd, timestamp=timestamp))
|
|
success = not output.startswith("Command '") and not output.startswith("Could not find")
|
|
out.append(command_result(hermes_session_id, actor, cmd, strip_ansi(output), success=success, timestamp=timestamp))
|
|
parsed = parse_room_output(output)
|
|
if parsed:
|
|
out.append(actor_located(actor, parsed["title"], parsed["title"], timestamp=timestamp))
|
|
out.append(room_snapshot(parsed["title"], parsed["title"], parsed["desc"], exits=parsed["exits"], objects=parsed["objects"], timestamp=timestamp))
|
|
return out
|
|
|
|
hermes_session_id = log_path.stem
|
|
async with websockets.connect(ws_url) as ws:
|
|
for line in log_path.read_text(encoding="utf-8").splitlines():
|
|
if not line.strip():
|
|
continue
|
|
raw = json.loads(line)
|
|
for event in normalize_event(raw, hermes_session_id):
|
|
await ws.send(json.dumps(event))
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Evennia -> Nexus WebSocket Bridge")
|
|
sub = parser.add_subparsers(dest="mode")
|
|
|
|
live = sub.add_parser("live", help="Live tail Evennia logs and stream to Nexus")
|
|
live.add_argument("--log-dir", default="/root/workspace/timmy-academy/server/logs", help="Evennia logs directory")
|
|
live.add_argument("--ws", default="ws://127.0.0.1:8765", help="Nexus WebSocket URL")
|
|
|
|
replay = sub.add_parser("playback", help="Replay a telemetry JSONL file")
|
|
replay.add_argument("log_path", help="Path to Evennia telemetry JSONL")
|
|
replay.add_argument("--ws", default="ws://127.0.0.1:8765", help="Nexus WebSocket URL")
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.mode == "live":
|
|
asyncio.run(live_bridge(args.log_dir, args.ws))
|
|
elif args.mode == "playback":
|
|
asyncio.run(playback(Path(args.log_path).expanduser(), args.ws))
|
|
else:
|
|
parser.print_help()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|