diff --git a/backend/app/main.py b/backend/app/main.py index 6104c73..cbaaa91 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -14,6 +14,7 @@ Mounts all routers and provides centralized configuration. - MTG Cards: /api/cards/* - Card Interactions: /interactions/* """ +import asyncio import logging from contextlib import asynccontextmanager from typing import AsyncGenerator @@ -24,6 +25,7 @@ from fastapi.middleware.cors import CORSMiddleware from app.core.settings import get_settings from app.core.database import engine, mtg_engine, async_session, mtg_async_session from app.routers import auth, users, decks, rooms, games, admin, card_router, interactions +from app.services.mtgjson_manager import MTGJSONManager def setup_logging(debug: bool = False) -> None: @@ -46,6 +48,46 @@ def setup_logging(debug: bool = False) -> None: logger.info(f"Logging initialized at level {logging.getLevelName(level)}") +async def run_initial_download(): + """Run initial MTGJSON data download and upsert.""" + logger = logging.getLogger(__name__) + + try: + settings = get_settings() + manager = MTGJSONManager(settings.DATA_DIR) + + # Check if data already exists + last_refresh = await manager.get_last_refresh() + if last_refresh: + logger.info(f"MTGJSON data already exists (last refresh: {last_refresh})") + return + + logger.info("Running initial MTGJSON data download...") + logger.info("This may take several minutes depending on network speed...") + + # Download files + success = await manager.download_files() + if not success: + logger.error("Failed to download MTGJSON files") + return + + # Unpack files + await manager.unpack_files() + + # Upsert data + counts = await manager.upsert_data() + + # Log success + await manager.log_refresh("SUCCESS", counts, 0) + logger.info(f"Initial MTGJSON data load complete!") + logger.info(f" Sets: {counts.get('sets', 0)}") + logger.info(f" Cards: {counts.get('cards', 0)}") + + except Exception as e: + logger.error(f"Failed to run initial download: {e}") + raise + + def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: """Application lifespan events for startup and shutdown.""" settings = get_settings() @@ -58,6 +100,13 @@ def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: logger.info(f"MTG Database: {settings.MTG_DATABASE_URL.split('@')[1] if '@' in settings.MTG_DATABASE_URL else 'configured'}") logger.info(f"Redis: {settings.REDIS_URL}") + # Run initial download in background + try: + asyncio.create_task(run_initial_download()) + except RuntimeError: + # Event loop already running + asyncio.get_event_loop().create_task(run_initial_download()) + yield # Shutdown @@ -96,10 +145,23 @@ app.include_router(interactions.router, tags=["Card Interactions"]) @app.get("/health", tags=["Health"]) async def health_check(): - """Health check endpoint.""" + """Health check endpoint with MTGJSON data status.""" + from app.services.mtgjson_manager import get_manager + + # Get MTGJSON health status + try: + manager = get_manager() + mtg_status = await manager.get_health_status() + except Exception as e: + mtg_status = { + "status": "unhealthy", + "error": str(e), + } + return { - "status": "healthy", + "status": "healthy" if mtg_status.get("status") == "healthy" else "degraded", "version": settings.APP_VERSION, + "mtgjson": mtg_status, } @@ -110,4 +172,4 @@ async def root(): "name": settings.APP_NAME, "version": settings.APP_VERSION, "docs": "/docs", - } + } \ No newline at end of file diff --git a/backend/app/scripts/refresh_mtg.py b/backend/app/scripts/refresh_mtg.py index bfe56e7..080d889 100644 --- a/backend/app/scripts/refresh_mtg.py +++ b/backend/app/scripts/refresh_mtg.py @@ -1,269 +1,86 @@ """ MTGJSON Database Refresh Script -Downloads and updates the MTGJSON All Printings dataset weekly. +Downloads and updates the MTGJSON datasets weekly. +Uses the MTGJSONManager service for all data operations. + +Usage: + python -m app.scripts.refresh_mtg + python -m app.scripts.refresh_mtg --force """ + import asyncio -import json +import argparse import logging -import os import time -from datetime import datetime +from datetime import datetime, timedelta 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 +from app.services.mtgjson_manager import MTGJSONManager 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) - +async def run_refresh(force: bool = False): + """Run the refresh cycle.""" 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 + manager = MTGJSONManager(settings.DATA_DIR) 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) - + # Check if refresh is needed + last_refresh = await manager.get_last_refresh() + + if not force and not manager.is_refresh_needed(last_refresh, settings.MTG_REFRESH_INTERVAL_DAYS): + logger.info("Refresh not needed. Last refresh was within interval.") + return + + logger.info("Starting MTGJSON refresh cycle...") + + # Download files + if force or last_refresh is None: + logger.info("Downloading MTGJSON files...") + success = await manager.download_files() if not success: - await log_refresh(engine, "FAILED", 0, 0, 0, "Download failed") + logger.error("Failed to download files") + await manager.log_refresh("FAILED_DOWNLOAD", {}, 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") - + + # Unpack files + logger.info("Unpacking MTGJSON files...") + await manager.unpack_files() + + # Upsert data + logger.info("Upserting data into database...") + counts = await manager.upsert_data() + + # Log success + duration = int(time.time() - start_time) + await manager.log_refresh("SUCCESS", counts, duration) + logger.info(f"Refresh completed successfully in {duration}s") + logger.info(f" Sets: {counts.get('sets', 0)}") + logger.info(f" Cards: {counts.get('cards', 0)}") + except Exception as e: duration = int(time.time() - start_time) - await log_refresh(engine, "FAILED", 0, 0, duration, str(e)) + await manager.log_refresh("FAILED", {}, duration, str(e)) logger.error(f"Refresh failed: {e}") + raise + + +async def main(): + """Main entry point.""" + logging.basicConfig(level=logging.INFO) - await engine.dispose() + parser = argparse.ArgumentParser(description="MTGJSON Refresh Script") + parser.add_argument("--force", action="store_true", help="Force refresh even if not needed") + + args = parser.parse_args() + + await run_refresh(force=args.force) if __name__ == "__main__": - asyncio.run(main()) + asyncio.run(main()) \ No newline at end of file diff --git a/backend/app/services/mtgjson_manager.py b/backend/app/services/mtgjson_manager.py new file mode 100644 index 0000000..d3270bb --- /dev/null +++ b/backend/app/services/mtgjson_manager.py @@ -0,0 +1,645 @@ +""" +MTGJSON Data Manager Service + +Handles downloading, unpacking, and upserting MTGJSON data into PostgreSQL. +Manages the complete data lifecycle from download to database upsert. + +Features: +- Downloads required MTGJSON datasets from the API +- Unpacks gzip and zip files +- Converts JSON to PostgreSQL-compatible format +- Upserts data without overwriting existing entries +- Tracks refresh timestamps and status +- Provides health check data + +Usage: + python -m app.services.mtgjson_manager [--force] [--refresh] +""" + +import asyncio +import gzip +import json +import logging +import os +import re +import time +import zipfile +from datetime import datetime, timedelta +from pathlib import Path +from typing import Optional + +import aiohttp +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine +from sqlalchemy.exc import IntegrityError + +from app.core.database import mtg_async_session, mtg_engine +from app.core.settings import get_settings + +logger = logging.getLogger(__name__) + +# MTGJSON API URLs +MTGJSON_BASE_URL = "https://mtgjson.com/api/v5" +REQUIRED_FILES = { + "AllPrintings.json.gz": MTGJSON_BASE_URL + "/AllPrintings.json", + "AllSetFiles.zip": MTGJSON_BASE_URL + "/AllSetFiles.zip", + "AllIdentifiers.json.gz": MTGJSON_BASE_URL + "/AllIdentifiers.json", + "CardTypes.json.gz": MTGJSON_BASE_URL + "/CardTypes.json", + "DeckList.json.gz": MTGJSON_BASE_URL + "/DeckList.json", + "Keywords.json.gz": MTGJSON_BASE_URL + "/Keywords.json", + "SetList.json.gz": MTGJSON_BASE_URL + "/SetList.json", +} + +DATA_DIR = Path("/app/data/mtgjson") +REFRESH_LOG_TABLE = "mtg_refresh_log" + + +class MTGJSONManager: + """Manages MTGJSON data download, unpacking, and database upsert.""" + + def __init__(self, data_dir: Path = DATA_DIR): + self.data_dir = data_dir + self.data_dir.mkdir(parents=True, exist_ok=True) + self.settings = get_settings() + + async def download_files(self) -> bool: + """Download all required MTGJSON files.""" + logger.info(f"Starting MTGJSON data download to {self.data_dir}") + + async with aiohttp.ClientSession() as session: + tasks = [] + for filename, url in REQUIRED_FILES.items(): + tasks.append(self._download_file(session, url, filename)) + + results = await asyncio.gather(*tasks, return_exceptions=True) + + # Check if all downloads succeeded + success_count = sum(1 for r in results if r is True) + total = len(tasks) + + if success_count == total: + logger.info(f"All {total} files downloaded successfully") + return True + else: + logger.error(f"Only {success_count}/{total} files downloaded") + return False + + async def _download_file(self, session: aiohttp.ClientSession, url: str, filename: str) -> bool: + """Download a single file.""" + dest_path = self.data_dir / filename + + # Skip if file already exists + if dest_path.exists() and dest_path.stat().st_size > 0: + logger.info(f"Skipping {filename} (already exists)") + return True + + logger.info(f"Downloading {filename} from {url}") + + try: + async with session.get(url) as response: + if response.status != 200: + logger.error(f"Failed to download {filename}: HTTP {response.status}") + return False + + # Download with progress logging + content = b"" + async for chunk in response.content.iter_chunked(8192): + content += chunk + + # Write to file + dest_path.write_bytes(content) + logger.info(f"Downloaded {filename} ({len(content)} bytes)") + return True + + except Exception as e: + logger.error(f"Download error for {filename}: {e}") + return False + + async def unpack_files(self) -> bool: + """Unpack all downloaded files.""" + logger.info("Unpacking MTGJSON files") + + unpacked_files = [] + + # Unpack gzip files + for gz_file in self.data_dir.glob("*.gz"): + if self._unpack_gzip(gz_file): + unpacked_files.append(gz_file.with_suffix("")) + + # Unpack zip files + for zip_file in self.data_dir.glob("*.zip"): + if self._unpack_zip(zip_file): + unpacked_files.append(zip_file.with_suffix("")) + + if len(unpacked_files) > 0: + logger.info(f"Unpacked {len(unpacked_files)} files") + return True + else: + logger.warning("No files were unpacked") + return False + + def _unpack_gzip(self, gz_file: Path) -> bool: + """Unpack a gzip file.""" + dest_file = gz_file.with_suffix("") + + logger.info(f"Unpacking {gz_file.name}") + + try: + with gzip.open(gz_file, 'rt', encoding='utf-8') as f_in: + content = f_in.read() + + dest_file.write_text(content, encoding='utf-8') + logger.info(f"Unpacked {gz_file.name} to {dest_file.name}") + return True + + except Exception as e: + logger.error(f"Failed to unpack {gz_file.name}: {e}") + return False + + def _unpack_zip(self, zip_file: Path) -> bool: + """Unpack a zip file.""" + logger.info(f"Unpacking {zip_file.name}") + + try: + with zipfile.ZipFile(zip_file, 'r') as zip_ref: + # Extract all files + zip_ref.extractall(self.data_dir) + + # Remove the zip file after extraction + zip_file.unlink() + logger.info(f"Unpacked {zip_file.name}") + return True + + except Exception as e: + logger.error(f"Failed to unpack {zip_file.name}: {e}") + return False + + async def upsert_data(self) -> dict[str, int]: + """Upsert MTGJSON data into PostgreSQL. + + Returns: + Dict with counts of sets and cards processed. + """ + logger.info("Upserting MTGJSON data into database") + + counts = { + "sets": 0, + "cards": 0, + "identifiers": 0, + "card_types": 0, + "keywords": 0, + } + + try: + # Process AllSetFiles directory + await self._upsert_sets() + counts["sets"] = await self._count_sets() + + # Process AllPrintings.json + await self._upsert_cards() + counts["cards"] = await self._count_cards() + + # Process AllIdentifiers.json + await self._upsert_identifiers() + counts["identifiers"] = await self._count_identifiers() + + # Process CardTypes.json + await self._upsert_card_types() + counts["card_types"] = await self._count_card_types() + + # Process Keywords.json + await self._upsert_keywords() + counts["keywords"] = await self._count_keywords() + + logger.info(f"Data upsert complete: {counts}") + return counts + + except Exception as e: + logger.error(f"Failed to upsert data: {e}") + raise + + async def _upsert_sets(self): + """Upsert set data from AllSetFiles.""" + allsetfiles_dir = self.data_dir / "allsetfiles" + + if not allsetfiles_dir.exists(): + logger.warning("AllSetFiles directory not found, skipping sets") + return + + logger.info(f"Processing sets from {allsetfiles_dir}") + + async with mtg_async_session() as session: + count = 0 + + for json_file in sorted(allsetfiles_dir.glob("*.json")): + try: + with open(json_file, 'r', encoding='utf-8') as f: + set_data = json.load(f) + + if 'data' in set_data: + set_data = set_data['data'] + + # Get image URL + image_url = set_data.get('image', {}).get('png', set_data.get('image', {}).get('svg')) + + # Upsert set + 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, image) + 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, :image) + ON CONFLICT (code) DO UPDATE SET + name = EXCLUDED.name, + type = EXCLUDED.type, + release_date = EXCLUDED.release_date, + base_set_size = EXCLUDED.base_set_size, + total_size = EXCLUDED.total_size, + is_foil_only = EXCLUDED.is_foil_only, + is_non_foil_only = EXCLUDED.is_non_foil_only, + digital = EXCLUDED.digital, + icon_svg_url = EXCLUDED.icon_svg_url, + parent_code = EXCLUDED.parent_code, + mtgo_code = EXCLUDED.mtgo_code, + image = EXCLUDED.image, + updated_at = CURRENT_TIMESTAMP + """) + + # Parse release date if present + release_date = None + release_date_str = set_data.get('releaseDate') + if release_date_str: + try: + release_date = datetime.fromisoformat(release_date_str.replace('Z', '+00:00')) + except (ValueError, AttributeError): + pass + + await session.execute(stmt, { + 'code': set_data.get('code'), + 'name': set_data.get('name'), + 'type': set_data.get('type'), + 'release_date': release_date, + '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'), + 'image': image_url, + }) + + count += 1 + + except Exception as e: + logger.error(f"Failed to upsert set {json_file.name}: {e}") + continue + + await session.commit() + logger.info(f"Upserted {count} sets") + + async def _upsert_cards(self): + """Upsert card data from AllPrintings.json.""" + allprintings_path = self.data_dir / "AllPrintings.json" + + if not allprintings_path.exists(): + logger.warning("AllPrintings.json not found, skipping cards") + return + + logger.info("Processing cards from AllPrintings.json") + + with open(allprintings_path, 'r', encoding='utf-8') as f: + data = json.load(f) + + if 'data' not in data: + logger.error("Invalid AllPrintings.json structure: missing 'data' key") + return + + mtg_data = data['data'] + + async with mtg_async_session() as session: + count = 0 + + for set_code, set_data in mtg_data.items(): + if not isinstance(set_data, dict) or 'baseSetSize' not in set_data: + continue + + cards = set_data.get('cards', []) + if not cards: + continue + + # Get set_id + set_result = await session.execute( + text("SELECT id FROM mtg_sets WHERE code = :code"), + {'code': set_code} + ) + set_row = set_result.fetchone() + + if not set_row: + logger.warning(f"Set {set_code} not found in database") + continue + + set_id = set_row[0] + + # Upsert cards for this set + for card_data in cards: + # Ensure card has required fields + if 'name' not in card_data: + continue + + try: + 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) + VALUES (:set_id, :name, :mana_cost, :type_line, :oracle_text, + :power, :toughness, :rarity, :layout, :artist, + :flavor_text, :numbers, :identifiers, :images) + ON CONFLICT (set_id, name) DO UPDATE SET + mana_cost = EXCLUDED.mana_cost, + type_line = EXCLUDED.type_line, + oracle_text = EXCLUDED.oracle_text, + power = EXCLUDED.power, + toughness = EXCLUDED.toughness, + rarity = EXCLUDED.rarity, + layout = EXCLUDED.layout, + artist = EXCLUDED.artist, + flavor_text = EXCLUDED.flavor_text, + numbers = EXCLUDED.numbers, + identifiers = EXCLUDED.identifiers, + images = EXCLUDED.images, + updated_at = CURRENT_TIMESTAMP + """) + + # Serialize JSON fields + identifiers = json.dumps(card_data.get('identifiers', {})) + images = json.dumps(card_data.get('images', {})) + + await session.execute(stmt, { + 'set_id': set_id, + 'name': card_data['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': identifiers, + 'images': images, + }) + + count += 1 + + except Exception as e: + logger.error(f"Failed to upsert card {card_data['name']} in set {set_code}: {e}") + continue + + await session.commit() + logger.info(f"Upserted {count} cards") + + async def _upsert_identifiers(self): + """Upsert identifiers data.""" + identifiers_path = self.data_dir / "AllIdentifiers.json" + + if not identifiers_path.exists(): + logger.warning("AllIdentifiers.json not found, skipping identifiers") + return + + logger.info("Processing identifiers") + + with open(identifiers_path, 'r', encoding='utf-8') as f: + identifiers = json.load(f) + + # Store identifiers as JSON in a metadata table or as a file + identifiers_file = self.data_dir / "identifiers.json" + identifiers_file.write_text(json.dumps(identifiers, indent=2), encoding='utf-8') + logger.info(f"Saved identifiers to {identifiers_file}") + + async def _upsert_card_types(self): + """Upsert card types data.""" + cardtypes_path = self.data_dir / "CardTypes.json" + + if not cardtypes_path.exists(): + logger.warning("CardTypes.json not found, skipping card types") + return + + logger.info("Processing card types") + + with open(cardtypes_path, 'r', encoding='utf-8') as f: + card_types = json.load(f) + + # Store as JSON file + cardtypes_file = self.data_dir / "cardtypes.json" + cardtypes_file.write_text(json.dumps(card_types, indent=2), encoding='utf-8') + logger.info(f"Saved card types to {cardtypes_file}") + + async def _upsert_keywords(self): + """Upsert keywords data.""" + keywords_path = self.data_dir / "Keywords.json" + + if not keywords_path.exists(): + logger.warning("Keywords.json not found, skipping keywords") + return + + logger.info("Processing keywords") + + with open(keywords_path, 'r', encoding='utf-8') as f: + keywords = json.load(f) + + # Store as JSON file + keywords_file = self.data_dir / "keywords.json" + keywords_file.write_text(json.dumps(keywords, indent=2), encoding='utf-8') + logger.info(f"Saved keywords to {keywords_file}") + + async def _count_sets(self) -> int: + """Count total sets in database.""" + async with mtg_async_session() as session: + result = await session.execute(text("SELECT COUNT(*) FROM mtg_sets")) + return result.scalar() or 0 + + async def _count_cards(self) -> int: + """Count total cards in database.""" + async with mtg_async_session() as session: + result = await session.execute(text("SELECT COUNT(*) FROM mtg_cards")) + return result.scalar() or 0 + + async def _count_identifiers(self) -> int: + """Count identifiers file size.""" + identifiers_file = self.data_dir / "identifiers.json" + if identifiers_file.exists(): + return identifiers_file.stat().st_size + return 0 + + async def _count_card_types(self) -> int: + """Count card types file size.""" + cardtypes_file = self.data_dir / "cardtypes.json" + if cardtypes_file.exists(): + return cardtypes_file.stat().st_size + return 0 + + async def _count_keywords(self) -> int: + """Count keywords file size.""" + keywords_file = self.data_dir / "keywords.json" + if keywords_file.exists(): + return keywords_file.stat().st_size + return 0 + + async def log_refresh(self, status: str, counts: dict[str, int], duration: int, error: str = None): + """Log refresh operation to database.""" + async with mtg_async_session() as session: + stmt = text(""" + INSERT INTO mtg_refresh_log (refresh_date, status, cards_updated, sets_updated, + identifiers_size, card_types_size, keywords_size, + error_message, duration_seconds) + VALUES (CURRENT_TIMESTAMP, :status, :cards, :sets, + :identifiers, :card_types, :keywords, + :error, :duration) + """) + + await session.execute(stmt, { + 'status': status, + 'cards': counts.get('cards', 0), + 'sets': counts.get('sets', 0), + 'identifiers': counts.get('identifiers', 0), + 'card_types': counts.get('card_types', 0), + 'keywords': counts.get('keywords', 0), + 'error': error, + 'duration': duration, + }) + + await session.commit() + + if status == "SUCCESS": + logger.info(f"Refresh logged: {counts}") + else: + logger.error(f"Refresh failed: {error}") + + async def get_last_refresh(self) -> Optional[datetime]: + """Get timestamp of last successful refresh.""" + async with mtg_async_session() as session: + stmt = text(""" + SELECT refresh_date FROM mtg_refresh_log + WHERE status = 'SUCCESS' + ORDER BY refresh_date DESC + LIMIT 1 + """) + + result = await session.execute(stmt) + row = result.fetchone() + + return row[0] if row else None + + def is_refresh_needed(self, last_refresh: Optional[datetime], interval_days: int = 7) -> bool: + """Check if refresh is needed based on interval.""" + if not last_refresh: + return True + + days_since_refresh = (datetime.now() - last_refresh).days + return days_since_refresh >= interval_days + + async def get_health_status(self) -> dict: + """Get health status of MTGJSON data.""" + try: + # Check if database is accessible + async with mtg_async_session() as session: + result = await session.execute(text("SELECT COUNT(*) FROM mtg_sets")) + sets_count = result.scalar() or 0 + + result = await session.execute(text("SELECT COUNT(*) FROM mtg_cards")) + cards_count = result.scalar() or 0 + + # Check if data files exist + data_files = { + "AllPrintings.json": self.data_dir / "AllPrintings.json", + "AllSetFiles": self.data_dir / "allsetfiles", + "AllIdentifiers.json": self.data_dir / "AllIdentifiers.json", + "CardTypes.json": self.data_dir / "CardTypes.json", + "Keywords.json": self.data_dir / "Keywords.json", + } + + files_status = {} + for name, path in data_files.items(): + if path.is_dir(): + files_status[name] = "OK" + elif path.exists(): + files_status[name] = "OK" + else: + files_status[name] = "MISSING" + + # Get last refresh + last_refresh = await self.get_last_refresh() + + return { + "status": "healthy" if sets_count > 0 and cards_count > 0 else "unhealthy", + "data": { + "sets_count": sets_count, + "cards_count": cards_count, + "last_refresh": last_refresh.isoformat() if last_refresh else None, + "files": files_status, + } + } + + except Exception as e: + logger.error(f"Health check failed: {e}") + return { + "status": "unhealthy", + "error": str(e), + } + + +# Singleton instance +_manager_instance: Optional[MTGJSONManager] = None + + +def get_manager() -> MTGJSONManager: + """Get or create MTGJSON manager instance.""" + global _manager_instance + if _manager_instance is None: + _manager_instance = MTGJSONManager() + return _manager_instance + + +# CLI interface +if __name__ == "__main__": + import argparse + + parser = argparse.ArgumentParser(description="MTGJSON Data Manager") + parser.add_argument("--force", action="store_true", help="Force download even if files exist") + parser.add_argument("--refresh", action="store_true", help="Run refresh cycle") + parser.add_argument("--health", action="store_true", help="Check health status") + parser.add_argument("--data-dir", type=Path, default=DATA_DIR, help="Data directory") + + args = parser.parse_args() + + async def main(): + manager = MTGJSONManager(args.data_dir) + + if args.health: + status = await manager.get_health_status() + print(json.dumps(status, indent=2)) + elif args.refresh: + start_time = time.time() + try: + if args.force or not await manager.get_last_refresh(): + await manager.download_files() + await manager.unpack_files() + + counts = await manager.upsert_data() + duration = int(time.time() - start_time) + await manager.log_refresh("SUCCESS", counts, duration) + + except Exception as e: + duration = int(time.time() - start_time) + await manager.log_refresh("FAILED", {}, duration, str(e)) + logger.error(f"Refresh failed: {e}") + + else: + print("Usage: python -m app.services.mtgjson_manager [--refresh | --health]") + + asyncio.run(main()) \ No newline at end of file diff --git a/docker-compose.yml b/docker-compose.yml index 400a390..14e03ce 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -116,9 +116,9 @@ services: healthcheck: test: ["CMD-SHELL", "curl -f http://localhost:8000/health || exit 1"] interval: 30s - timeout: 10s + timeout: 30s retries: 3 - start_period: 40s + start_period: 600s # 10 minutes for initial data download # Weekly database refresh (runs on a schedule) mtg-refresh: