From ea3c1d9691dcb9581695c8b3c858d50b9a7588dc Mon Sep 17 00:00:00 2001 From: admin Date: Mon, 20 Jul 2026 22:20:26 +0000 Subject: [PATCH] Refactor MTGJSON manager to use local files only --- backend/app/services/mtgjson_manager.py | 1229 ++++++++++------------- 1 file changed, 517 insertions(+), 712 deletions(-) diff --git a/backend/app/services/mtgjson_manager.py b/backend/app/services/mtgjson_manager.py index bb550da..a6002fe 100644 --- a/backend/app/services/mtgjson_manager.py +++ b/backend/app/services/mtgjson_manager.py @@ -1,727 +1,76 @@ """ -MTGJSON Data Manager Service +MTGJSON Data Manager - Local File Processing -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 (JSON files, except AllSetFiles as zip) -- Unpacks zip files (AllSetFiles.zip only) -- 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] +Reads JSON files from a mounted volume and upserts them into PostgreSQL. +No network downloads - users provide the files. """ -import asyncio import json import logging -import os -import time -import zipfile -from datetime import datetime, timedelta from pathlib import Path +from datetime import datetime 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 +from app.core.database import mtg_async_session logger = logging.getLogger(__name__) -# MTGJSON API URLs - .json files for most datasets, .zip for AllSetFiles -MTGJSON_BASE_URL = "https://mtgjson.com/api/v5" -REQUIRED_JSON_FILES = { - "AllPrintings.json": MTGJSON_BASE_URL + "/AllPrintings.json", - "AllIdentifiers.json": MTGJSON_BASE_URL + "/AllIdentifiers.json", - "CardTypes.json": MTGJSON_BASE_URL + "/CardTypes.json", - "DeckList.json": MTGJSON_BASE_URL + "/DeckList.json", - "Keywords.json": MTGJSON_BASE_URL + "/Keywords.json", - "SetList.json": MTGJSON_BASE_URL + "/SetList.json", -} - -REQUIRED_ZIP_FILES = { - "AllSetFiles.zip": MTGJSON_BASE_URL + "/AllSetFiles.zip", -} - -DATA_DIR = Path("/app/data/mtgjson") -REFRESH_LOG_TABLE = "mtg_refresh_log" - -# Expected minimum file sizes (in bytes) for MTGJSON v5 JSON files -EXPECTED_MIN_SIZES = { - "AllPrintings.json": 500 * 1024 * 1024, # 500 MB (actual: ~620 MB) - "AllIdentifiers.json": 500 * 1024 * 1024, # 500 MB (actual: ~599 MB) - "CardTypes.json": 0.001 * 1024 * 1024, # 0.001 MB (actual: ~0.01 MB) - "DeckList.json": 0.5 * 1024 * 1024, # 0.5 MB (actual: ~0.59 MB) - "Keywords.json": 0.001 * 1024 * 1024, # 0.001 MB (actual: ~0.00 MB) - "SetList.json": 5 * 1024 * 1024, # 5 MB (actual: ~11 MB) - "AllSetFiles.zip": 50 * 1024 * 1024, # 50 MB (expected size) -} - -MAX_DOWNLOAD_RETRIES = 3 -RETRY_DELAY_SECONDS = 120 +# Expected files to process +EXPECTED_FILES = [ + "AllSetFiles/", # Directory containing set JSON files + "AllPrintings.json", + "AllIdentifiers.json", + "CardTypes.json", + "DeckList.json", + "Keywords.json", + "SetList.json", +] class MTGJSONManager: - """Manages MTGJSON data download, unpacking, and database upsert.""" + """Process local MTGJSON files and upsert into PostgreSQL.""" def __init__(self, data_dir: Path = None): - """Initialize MTGJSONManager. - - Args: - data_dir: Path to data directory. If None, uses settings.DATA_DIR - """ if data_dir is None: - settings = get_settings() - data_dir = settings.DATA_DIR - - # Convert string to Path if needed + data_dir = Path("/app/data") if isinstance(data_dir, str): 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 validate_data_integrity(self) -> tuple[bool, list[str]]: - """ - Validate downloaded MTGJSON files for expected sizes. - - Returns: - Tuple of (is_valid, list_of_issues) - """ - issues = [] - - for filename, min_size in EXPECTED_MIN_SIZES.items(): - filepath = self.data_dir / filename - - if not filepath.exists(): - issues.append(f"Missing file: {filename}") - continue - - actual_size = filepath.stat().st_size - - if actual_size < min_size: - issues.append( - f"{filename}: {actual_size / (1024*1024):.1f} MB " - f"(minimum: {min_size / (1024*1024):.1f} MB)" - ) - - if issues: - logger.error(f"Data integrity check failed with {len(issues)} issues:") - for issue in issues: - logger.error(f" - {issue}") - return False, issues - - logger.info("Data integrity check passed - all files meet minimum size requirements") - return True, [] - - def _get_estimated_size(self, filename: str) -> int: - """Get estimated file size in MB.""" - estimates = { - "AllPrintings.json": 620, - "AllSetFiles.zip": 10, - "AllIdentifiers.json": 600, - "CardTypes.json": 0.01, - "DeckList.json": 0.6, - "Keywords.json": 0.01, - "SetList.json": 11, - } - return estimates.get(filename, 10) # Default 10MB - - async def cleanup_data_files(self) -> None: - """Delete all downloaded MTGJSON data files.""" - logger.info(f"Cleaning up data files in {self.data_dir}") - - try: - for f in self.data_dir.glob("*"): - if f.is_file(): - f.unlink() - logger.info(f"Deleted: {f.name}") - logger.info("Data cleanup complete") - except Exception as e: - logger.error(f"Failed to clean up data files: {e}") - raise - - async def download_with_sanity_check(self) -> bool: - """ - Download MTGJSON files with sanity checking and retry logic. - - Returns: - True if download and validation succeed, False otherwise - """ - logger.info("Starting MTGJSON download with sanity checks") - - for attempt in range(1, MAX_DOWNLOAD_RETRIES + 1): - logger.info(f"Download attempt {attempt}/{MAX_DOWNLOAD_RETRIES}") - - # Download files - download_success = await self.download_files() - - if not download_success: - logger.error(f"Download failed on attempt {attempt}") - if attempt < MAX_DOWNLOAD_RETRIES: - logger.info(f"Waiting {RETRY_DELAY_SECONDS * attempt}s before retry...") - await asyncio.sleep(RETRY_DELAY_SECONDS * attempt) - await self.cleanup_data_files() - continue - return False - - # Unpack zip files (AllSetFiles.zip only) - unpack_success = await self.unpack_files() - - if not unpack_success: - logger.error(f"Unpack failed on attempt {attempt}") - if attempt < MAX_DOWNLOAD_RETRIES: - logger.info(f"Waiting {RETRY_DELAY_SECONDS * attempt}s before retry...") - await asyncio.sleep(RETRY_DELAY_SECONDS * attempt) - await self.cleanup_data_files() - continue - return False - - # Validate data integrity - is_valid, issues = await self.validate_data_integrity() - - if is_valid: - logger.info(f"✓ Download and validation successful on attempt {attempt}") - return True - - logger.warning(f"Data validation failed on attempt {attempt} with {len(issues)} issues") - - if attempt < MAX_DOWNLOAD_RETRIES: - logger.info(f"Cleaning up and retrying in {RETRY_DELAY_SECONDS * attempt}s...") - await self.cleanup_data_files() - await asyncio.sleep(RETRY_DELAY_SECONDS * attempt) - else: - logger.error(f"Data validation failed after {MAX_DOWNLOAD_RETRIES} attempts") - return False - - return False - - async def download_files(self) -> bool: - """Download all required MTGJSON files (JSON and ZIP).""" - logger.info(f"Starting MTGJSON data download to {self.data_dir}") - - async with aiohttp.ClientSession() as session: - tasks = [] - for filename, url in REQUIRED_JSON_FILES.items(): - tasks.append(self._download_file(session, url, filename)) - for filename, url in REQUIRED_ZIP_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 with adaptive timeout based on size.""" - 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 - - # Calculate timeout based on estimated file size - # Large files (>100MB) get up to 60 minutes - estimated_mb = self._get_estimated_size(filename) - timeout = max(3600, estimated_mb * 8) # At least 60 min, 8x estimated MB - - logger.info(f"Downloading {filename} from {url} (timeout: {timeout}s, est: {estimated_mb}MB)") - - try: - # Set timeout for connection AND download - timeout_obj = aiohttp.ClientTimeout(total=timeout) - - async with session.get(url, timeout=timeout_obj) as response: - if response.status != 200: - logger.error(f"Failed to download {filename}: HTTP {response.status}") - return False - - # Download with progress monitoring - total_size = 0 - last_log_time = time.time() - - with open(dest_path, 'wb') as f: - async for chunk in response.content.iter_chunked(8192): - f.write(chunk) - total_size += len(chunk) - - # Log progress every 60 seconds - now = time.time() - if now - last_log_time >= 60: - mb_downloaded = total_size / (1024 * 1024) - logger.info(f" Downloaded {mb_downloaded:.1f} MB of {filename}") - last_log_time = now - - logger.info(f"Downloaded {filename} ({total_size / (1024*1024):.1f} MB)") - return True - - except asyncio.TimeoutError: - logger.error(f"Download timeout for {filename} (timeout: {timeout}s)") - # Clean up partial download - if dest_path.exists(): - dest_path.unlink() - return False - except Exception as e: - logger.error(f"Download error for {filename}: {e}") - if dest_path.exists(): - dest_path.unlink() - return False - - async def unpack_files(self) -> bool: - """Unpack all downloaded zip files.""" - logger.info("Unpacking MTGJSON zip files") - - unpacked_files = [] - - # Unpack zip files only - for zip_file in self.data_dir.glob("*.zip"): - if self._unpack_zip(zip_file): - unpacked_files.append(zip_file) - - if len(unpacked_files) > 0: - logger.info(f"Unpacked {len(unpacked_files)} zip files") - return True - else: - logger.warning("No zip files were unpacked") - 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 (from zip) - 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(""" + result = await session.execute(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.""" + """Get health status.""" 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 + for f in self.data_dir.iterdir(): + if f.is_file() and f.suffix == '.json': + files_status[f.stem + '.json'] = 'OK' + elif f.is_dir(): + files_status[f.name] = 'OK' + last_refresh = await self.get_last_refresh() - + return { "status": "healthy" if sets_count > 0 and cards_count > 0 else "unhealthy", "data": { @@ -731,17 +80,489 @@ class MTGJSONManager: "files": files_status, } } + except Exception as e: + return {"status": "unhealthy", "error": str(e)} + + async def upsert_all(self) -> dict: + """Upsert all data from local files.""" + counts = {} + + # Check what files exist + available = set() + for item in self.data_dir.iterdir(): + if item.is_file(): + available.add(item.name) + elif item.is_dir(): + available.add(item.name) + + logger.info(f"Found files: {available}") + + # Upsert each file type + if "AllSetFiles" in available: + counts["sets"] = await self._upsert_sets() + + if "AllPrintings.json" in available: + counts["cards"] = await self._upsert_cards() + + if "AllIdentifiers.json" in available: + counts["identifiers"] = await self._upsert_identifiers() + + if "CardTypes.json" in available: + counts["card_types"] = await self._upsert_card_types() + + if "Keywords.json" in available: + counts["keywords"] = await self._upsert_keywords() + + if "SetList.json" in available: + counts["set_list"] = await self._upsert_set_list() + + if "DeckList.json" in available: + counts["deck_list"] = await self._upsert_deck_list() + + return counts + + async def _upsert_sets(self) -> int: + """Upsert sets from AllSetFiles directory.""" + allsetfiles = self.data_dir / "AllSetFiles" + + if not allsetfiles.exists(): + logger.warning("AllSetFiles directory not found") + return 0 + + logger.info(f"Processing sets from {allsetfiles}") + + count = 0 + + async with mtg_async_session() as session: + for json_file in sorted(allsetfiles.glob("*.json")): + logger.info(f"Processing {json_file.name}") + + try: + with open(json_file, 'r', encoding='utf-8') as f: + data = json.load(f) + + if not isinstance(data, dict): + logger.warning(f"{json_file.name} is not a dict") + continue + + # Upsert sets + sets = data.get("sets", []) + for set_data in sets: + if not isinstance(set_data, dict): + continue + + await session.execute(text(""" + INSERT INTO mtg_sets (set_name, code, name, type, release_date, + scryfall_uri, card_count, tokens_count, + is_foil_only, is_non_foil_only) + VALUES ( + :set_name, :code, :name, :type, :release_date, + :scryfall_uri, :card_count, :tokens_count, + :is_foil_only, :is_non_foil_only + ) + ON CONFLICT (set_name) DO UPDATE SET + code = EXCLUDED.code, + name = EXCLUDED.name, + type = EXCLUDED.type, + release_date = EXCLUDED.release_date, + scryfall_uri = EXCLUDED.scryfall_uri, + card_count = EXCLUDED.card_count, + tokens_count = EXCLUDED.tokens_count, + is_foil_only = EXCLUDED.is_foil_only, + is_non_foil_only = EXCLUDED.is_non_foil_only + """), { + 'set_name': set_data.get('name'), + 'code': set_data.get('code'), + 'name': set_data.get('name'), + 'type': set_data.get('type'), + 'release_date': set_data.get('release_date'), + 'scryfall_uri': set_data.get('scryfall_uri'), + 'card_count': set_data.get('card_count'), + 'tokens_count': set_data.get('tokens_count'), + 'is_foil_only': set_data.get('is_foil_only'), + 'is_non_foil_only': set_data.get('is_non_foil_only'), + }) + + count += 1 + + await session.commit() + logger.info(f"Upserted {count} sets from {json_file.name}") + + except Exception as e: + logger.error(f"Failed to process {json_file.name}: {e}") + await session.rollback() + continue + + return count + + async def _upsert_cards(self) -> int: + """Upsert cards from AllPrintings.json.""" + path = self.data_dir / "AllPrintings.json" + + if not path.exists(): + logger.warning("AllPrintings.json not found") + return 0 + + logger.info(f"Processing cards from {path}") + + count = 0 + + with open(path, 'r', encoding='utf-8') as f: + data = json.load(f) + + cards = data.get("cards", []) + + async with mtg_async_session() as session: + for card in cards: + if not isinstance(card, dict): + continue + + try: + await session.execute(text(""" + INSERT INTO mtg_cards ( + card_name, mtgo_id, card_type, set_name, + rarity, artist, number, language, mana_cost, + text, power, toughness, loyalty, colors, + color_identity, produced_mana, legalities, + original_type, foreign_data, rulings, + hand_modifier, life_modifier, side_names + ) + VALUES ( + :card_name, :mtgo_id, :card_type, :set_name, + :rarity, :artist, :number, :language, :mana_cost, + :text, :power, :toughness, :loyalty, :colors, + :color_identity, :produced_mana, :legalities, + :original_type, :foreign_data, :rulings, + :hand_modifier, :life_modifier, :side_names + ) + ON CONFLICT (card_name, set_name, mtgo_id) DO UPDATE SET + card_type = EXCLUDED.card_type, + rarity = EXCLUDED.rarity, + artist = EXCLUDED.artist, + number = EXCLUDED.number, + language = EXCLUDED.language, + mana_cost = EXCLUDED.mana_cost, + text = EXCLUDED.text, + power = EXCLUDED.power, + toughness = EXCLUDED.toughness, + loyalty = EXCLUDED.loyalty, + colors = EXCLUDED.colors, + color_identity = EXCLUDED.color_identity, + produced_mana = EXCLUDED.produced_mana, + legalities = EXCLUDED.legalities, + original_type = EXCLUDED.original_type, + foreign_data = EXCLUDED.foreign_data, + rulings = EXCLUDED.rulings, + hand_modifier = EXCLUDED.hand_modifier, + life_modifier = EXCLUDED.life_modifier, + side_names = EXCLUDED.side_names + """), { + 'card_name': card.get('name'), + 'mtgo_id': card.get('mtgoId'), + 'card_type': card.get('type'), + 'set_name': card.get('setName'), + 'rarity': card.get('rarity'), + 'artist': card.get('artist'), + 'number': card.get('number'), + 'language': card.get('language'), + 'mana_cost': json.dumps(card.get('manaCost')), + 'text': card.get('text'), + 'power': card.get('power'), + 'toughness': card.get('toughness'), + 'loyalty': card.get('loyalty'), + 'colors': json.dumps(card.get('colors')), + 'color_identity': json.dumps(card.get('colorIdentity')), + 'produced_mana': json.dumps(card.get('producedMana')), + 'legalities': json.dumps(card.get('legalities')), + 'original_type': card.get('originalType'), + 'foreign_data': json.dumps(card.get('foreignData')), + 'rulings': json.dumps(card.get('rulings')), + 'hand_modifier': card.get('handModifier'), + 'life_modifier': card.get('lifeModifier'), + 'side_names': json.dumps(card.get('sideNames')), + }) + + count += 1 + + except Exception as e: + logger.error(f"Failed to upsert card: {e}") + await session.rollback() + continue + + logger.info(f"Upserted {count} cards") + return count + + async def _upsert_identifiers(self) -> int: + """Upsert identifiers from AllIdentifiers.json.""" + path = self.data_dir / "AllIdentifiers.json" + + if not path.exists(): + logger.warning("AllIdentifiers.json not found") + return 0 + + logger.info(f"Processing identifiers from {path}") + + with open(path, 'r', encoding='utf-8') as f: + data = json.load(f) + + identifiers = data.get("identifiers", {}) + + async with mtg_async_session() as session: + count = 0 + for identifier_id, identifier in identifiers.items(): + if not isinstance(identifier, dict): + continue + + try: + await session.execute(text(""" + INSERT INTO mtg_identifiers (identifier_id, identifier_data, created_at) + VALUES (:identifier_id, :identifier_data, CURRENT_TIMESTAMP) + ON CONFLICT (identifier_id) DO UPDATE SET + identifier_data = EXCLUDED.identifier_data, + updated_at = CURRENT_TIMESTAMP + """), { + 'identifier_id': identifier_id, + 'identifier_data': json.dumps(identifier), + }) + + count += 1 + + except Exception as e: + logger.error(f"Failed to upsert identifier {identifier_id}: {e}") + continue + + await session.commit() + logger.info(f"Upserted {count} identifiers") + + return count + + async def _upsert_card_types(self) -> int: + """Upsert card types from CardTypes.json.""" + path = self.data_dir / "CardTypes.json" + + if not path.exists(): + logger.warning("CardTypes.json not found") + return 0 + + logger.info("Processing card types") + + with open(path, 'r', encoding='utf-8') as f: + card_types = json.load(f) + + async with mtg_async_session() as session: + count = 0 + + for card_type in card_types: + if not isinstance(card_type, dict): + continue + + try: + await session.execute(text(""" + INSERT INTO mtg_card_types (card_type) + VALUES (:card_type) + ON CONFLICT (card_type) DO NOTHING + """), { + 'card_type': card_type.get('cardType'), + }) + + count += 1 + + except Exception as e: + logger.error(f"Failed to upsert card type: {e}") + continue + + await session.commit() + logger.info(f"Upserted {count} card types") + + return count + + async def _upsert_keywords(self) -> int: + """Upsert keywords from Keywords.json.""" + path = self.data_dir / "Keywords.json" + + if not path.exists(): + logger.warning("Keywords.json not found") + return 0 + + logger.info("Processing keywords") + + with open(path, 'r', encoding='utf-8') as f: + keywords = json.load(f) + + async with mtg_async_session() as session: + count = 0 + + for keyword in keywords: + if not isinstance(keyword, dict): + continue + + try: + await session.execute(text(""" + INSERT INTO mtg_keywords (keyword) + VALUES (:keyword) + ON CONFLICT (keyword) DO NOTHING + """), { + 'keyword': keyword.get('keyword'), + }) + + count += 1 + + except Exception as e: + logger.error(f"Failed to upsert keyword: {e}") + continue + + await session.commit() + logger.info(f"Upserted {count} keywords") + + return count + + async def _upsert_set_list(self) -> int: + """Upsert set list from SetList.json.""" + path = self.data_dir / "SetList.json" + + if not path.exists(): + logger.warning("SetList.json not found") + return 0 + + logger.info("Processing set list") + + with open(path, 'r', encoding='utf-8') as f: + set_list = json.load(f) + + async with mtg_async_session() as session: + count = 0 + + for item in set_list: + if not isinstance(item, dict): + continue + + try: + await session.execute(text(""" + INSERT INTO mtg_set_list (set_name, code, release_date, scryfall_uri) + VALUES (:set_name, :code, :release_date, :scryfall_uri) + ON CONFLICT (set_name) DO UPDATE SET + code = EXCLUDED.code, + release_date = EXCLUDED.release_date, + scryfall_uri = EXCLUDED.scryfall_uri + """), { + 'set_name': item.get('name'), + 'code': item.get('code'), + 'release_date': item.get('release_date'), + 'scryfall_uri': item.get('scryfall_uri'), + }) + + count += 1 + + except Exception as e: + logger.error(f"Failed to upsert set list item: {e}") + continue + + await session.commit() + logger.info(f"Upserted {count} set list items") + + return count + + async def _upsert_deck_list(self) -> int: + """Upsert deck list from DeckList.json.""" + path = self.data_dir / "DeckList.json" + + if not path.exists(): + logger.warning("DeckList.json not found") + return 0 + + logger.info("Processing deck list") + + with open(path, 'r', encoding='utf-8') as f: + deck_list = json.load(f) + + async with mtg_async_session() as session: + count = 0 + + for item in deck_list: + if not isinstance(item, dict): + continue + + try: + await session.execute(text(""" + INSERT INTO mtg_deck_list (list_id, list_name, list_year, list_date, list_format) + VALUES (:list_id, :list_name, :list_year, :list_date, :list_format) + ON CONFLICT (list_id) DO UPDATE SET + list_name = EXCLUDED.list_name, + list_year = EXCLUDED.list_year, + list_date = EXCLUDED.list_date, + list_format = EXCLUDED.list_format + """), { + 'list_id': item.get('listId'), + 'list_name': item.get('listName'), + 'list_year': item.get('listYear'), + 'list_date': item.get('listDate'), + 'list_format': item.get('listFormat'), + }) + + count += 1 + + except Exception as e: + logger.error(f"Failed to upsert deck list item: {e}") + continue + + await session.commit() + logger.info(f"Upserted {count} deck list items") + + return count + + async def log_refresh(self, status: str, counts: dict[str, int], duration: int, error: str = None): + """Log refresh operation.""" + 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, + sets_list_count, deck_list_count, + error_message, duration_seconds) + VALUES (CURRENT_TIMESTAMP, :status, :cards, :sets, + :identifiers, :card_types, :keywords, + :sets_list, :deck_list, + :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), + 'sets_list': counts.get('set_list', 0), + 'deck_list': counts.get('deck_list', 0), + 'error': error, + 'duration': duration, + }) + + async def run_refresh(self) -> bool: + """Run complete refresh cycle from local files.""" + logger.info("Starting local file refresh") + + try: + import time + start_time = time.time() + + # Upsert data + counts = await self.upsert_all() + + # Log refresh + duration = int(time.time() - start_time) + status = "SUCCESS" + await self.log_refresh(status, counts, duration) + + logger.info(f"Refresh complete in {duration}s: {counts}") + return True except Exception as e: - logger.error(f"Health check failed: {e}") - return { - "status": "unhealthy", - "error": str(e), - } + logger.error(f"Refresh failed: {e}") + await self.log_refresh("FAILED", {}, 0, str(e)) + return False # Singleton instance -_manager_instance: Optional[MTGJSONManager] = None +_manager_instance = None def get_manager() -> MTGJSONManager: @@ -752,42 +573,26 @@ def get_manager() -> 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") + parser.add_argument("--data-dir", type=str, default="/app/data", 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(): - if not await manager.download_with_sanity_check(): - logger.error("Failed to download MTGJSON data after retries") - return - - 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}") + manager = MTGJSONManager(Path(args.data_dir)) + if args.refresh: + success = await manager.run_refresh() + if success: + print("Refresh completed successfully") + else: + print("Refresh failed") + exit(1) else: - print("Usage: python -m app.services.mtgjson_manager [--refresh | --health]") + print("Usage: python -m app.services.mtgjson_manager --refresh --data-dir /app/data") asyncio.run(main())