658 lines
25 KiB
Python
658 lines
25 KiB
Python
"""
|
|
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()
|