""" MTGJSON Data Manager - Local File Processing Reads JSON files from a mounted volume and upserts them into PostgreSQL. No network downloads - users provide the files. """ import json import logging from pathlib import Path from datetime import datetime from typing import Optional from sqlalchemy import text from app.core.database import mtg_async_session logger = logging.getLogger(__name__) # 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: """Process local MTGJSON files and upsert into PostgreSQL.""" def __init__(self, data_dir: Path = None): if data_dir is None: 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) async def get_last_refresh(self) -> Optional[datetime]: """Get timestamp of last successful refresh.""" async with mtg_async_session() as session: result = await session.execute(text(""" SELECT refresh_date FROM mtg_refresh_log WHERE status = 'SUCCESS' ORDER BY refresh_date DESC LIMIT 1 """)) row = result.fetchone() return row[0] if row else None async def get_health_status(self) -> dict: """Get health status.""" try: 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 files_status = {} 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": { "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: 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"Refresh failed: {e}") await self.log_refresh("FAILED", {}, 0, str(e)) return False # Singleton instance _manager_instance = 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 if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="MTGJSON Data Manager") parser.add_argument("--refresh", action="store_true", help="Run refresh cycle") parser.add_argument("--data-dir", type=str, default="/app/data", help="Data directory") args = parser.parse_args() async def main(): 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 --data-dir /app/data") asyncio.run(main())