""" MTGJSON Database Refresh Script Downloads and updates the MTGJSON All Printings dataset weekly. """ import asyncio import json import logging import os import time from datetime import datetime from pathlib import Path import aiohttp from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy import text from sqlalchemy.orm import sessionmaker from app.core.settings import get_settings logger = logging.getLogger(__name__) MTGJSON_API_URL = "https://mtgjson.com/api/v5/AllPrintings.json" DATA_DIR = Path("/app/data") REFRESH_INTERVAL_DAYS = 7 async def download_mtgjson(session: aiohttp.ClientSession, output_path: Path) -> bool: """Download the latest AllPrintings dataset.""" try: logger.info(f"Downloading MTGJSON from {MTGJSON_API_URL}...") async with session.get(MTGJSON_API_URL) as response: if response.status != 200: logger.error(f"Failed to download: {response.status}") return False with open(output_path, 'wb') as f: async for chunk in response.content.iter_chunked(8192): f.write(chunk) logger.info(f"Downloaded to {output_path}") return True except Exception as e: logger.error(f"Download error: {e}") return False async def parse_mtgjson(filepath: Path) -> tuple[dict, dict]: """Parse the AllPrintings JSON file. MTGJSON v5 AllPrintings structure: { "meta": {...}, "data": { "10E": {"baseSetSize": 383, "block": "Core Set", "cards": [...]}, "UNH": {...} } } Returns: (sets_dict, cards_dict) where sets_dict maps set_code -> set_data and cards_dict maps set_code -> list of card dicts """ try: with open(filepath, 'r', encoding='utf-8') as f: data = json.load(f) # Verify structure - data key is required if 'data' not in data: raise ValueError("Invalid MTGJSON structure: missing 'data' key") mtg_data = data['data'] # MTGJSON v5: data contains set codes directly as keys # Each set code maps to {baseSetSize, block, cards: [...]} sets_dict = {} cards_dict = {} for set_code, set_data in mtg_data.items(): # Skip if it looks like metadata, not a set if isinstance(set_data, dict) and 'baseSetSize' in set_data: # Convert release_date string to datetime object release_date_str = set_data.get('releaseDate') if release_date_str: try: set_data['releaseDate'] = datetime.fromisoformat(release_date_str.replace('Z', '+00:00')) except (ValueError, AttributeError): pass # Keep as string if parsing fails sets_dict[set_code] = set_data # Extract cards for this set if 'cards' in set_data and isinstance(set_data['cards'], list): cards_dict[set_code] = set_data['cards'] if not sets_dict: raise ValueError("No sets found in MTGJSON data") logger.info(f"Parsed {len(sets_dict)} sets, {sum(len(c) for c in cards_dict.values())} cards") return sets_dict, cards_dict except Exception as e: logger.error(f"Parse error: {e}") return {}, {} async def update_database(session: AsyncSession, sets_dict: dict, cards_dict: dict) -> tuple[int, int]: """Update the database with parsed MTGJSON data.""" cards_updated = 0 sets_updated = 0 try: # Process sets for set_code, set_data in sets_dict.items(): stmt = text(""" INSERT INTO mtg_sets (code, name, type, release_date, base_set_size, total_size, is_foil_only, is_non_foil_only, digital, icon_svg_url, parent_code, mtgo_code) VALUES (:code, :name, :type, :release_date, :base_set_size, :total_size, :is_foil_only, :is_non_foil_only, :digital, :icon_svg_url, :parent_code, :mtgo_code) ON CONFLICT (code) DO UPDATE SET name = EXCLUDED.name, updated_at = CURRENT_TIMESTAMP """) await session.execute(stmt, { 'code': set_code, 'name': set_data.get('name'), 'type': set_data.get('type'), 'release_date': set_data.get('releaseDate'), 'base_set_size': set_data.get('baseSetSize'), 'total_size': set_data.get('totalSize'), 'is_foil_only': set_data.get('isFoilOnly'), 'is_non_foil_only': set_data.get('isNonFoilOnly'), 'digital': set_data.get('digital'), 'icon_svg_url': set_data.get('iconSvgUri'), 'parent_code': set_data.get('parentCode'), 'mtgo_code': set_data.get('mtgoCode'), }) sets_updated += 1 # Process cards grouped by set for set_code, cards in cards_dict.items(): for card_data in cards: stmt = text(""" INSERT INTO mtg_cards (set_id, name, mana_cost, type_line, oracle_text, power, toughness, rarity, layout, artist, flavor_text, numbers, identifiers, images) SELECT s.id, :name, :mana_cost, :type_line, :oracle_text, :power, :toughness, :rarity, :layout, :artist, :flavor_text, :numbers, :identifiers, :images FROM mtg_sets s WHERE s.code = :set_code ON CONFLICT DO NOTHING """) await session.execute(stmt, { 'set_code': set_code, 'name': card_data.get('name'), 'mana_cost': card_data.get('manaCost'), 'type_line': card_data.get('type'), 'oracle_text': card_data.get('text'), 'power': card_data.get('power'), 'toughness': card_data.get('toughness'), 'rarity': card_data.get('rarity'), 'layout': card_data.get('layout'), 'artist': card_data.get('artist'), 'flavor_text': card_data.get('flavorText'), 'numbers': str(card_data.get('numbers', '')), 'identifiers': json.dumps(card_data.get('identifiers', {})), 'images': json.dumps(card_data.get('images', {})), }) cards_updated += 1 await session.commit() return cards_updated, sets_updated except Exception as e: logger.error(f"Database update error: {e}") await session.rollback() raise async def check_last_refresh(engine: create_async_engine) -> datetime: """Check when the last refresh occurred.""" async with AsyncSession(engine) as session: stmt = text("SELECT refresh_date FROM mtg_refresh_log ORDER BY refresh_date DESC LIMIT 1") result = await session.execute(stmt) row = result.fetchone() if row: return row[0] return datetime.min async def log_refresh(engine: create_async_engine, status: str, cards: int, sets: int, duration: int, error: str = None): """Log the refresh operation.""" async with AsyncSession(engine) as session: stmt = text(""" INSERT INTO mtg_refresh_log (status, cards_updated, sets_updated, error_message, duration_seconds) VALUES (:status, :cards, :sets, :error, :duration) """) await session.execute(stmt, { 'status': status, 'cards': cards, 'sets': sets, 'error': error, 'duration': duration, }) await session.commit() async def main(): """Main refresh logic.""" logging.basicConfig(level=logging.INFO) settings = get_settings() DATA_DIR = Path(settings.DATA_DIR) REFRESH_INTERVAL_DAYS = settings.MTG_REFRESH_INTERVAL_DAYS # Use database URL from settings engine = create_async_engine(settings.MTG_DATABASE_URL) last_refresh = await check_last_refresh(engine) refresh_needed = (datetime.now() - last_refresh).days >= REFRESH_INTERVAL_DAYS if not refresh_needed: logger.info("Refresh not needed. Last refresh was within interval.") return start_time = time.time() try: async with aiohttp.ClientSession() as session: # Download dataset download_path = DATA_DIR / "AllPrintings.json" success = await download_mtgjson(session, download_path) if not success: await log_refresh(engine, "FAILED", 0, 0, 0, "Download failed") return # Parse data (returns sets_dict, cards_dict) sets_dict, cards_dict = await parse_mtgjson(download_path) if not sets_dict: await log_refresh(engine, "FAILED", 0, 0, 0, "Parse failed") return # Update database async with AsyncSession(engine) as db_session: cards_updated, sets_updated = await update_database(db_session, sets_dict, cards_dict) # Log success duration = int(time.time() - start_time) await log_refresh(engine, "SUCCESS", cards_updated, sets_updated, duration) logger.info(f"Refresh completed: {cards_updated} cards, {sets_updated} sets in {duration}s") except Exception as e: duration = int(time.time() - start_time) await log_refresh(engine, "FAILED", 0, 0, duration, str(e)) logger.error(f"Refresh failed: {e}") await engine.dispose() if __name__ == "__main__": asyncio.run(main())