""" DungeonMaster AI - Adventure Loader Loads and manages adventure JSON files for pre-made scenarios. """ from __future__ import annotations import json import logging from pathlib import Path from typing import TYPE_CHECKING from .models import ( AdventureData, EncounterData, NPCInfo, SceneInfo, ) if TYPE_CHECKING: from .game_state_manager import GameStateManager logger = logging.getLogger(__name__) # Default adventures directory relative to project root DEFAULT_ADVENTURES_DIR = Path(__file__).parent.parent.parent / "adventures" class AdventureLoader: """ Loads and manages adventure JSON files. Adventures are pre-made scenarios with scenes, NPCs, encounters, and victory conditions. This class handles loading, parsing, and initializing games from adventure files. """ def __init__( self, adventures_dir: Path | str | None = None, ) -> None: """ Initialize the adventure loader. Args: adventures_dir: Directory containing adventure JSON files. Defaults to 'adventures/' in project root. """ if adventures_dir is None: self._adventures_dir = DEFAULT_ADVENTURES_DIR else: self._adventures_dir = Path(adventures_dir) # Cache loaded adventures self._cache: dict[str, AdventureData] = {} logger.debug(f"AdventureLoader initialized with dir: {self._adventures_dir}") @property def adventures_dir(self) -> Path: """Get the adventures directory path.""" return self._adventures_dir # ========================================================================= # Adventure Discovery # ========================================================================= def list_adventures(self) -> list[dict[str, str]]: """ List all available adventures. Returns: List of adventure info dicts with keys: - name: Adventure name - file: Filename - description: Adventure description - difficulty: Difficulty level - estimated_time: Estimated play time """ adventures: list[dict[str, str]] = [] if not self._adventures_dir.exists(): logger.warning(f"Adventures directory not found: {self._adventures_dir}") return adventures for json_file in self._adventures_dir.glob("*.json"): # Skip sample_characters directory if json_file.is_dir(): continue try: with open(json_file, encoding="utf-8") as f: data = json.load(f) metadata = data.get("metadata", {}) adventures.append( { "name": metadata.get("name", json_file.stem), "file": json_file.name, "description": metadata.get("description", ""), "difficulty": metadata.get("difficulty", "medium"), "estimated_time": metadata.get("estimated_time", "Unknown"), "recommended_level": str( metadata.get("recommended_level", 1) ), } ) except (json.JSONDecodeError, OSError) as e: logger.warning(f"Failed to load adventure {json_file}: {e}") continue return adventures # ========================================================================= # Adventure Loading # ========================================================================= def load(self, adventure_name: str) -> AdventureData | None: """ Load an adventure by name. Checks cache first, then loads from file. Args: adventure_name: Name of adventure or filename (with/without .json) Returns: AdventureData if found, None otherwise """ # Check cache if adventure_name in self._cache: return self._cache[adventure_name] # Find the file json_file = self._find_adventure_file(adventure_name) if json_file is None: logger.warning(f"Adventure not found: {adventure_name}") return None # Load and parse try: with open(json_file, encoding="utf-8") as f: data = json.load(f) adventure = AdventureData.from_json(data) # Cache by both name and filename self._cache[adventure_name] = adventure self._cache[adventure.metadata.name] = adventure self._cache[json_file.stem] = adventure logger.info(f"Loaded adventure: {adventure.metadata.name}") return adventure except (json.JSONDecodeError, OSError) as e: logger.error(f"Failed to load adventure {json_file}: {e}") return None except Exception as e: logger.error(f"Failed to parse adventure {json_file}: {e}") return None def _find_adventure_file(self, adventure_name: str) -> Path | None: """ Find an adventure file by name. Args: adventure_name: Name or filename to search for Returns: Path to file if found, None otherwise """ if not self._adventures_dir.exists(): return None # Try exact filename exact_path = self._adventures_dir / adventure_name if exact_path.exists(): return exact_path # Try with .json extension json_path = self._adventures_dir / f"{adventure_name}.json" if json_path.exists(): return json_path # Try matching by metadata name for json_file in self._adventures_dir.glob("*.json"): try: with open(json_file, encoding="utf-8") as f: data = json.load(f) metadata = data.get("metadata", {}) if metadata.get("name", "").lower() == adventure_name.lower(): return json_file except (json.JSONDecodeError, OSError): continue return None # ========================================================================= # Game Initialization # ========================================================================= async def initialize_game( self, manager: GameStateManager, adventure_name: str, ) -> bool: """ Initialize a game session with an adventure. Args: manager: GameStateManager to initialize adventure_name: Name of adventure to load Returns: True if successful, False otherwise """ adventure = self.load(adventure_name) if adventure is None: return False try: # Start new game with adventure name await manager.new_game(adventure=adventure.metadata.name) # Set starting location starting_scene = adventure.starting_scene scene_id = str(starting_scene.get("scene_id", "")) scene_info = self.get_scene(adventure_name, scene_id) if scene_info: manager.set_location(scene_info.name, scene_info) else: # Fallback to basic location from starting scene manager.set_location( str(starting_scene.get("scene_id", "Unknown")), None, ) # Add all NPCs to manager for npc_data in adventure.npcs: if isinstance(npc_data, dict): npc = self._parse_npc(npc_data) if npc: manager.add_known_npc(npc) # Set initial story flags manager.set_story_flag("adventure_started", True) manager.set_story_flag("adventure_name", adventure.metadata.name) logger.info( f"Initialized game with adventure: {adventure.metadata.name}" ) return True except Exception as e: logger.error(f"Failed to initialize game with adventure: {e}") return False # ========================================================================= # Content Retrieval # ========================================================================= def get_starting_narrative(self, adventure_name: str) -> str: """ Get the opening narrative for an adventure. Args: adventure_name: Name of adventure Returns: Opening narrative text, or empty string if not found """ adventure = self.load(adventure_name) if adventure is None: return "" starting_scene = adventure.starting_scene return str(starting_scene.get("narrative", "")) def get_scene( self, adventure_name: str, scene_id: str, ) -> SceneInfo | None: """ Get a scene from an adventure. Args: adventure_name: Name of adventure scene_id: Scene ID to find Returns: SceneInfo if found, None otherwise """ adventure = self.load(adventure_name) if adventure is None: return None scene_data = adventure.get_scene(scene_id) if scene_data is None: return None return self._parse_scene(scene_data) def get_encounter( self, adventure_name: str, encounter_id: str, ) -> EncounterData | None: """ Get an encounter from an adventure. Args: adventure_name: Name of adventure encounter_id: Encounter ID to find Returns: EncounterData if found, None otherwise """ adventure = self.load(adventure_name) if adventure is None: return None return adventure.get_encounter(encounter_id) def get_npc( self, adventure_name: str, npc_id: str, ) -> NPCInfo | None: """ Get an NPC from an adventure. Args: adventure_name: Name of adventure npc_id: NPC ID to find Returns: NPCInfo if found, None otherwise """ adventure = self.load(adventure_name) if adventure is None: return None npc_data = adventure.get_npc(npc_id) if npc_data is None: return None return self._parse_npc(npc_data) def get_all_scenes(self, adventure_name: str) -> list[SceneInfo]: """ Get all scenes from an adventure. Args: adventure_name: Name of adventure Returns: List of all SceneInfo objects """ adventure = self.load(adventure_name) if adventure is None: return [] scenes: list[SceneInfo] = [] for scene_data in adventure.scenes: if isinstance(scene_data, dict): scene = self._parse_scene(scene_data) if scene: scenes.append(scene) return scenes def get_all_npcs(self, adventure_name: str) -> list[NPCInfo]: """ Get all NPCs from an adventure. Args: adventure_name: Name of adventure Returns: List of all NPCInfo objects """ adventure = self.load(adventure_name) if adventure is None: return [] npcs: list[NPCInfo] = [] for npc_data in adventure.npcs: if isinstance(npc_data, dict): npc = self._parse_npc(npc_data) if npc: npcs.append(npc) return npcs # ========================================================================= # Parsing Helpers # ========================================================================= def _parse_scene(self, data: dict[str, object]) -> SceneInfo | None: """ Parse a scene dict into SceneInfo. Args: data: Raw scene data Returns: SceneInfo if valid, None otherwise """ try: scene_id = str(data.get("scene_id", "")) if not scene_id: return None # Extract sensory details details = data.get("details", {}) sensory: dict[str, str] = {} if isinstance(details, dict): sensory_raw = details.get("sensory", {}) if isinstance(sensory_raw, dict): for key, value in sensory_raw.items(): sensory[str(key)] = str(value) # Extract searchable objects searchable: list[dict[str, object]] = [] if isinstance(details, dict): searchable_raw = details.get("searchable", []) if isinstance(searchable_raw, list): searchable = [ dict(obj) for obj in searchable_raw if isinstance(obj, dict) ] # Extract encounter ID encounter = data.get("encounter") encounter_id: str | None = None if isinstance(encounter, dict): encounter_id = str(encounter.get("encounter_id", "")) elif isinstance(encounter, str): encounter_id = encounter return SceneInfo( scene_id=scene_id, name=str(data.get("name", scene_id)), description=str(data.get("description", "")), sensory_details=sensory, exits=dict(data.get("exits", {})), npcs_present=list(data.get("npcs_present", [])), items=list(data.get("items", [])), encounter_id=encounter_id if encounter_id else None, searchable_objects=searchable, ) except Exception as e: logger.warning(f"Failed to parse scene: {e}") return None def _parse_npc(self, data: dict[str, object]) -> NPCInfo | None: """ Parse an NPC dict into NPCInfo. Args: data: Raw NPC data Returns: NPCInfo if valid, None otherwise """ try: npc_id = str(data.get("npc_id", "")) if not npc_id: return None return NPCInfo( npc_id=npc_id, name=str(data.get("name", "Unknown")), description=str(data.get("description", "")), personality=str(data.get("personality", "")), voice_profile=str(data.get("voice_profile", "dm")), dialogue_hooks=list(data.get("dialogue_hooks", [])), monster_stat_block=data.get("monster_stat_block"), relationship="hostile" if data.get("monster_stat_block") else "neutral", ) except Exception as e: logger.warning(f"Failed to parse NPC: {e}") return None # ========================================================================= # Cache Management # ========================================================================= def clear_cache(self) -> None: """Clear the adventure cache.""" self._cache.clear() logger.debug("Adventure cache cleared") def is_cached(self, adventure_name: str) -> bool: """ Check if an adventure is cached. Args: adventure_name: Name of adventure Returns: True if cached, False otherwise """ return adventure_name in self._cache def preload(self, adventure_names: list[str]) -> int: """ Preload multiple adventures into cache. Args: adventure_names: List of adventure names to load Returns: Number of successfully loaded adventures """ loaded = 0 for name in adventure_names: if self.load(name) is not None: loaded += 1 return loaded def __len__(self) -> int: """Return number of cached adventures.""" return len(self._cache)