599 lines
23 KiB
Python
599 lines
23 KiB
Python
"""
|
|
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())
|