492 lines
17 KiB
Python
492 lines
17 KiB
Python
"""
|
|
MTG Card Interaction Pipeline
|
|
|
|
Orchestrates the full interaction determination and recommendation pipeline.
|
|
Handles initial loads and rolling updates.
|
|
"""
|
|
import json
|
|
import logging
|
|
from datetime import datetime
|
|
from typing import List, Dict, Optional, Tuple
|
|
|
|
from sqlalchemy import create_engine, text
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from card_profile_extractor import CardProfileExtractor, CardProfile
|
|
from interaction_determinator import InteractionDeterminator, InteractionResult
|
|
|
|
|
|
# Configure logging
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class MTGInteractionPipeline:
|
|
"""
|
|
Main pipeline for processing card interactions.
|
|
|
|
Handles:
|
|
1. Loading cards from database
|
|
2. Extracting card profiles
|
|
3. Determining interactions (synergies, counters, evolutions)
|
|
4. Storing interactions in database
|
|
5. Updating interaction statistics
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
db_url: str,
|
|
min_confidence: float = 0.5,
|
|
):
|
|
"""
|
|
Initialize the pipeline.
|
|
|
|
Args:
|
|
db_url: PostgreSQL database URL
|
|
min_confidence: Minimum confidence to auto-store interactions
|
|
"""
|
|
self.db_url = db_url
|
|
self.min_confidence = min_confidence
|
|
self.engine = create_engine(db_url)
|
|
self.SessionLocal = sessionmaker(bind=self.engine)
|
|
|
|
self.profile_extractor = CardProfileExtractor()
|
|
self.determinator = InteractionDeterminator()
|
|
|
|
# Statistics
|
|
self.stats = {
|
|
'cards_processed': 0,
|
|
'interactions_determined': 0,
|
|
'interactions_stored': 0,
|
|
'interactions_review_queue': 0,
|
|
'errors': 0,
|
|
}
|
|
|
|
def load_cards_from_db(self, set_code: Optional[str] = None) -> List[Dict]:
|
|
"""
|
|
Load cards from the database.
|
|
|
|
Args:
|
|
set_code: Optional set code to filter by
|
|
|
|
Returns:
|
|
List of card dictionaries
|
|
"""
|
|
db = self.SessionLocal()
|
|
try:
|
|
if set_code:
|
|
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 s.code = :set_code
|
|
""")
|
|
cards = [dict(row._mapping) for row in
|
|
db.execute(query, {"set_code": set_code}).fetchall()]
|
|
else:
|
|
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
|
|
""")
|
|
cards = [dict(row._mapping) for row in db.execute(query).fetchall()]
|
|
|
|
logger.info(f"Loaded {len(cards)} cards from database")
|
|
return cards
|
|
finally:
|
|
db.close()
|
|
|
|
def extract_profiles(self, cards: List[Dict]) -> List[CardProfile]:
|
|
"""
|
|
Extract card profiles from card data.
|
|
|
|
Args:
|
|
cards: List of card dictionaries
|
|
|
|
Returns:
|
|
List of CardProfile objects
|
|
"""
|
|
return self.profile_extractor.extract_profiles_batch(cards)
|
|
|
|
def determine_interactions(self, profiles: List[CardProfile]) -> dict:
|
|
"""
|
|
Determine interactions for a batch of card profiles.
|
|
|
|
Args:
|
|
profiles: List of CardProfile objects
|
|
|
|
Returns:
|
|
Dictionary with synergies, counters, and evolutions
|
|
"""
|
|
return self.determinator.determine_all_interactions(profiles)
|
|
|
|
def _determine_synergy_type(self, interaction: InteractionResult) -> str:
|
|
"""
|
|
Determine synergy type from interaction metadata.
|
|
|
|
Args:
|
|
interaction: Interaction result
|
|
|
|
Returns:
|
|
Synergy type string
|
|
"""
|
|
metadata = interaction.metadata or {}
|
|
|
|
if 'common_archetypes' in metadata:
|
|
return 'archetype'
|
|
elif 'mechanics' in metadata:
|
|
return 'mechanic'
|
|
elif 'colors' in metadata:
|
|
return 'mana'
|
|
elif 'card_a_targets' in metadata:
|
|
return 'combo'
|
|
else:
|
|
return 'support'
|
|
|
|
def _determine_counter_type(self, interaction: InteractionResult) -> str:
|
|
"""
|
|
Determine counter type from interaction metadata.
|
|
|
|
Args:
|
|
interaction: Interaction result
|
|
|
|
Returns:
|
|
Counter type string
|
|
"""
|
|
metadata = interaction.metadata or {}
|
|
|
|
if 'colors_a' in metadata:
|
|
return 'color'
|
|
elif 'power_a' in metadata:
|
|
return 'stats'
|
|
else:
|
|
return 'keyword'
|
|
|
|
def _determine_evolution_type(self, interaction: InteractionResult) -> str:
|
|
"""
|
|
Determine evolution type from interaction metadata.
|
|
|
|
Args:
|
|
interaction: Interaction result
|
|
|
|
Returns:
|
|
Evolution type string
|
|
"""
|
|
metadata = interaction.metadata or {}
|
|
|
|
if 'card_name' in metadata:
|
|
return 'reprint'
|
|
else:
|
|
return 'evolution'
|
|
|
|
def store_interactions(self, interactions: dict) -> Tuple[int, int]:
|
|
"""
|
|
Store interactions in the database.
|
|
|
|
Args:
|
|
interactions: Dictionary with synergies, counters, evolutions
|
|
|
|
Returns:
|
|
Tuple of (stored_count, review_queue_count)
|
|
"""
|
|
db = self.SessionLocal()
|
|
stored = 0
|
|
review_queue = 0
|
|
|
|
try:
|
|
# Store synergies
|
|
for interaction in interactions['synergies']:
|
|
if interaction.confidence >= self.min_confidence:
|
|
synergy_type = self._determine_synergy_type(interaction)
|
|
db.execute(text("""
|
|
INSERT INTO mtg_card_synergies (
|
|
card_a_id, card_b_id, synergy_type, strength, notes, confidence
|
|
) VALUES (
|
|
:card_a, :card_b, :synergy_type, :strength, :notes, :confidence
|
|
) ON CONFLICT DO NOTHING
|
|
"""), {
|
|
"card_a": interaction.card_a_id,
|
|
"card_b": interaction.card_b_id,
|
|
"synergy_type": synergy_type,
|
|
"strength": interaction.strength,
|
|
"notes": interaction.notes,
|
|
"confidence": interaction.confidence,
|
|
})
|
|
stored += 1
|
|
else:
|
|
review_queue += 1
|
|
|
|
# Store counters
|
|
for interaction in interactions['counters']:
|
|
if interaction.confidence >= self.min_confidence:
|
|
counter_type = self._determine_counter_type(interaction)
|
|
db.execute(text("""
|
|
INSERT INTO mtg_card_counters (
|
|
card_a_id, card_b_id, counter_type, strength, notes, confidence
|
|
) VALUES (
|
|
:card_a, :card_b, :counter_type, :strength, :notes, :confidence
|
|
) ON CONFLICT DO NOTHING
|
|
"""), {
|
|
"card_a": interaction.card_a_id,
|
|
"card_b": interaction.card_b_id,
|
|
"counter_type": counter_type,
|
|
"strength": interaction.strength,
|
|
"notes": interaction.notes,
|
|
"confidence": interaction.confidence,
|
|
})
|
|
stored += 1
|
|
else:
|
|
review_queue += 1
|
|
|
|
# Store evolutions
|
|
for interaction in interactions['evolutions']:
|
|
if interaction.confidence >= self.min_confidence:
|
|
evolution_type = self._determine_evolution_type(interaction)
|
|
db.execute(text("""
|
|
INSERT INTO mtg_card_evolution (
|
|
card_id, evolved_card_id, evolution_type, strength, notes, confidence
|
|
) VALUES (
|
|
:card_id, :evolved_card_id, :evolution_type, :strength, :notes, :confidence
|
|
) ON CONFLICT DO NOTHING
|
|
"""), {
|
|
"card_id": interaction.card_a_id,
|
|
"evolved_card_id": interaction.card_b_id,
|
|
"evolution_type": evolution_type,
|
|
"strength": interaction.strength,
|
|
"notes": interaction.notes,
|
|
"confidence": interaction.confidence,
|
|
})
|
|
stored += 1
|
|
else:
|
|
review_queue += 1
|
|
|
|
db.commit()
|
|
logger.info(f"Stored {stored} interactions, {review_queue} sent to review queue")
|
|
|
|
except Exception as e:
|
|
db.rollback()
|
|
logger.error(f"Error storing interactions: {e}")
|
|
self.stats['errors'] += 1
|
|
finally:
|
|
db.close()
|
|
|
|
return stored, review_queue
|
|
|
|
def update_interaction_stats(self):
|
|
"""Update interaction statistics for all cards."""
|
|
db = self.SessionLocal()
|
|
try:
|
|
# Delete existing stats
|
|
db.execute(text("DELETE FROM mtg_card_interaction_stats"))
|
|
|
|
# Recalculate stats
|
|
db.execute(text("""
|
|
INSERT INTO mtg_card_interaction_stats (
|
|
card_id, total_synergies, total_counters, total_evolutions,
|
|
total_synergy_strength, avg_synergy_strength
|
|
)
|
|
SELECT
|
|
c.id,
|
|
COALESCE(synergies.synergy_count, 0),
|
|
COALESCE(counters.counter_count, 0),
|
|
COALESCE(evolution.evolution_count, 0),
|
|
COALESCE(synergies.total_strength, 0),
|
|
COALESCE(synergies.avg_strength, 0)
|
|
FROM mtg_cards c
|
|
LEFT JOIN (
|
|
SELECT card_a_id as card_id, COUNT(*) as synergy_count,
|
|
SUM(strength) as total_strength,
|
|
AVG(strength) as avg_strength
|
|
FROM mtg_card_synergies
|
|
GROUP BY card_a_id
|
|
) synergies ON c.id = synergies.card_id
|
|
LEFT JOIN (
|
|
SELECT card_a_id as card_id, COUNT(*) as counter_count
|
|
FROM mtg_card_counters
|
|
GROUP BY card_a_id
|
|
) counters ON c.id = counters.card_id
|
|
LEFT JOIN (
|
|
SELECT card_id as card_id, COUNT(*) as evolution_count
|
|
FROM mtg_card_evolution
|
|
GROUP BY card_id
|
|
) evolution ON c.id = evolution.card_id
|
|
"""))
|
|
|
|
db.commit()
|
|
logger.info("Updated interaction statistics")
|
|
|
|
except Exception as e:
|
|
db.rollback()
|
|
logger.error(f"Error updating interaction stats: {e}")
|
|
self.stats['errors'] += 1
|
|
finally:
|
|
db.close()
|
|
|
|
def run_initial_load(self, set_code: Optional[str] = None):
|
|
"""
|
|
Run initial load for all cards or a specific set.
|
|
|
|
This is used for the first time data is loaded into the database.
|
|
|
|
Args:
|
|
set_code: Optional set code to process
|
|
"""
|
|
logger.info("=" * 60)
|
|
logger.info("Starting Initial Load")
|
|
logger.info("=" * 60)
|
|
|
|
# Load all cards
|
|
all_cards = self.load_cards_from_db(set_code)
|
|
|
|
if not all_cards:
|
|
logger.warning("No cards found in database")
|
|
return
|
|
|
|
# Extract profiles
|
|
logger.info(f"Extracting profiles for {len(all_cards)} cards...")
|
|
profiles = self.extract_profiles(all_cards)
|
|
|
|
# Determine interactions
|
|
logger.info(f"Determining interactions for {len(profiles)} cards...")
|
|
interactions = self.determine_interactions(profiles)
|
|
|
|
logger.info(
|
|
f"Determined {len(interactions['synergies'])} synergies, "
|
|
f"{len(interactions['counters'])} counters, "
|
|
f"{len(interactions['evolutions'])} evolutions"
|
|
)
|
|
|
|
# Store interactions
|
|
stored, review_queue = self.store_interactions(interactions)
|
|
|
|
# Update statistics
|
|
self.update_interaction_stats()
|
|
|
|
# Update stats
|
|
self.stats['cards_processed'] = len(all_cards)
|
|
self.stats['interactions_determined'] = len(interactions['synergies']) + len(interactions['counters']) + len(interactions['evolutions'])
|
|
self.stats['interactions_stored'] = stored
|
|
self.stats['interactions_review_queue'] = review_queue
|
|
|
|
logger.info("=" * 60)
|
|
logger.info(f"Initial Load Complete")
|
|
logger.info(f" Cards processed: {self.stats['cards_processed']}")
|
|
logger.info(f" Interactions determined: {self.stats['interactions_determined']}")
|
|
logger.info(f" Interactions stored: {self.stats['interactions_stored']}")
|
|
logger.info(f" Interactions in review queue: {self.stats['interactions_review_queue']}")
|
|
logger.info("=" * 60)
|
|
|
|
def run_rolling_update(self, new_cards: List[Dict], set_code: Optional[str] = None):
|
|
"""
|
|
Run rolling update for new cards.
|
|
|
|
This is used when new cards are added via MTGJSON updates.
|
|
|
|
Args:
|
|
new_cards: List of new card dictionaries
|
|
set_code: Optional set code
|
|
"""
|
|
logger.info("=" * 60)
|
|
logger.info("Starting Rolling Update")
|
|
logger.info(f"New cards: {len(new_cards)}")
|
|
logger.info("=" * 60)
|
|
|
|
# Load existing cards
|
|
existing_cards = self.load_cards_from_db(set_code)
|
|
|
|
# Combine existing and new cards
|
|
all_cards = existing_cards + new_cards
|
|
|
|
# Extract profiles
|
|
logger.info(f"Extracting profiles for {len(all_cards)} cards...")
|
|
profiles = self.extract_profiles(all_cards)
|
|
|
|
# Determine interactions
|
|
logger.info(f"Determining interactions for {len(profiles)} cards...")
|
|
interactions = self.determine_interactions(profiles)
|
|
|
|
logger.info(
|
|
f"Determined {len(interactions['synergies'])} synergies, "
|
|
f"{len(interactions['counters'])} counters, "
|
|
f"{len(interactions['evolutions'])} evolutions"
|
|
)
|
|
|
|
# Store interactions
|
|
stored, review_queue = self.store_interactions(interactions)
|
|
|
|
# Update statistics
|
|
self.update_interaction_stats()
|
|
|
|
# Update stats
|
|
self.stats['cards_processed'] = len(new_cards)
|
|
self.stats['interactions_determined'] = len(interactions['synergies']) + len(interactions['counters']) + len(interactions['evolutions'])
|
|
self.stats['interactions_stored'] = stored
|
|
self.stats['interactions_review_queue'] = review_queue
|
|
|
|
logger.info("=" * 60)
|
|
logger.info(f"Rolling Update Complete")
|
|
logger.info(f" New cards processed: {self.stats['cards_processed']}")
|
|
logger.info(f" Interactions determined: {self.stats['interactions_determined']}")
|
|
logger.info(f" Interactions stored: {self.stats['interactions_stored']}")
|
|
logger.info(f" Interactions in review queue: {self.stats['interactions_review_queue']}")
|
|
logger.info("=" * 60)
|
|
|
|
def get_pipeline_stats(self) -> Dict:
|
|
"""Get pipeline statistics."""
|
|
return {
|
|
**self.stats,
|
|
'timestamp': datetime.now().isoformat(),
|
|
}
|
|
|
|
def close(self):
|
|
"""Close database connection."""
|
|
self.engine.dispose()
|
|
|
|
|
|
def main():
|
|
"""Main entry point for pipeline execution."""
|
|
import sys
|
|
|
|
# Database URL from environment or default
|
|
db_url = "postgresql+psycopg2://mtgonline:mtgonline_pass@172.18.0.2:5432/mtgdata"
|
|
|
|
# Get command line arguments
|
|
if len(sys.argv) < 2:
|
|
print("Usage: python pipeline.py [initial|rolling] [set_code]")
|
|
print(" initial: Run initial load for all cards or a specific set")
|
|
print(" rolling: Run rolling update for new cards (requires JSON input)")
|
|
sys.exit(1)
|
|
|
|
command = sys.argv[1]
|
|
set_code = sys.argv[2] if len(sys.argv) > 2 else None
|
|
|
|
# Initialize pipeline
|
|
pipeline = MTGInteractionPipeline(db_url)
|
|
|
|
try:
|
|
if command == "initial":
|
|
pipeline.run_initial_load(set_code)
|
|
elif command == "rolling":
|
|
# Read new cards from stdin (JSON)
|
|
new_cards = json.loads(sys.stdin.read())
|
|
pipeline.run_rolling_update(new_cards, set_code)
|
|
else:
|
|
print(f"Unknown command: {command}")
|
|
sys.exit(1)
|
|
|
|
# Print statistics
|
|
stats = pipeline.get_pipeline_stats()
|
|
print("\nPipeline Statistics:")
|
|
for key, value in stats.items():
|
|
print(f" {key}: {value}")
|
|
|
|
finally:
|
|
pipeline.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|