from __future__ import annotations from enum import Enum from uuid import UUID, uuid4 from datetime import datetime, timezone from dataclasses import dataclass, field, asdict from typing import Callable, List, Optional, Dict, Any, Iterable, Mapping, TYPE_CHECKING from app.models.enums import HeroClass, Race from app.models.primitives import Attributes, Resources, Collections from app.utils.merging import merge_catalogs, merge_sheets if TYPE_CHECKING: from app.utils.hero_catalog import HeroDataRegistry from app.utils.race_catalog import RaceDataRegistry # ----------------------------- # Small utilities # ----------------------------- def _utcnow_iso() -> str: """Return current UTC time as ISO 8601 string.""" return datetime.now(timezone.utc).isoformat() def _clamp(value: int, lo: int, hi: int) -> int: """Clamp integer between lo and hi.""" return max(lo, min(hi, value)) # ----------------------------- # Component dataclasses # ----------------------------- @dataclass class Progress: """Economy, experience, level, and other progression flags.""" level: int = 1 gold: int = 50 xp: int = 0 xp_next_level: int = 100 # will be recomputed on init needs_sleep: int = 0 fame: int = 0 alignment: int = 0 # -100..+100 works nicely battlecount: int = 0 def gain_xp(self, amount: int, xp_curve: Callable[[int, int], int]) -> int: """ Add XP and compute how many levels are gained using the provided xp curve. Returns the number of levels gained. """ amount = max(0, int(amount)) if amount == 0: return 0 self.xp += amount levels_gained = 0 # Level up loop (handles large XP chunks) while self.xp >= self.xp_next_level: self.level += 1 levels_gained += 1 self.xp_next_level = xp_curve(self.xp, self.level) return levels_gained @dataclass class Location: current_town: str = "Brindlemark" current_nation: str = "Brindlemark" current_continent: str = "Brindlemark" @dataclass class Equipment: """References to equipped items (IDs or simple names for now).""" weapon: Optional[str] = None armor: Optional[str] = None # ----------------------------- # Main model # ----------------------------- @dataclass class Hero: # Identity hero_id: UUID = field(default_factory=uuid4) name: str = "Unnamed" # Build race: Optional[Race] = None hero_class: Optional[HeroClass] = None # Stats & state attributes: Attributes = field(default_factory=Attributes) resources: Resources = field(default_factory=Resources) progress: Progress = field(default_factory=Progress) location: Location = field(default_factory=Location) equipment: Equipment = field(default_factory=Equipment) collections: Collections = field(default_factory=Collections) # RP bits origin_story: str = "" # Auditing & versioning version: int = 1 created_at: str = field(default_factory=_utcnow_iso) updated_at: str = field(default_factory=_utcnow_iso) # ------------------------- # Lifecycle helpers # ------------------------- def initialize(self, xp_curve: Callable[[int, int], int]) -> None: """ Call once after creation or after loading if you need to recompute derived fields. For example: set xp_next_level using your curve. """ self.progress.xp_next_level = xp_curve(self.progress.xp, self.progress.level) self.touch() def touch(self) -> None: """Update the `updated_at` timestamp.""" self.updated_at = _utcnow_iso() @classmethod def from_race_and_class( cls, name: str, race: Race | None, hero_class: HeroClass | None, class_registry: HeroDataRegistry | None = None, race_registry: RaceDataRegistry | None = None, *, # Optional global registries (if you’ve split catalogs): skill_registry: Any | None = None, spell_registry: Any | None = None, ) -> Hero: """ Factory that builds a Hero and seeds starting skills/spells from race/class. Validates starting IDs against global registries when provided, otherwise falls back to merged local catalogs from the race/class sheets. """ # Always define locals so later checks are safe skills_catalog: Dict[str, Any] = {} spells_catalog: Dict[str, Any] = {} race_sheet = None cls_sheet = None if class_registry and race_registry and hero_class and race: cls_sheet = class_registry.for_class(hero_class) race_sheet = race_registry.for_race(race) # Merge any local catalogs for fallback validation skills_catalog, spells_catalog = merge_catalogs( race_sheet.skills, race_sheet.spells, cls_sheet.skills, cls_sheet.spells ) starting_skills: list[str] = [] starting_spells: list[str] = [] if race_sheet: starting_skills.extend(race_sheet.starting_skills or []) starting_spells.extend(race_sheet.starting_spells or []) if cls_sheet: starting_skills.extend(cls_sheet.starting_skills or []) starting_spells.extend(cls_sheet.starting_spells or []) # Dedup while preserving order def _dedup(seq: list[str]) -> list[str]: seen: set[str] = set() out: list[str] = [] for x in seq: if x not in seen: out.append(x) seen.add(x) return out starting_skills = _dedup(starting_skills) starting_spells = _dedup(starting_spells) # Validate against global registries first, fallback to locals _validate_ids_exist( starting_skills, skills_catalog, "skill", owner=f"{race.value if race else 'unknown'}+{hero_class.value if hero_class else 'unknown'}", registry=skill_registry ) _validate_ids_exist( starting_spells, spells_catalog, "spell", owner=f"{race.value if race else 'unknown'}+{hero_class.value if hero_class else 'unknown'}", registry=spell_registry ) hero = cls(name=name, race=race, hero_class=hero_class) hero.collections.skills = starting_skills hero.collections.spells = starting_spells hero.touch() return hero # ------------------------- # Progression # ------------------------- def gain_xp( self, amount: int, xp_curve: Callable[[int, int], int], class_registry: HeroDataRegistry | None = None, race_registry: RaceDataRegistry | None = None, *, # Optional global registries (preferred for validation) skill_registry: Any | None = None, spell_registry: Any | None = None, ) -> int: """ Gain XP and apply level-ups via the provided xp_curve. If registries are provided, also grant any class/race skills & spells unlocked by the newly reached level(s). Returns number of levels gained. """ prev_level = self.progress.level levels = self.progress.gain_xp(amount, xp_curve) # Always prepare empty local catalogs; may be filled below skills_catalog: Dict[str, Any] = {} spells_catalog: Dict[str, Any] = {} if levels > 0 and class_registry and race_registry and self.hero_class and self.race: cls_sheet = class_registry.for_class(self.hero_class) race_sheet = race_registry.for_race(self.race) # Optional: merge local catalogs for fallback validation skills_catalog, spells_catalog = merge_catalogs( race_sheet.skills, race_sheet.spells, cls_sheet.skills, cls_sheet.spells ) unlocked_skills: list[str] = [] unlocked_spells: list[str] = [] now_lvl = self.progress.level # Unlock skills whose min_level is newly reached for nodes in (cls_sheet.skill_trees or {}).values(): for ref in nodes: if ref.min_level is not None and prev_level < ref.min_level <= now_lvl: unlocked_skills.append(ref.id) # Unlock spells by each level reached in the jump (handles 1→6, etc.) for L in range(prev_level + 1, now_lvl + 1): unlocked_spells.extend((cls_sheet.spells_by_level.get(L) or [])) # Validate using global registries when provided, fallback to locals owner = f"{self.race.value}+{self.hero_class.value}" _validate_ids_exist(unlocked_skills, skills_catalog, "skill", owner, registry=skill_registry) _validate_ids_exist(unlocked_spells, spells_catalog, "spell", owner, registry=spell_registry) # Grant only new ones (dedupe) _unique_extend(self.collections.skills, unlocked_skills) _unique_extend(self.collections.spells, unlocked_spells) # Your existing resource gains hp_gain = 8 * levels mp_gain = 6 * levels self.resources.maxhp += hp_gain self.resources.maxmp += mp_gain self.resources.hp = self.resources.maxhp self.resources.mp = self.resources.maxmp self.touch() return levels # ------------------------- # (De)serialization helpers # ------------------------- def to_dict(self) -> Dict[str, Any]: """ Convert to a pure-JSON-friendly dict (Enums → value, UUID → str). """ def _enum_to_value(obj: Any) -> Any: if isinstance(obj, Enum): return obj.value if isinstance(obj, UUID): return str(obj) return obj raw = asdict(self) # Fix enum & UUID fields: raw["hero_id"] = str(self.hero_id) raw["hero_class"] = self.hero_class.value if self.hero_class else None return _walk_map(raw, _enum_to_value) @classmethod def from_dict(cls, data: Dict[str, Any]) -> Hero: """ Build Hero from a dict (reverse of to_dict). Accepts hero_class as string or None; hero_id as str or UUID. """ data = dict(data) # shallow copy # Parse UUID hero_id_val = data.get("hero_id") if hero_id_val and not isinstance(hero_id_val, UUID): data["hero_id"] = UUID(hero_id_val) # Parse enum hc = data.get("hero_class") if isinstance(hc, str) and hc: data["hero_class"] = HeroClass(hc) elif hc is None: data["hero_class"] = None # Rebuild nested dataclasses if passed as dicts for k, typ in { "attributes": Attributes, "resources": Resources, "progress": Progress, "location": Location, "equipment": Equipment, "collections": Collections, }.items(): if isinstance(data.get(k), dict): data[k] = typ(**data[k]) hero: Hero = cls(**data) # type: ignore[arg-type] return hero # ----------------------------- # Helpers # ----------------------------- def _walk_map(obj: Any, f) -> Any: """Recursively map Python structures with function f applied to leaf values as needed.""" if isinstance(obj, dict): return {k: _walk_map(v, f) for k, v in obj.items()} if isinstance(obj, list): return [_walk_map(x, f) for x in obj] return f(obj) def _unique_extend(target: list[str], new_items: Iterable[str]) -> list[str]: """Append only items not already present; return the list for convenience.""" seen = set(target) for x in new_items: if x not in seen: target.append(x) seen.add(x) return target def default_xp_curve(current_xp: int, level: int) -> int: """ Replace this with your real XP curve (e.g., g_utils.get_xp_for_next_level). Simple example: mild exponential growth. """ base = 100 return base + int(50 * (level ** 1.5)) + (level * 25) def _validate_ids_exist( ids: Iterable[str], local_catalog: Mapping[str, Any] | None, kind: str, owner: str, *, registry: Any | None = None, # e.g., SkillRegistry or SpellRegistry with .get() ) -> None: """ Validate that each id is present either in the provided local_catalog OR resolvable via the optional global registry (registry.get(id) not None). """ local_catalog = local_catalog or {} reg_get = getattr(registry, "get", None) if registry is not None else None missing: list[str] = [] for _id in ids: in_local = _id in local_catalog in_global = bool(reg_get(_id)) if reg_get else False if not (in_local or in_global): missing.append(_id) if missing: raise ValueError(f"{owner}: undefined {kind} id(s): {', '.join(missing)}")