934aa4c443
- Rename data/cards/ to data/raw/ for scraped data - Add data/definitions/ as authoritative card data source - Add convert_cards.py script to transform raw -> definitions - Generate 378 card definitions (344 Pokemon, 24 Trainers, 10 Energy) - Add CardService for loading and querying card definitions - In-memory indexes for fast lookups by type, set, pokemon_type - search() with multiple filter criteria - get_all_cards() for GameEngine integration - Add SetInfo model for set metadata - Update Attack model with damage_display field for variable damage - Update CardDefinition with image_path, illustrator, flavor_text - Add 45 tests (21 converter + 24 CardService) - Update scraper output path to data/raw/ Card data is JSON-authoritative (no database) to support offline fork goal.
380 lines
12 KiB
Python
380 lines
12 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://cdn.mantimon.com/cards"
|
|
|
|
|
|
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 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.
|
|
ValueError: If any card definition is invalid.
|
|
"""
|
|
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),
|
|
)
|
|
)
|
|
|
|
# Load Pokemon cards
|
|
await self._load_card_type("pokemon", CardType.POKEMON)
|
|
|
|
# Load Trainer cards
|
|
await self._load_card_type("trainer", CardType.TRAINER)
|
|
|
|
# Load Energy cards
|
|
await self._load_energy_cards()
|
|
|
|
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) -> None:
|
|
"""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.
|
|
"""
|
|
type_dir = self._definitions_dir / subdir
|
|
if not type_dir.exists():
|
|
logger.debug(f"Directory {type_dir} does not exist, skipping")
|
|
return
|
|
|
|
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"):
|
|
card = await self._load_card_file(card_file)
|
|
if card:
|
|
self._index_card(card)
|
|
|
|
async def _load_energy_cards(self) -> None:
|
|
"""Load energy cards from the energy subdirectory."""
|
|
energy_dir = self._definitions_dir / "energy"
|
|
if not energy_dir.exists():
|
|
logger.debug(f"Directory {energy_dir} does not exist, skipping")
|
|
return
|
|
|
|
# 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":
|
|
card = await self._load_card_file(item)
|
|
if card:
|
|
self._index_card(card)
|
|
elif item.is_dir():
|
|
for card_file in item.glob("*.json"):
|
|
card = await self._load_card_file(card_file)
|
|
if card:
|
|
self._index_card(card)
|
|
|
|
async def _load_card_file(self, file_path: Path) -> CardDefinition | None:
|
|
"""Load and validate a single card definition file.
|
|
|
|
Args:
|
|
file_path: Path to the JSON file.
|
|
|
|
Returns:
|
|
CardDefinition if valid, None if loading failed.
|
|
"""
|
|
try:
|
|
with open(file_path) as f:
|
|
card_data = json.load(f)
|
|
return CardDefinition.model_validate(card_data)
|
|
except Exception as e:
|
|
logger.error(f"Failed to load card from {file_path}: {e}")
|
|
return None
|
|
|
|
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
|