""" Affix Loader Service - YAML-based affix pool loading. This service loads prefix and suffix affix definitions from YAML files, providing a data-driven approach to item generation. """ from pathlib import Path from typing import Dict, List, Optional import random import yaml from app.models.affixes import Affix from app.models.enums import AffixType, AffixTier from app.utils.logging import get_logger logger = get_logger(__file__) class AffixLoader: """ Loads and manages item affixes from YAML configuration files. This allows game designers to define affixes without touching code. Affixes are organized into prefixes.yaml and suffixes.yaml files. """ def __init__(self, data_dir: Optional[str] = None): """ Initialize the affix loader. Args: data_dir: Path to directory containing affix YAML files Defaults to /app/data/affixes/ """ if data_dir is None: # Default to app/data/affixes relative to this file current_file = Path(__file__) app_dir = current_file.parent.parent # Go up to /app data_dir = str(app_dir / "data" / "affixes") self.data_dir = Path(data_dir) self._prefix_cache: Dict[str, Affix] = {} self._suffix_cache: Dict[str, Affix] = {} self._loaded = False logger.info("AffixLoader initialized", data_dir=str(self.data_dir)) def _ensure_loaded(self) -> None: """Ensure affixes are loaded before any operation.""" if not self._loaded: self.load_all() def load_all(self) -> None: """Load all affixes from YAML files.""" if not self.data_dir.exists(): logger.warning("Affix data directory not found", path=str(self.data_dir)) return # Load prefixes prefixes_file = self.data_dir / "prefixes.yaml" if prefixes_file.exists(): self._load_affixes_from_file(prefixes_file, self._prefix_cache) # Load suffixes suffixes_file = self.data_dir / "suffixes.yaml" if suffixes_file.exists(): self._load_affixes_from_file(suffixes_file, self._suffix_cache) self._loaded = True logger.info( "Affixes loaded", prefix_count=len(self._prefix_cache), suffix_count=len(self._suffix_cache) ) def _load_affixes_from_file( self, yaml_file: Path, cache: Dict[str, Affix] ) -> None: """ Load affixes from a YAML file into the cache. Args: yaml_file: Path to the YAML file cache: Cache dictionary to populate """ try: with open(yaml_file, 'r') as f: data = yaml.safe_load(f) # Get the top-level key (prefixes or suffixes) affix_key = "prefixes" if "prefixes" in data else "suffixes" affixes_data = data.get(affix_key, {}) for affix_id, affix_data in affixes_data.items(): # Ensure affix_id is set affix_data["affix_id"] = affix_id # Set defaults for missing optional fields affix_data.setdefault("stat_bonuses", {}) affix_data.setdefault("defense_bonus", 0) affix_data.setdefault("resistance_bonus", 0) affix_data.setdefault("damage_bonus", 0) affix_data.setdefault("elemental_ratio", 0.0) affix_data.setdefault("crit_chance_bonus", 0.0) affix_data.setdefault("crit_multiplier_bonus", 0.0) affix_data.setdefault("allowed_item_types", []) affix_data.setdefault("required_rarity", None) affix = Affix.from_dict(affix_data) cache[affix.affix_id] = affix logger.debug( "Affixes loaded from file", file=str(yaml_file), count=len(affixes_data) ) except Exception as e: logger.error( "Failed to load affix file", file=str(yaml_file), error=str(e) ) def get_affix(self, affix_id: str) -> Optional[Affix]: """ Get a specific affix by ID. Args: affix_id: Unique affix identifier Returns: Affix instance or None if not found """ self._ensure_loaded() if affix_id in self._prefix_cache: return self._prefix_cache[affix_id] if affix_id in self._suffix_cache: return self._suffix_cache[affix_id] return None def get_eligible_prefixes( self, item_type: str, rarity: str, tier: Optional[AffixTier] = None ) -> List[Affix]: """ Get all prefixes eligible for an item. Args: item_type: Type of item ("weapon", "armor") rarity: Item rarity ("rare", "epic", "legendary") tier: Optional tier filter Returns: List of eligible Affix instances """ self._ensure_loaded() eligible = [] for affix in self._prefix_cache.values(): # Check if affix can apply to this item if not affix.can_apply_to(item_type, rarity): continue # Apply tier filter if specified if tier and affix.tier != tier: continue eligible.append(affix) return eligible def get_eligible_suffixes( self, item_type: str, rarity: str, tier: Optional[AffixTier] = None ) -> List[Affix]: """ Get all suffixes eligible for an item. Args: item_type: Type of item ("weapon", "armor") rarity: Item rarity ("rare", "epic", "legendary") tier: Optional tier filter Returns: List of eligible Affix instances """ self._ensure_loaded() eligible = [] for affix in self._suffix_cache.values(): # Check if affix can apply to this item if not affix.can_apply_to(item_type, rarity): continue # Apply tier filter if specified if tier and affix.tier != tier: continue eligible.append(affix) return eligible def get_random_prefix( self, item_type: str, rarity: str, tier: Optional[AffixTier] = None, exclude_ids: Optional[List[str]] = None ) -> Optional[Affix]: """ Get a random eligible prefix. Args: item_type: Type of item ("weapon", "armor") rarity: Item rarity tier: Optional tier filter exclude_ids: Affix IDs to exclude (for avoiding duplicates) Returns: Random eligible Affix or None if none available """ eligible = self.get_eligible_prefixes(item_type, rarity, tier) # Filter out excluded IDs if exclude_ids: eligible = [a for a in eligible if a.affix_id not in exclude_ids] if not eligible: return None return random.choice(eligible) def get_random_suffix( self, item_type: str, rarity: str, tier: Optional[AffixTier] = None, exclude_ids: Optional[List[str]] = None ) -> Optional[Affix]: """ Get a random eligible suffix. Args: item_type: Type of item ("weapon", "armor") rarity: Item rarity tier: Optional tier filter exclude_ids: Affix IDs to exclude (for avoiding duplicates) Returns: Random eligible Affix or None if none available """ eligible = self.get_eligible_suffixes(item_type, rarity, tier) # Filter out excluded IDs if exclude_ids: eligible = [a for a in eligible if a.affix_id not in exclude_ids] if not eligible: return None return random.choice(eligible) def get_all_prefixes(self) -> Dict[str, Affix]: """ Get all cached prefixes. Returns: Dictionary of prefix affixes """ self._ensure_loaded() return self._prefix_cache.copy() def get_all_suffixes(self) -> Dict[str, Affix]: """ Get all cached suffixes. Returns: Dictionary of suffix affixes """ self._ensure_loaded() return self._suffix_cache.copy() def clear_cache(self) -> None: """Clear the affix cache, forcing reload on next access.""" self._prefix_cache.clear() self._suffix_cache.clear() self._loaded = False logger.debug("Affix cache cleared") # Global instance for convenience _loader_instance: Optional[AffixLoader] = None def get_affix_loader() -> AffixLoader: """ Get the global AffixLoader instance. Returns: Singleton AffixLoader instance """ global _loader_instance if _loader_instance is None: _loader_instance = AffixLoader() return _loader_instance