55e02ceb21
Three changes to fail fast instead of silently degrading: 1. GameService.create_game: Raise GameCreationError when energy card definition not found instead of logging warning and continuing. A deck with missing energy cards is fundamentally broken. 2. CardService.load_all: Collect all card file load failures and raise CardServiceLoadError at end with comprehensive error report. Prevents startup with partial card data that causes cryptic runtime errors. New exceptions: CardLoadError, CardServiceLoadError 3. GameStateManager.recover_active_games: Return RecoveryResult dataclass with recovered count, failed game IDs with error messages, and total. Enables proper monitoring and alerting for corrupted game state. Tests added for energy card error case. Existing tests updated for new RecoveryResult return type. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
438 lines
14 KiB
Python
438 lines
14 KiB
Python
"""Card service for loading and serving card definitions.
|
|
|
|
This service loads card definitions from JSON files in data/definitions/ and
|
|
serves them to the game engine and other parts of the application.
|
|
|
|
Cards are loaded into memory at startup for fast lookups. The service maintains
|
|
several indexes for efficient querying by type, set, and other criteria.
|
|
|
|
Usage:
|
|
card_service = CardService()
|
|
await card_service.load_all()
|
|
|
|
# Get a single card
|
|
pikachu = card_service.get_card("a1-025-pikachu")
|
|
|
|
# Get all cards for game engine
|
|
registry = card_service.get_all_cards()
|
|
|
|
# Search cards
|
|
grass_pokemon = card_service.search(
|
|
card_type=CardType.POKEMON,
|
|
pokemon_type=EnergyType.GRASS,
|
|
)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import TYPE_CHECKING
|
|
|
|
from pydantic import BaseModel
|
|
|
|
from app.core.enums import CardType, EnergyType, PokemonStage, PokemonVariant
|
|
from app.core.models.card import CardDefinition
|
|
|
|
if TYPE_CHECKING:
|
|
pass
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# TODO: Update CDN_BASE_URL when CDN is configured
|
|
CDN_BASE_URL = "https://mantipocket.s3.us-east-1.amazonaws.com/card-images/"
|
|
|
|
|
|
class SetInfo(BaseModel):
|
|
"""Metadata about a card set.
|
|
|
|
Attributes:
|
|
code: Unique identifier for the set (e.g., "a1").
|
|
name: Display name of the set (e.g., "Genetic Apex").
|
|
card_count: Total number of cards in this set.
|
|
"""
|
|
|
|
code: str
|
|
name: str
|
|
card_count: int
|
|
|
|
|
|
class CardLoadError(Exception):
|
|
"""Raised when a card definition file fails to load.
|
|
|
|
Attributes:
|
|
file_path: Path to the file that failed to load.
|
|
reason: Description of why loading failed.
|
|
"""
|
|
|
|
def __init__(self, file_path: str, reason: str) -> None:
|
|
self.file_path = file_path
|
|
self.reason = reason
|
|
super().__init__(f"Failed to load card from {file_path}: {reason}")
|
|
|
|
|
|
class CardServiceLoadError(Exception):
|
|
"""Raised when CardService fails to load one or more card files.
|
|
|
|
Collects all individual load failures for reporting.
|
|
|
|
Attributes:
|
|
failures: List of (file_path, error_message) tuples.
|
|
"""
|
|
|
|
def __init__(self, failures: list[tuple[str, str]]) -> None:
|
|
self.failures = failures
|
|
file_list = "\n ".join(f"{path}: {err}" for path, err in failures[:5])
|
|
more = f"\n ... and {len(failures) - 5} more" if len(failures) > 5 else ""
|
|
super().__init__(f"Failed to load {len(failures)} card definition(s):\n {file_list}{more}")
|
|
|
|
|
|
class CardService:
|
|
"""Load and serve card definitions from JSON files.
|
|
|
|
This service loads all card definitions into memory at startup for fast
|
|
lookups. Cards are immutable after loading - all lookups are in-memory.
|
|
|
|
The service maintains several indexes for efficient querying:
|
|
- By ID: O(1) lookup for any card
|
|
- By type: O(1) to get all Pokemon, Trainers, or Energy
|
|
- By set: O(1) to get all cards from a set
|
|
- By Pokemon type: O(1) to get all Pokemon of a specific type
|
|
|
|
Attributes:
|
|
definitions_dir: Path to the definitions directory.
|
|
"""
|
|
|
|
def __init__(self, definitions_dir: Path | None = None) -> None:
|
|
"""Initialize the card service.
|
|
|
|
Args:
|
|
definitions_dir: Path to the data/definitions directory.
|
|
If None, uses the default location relative to this file.
|
|
"""
|
|
if definitions_dir is None:
|
|
# Default: backend/data/definitions/
|
|
self._definitions_dir = Path(__file__).parent.parent.parent / "data" / "definitions"
|
|
else:
|
|
self._definitions_dir = definitions_dir
|
|
|
|
# Primary storage
|
|
self._cards: dict[str, CardDefinition] = {}
|
|
|
|
# Indexes for fast querying
|
|
self._by_type: dict[CardType, list[str]] = {t: [] for t in CardType}
|
|
self._by_set: dict[str, list[str]] = {}
|
|
self._by_pokemon_type: dict[EnergyType, list[str]] = {t: [] for t in EnergyType}
|
|
|
|
# Set metadata
|
|
self._sets: list[SetInfo] = []
|
|
|
|
# Load state
|
|
self._loaded: bool = False
|
|
|
|
async def load_all(self) -> None:
|
|
"""Load all card definitions into memory.
|
|
|
|
This should be called once at application startup. After loading,
|
|
all card lookups are synchronous in-memory operations.
|
|
|
|
Raises:
|
|
FileNotFoundError: If the definitions directory doesn't exist.
|
|
CardServiceLoadError: If any card definition files fail to load.
|
|
"""
|
|
if self._loaded:
|
|
logger.warning("CardService.load_all() called but cards already loaded")
|
|
return
|
|
|
|
if not self._definitions_dir.exists():
|
|
raise FileNotFoundError(f"Definitions directory not found: {self._definitions_dir}")
|
|
|
|
# Load index file for set metadata
|
|
index_path = self._definitions_dir / "_index.json"
|
|
if index_path.exists():
|
|
with open(index_path) as f:
|
|
index_data = json.load(f)
|
|
for set_code, set_info in index_data.get("sets", {}).items():
|
|
self._sets.append(
|
|
SetInfo(
|
|
code=set_code,
|
|
name=set_info.get("name", set_code),
|
|
card_count=set_info.get("card_count", 0),
|
|
)
|
|
)
|
|
|
|
# Collect all failures across card types
|
|
all_failures: list[tuple[str, str]] = []
|
|
|
|
# Load Pokemon cards
|
|
all_failures.extend(await self._load_card_type("pokemon", CardType.POKEMON))
|
|
|
|
# Load Trainer cards
|
|
all_failures.extend(await self._load_card_type("trainer", CardType.TRAINER))
|
|
|
|
# Load Energy cards
|
|
all_failures.extend(await self._load_energy_cards())
|
|
|
|
# Fail fast if any card files couldn't be loaded
|
|
if all_failures:
|
|
raise CardServiceLoadError(all_failures)
|
|
|
|
self._loaded = True
|
|
logger.info(
|
|
f"CardService loaded {len(self._cards)} cards "
|
|
f"({len(self._by_type[CardType.POKEMON])} Pokemon, "
|
|
f"{len(self._by_type[CardType.TRAINER])} Trainers, "
|
|
f"{len(self._by_type[CardType.ENERGY])} Energy)"
|
|
)
|
|
|
|
async def _load_card_type(self, subdir: str, card_type: CardType) -> list[tuple[str, str]]:
|
|
"""Load all cards of a specific type from a subdirectory.
|
|
|
|
Args:
|
|
subdir: Subdirectory name (e.g., "pokemon", "trainer").
|
|
card_type: The CardType these cards should be.
|
|
|
|
Returns:
|
|
List of (file_path, error_message) tuples for any files that failed.
|
|
"""
|
|
failures: list[tuple[str, str]] = []
|
|
type_dir = self._definitions_dir / subdir
|
|
if not type_dir.exists():
|
|
logger.debug(f"Directory {type_dir} does not exist, skipping")
|
|
return failures
|
|
|
|
for set_dir in type_dir.iterdir():
|
|
if not set_dir.is_dir():
|
|
continue
|
|
|
|
set_code = set_dir.name
|
|
if set_code not in self._by_set:
|
|
self._by_set[set_code] = []
|
|
|
|
for card_file in set_dir.glob("*.json"):
|
|
try:
|
|
card = await self._load_card_file(card_file)
|
|
self._index_card(card)
|
|
except CardLoadError as e:
|
|
failures.append((e.file_path, e.reason))
|
|
|
|
return failures
|
|
|
|
async def _load_energy_cards(self) -> list[tuple[str, str]]:
|
|
"""Load energy cards from the energy subdirectory.
|
|
|
|
Returns:
|
|
List of (file_path, error_message) tuples for any files that failed.
|
|
"""
|
|
failures: list[tuple[str, str]] = []
|
|
energy_dir = self._definitions_dir / "energy"
|
|
if not energy_dir.exists():
|
|
logger.debug(f"Directory {energy_dir} does not exist, skipping")
|
|
return failures
|
|
|
|
# Energy can be in subdirectories (e.g., energy/basic/) or directly in energy/
|
|
for item in energy_dir.iterdir():
|
|
if item.is_file() and item.suffix == ".json":
|
|
try:
|
|
card = await self._load_card_file(item)
|
|
self._index_card(card)
|
|
except CardLoadError as e:
|
|
failures.append((e.file_path, e.reason))
|
|
elif item.is_dir():
|
|
for card_file in item.glob("*.json"):
|
|
try:
|
|
card = await self._load_card_file(card_file)
|
|
self._index_card(card)
|
|
except CardLoadError as e:
|
|
failures.append((e.file_path, e.reason))
|
|
|
|
return failures
|
|
|
|
async def _load_card_file(self, file_path: Path) -> CardDefinition:
|
|
"""Load and validate a single card definition file.
|
|
|
|
Args:
|
|
file_path: Path to the JSON file.
|
|
|
|
Returns:
|
|
CardDefinition if valid.
|
|
|
|
Raises:
|
|
CardLoadError: If loading or validation fails.
|
|
"""
|
|
try:
|
|
with open(file_path) as f:
|
|
card_data = json.load(f)
|
|
return CardDefinition.model_validate(card_data)
|
|
except Exception as e:
|
|
raise CardLoadError(str(file_path), str(e)) from e
|
|
|
|
def _index_card(self, card: CardDefinition) -> None:
|
|
"""Add a card to all indexes.
|
|
|
|
Args:
|
|
card: The CardDefinition to index.
|
|
"""
|
|
# Primary storage
|
|
self._cards[card.id] = card
|
|
|
|
# Type index
|
|
self._by_type[card.card_type].append(card.id)
|
|
|
|
# Set index
|
|
if card.set_id:
|
|
if card.set_id not in self._by_set:
|
|
self._by_set[card.set_id] = []
|
|
self._by_set[card.set_id].append(card.id)
|
|
|
|
# Pokemon type index (for Pokemon cards only)
|
|
if card.card_type == CardType.POKEMON and card.pokemon_type:
|
|
self._by_pokemon_type[card.pokemon_type].append(card.id)
|
|
|
|
def get_card(self, card_id: str) -> CardDefinition | None:
|
|
"""Get a card by its ID.
|
|
|
|
Args:
|
|
card_id: Unique card identifier (e.g., "a1-025-pikachu").
|
|
|
|
Returns:
|
|
CardDefinition if found, None otherwise.
|
|
"""
|
|
return self._cards.get(card_id)
|
|
|
|
def get_all_cards(self) -> dict[str, CardDefinition]:
|
|
"""Get all cards as a registry dict.
|
|
|
|
This is the format expected by GameEngine.create_game().
|
|
|
|
Returns:
|
|
Dictionary mapping card ID to CardDefinition.
|
|
"""
|
|
return self._cards.copy()
|
|
|
|
def get_cards_by_ids(self, card_ids: list[str]) -> list[CardDefinition]:
|
|
"""Get multiple cards by their IDs.
|
|
|
|
Args:
|
|
card_ids: List of card IDs to retrieve.
|
|
|
|
Returns:
|
|
List of CardDefinitions in the same order as input.
|
|
|
|
Raises:
|
|
KeyError: If any card ID is not found.
|
|
"""
|
|
result = []
|
|
for card_id in card_ids:
|
|
card = self._cards.get(card_id)
|
|
if card is None:
|
|
raise KeyError(f"Card not found: {card_id}")
|
|
result.append(card)
|
|
return result
|
|
|
|
def search(
|
|
self,
|
|
name: str | None = None,
|
|
card_type: CardType | None = None,
|
|
pokemon_type: EnergyType | None = None,
|
|
set_id: str | None = None,
|
|
stage: PokemonStage | None = None,
|
|
variant: PokemonVariant | None = None,
|
|
) -> list[CardDefinition]:
|
|
"""Search cards by various criteria.
|
|
|
|
All provided criteria are AND-ed together. Omitted criteria match all cards.
|
|
|
|
Args:
|
|
name: Partial name match (case-insensitive).
|
|
card_type: Filter by card type (Pokemon, Trainer, Energy).
|
|
pokemon_type: Filter by Pokemon energy type (e.g., GRASS, FIRE).
|
|
set_id: Filter by set code (e.g., "a1").
|
|
stage: Filter by Pokemon stage (BASIC, STAGE_1, STAGE_2).
|
|
variant: Filter by Pokemon variant (NORMAL, EX, etc.).
|
|
|
|
Returns:
|
|
List of matching CardDefinitions.
|
|
"""
|
|
# Start with the smallest candidate set based on indexed filters
|
|
if pokemon_type is not None:
|
|
candidate_ids = set(self._by_pokemon_type.get(pokemon_type, []))
|
|
elif card_type is not None:
|
|
candidate_ids = set(self._by_type.get(card_type, []))
|
|
elif set_id is not None:
|
|
candidate_ids = set(self._by_set.get(set_id, []))
|
|
else:
|
|
candidate_ids = set(self._cards.keys())
|
|
|
|
results = []
|
|
for card_id in candidate_ids:
|
|
card = self._cards[card_id]
|
|
|
|
# Apply all filters
|
|
if card_type is not None and card.card_type != card_type:
|
|
continue
|
|
if pokemon_type is not None and card.pokemon_type != pokemon_type:
|
|
continue
|
|
if set_id is not None and card.set_id != set_id:
|
|
continue
|
|
if stage is not None and card.stage != stage:
|
|
continue
|
|
if variant is not None and card.variant != variant:
|
|
continue
|
|
if name is not None and name.lower() not in card.name.lower():
|
|
continue
|
|
|
|
results.append(card)
|
|
|
|
return results
|
|
|
|
def get_set_cards(self, set_id: str) -> list[CardDefinition]:
|
|
"""Get all cards from a specific set.
|
|
|
|
Args:
|
|
set_id: Set code (e.g., "a1", "a1a").
|
|
|
|
Returns:
|
|
List of CardDefinitions from that set.
|
|
"""
|
|
card_ids = self._by_set.get(set_id, [])
|
|
return [self._cards[card_id] for card_id in card_ids]
|
|
|
|
def get_sets(self) -> list[SetInfo]:
|
|
"""Get list of available sets with metadata.
|
|
|
|
Returns:
|
|
List of SetInfo objects.
|
|
"""
|
|
return self._sets.copy()
|
|
|
|
@property
|
|
def card_count(self) -> int:
|
|
"""Total number of loaded cards."""
|
|
return len(self._cards)
|
|
|
|
@property
|
|
def is_loaded(self) -> bool:
|
|
"""Whether cards have been loaded."""
|
|
return self._loaded
|
|
|
|
|
|
# Singleton instance for application use
|
|
_card_service: CardService | None = None
|
|
|
|
|
|
def get_card_service() -> CardService:
|
|
"""Get the global CardService instance.
|
|
|
|
Creates the instance if it doesn't exist. Note that load_all() must still
|
|
be called before using the service.
|
|
|
|
Returns:
|
|
The global CardService instance.
|
|
"""
|
|
global _card_service
|
|
if _card_service is None:
|
|
_card_service = CardService()
|
|
return _card_service
|