""" OriginService for loading character origin definitions from YAML files. This service reads origin configuration and converts it into Origin dataclass instances, providing caching for performance. """ import yaml from pathlib import Path from typing import Dict, List, Optional import structlog from app.models.origins import Origin, StartingLocation, StartingBonus logger = structlog.get_logger(__name__) class OriginService: """ Loads character origin definitions from YAML configuration. Origins define character backstories, starting locations, and narrative hooks that the AI DM uses to create personalized gameplay experiences. All origin definitions are stored in /app/data/origins.yaml. """ def __init__(self, data_file: Optional[str] = None): """ Initialize the origin service. Args: data_file: Path to origins YAML file. Defaults to /app/data/origins.yaml """ if data_file is None: # Default to app/data/origins.yaml relative to this file current_file = Path(__file__) app_dir = current_file.parent.parent # Go up to /app data_file = str(app_dir / "data" / "origins.yaml") self.data_file = Path(data_file) self._origins_cache: Dict[str, Origin] = {} self._all_origins_loaded = False logger.info("OriginService initialized", data_file=str(self.data_file)) def load_origin(self, origin_id: str) -> Optional[Origin]: """ Load a single origin by ID. Args: origin_id: Unique origin identifier (e.g., "soul_revenant") Returns: Origin instance or None if not found """ # Check cache first if origin_id in self._origins_cache: logger.debug("Origin loaded from cache", origin_id=origin_id) return self._origins_cache[origin_id] # Load all origins if not already loaded if not self._all_origins_loaded: self._load_all_origins() # Return from cache after loading origin = self._origins_cache.get(origin_id) if origin: logger.info("Origin loaded successfully", origin_id=origin_id) else: logger.warning("Origin not found", origin_id=origin_id) return origin def load_all_origins(self) -> List[Origin]: """ Load all origins from the data file. Returns: List of Origin instances """ if self._all_origins_loaded and self._origins_cache: logger.debug("All origins loaded from cache") return list(self._origins_cache.values()) return self._load_all_origins() def _load_all_origins(self) -> List[Origin]: """ Internal method to load all origins from YAML. Returns: List of Origin instances """ if not self.data_file.exists(): logger.error("Origins data file does not exist", data_file=str(self.data_file)) return [] try: # Load YAML file with open(self.data_file, 'r') as f: data = yaml.safe_load(f) origins_data = data.get("origins", {}) origins = [] # Parse each origin for origin_id, origin_data in origins_data.items(): try: origin = self._parse_origin_data(origin_id, origin_data) self._origins_cache[origin_id] = origin origins.append(origin) except Exception as e: logger.error("Failed to parse origin", origin_id=origin_id, error=str(e)) continue self._all_origins_loaded = True logger.info("All origins loaded successfully", count=len(origins)) return origins except Exception as e: logger.error("Failed to load origins file", error=str(e)) return [] def get_origin_by_id(self, origin_id: str) -> Optional[Origin]: """ Get an origin by ID (alias for load_origin). Args: origin_id: Unique origin identifier Returns: Origin instance or None if not found """ return self.load_origin(origin_id) def get_all_origin_ids(self) -> List[str]: """ Get a list of all available origin IDs. Returns: List of origin IDs (e.g., ["soul_revenant", "memory_thief"]) """ if not self._all_origins_loaded: self._load_all_origins() return list(self._origins_cache.keys()) def reload_origins(self) -> List[Origin]: """ Force reload all origins from disk, bypassing cache. Useful for development/testing when origin definitions change. Returns: List of Origin instances """ self.clear_cache() return self._load_all_origins() def clear_cache(self): """Clear the origins cache. Useful for testing.""" self._origins_cache.clear() self._all_origins_loaded = False logger.info("Origins cache cleared") def _parse_origin_data(self, origin_id: str, data: Dict) -> Origin: """ Parse YAML data into an Origin dataclass. Args: origin_id: The origin's unique identifier data: Dictionary loaded from YAML file Returns: Origin instance Raises: ValueError: If data is invalid or missing required fields """ # Validate required fields required_fields = ["name", "description", "starting_location"] for field in required_fields: if field not in data: raise ValueError(f"Missing required field in origin '{origin_id}': {field}") # Parse starting location location_data = data["starting_location"] starting_location = StartingLocation( id=location_data.get("id", ""), name=location_data.get("name", ""), region=location_data.get("region", ""), description=location_data.get("description", "") ) # Parse starting bonus (optional) starting_bonus = None if "starting_bonus" in data: bonus_data = data["starting_bonus"] starting_bonus = StartingBonus( trait=bonus_data.get("trait", ""), description=bonus_data.get("description", ""), effect=bonus_data.get("effect", "") ) # Parse narrative hooks (optional) narrative_hooks = data.get("narrative_hooks", []) # Create Origin instance origin = Origin( id=data.get("id", origin_id), # Use provided ID or fall back to key name=data["name"], description=data["description"], starting_location=starting_location, narrative_hooks=narrative_hooks, starting_bonus=starting_bonus ) return origin # Global instance for convenience _service_instance: Optional[OriginService] = None def get_origin_service() -> OriginService: """ Get the global OriginService instance. Returns: Singleton OriginService instance """ global _service_instance if _service_instance is None: _service_instance = OriginService() return _service_instance