"""Interactive fiction session management via dfrotz subprocess.""" import asyncio import re from dataclasses import dataclass from pathlib import Path from typing import TYPE_CHECKING if TYPE_CHECKING: from mudlib.player import Player @dataclass class IFResponse: """Response from IF session input handling.""" output: str done: bool class IFSession: """Manages an interactive fiction session via dfrotz subprocess.""" def __init__(self, player, story_path: str, game_name: str = ""): self.player = player self.story_path = story_path self.game_name = game_name or Path(story_path).stem self.process: asyncio.subprocess.Process | None = None # data/ directory is at project root (2 levels up from src/mudlib/) self._data_dir = Path(__file__).resolve().parents[2] / "data" # Track whether we've saved (prevents double-save on quit+stop) self._saved = False @property def save_path(self) -> Path: """Return path to save file for this player/game combo.""" # Sanitize player name to prevent path traversal attacks # Account creation doesn't validate names, so defensive check here safe_name = re.sub(r"[^a-zA-Z0-9_-]", "_", self.player.name) return self._data_dir / "if_saves" / safe_name / f"{self.game_name}.qzl" def _ensure_save_dir(self) -> None: """Create save directory if it doesn't exist.""" self.save_path.parent.mkdir(parents=True, exist_ok=True) async def start(self) -> str: """Spawn dfrotz and return intro text.""" self.process = await asyncio.create_subprocess_exec( "dfrotz", "-p", # plain ASCII, no formatting "-w", "80", # screen width "-m", # no MORE prompts self.story_path, stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) # Read intro text until we see the prompt intro = await self._read_response() return intro async def handle_input(self, text: str) -> IFResponse: """Handle player input. Route to dfrotz or handle escape commands.""" # Check for escape commands (:: prefix) if text.lower() == "::quit": # Auto-save before quitting await self._do_save() return IFResponse(output="game saved.", done=True) if text.lower() == "::help": help_text = """escape commands: ::quit - exit the game ::save - save game progress ::help - show this help""" return IFResponse(output=help_text, done=False) if text.lower() == "::save": confirmation = await self._do_save() return IFResponse(output=confirmation, done=False) # Regular game input - send to dfrotz # Reset saved flag since game state has changed self._saved = False if self.process and self.process.stdin: stripped = text.strip() self.process.stdin.write(f"{stripped}\n".encode()) await self.process.stdin.drain() # Read response output = await self._read_response() return IFResponse(output=output, done=False) # Process not running return IFResponse(output="error: game not running", done=True) async def stop(self): """Terminate dfrotz process.""" if self.process and self.process.returncode is None: # Auto-save before terminating await self._do_save() self.process.terminate() await self.process.wait() async def _do_save(self) -> str: """Save game state to disk. Returns confirmation message.""" # Skip if already saved (prevents double-save on quit+stop) if self._saved: return "already saved" if not self.process or not self.process.stdin: return "error: game not running" try: # Ensure save directory exists self._ensure_save_dir() # Send "save" command to dfrotz self.process.stdin.write(b"save\n") await self.process.stdin.drain() # Read filename prompt await self._read_response() # Send save file path save_path_str = str(self.save_path) self.process.stdin.write(f"{save_path_str}\n".encode()) await self.process.stdin.drain() # Read response - might be "Ok." or "Overwrite existing file?" response = await self._read_response() # Auto-confirm overwrite if file already exists if "overwrite" in response.lower(): self.process.stdin.write(b"yes\n") await self.process.stdin.drain() response = await self._read_response() # Check for failure if "failed" in response.lower(): return "error: save failed" # Mark as saved self._saved = True return "saved." except Exception as e: return f"error: save failed ({e})" async def _do_restore(self) -> str: """Restore game state from disk. Returns game text or empty string.""" # Check if save file exists if not self.save_path.exists(): return "" if not self.process or not self.process.stdin: return "" try: # Send "restore" command to dfrotz self.process.stdin.write(b"restore\n") await self.process.stdin.drain() # Read filename prompt await self._read_response() # Send save file path save_path_str = str(self.save_path) self.process.stdin.write(f"{save_path_str}\n".encode()) await self.process.stdin.drain() # Read game text response game_text = await self._read_response() return game_text except Exception: # If restore fails, return empty string (player starts fresh) return "" async def _read_response(self) -> str: """Read dfrotz output until the '>' prompt appears.""" if not self.process or not self.process.stdout: return "" buf = b"" idle_timeout = 0.1 # return quickly once data stops flowing loop = asyncio.get_running_loop() overall_deadline = loop.time() + 5.0 try: while True: remaining = overall_deadline - loop.time() if remaining <= 0: break timeout = min(idle_timeout, remaining) try: chunk = await asyncio.wait_for( self.process.stdout.read(1024), timeout=timeout ) except TimeoutError: # No data for idle_timeout - dfrotz is done talking break if not chunk: break buf += chunk # Check for prompt at end of buffer # dfrotz prompt is always "> " (with trailing space/newline) raw = buf.decode("latin-1") has_prompt = ( raw.endswith("> ") or raw.endswith(">\n") or raw.endswith(">\r\n") ) text = raw.rstrip() # \n> is always the dfrotz prompt — strip unconditionally if text.endswith("\n>"): return text[:-2].rstrip() if text == ">": return "" # bare > (no preceding newline) — only strip when raw confirms prompt if has_prompt and text.endswith(">"): return text[:-1].rstrip() except TimeoutError: pass raw = buf.decode("latin-1") has_prompt = raw.endswith("> ") or raw.endswith(">\n") or raw.endswith(">\r\n") text = raw.rstrip() # \n> is always the dfrotz prompt — strip unconditionally if text.endswith("\n>"): return text[:-2].rstrip() if text == ">": return "" # bare > (no preceding newline) — only strip when raw confirms prompt if has_prompt and text.endswith(">"): return text[:-1].rstrip() return text async def broadcast_to_spectators(player: "Player", message: str) -> None: """Send message to all other players at the same location.""" from mudlib.entity import Entity from mudlib.zone import Zone # Use zone spatial query to find all objects at player's exact coordinates assert isinstance(player.location, Zone), "Player must be in a zone" for obj in player.location.contents_at(player.x, player.y): # Filter for Entity instances (players/mobs) and exclude self if obj is not player and isinstance(obj, Entity): await obj.send(message)