- Add card import feature with fuzzy matching - Implement deck CRUD and management endpoints - Add user data APIs for groups, networks, preferences, activity, replays - Create comprehensive API documentation (API_DOCUMENTATION.md) - Add ENDPOINT_AUDIT.md for endpoint verification - Update documentation (README, ROADMAP, state.json) - Update architecture blueprint and Cockatrice analysis - All Phase 2 deliverables complete and documented
191 lines
5.6 KiB
Python
191 lines
5.6 KiB
Python
"""Card search service with filters."""
|
|
from typing import List, Dict, Any, Optional
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select, or_, and_
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from app.models.mtg_models import MtgCard, MtgSet
|
|
from app.models.mirror_models import MtgCardMirror
|
|
|
|
|
|
class CardSearchService:
|
|
"""Card search service with filters."""
|
|
|
|
@staticmethod
|
|
async def search_cards(
|
|
db: AsyncSession,
|
|
query: str,
|
|
card_type: Optional[str] = None,
|
|
set_code: Optional[str] = None,
|
|
color: Optional[str] = None,
|
|
limit: int = 100,
|
|
offset: int = 0,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Search cards with filters.
|
|
|
|
Args:
|
|
db: Database session
|
|
query: Search query (name, type, mana cost)
|
|
card_type: Filter by card type
|
|
set_code: Filter by set code
|
|
color: Filter by card color
|
|
limit: Maximum results
|
|
offset: Number of results to skip
|
|
|
|
Returns:
|
|
Dictionary with search results and metadata
|
|
"""
|
|
# Build conditions
|
|
conditions = [
|
|
or_(
|
|
MtgCard.name.ilike(f"%{query}%"),
|
|
MtgCard.type_line.ilike(f"%{query}%"),
|
|
MtgCard.mana_cost.ilike(f"%{query}%"),
|
|
)
|
|
]
|
|
|
|
if card_type:
|
|
conditions.append(MtgCard.type_line.ilike(f"%{card_type}%"))
|
|
|
|
if set_code:
|
|
conditions.append(MtgCard.set_code == set_code)
|
|
|
|
if color:
|
|
# Parse color string (e.g., "WU" for white-blue)
|
|
colors = [c.strip() for c in color.upper().split(",")]
|
|
for c in colors:
|
|
if c in ["W", "U", "B", "R", "G"]:
|
|
conditions.append(MtgCard.colors.ilike(f"%{c}%"))
|
|
|
|
# Count total results
|
|
count_stmt = select(MtgCard).where(*conditions)
|
|
total_result = await db.execute(count_stmt)
|
|
total = len(total_result.scalars().all())
|
|
|
|
# Fetch results with pagination
|
|
stmt = select(MtgCard).where(*conditions).offset(offset).limit(limit)
|
|
result = await db.execute(stmt)
|
|
cards = result.scalars().all()
|
|
|
|
# Format results
|
|
card_list = []
|
|
for card in cards:
|
|
card_data = {
|
|
"id": card.id,
|
|
"name": card.name,
|
|
"mana_cost": card.mana_cost,
|
|
"type_line": card.type_line,
|
|
"oracle_text": card.oracle_text,
|
|
"power": card.power,
|
|
"toughness": card.toughness,
|
|
"rarity": card.rarity,
|
|
"layout": card.layout,
|
|
"colors": card.colors,
|
|
"set_code": card.set_code,
|
|
"set_name": card.set_name,
|
|
}
|
|
card_list.append(card_data)
|
|
|
|
return {
|
|
"cards": card_list,
|
|
"total": total,
|
|
"page": offset // limit + 1,
|
|
"page_size": limit,
|
|
"total_pages": (total + limit - 1) // limit,
|
|
}
|
|
|
|
@staticmethod
|
|
async def get_card_by_id(db: AsyncSession, card_id: int) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Get a card by its ID.
|
|
|
|
Args:
|
|
db: Database session
|
|
card_id: Card ID
|
|
|
|
Returns:
|
|
Card data dictionary or None
|
|
"""
|
|
stmt = select(MtgCard).where(MtgCard.id == card_id)
|
|
result = await db.execute(stmt)
|
|
card = result.scalar_one_or_none()
|
|
|
|
if not card:
|
|
return None
|
|
|
|
return {
|
|
"id": card.id,
|
|
"name": card.name,
|
|
"mana_cost": card.mana_cost,
|
|
"type_line": card.type_line,
|
|
"oracle_text": card.oracle_text,
|
|
"power": card.power,
|
|
"toughness": card.toughness,
|
|
"rarity": card.rarity,
|
|
"layout": card.layout,
|
|
"colors": card.colors,
|
|
"set_code": card.set_code,
|
|
"set_name": card.set_name,
|
|
"identifiers": card.identifiers,
|
|
"images": card.images,
|
|
}
|
|
|
|
@staticmethod
|
|
async def get_sets(db: AsyncSession) -> List[Dict[str, Any]]:
|
|
"""
|
|
Get all available sets.
|
|
|
|
Args:
|
|
db: Database session
|
|
|
|
Returns:
|
|
List of set data dictionaries
|
|
"""
|
|
stmt = select(MtgSet).order_by(MtgSet.name)
|
|
result = await db.execute(stmt)
|
|
sets = result.scalars().all()
|
|
|
|
return [
|
|
{
|
|
"id": s.id,
|
|
"name": s.name,
|
|
"code": s.code,
|
|
"release_date": s.release_date,
|
|
"card_count": s.card_count,
|
|
}
|
|
for s in sets
|
|
]
|
|
|
|
@staticmethod
|
|
async def get_card_types(db: AsyncSession) -> List[str]:
|
|
"""
|
|
Get all unique card types.
|
|
|
|
Args:
|
|
db: Database session
|
|
|
|
Returns:
|
|
List of unique card types
|
|
"""
|
|
stmt = select(MtgCard.type_line).distinct()
|
|
result = await db.execute(stmt)
|
|
types = result.scalars().all()
|
|
return list(types)
|
|
|
|
@staticmethod
|
|
async def get_card_rarities(db: AsyncSession) -> List[str]:
|
|
"""
|
|
Get all unique card rarities.
|
|
|
|
Args:
|
|
db: Database session
|
|
|
|
Returns:
|
|
List of unique rarities
|
|
"""
|
|
stmt = select(MtgCard.rarity).distinct()
|
|
result = await db.execute(stmt)
|
|
rarities = result.scalars().all()
|
|
return list(rarities)
|