156 lines
4.1 KiB
Python
156 lines
4.1 KiB
Python
"""
|
|
MTG Card Data Loader Trigger
|
|
|
|
Integrates with the card data loading process to automatically:
|
|
1. Determine interactions for new cards
|
|
2. Store interactions in the database
|
|
3. Update interaction statistics
|
|
4. Queue low-confidence interactions for review
|
|
|
|
This script is called after cards are loaded into the database.
|
|
"""
|
|
import json
|
|
import logging
|
|
import sys
|
|
import os
|
|
from datetime import datetime
|
|
|
|
# Add parent directory to path for imports
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
from interaction_pipeline import MTGInteractionPipeline
|
|
|
|
|
|
# Configure logging
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def load_cards_from_file(file_path: str) -> list:
|
|
"""
|
|
Load cards from a JSON file.
|
|
|
|
Args:
|
|
file_path: Path to JSON file containing card data
|
|
|
|
Returns:
|
|
List of card dictionaries
|
|
"""
|
|
with open(file_path, 'r') as f:
|
|
cards = json.load(f)
|
|
|
|
logger.info(f"Loaded {len(cards)} cards from {file_path}")
|
|
return cards
|
|
|
|
|
|
def trigger_interaction_processing(
|
|
new_cards: list,
|
|
set_code: str = None,
|
|
db_url: str = None,
|
|
min_confidence: float = 0.5,
|
|
):
|
|
"""
|
|
Trigger interaction processing for new cards.
|
|
|
|
This function is called after cards are loaded into the database.
|
|
|
|
Args:
|
|
new_cards: List of new card dictionaries
|
|
set_code: Set code for the new cards
|
|
db_url: Database URL (optional, uses default if not provided)
|
|
min_confidence: Minimum confidence to auto-store interactions
|
|
"""
|
|
# Database URL from environment or default
|
|
if not db_url:
|
|
db_url = "postgresql+psycopg2://mtgonline:mtgonline_pass@172.18.0.2:5432/mtgdata"
|
|
|
|
logger.info("=" * 60)
|
|
logger.info("MTG Card Interaction Processing Trigger")
|
|
logger.info("=" * 60)
|
|
logger.info(f"New cards: {len(new_cards)}")
|
|
logger.info(f"Set code: {set_code or 'all sets'}")
|
|
logger.info(f"Min confidence: {min_confidence}")
|
|
logger.info("=" * 60)
|
|
|
|
# Initialize pipeline
|
|
pipeline = MTGInteractionPipeline(db_url)
|
|
|
|
try:
|
|
# Run rolling update
|
|
pipeline.run_rolling_update(new_cards, set_code)
|
|
|
|
# Get statistics
|
|
stats = pipeline.get_pipeline_stats()
|
|
|
|
# Log results
|
|
logger.info("=" * 60)
|
|
logger.info("Processing Complete")
|
|
logger.info(f" Cards processed: {stats['cards_processed']}")
|
|
logger.info(f" Interactions determined: {stats['interactions_determined']}")
|
|
logger.info(f" Interactions stored: {stats['interactions_stored']}")
|
|
logger.info(f" Interactions in review queue: {stats['interactions_review_queue']}")
|
|
logger.info(f" Errors: {stats['errors']}")
|
|
logger.info("=" * 60)
|
|
|
|
return stats
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error processing interactions: {e}")
|
|
raise
|
|
|
|
finally:
|
|
pipeline.close()
|
|
|
|
|
|
def main():
|
|
"""Main entry point."""
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser(
|
|
description="Process interactions for new MTG cards"
|
|
)
|
|
parser.add_argument(
|
|
"--cards-file",
|
|
type=str,
|
|
help="Path to JSON file containing new cards",
|
|
required=True,
|
|
)
|
|
parser.add_argument(
|
|
"--set-code",
|
|
type=str,
|
|
help="Set code for the new cards",
|
|
default=None,
|
|
)
|
|
parser.add_argument(
|
|
"--db-url",
|
|
type=str,
|
|
help="Database URL (optional)",
|
|
default=None,
|
|
)
|
|
parser.add_argument(
|
|
"--min-confidence",
|
|
type=float,
|
|
help="Minimum confidence to auto-store interactions",
|
|
default=0.5,
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
|
|
# Load cards from file
|
|
new_cards = load_cards_from_file(args.cards_file)
|
|
|
|
# Trigger processing
|
|
trigger_interaction_processing(
|
|
new_cards=new_cards,
|
|
set_code=args.set_code,
|
|
db_url=args.db_url,
|
|
min_confidence=args.min_confidence,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|