"""Deck suggestion service.""" from typing import List, Dict, Any, Optional, Tuple from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, or_, and_, func, case from sqlalchemy.orm import selectinload from app.models.user_deck import UserDeck, UserDeckCard, CardSuggestion from app.models.mtg_models import MtgCard from app.models.mirror_models import MtgCardMirror from app.services.fuzzy_card_matcher import FuzzyCardMatcher class DeckSuggestionService: """Deck suggestion service.""" @staticmethod async def suggest_cards( db: AsyncSession, deck_id: int, limit: int = 20, ) -> List[Dict[str, Any]]: """ Suggest similar cards for a deck. Args: db: Database session deck_id: Deck ID to suggest cards for limit: Maximum number of suggestions Returns: List of suggested card data dictionaries """ # Get deck cards deck_cards_stmt = select(UserDeckCard).where(UserDeckCard.deck_id == deck_id) deck_cards_result = await db.execute(deck_cards_stmt) deck_cards = deck_cards_result.scalars().all() if not deck_cards: return [] # Get card IDs in the deck card_ids = [dc.card_id for dc in deck_cards] # Get deck card details card_details_stmt = select(MtgCard).where(MtgCard.id.in_(card_ids)) card_details_result = await db.execute(card_details_stmt) deck_card_details = card_details_result.scalars().all() # Analyze deck characteristics deck_types = set() deck_colors = set() deck_sets = set() deck_mana_costs = [] for card in deck_card_details: if card.type_line: # Extract main type (e.g., "Creature" from "Creature — Elf") main_type = card.type_line.split(" — ")[0].strip() deck_types.add(main_type) if card.colors: deck_colors.update(card.colors) if card.set_code: deck_sets.add(card.set_code) if card.mana_cost: deck_mana_costs.append(card.mana_cost) # Search for similar cards suggestions = [] # Strategy 1: Same type, not already in deck if deck_types: type_conditions = [MtgCard.type_line.ilike(f"%{t}%") for t in deck_types] type_search_stmt = select(MtgCard).where( or_(*type_conditions), MtgCard.id.notin_(card_ids), ) type_results = await db.execute(type_search_stmt) type_cards = type_results.scalars().all() for card in type_cards: suggestions.append({ "card": card, "reason": "same_type", "confidence": 0.8, }) # Strategy 2: Same color, not already in deck if deck_colors: color_conditions = [] for color in deck_colors: color_conditions.append(MtgCard.colors.ilike(f"%{color}%")) color_search_stmt = select(MtgCard).where( or_(*color_conditions), MtgCard.id.notin_(card_ids), ) color_results = await db.execute(color_search_stmt) color_cards = color_results.scalars().all() for card in color_cards: # Check if already added if not any(s["card"].id == card.id for s in suggestions): suggestions.append({ "card": card, "reason": "same_color", "confidence": 0.7, }) # Strategy 3: Same set, not already in deck if deck_sets: set_search_stmt = select(MtgCard).where( MtgCard.set_code.in_(list(deck_sets)), MtgCard.id.notin_(card_ids), ) set_results = await db.execute(set_search_stmt) set_cards = set_results.scalars().all() for card in set_cards: # Check if already added if not any(s["card"].id == card.id for s in suggestions): suggestions.append({ "card": card, "reason": "same_set", "confidence": 0.6, }) # Sort by confidence and limit results suggestions.sort(key=lambda x: x["confidence"], reverse=True) suggestions = suggestions[:limit] # Format results result = [] for suggestion in suggestions: card = suggestion["card"] result.append({ "card_id": card.id, "name": card.name, "mana_cost": card.mana_cost, "type_line": card.type_line, "colors": card.colors, "reason": suggestion["reason"], "confidence": suggestion["confidence"], }) return result @staticmethod async def add_suggestion( db: AsyncSession, deck_id: int, card_id: int, source_card_id: Optional[int] = None, suggestion_type: str = "SIMILAR", confidence: Optional[float] = None, notes: Optional[str] = None, ) -> CardSuggestion: """ Add a card suggestion to a deck. Args: db: Database session deck_id: Deck ID card_id: Card ID to suggest source_card_id: Source card ID that triggered the suggestion suggestion_type: Type of suggestion confidence: Confidence score notes: Additional notes Returns: Created CardSuggestion record """ suggestion = CardSuggestion( deck_id=deck_id, card_id=card_id, source_card_id=source_card_id, suggestion_type=suggestion_type, confidence=confidence, notes=notes, ) db.add(suggestion) await db.flush() return suggestion @staticmethod async def get_deck_suggestions( db: AsyncSession, deck_id: int, suggestion_type: Optional[str] = None, ) -> List[CardSuggestion]: """ Get suggestions for a deck. Args: db: Database session deck_id: Deck ID suggestion_type: Filter by suggestion type Returns: List of CardSuggestion records """ conditions = [CardSuggestion.deck_id == deck_id] if suggestion_type: conditions.append(CardSuggestion.suggestion_type == suggestion_type) stmt = select(CardSuggestion).where(*conditions).order_by(CardSuggestion.created_at.desc()) result = await db.execute(stmt) return result.scalars().all()