""" Card Interaction Recommendation Engine Uses the interaction graph to provide: - Synergy-based card recommendations - Deck archetype suggestions - Card combination suggestions - "Cards like this" recommendations """ from typing import List, Dict, Optional, Tuple from dataclasses import dataclass from enum import Enum import json from sqlalchemy import create_engine, text from sqlalchemy.orm import sessionmaker class RecommendationType(Enum): """Types of recommendations.""" SYNERGY = "synergy" ARCHETYPE = "archetype" COMBO = "combo" COUNTER = "counter" EVOLUTION = "evolution" CARD_LIKE_THIS = "card_like_this" @dataclass class Recommendation: """A single recommendation.""" recommendation_type: str card_id: int card_name: str card_type_line: str confidence: float score: float # Weighted score for ranking reason: str metadata: Dict[str, any] = None def __post_init__(self): if self.metadata is None: self.metadata = {} def to_dict(self) -> Dict: """Convert to dictionary for JSON serialization.""" return { 'recommendation_type': self.recommendation_type, 'card_id': self.card_id, 'card_name': self.card_name, 'card_type_line': self.card_type_line, 'confidence': self.confidence, 'score': self.score, 'reason': self.reason, 'metadata': self.metadata, } class RecommendationEngine: """ Generates card recommendations based on interaction graph data. Uses: - Interaction graph for synergy matching - Card profiles for archetype/mana curve matching - Confidence scoring for ranking recommendations """ def __init__(self, db_url: str, config: Optional[Dict] = None): """Initialize with database URL and configuration.""" self.db_url = db_url self.config = config or { 'max_recommendations': 50, 'min_confidence': 0.5, 'min_score': 1.0, 'synergy_weight': 1.0, 'archetype_weight': 0.8, 'combo_weight': 1.2, 'counter_weight': 0.6, 'evolution_weight': 0.7, } # Initialize database connection self.engine = create_engine(db_url) self.SessionLocal = sessionmaker(bind=self.engine) def get_card_profile(self, card_id: int) -> Optional[Dict]: """Get full card profile from database.""" db = self.SessionLocal() try: query = text(""" SELECT c.*, s.code as set_code, s.name as set_name FROM mtg_cards c JOIN mtg_sets s ON c.set_id = s.id WHERE c.id = :card_id """) result = db.execute(query, {"card_id": card_id}).fetchone() if result: return dict(result._mapping) return None finally: db.close() def get_interactions_for_card(self, card_id: int) -> Dict[str, List[Dict]]: """Get all interactions for a specific card.""" db = self.SessionLocal() try: # Get synergies synergies_query = text(""" SELECT card_a_id, card_b_id, synergy_type, strength, notes FROM mtg_card_synergies WHERE card_a_id = :card_id OR card_b_id = :card_id """) synergies = [dict(row._mapping) for row in db.execute(synergies_query, {"card_id": card_id}).fetchall()] # Get counters counters_query = text(""" SELECT card_a_id, card_b_id, counter_type, strength, notes FROM mtg_card_counters WHERE card_a_id = :card_id OR card_b_id = :card_id """) counters = [dict(row._mapping) for row in db.execute(counters_query, {"card_id": card_id}).fetchall()] # Get evolutions evolutions_query = text(""" SELECT card_id, evolved_card_id, evolution_type, strength, notes FROM mtg_card_evolution WHERE card_id = :card_id OR evolved_card_id = :card_id """) evolutions = [dict(row._mapping) for row in db.execute(evolutions_query, {"card_id": card_id}).fetchall()] # Get archetypes archetypes_query = text(""" SELECT archetype, strength FROM mtg_card_archetypes WHERE card_id = :card_id """) archetypes = [dict(row._mapping) for row in db.execute(archetypes_query, {"card_id": card_id}).fetchall()] # Get mechanics mechanics_query = text(""" SELECT mechanic, strength FROM mtg_card_mechanics WHERE card_id = :card_id """) mechanics = [dict(row._mapping) for row in db.execute(mechanics_query, {"card_id": card_id}).fetchall()] return { 'synergies': synergies, 'counters': counters, 'evolutions': evolutions, 'archetypes': archetypes, 'mechanics': mechanics, } finally: db.close() def recommend_card_synergies( self, card_id: int, max_results: int = 20 ) -> List[Recommendation]: """ Recommend cards that synergize with a given card. Looks for cards with: - Same archetype - Supporting mechanics - Compatible mana costs - Combo potential """ recommendations = [] card_profile = self.get_card_profile(card_id) if not card_profile: return recommendations db = self.SessionLocal() try: # Get archetypes for this card archetypes_query = text(""" SELECT archetype, strength FROM mtg_card_archetypes WHERE card_id = :card_id """) card_archetypes = [dict(row._mapping) for row in db.execute(archetypes_query, {"card_id": card_id}).fetchall()] # Get mechanics for this card mechanics_query = text(""" SELECT mechanic, strength FROM mtg_card_mechanics WHERE card_id = :card_id """) card_mechanics = [dict(row._mapping) for row in db.execute(mechanics_query, {"card_id": card_id}).fetchall()] # Get synergies for this card synergies_query = text(""" SELECT card_b_id as card_id, synergy_type, strength, notes FROM mtg_card_synergies WHERE card_a_id = :card_id ORDER BY strength DESC """) synergy_cards = [dict(row._mapping) for row in db.execute(synergies_query, {"card_id": card_id}).fetchall()] # Score each synergizing card for synergy in synergy_cards: synergy_card_id = synergy['card_id'] # Get the other card's profile other_card = self.get_card_profile(synergy_card_id) if not other_card: continue # Calculate score based on synergy strength and other factors score = synergy['strength'] * self.config['synergy_weight'] # Boost score if same archetype archetype_match = False for archetype in card_archetypes: if archetype['archetype'] in other_card.get('subtypes', ''): archetype_match = True score *= 1.2 break # Boost score if shared mechanic mechanic_match = False for mechanic in card_mechanics: if mechanic['mechanic'] in other_card.get('oracle_text', '').lower(): mechanic_match = True score *= 1.1 break recommendations.append(Recommendation( recommendation_type=RecommendationType.SYNERGY.value, card_id=synergy_card_id, card_name=other_card['name'], card_type_line=other_card['type_line'], confidence=0.9, score=score, reason=f"Synergizes with {card_profile['name']} ({synergy['synergy_type']})", metadata={ 'synergy_type': synergy['synergy_type'], 'synergy_strength': synergy['strength'], 'archetype_match': archetype_match, 'mechanic_match': mechanic_match, } )) # Sort by score and return top results recommendations.sort(key=lambda r: r.score, reverse=True) return recommendations[:max_results] finally: db.close() def recommend_archetype_cards( self, archetype: str, max_results: int = 20 ) -> List[Recommendation]: """ Recommend cards that fit a specific archetype. Looks for cards with: - Matching subtype - Supporting mechanics - Compatible mana costs """ recommendations = [] db = self.SessionLocal() try: # Get cards with this archetype cards_query = text(""" SELECT c.*, s.code as set_code, s.name as set_name FROM mtg_cards c JOIN mtg_sets s ON c.set_id = s.id WHERE c.subtypes LIKE :archetype LIMIT :limit """) cards = [dict(row._mapping) for row in db.execute(cards_query, { "archetype": f"%{archetype}%", "limit": max_results * 2 }).fetchall()] # Score each card for card in cards: # Calculate base score from archetype match score = 1.0 # Boost score for cards with supporting mechanics supporting_mechanics = [] if archetype.lower() == 'elf': supporting_mechanics = ['landfall', 'vigilance', 'trample'] elif archetype.lower() == 'goblin': supporting_mechanics = ['haste', 'trample', 'damage'] elif archetype.lower() == 'vampire': supporting_mechanics = ['lifelink', 'first_strike', 'deathtouch'] elif archetype.lower() == 'angel': supporting_mechanics = ['flying', 'lifelink', 'indestructible'] elif archetype.lower() == 'dragon': supporting_mechanics = ['flying', 'trample', 'menace'] elif archetype.lower() == 'zombie': supporting_mechanics = ['deathtouch', 'first_strike', 'haste'] for mechanic in supporting_mechanics: if mechanic in card.get('oracle_text', '').lower(): score += 0.5 # Boost score for cards with good power/toughness try: power = int(card.get('power', 0) or 0) toughness = int(card.get('toughness', 0) or 0) if power >= 3 and toughness >= 3: score += 0.5 except (ValueError, TypeError): pass recommendations.append(Recommendation( recommendation_type=RecommendationType.ARCHETYPE.value, card_id=card['id'], card_name=card['name'], card_type_line=card['type_line'], confidence=0.8, score=score, reason=f"Matches {archetype} archetype", metadata={ 'archetype': archetype, 'supporting_mechanics': supporting_mechanics, } )) # Sort by score and return top results recommendations.sort(key=lambda r: r.score, reverse=True) return recommendations[:max_results] finally: db.close() def recommend_card_combos( self, card_id: int, max_results: int = 10 ) -> List[Recommendation]: """ Recommend card combos involving a specific card. Looks for cards that: - Target the same creature - Create powerful combinations - Have complementary effects """ recommendations = [] card_profile = self.get_card_profile(card_id) if not card_profile: return recommendations db = self.SessionLocal() try: # Get synergies that are combo partners combos_query = text(""" SELECT card_b_id as card_id, synergy_type, strength, notes FROM mtg_card_synergies WHERE card_a_id = :card_id AND synergy_type = 'COMBO_PARTNER' ORDER BY strength DESC """) combo_cards = [dict(row._mapping) for row in db.execute(combos_query, {"card_id": card_id}).fetchall()] for combo in combo_cards: combo_card_id = combo['card_id'] # Get the other card's profile other_card = self.get_card_profile(combo_card_id) if not other_card: continue # Calculate score based on combo strength score = combo['strength'] * self.config['combo_weight'] recommendations.append(Recommendation( recommendation_type=RecommendationType.COMBO.value, card_id=combo_card_id, card_name=other_card['name'], card_type_line=other_card['type_line'], confidence=0.85, score=score, reason=f"Combo with {card_profile['name']} ({combo['notes']})", metadata={ 'combo_notes': combo['notes'], 'combo_strength': combo['strength'], } )) # Sort by score and return top results recommendations.sort(key=lambda r: r.score, reverse=True) return recommendations[:max_results] finally: db.close() def recommend_counter_cards( self, card_id: int, max_results: int = 10 ) -> List[Recommendation]: """ Recommend cards that counter a specific card. Looks for cards that: - Have counter spells - Target the same card types - Have relevant keywords """ recommendations = [] card_profile = self.get_card_profile(card_id) if not card_profile: return recommendations db = self.SessionLocal() try: # Get cards that counter this card counters_query = text(""" SELECT card_b_id as card_id, counter_type, strength, notes FROM mtg_card_counters WHERE card_a_id = :card_id ORDER BY strength DESC """) counter_cards = [dict(row._mapping) for row in db.execute(counters_query, {"card_id": card_id}).fetchall()] for counter in counter_cards: counter_card_id = counter['card_id'] # Get the counter card's profile counter_card = self.get_card_profile(counter_card_id) if not counter_card: continue # Calculate score based on counter strength score = counter['strength'] * self.config['counter_weight'] recommendations.append(Recommendation( recommendation_type=RecommendationType.COUNTER.value, card_id=counter_card_id, card_name=counter_card['name'], card_type_line=counter_card['type_line'], confidence=0.75, score=score, reason=f"Counters {card_profile['name']} ({counter['counter_type']})", metadata={ 'counter_type': counter['counter_type'], 'counter_strength': counter['strength'], } )) # Sort by score and return top results recommendations.sort(key=lambda r: r.score, reverse=True) return recommendations[:max_results] finally: db.close() def recommend_card_like_this( self, card_id: int, max_results: int = 20 ) -> List[Recommendation]: """ Recommend cards similar to a given card. Looks for cards with: - Similar archetypes - Similar mechanics - Similar mana costs - Similar power/toughness """ recommendations = [] card_profile = self.get_card_profile(card_id) if not card_profile: return recommendations db = self.SessionLocal() try: # Get this card's archetypes archetypes_query = text(""" SELECT archetype, strength FROM mtg_card_archetypes WHERE card_id = :card_id """) card_archetypes = [dict(row._mapping) for row in db.execute(archetypes_query, {"card_id": card_id}).fetchall()] # Get this card's mechanics mechanics_query = text(""" SELECT mechanic, strength FROM mtg_card_mechanics WHERE card_id = :card_id """) card_mechanics = [dict(row._mapping) for row in db.execute(mechanics_query, {"card_id": card_id}).fetchall()] # Search for similar cards similar_cards_query = text(""" SELECT c.*, s.code as set_code, s.name as set_name FROM mtg_cards c JOIN mtg_sets s ON c.set_id = s.id WHERE c.id != :card_id AND (c.subtypes LIKE :archetype OR c.oracle_text LIKE :mechanic) LIMIT :limit """) # Get cards with matching archetypes archetype_matches = [] for archetype in card_archetypes: archetype_matches.extend( [dict(row._mapping) for row in db.execute(similar_cards_query, { "card_id": card_id, "archetype": f"%{archetype['archetype']}%", "mechanic": "%", "limit": max_results * 2 }).fetchall()] ) # Get cards with matching mechanics mechanic_matches = [] for mechanic in card_mechanics: mechanic_matches.extend( [dict(row._mapping) for row in db.execute(similar_cards_query, { "card_id": card_id, "archetype": "%", "mechanic": f"%{mechanic['mechanic']}%", "limit": max_results * 2 }).fetchall()] ) # Deduplicate seen_cards = set() all_matches = [] for card in archetype_matches + mechanic_matches: if card['id'] not in seen_cards: seen_cards.add(card['id']) all_matches.append(card) # Score each similar card for card in all_matches: score = 0.5 # Boost for archetype match for archetype in card_archetypes: if archetype['archetype'] in card.get('subtypes', ''): score += 1.0 break # Boost for mechanic match for mechanic in card_mechanics: if mechanic['mechanic'] in card.get('oracle_text', '').lower(): score += 0.5 break # Boost for similar mana cost try: mana_a = int(card_profile.get('mana_cost', '0').replace('{', '').replace('}', '').replace('W', '').replace('U', '').replace('B', '').replace('R', '').replace('G', '').replace('X', '').replace('Y', '')) mana_b = int(card.get('mana_cost', '0').replace('{', '').replace('}', '').replace('W', '').replace('U', '').replace('B', '').replace('R', '').replace('G', '').replace('X', '').replace('Y', '')) if abs(mana_a - mana_b) <= 1: score += 0.5 except (ValueError, TypeError): pass # Boost for similar power/toughness try: power_a = int(card_profile.get('power', 0) or 0) power_b = int(card.get('power', 0) or 0) toughness_a = int(card_profile.get('toughness', 0) or 0) toughness_b = int(card.get('toughness', 0) or 0) if abs(power_a - power_b) <= 1 and abs(toughness_a - toughness_b) <= 1: score += 0.5 except (ValueError, TypeError): pass recommendations.append(Recommendation( recommendation_type=RecommendationType.CARD_LIKE_THIS.value, card_id=card['id'], card_name=card['name'], card_type_line=card['type_line'], confidence=0.7, score=score, reason=f"Similar to {card_profile['name']}", metadata={ 'archetype_match': any(a['archetype'] in card.get('subtypes', '') for a in card_archetypes), 'mechanic_match': any(m['mechanic'] in card.get('oracle_text', '').lower() for m in card_mechanics), } )) # Sort by score and return top results recommendations.sort(key=lambda r: r.score, reverse=True) return recommendations[:max_results] finally: db.close() def get_full_recommendations( self, card_id: int, max_results: int = 50 ) -> List[Recommendation]: """ Get all recommendations for a card. Combines synergies, archetypes, combos, counters, and similar cards. """ all_recommendations = [] # Get synergies synergies = self.recommend_card_synergies(card_id, max_results) all_recommendations.extend(synergies) # Get archetype cards card_profile = self.get_card_profile(card_id) if card_profile and card_profile.get('subtypes'): archetypes = card_profile['subtypes'].split(',') for archetype in archetypes: archetype_cards = self.recommend_archetype_cards(archetype.strip(), max_results) all_recommendations.extend(archetype_cards) # Get combos combos = self.recommend_card_combos(card_id, max_results) all_recommendations.extend(combos) # Get counters counters = self.recommend_counter_cards(card_id, max_results) all_recommendations.extend(counters) # Get similar cards similar = self.recommend_card_like_this(card_id, max_results) all_recommendations.extend(similar) # Deduplicate by card_id seen_cards = set() unique_recommendations = [] for rec in all_recommendations: if rec.card_id not in seen_cards: seen_cards.add(rec.card_id) unique_recommendations.append(rec) # Sort by score and return top results unique_recommendations.sort(key=lambda r: r.score, reverse=True) return unique_recommendations[:max_results] def close(self): """Close database connection.""" self.engine.dispose()