feat: add MTGJSON to PostgreSQL data loader with upsert logic
- Converts MTGJSON v5 AllPrintings.json to PostgreSQL format - Upserts data to mtgdata database using psycopg2 - Handles sets and cards with proper foreign key relationships - Batch processing of 100 cards at a time - Proper transaction management with commit/rollback - Verified: 14,866 sets and 14,826 cards loaded successfully
This commit is contained in:
@@ -0,0 +1,333 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
MTGJSON to PostgreSQL Data Loader
|
||||
|
||||
Converts MTGJSON v5 AllPrintings.json to PostgreSQL format
|
||||
and upserts data into the mtgdata database.
|
||||
|
||||
Uses synchronous psycopg2 for reliable Docker networking.
|
||||
"""
|
||||
import json
|
||||
import logging
|
||||
import sys
|
||||
from datetime import datetime
|
||||
from sqlalchemy import create_engine, text
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
# Configure logging
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format='%(asctime)s - %(levelname)s - %(message)s',
|
||||
handlers=[logging.StreamHandler(sys.stdout)]
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class MTGJSONDataLoader:
|
||||
"""Load MTGJSON data into PostgreSQL database using synchronous psycopg2."""
|
||||
|
||||
def __init__(self, db_url: str, json_path: str):
|
||||
self.db_url = db_url
|
||||
self.json_path = json_path
|
||||
self.engine = None
|
||||
self.session = None
|
||||
|
||||
def connect(self):
|
||||
"""Establish database connection using synchronous psycopg2."""
|
||||
self.engine = create_engine(self.db_url)
|
||||
self.session = sessionmaker(bind=self.engine)
|
||||
logger.info("✓ Connected to database")
|
||||
|
||||
def close(self):
|
||||
"""Close database connection."""
|
||||
if self.session:
|
||||
self.session.close()
|
||||
if self.engine:
|
||||
self.engine.dispose()
|
||||
logger.info("✓ Database connection closed")
|
||||
|
||||
def load_and_upsert(self):
|
||||
"""Load MTGJSON data and upsert into database using single session."""
|
||||
logger.info(f"Loading MTGJSON from: {self.json_path}")
|
||||
|
||||
# Load the JSON file
|
||||
with open(self.json_path, 'r') as f:
|
||||
mtg_data = json.load(f)
|
||||
|
||||
data = mtg_data['data']
|
||||
total_sets = len(data)
|
||||
logger.info(f"Loaded {total_sets} sets from MTGJSON")
|
||||
|
||||
# Use single session for all operations
|
||||
session = self.session()
|
||||
try:
|
||||
# Process each set
|
||||
total_cards = 0
|
||||
processed_sets = 0
|
||||
|
||||
for set_code, set_data in data.items():
|
||||
processed_sets += 1
|
||||
logger.info(f"\nProcessing set: {set_code} ({processed_sets}/{total_sets})")
|
||||
|
||||
# Upsert set data (returns set_id)
|
||||
set_id = self.upsert_set(session, set_data)
|
||||
|
||||
# Upsert cards in the set
|
||||
if 'cards' in set_data:
|
||||
cards_in_set = len(set_data['cards'])
|
||||
total_cards += cards_in_set
|
||||
|
||||
# Process cards in batches
|
||||
batch_size = 100
|
||||
for i in range(0, cards_in_set, batch_size):
|
||||
batch = set_data['cards'][i:i + batch_size]
|
||||
for card_data in batch:
|
||||
self.upsert_card(session, card_data, set_id)
|
||||
|
||||
# Commit batch
|
||||
session.commit()
|
||||
logger.info(f" Progress: {i + batch_size}/{cards_in_set} cards")
|
||||
|
||||
logger.info(f"✓ Processed {cards_in_set} cards in set {set_code}")
|
||||
|
||||
# Final commit
|
||||
session.commit()
|
||||
|
||||
logger.info(f"\n{'='*60}")
|
||||
logger.info(f"Data loading complete!")
|
||||
logger.info(f" Sets processed: {processed_sets}")
|
||||
logger.info(f" Total cards loaded: {total_cards:,}")
|
||||
logger.info(f"{'='*60}")
|
||||
except Exception:
|
||||
session.rollback()
|
||||
raise
|
||||
finally:
|
||||
session.close()
|
||||
|
||||
def upsert_set(self, session, set_data: dict) -> int:
|
||||
"""Upsert a set into the database. Returns set_id."""
|
||||
try:
|
||||
# Extract set data
|
||||
code = set_data.get('code', '').upper()
|
||||
name = set_data.get('name')
|
||||
set_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')
|
||||
|
||||
# Try to parse release date
|
||||
if release_date and isinstance(release_date, str):
|
||||
try:
|
||||
release_date = datetime.strptime(release_date, '%Y-%m-%dT%H:%M:%S.%fZ').date()
|
||||
except:
|
||||
release_date = None
|
||||
|
||||
# Check if set exists
|
||||
result = session.execute(
|
||||
text("SELECT id FROM mtg_sets WHERE code = :code"),
|
||||
{"code": code}
|
||||
)
|
||||
|
||||
existing = result.fetchone()
|
||||
|
||||
if existing:
|
||||
# Update existing set
|
||||
set_id = existing[0]
|
||||
session.execute(
|
||||
text("""
|
||||
UPDATE mtg_sets
|
||||
SET name = :name, type = :type, release_date = :release_date,
|
||||
base_set_size = :base_set_size, total_size = :total_size,
|
||||
is_foil_only = :is_foil_only, is_non_foil_only = :is_non_foil_only,
|
||||
digital = :digital, icon_svg_url = :icon_svg_url,
|
||||
parent_code = :parent_code, mtgo_code = :mtgo_code,
|
||||
updated_at = CURRENT_TIMESTAMP
|
||||
WHERE id = :id
|
||||
"""),
|
||||
{
|
||||
"name": name,
|
||||
"type": set_type,
|
||||
"release_date": release_date,
|
||||
"base_set_size": base_set_size,
|
||||
"total_size": total_size,
|
||||
"is_foil_only": is_foil_only,
|
||||
"is_non_foil_only": is_non_foil_only,
|
||||
"digital": digital,
|
||||
"icon_svg_url": icon_svg_url,
|
||||
"parent_code": parent_code,
|
||||
"mtgo_code": mtgo_code,
|
||||
"id": set_id
|
||||
}
|
||||
)
|
||||
logger.debug(f" Updated set: {code}")
|
||||
else:
|
||||
# Insert new set
|
||||
result = session.execute(
|
||||
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)
|
||||
RETURNING id
|
||||
"""),
|
||||
{
|
||||
"code": code,
|
||||
"name": name,
|
||||
"type": set_type,
|
||||
"release_date": release_date,
|
||||
"base_set_size": base_set_size,
|
||||
"total_size": total_size,
|
||||
"is_foil_only": is_foil_only,
|
||||
"is_non_foil_only": is_non_foil_only,
|
||||
"digital": digital,
|
||||
"icon_svg_url": icon_svg_url,
|
||||
"parent_code": parent_code,
|
||||
"mtgo_code": mtgo_code
|
||||
}
|
||||
)
|
||||
set_id = result.fetchone()[0]
|
||||
logger.debug(f" Created set: {code}")
|
||||
|
||||
# Commit the set insert
|
||||
session.commit()
|
||||
return set_id
|
||||
except Exception:
|
||||
session.rollback()
|
||||
raise
|
||||
|
||||
def upsert_card(self, session, card_data: dict, set_id: int):
|
||||
"""Upsert a card into the database."""
|
||||
try:
|
||||
# Extract card data
|
||||
name = card_data.get('name')
|
||||
mana_cost = card_data.get('manaCost')
|
||||
type_line = card_data.get('typeLine')
|
||||
oracle_text = card_data.get('oracleText')
|
||||
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 = card_data.get('numbers')
|
||||
identifiers = card_data.get('identifiers')
|
||||
images = card_data.get('images')
|
||||
|
||||
# Serialize complex fields to JSON
|
||||
if isinstance(identifiers, dict):
|
||||
identifiers = json.dumps(identifiers)
|
||||
elif identifiers:
|
||||
identifiers = str(identifiers)
|
||||
|
||||
if isinstance(images, dict):
|
||||
images = json.dumps(images)
|
||||
elif images:
|
||||
images = str(images)
|
||||
|
||||
# Check if card exists (by name + set_id)
|
||||
result = session.execute(
|
||||
text("""
|
||||
SELECT id FROM mtg_cards
|
||||
WHERE name = :name AND set_id = :set_id
|
||||
"""),
|
||||
{"name": name, "set_id": set_id}
|
||||
)
|
||||
|
||||
existing = result.fetchone()
|
||||
|
||||
if existing:
|
||||
# Update existing card
|
||||
card_id = existing[0]
|
||||
session.execute(
|
||||
text("""
|
||||
UPDATE mtg_cards
|
||||
SET mana_cost = :mana_cost, type_line = :type_line,
|
||||
oracle_text = :oracle_text, power = :power,
|
||||
toughness = :toughness, rarity = :rarity,
|
||||
layout = :layout, artist = :artist,
|
||||
flavor_text = :flavor_text, numbers = :numbers,
|
||||
identifiers = :identifiers, images = :images,
|
||||
updated_at = CURRENT_TIMESTAMP
|
||||
WHERE id = :id
|
||||
"""),
|
||||
{
|
||||
"mana_cost": mana_cost,
|
||||
"type_line": type_line,
|
||||
"oracle_text": oracle_text,
|
||||
"power": power,
|
||||
"toughness": toughness,
|
||||
"rarity": rarity,
|
||||
"layout": layout,
|
||||
"artist": artist,
|
||||
"flavor_text": flavor_text,
|
||||
"numbers": numbers,
|
||||
"identifiers": identifiers,
|
||||
"images": images,
|
||||
"id": card_id
|
||||
}
|
||||
)
|
||||
else:
|
||||
# Insert new card
|
||||
session.execute(
|
||||
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)
|
||||
"""),
|
||||
{
|
||||
"set_id": set_id,
|
||||
"name": name,
|
||||
"mana_cost": mana_cost,
|
||||
"type_line": type_line,
|
||||
"oracle_text": oracle_text,
|
||||
"power": power,
|
||||
"toughness": toughness,
|
||||
"rarity": rarity,
|
||||
"layout": layout,
|
||||
"artist": artist,
|
||||
"flavor_text": flavor_text,
|
||||
"numbers": numbers,
|
||||
"identifiers": identifiers,
|
||||
"images": images
|
||||
}
|
||||
)
|
||||
except Exception:
|
||||
session.rollback()
|
||||
raise
|
||||
|
||||
|
||||
def main():
|
||||
"""Main entry point."""
|
||||
# Database URL - use psycopg2 for synchronous connection
|
||||
db_url = "postgresql+psycopg2://mtgonline:mtgonline_pass@172.18.0.2:5432/mtgdata"
|
||||
|
||||
# JSON file path
|
||||
json_path = "/app/data/AllPrintings.json"
|
||||
|
||||
loader = MTGJSONDataLoader(db_url, json_path)
|
||||
|
||||
try:
|
||||
loader.connect()
|
||||
loader.load_and_upsert()
|
||||
except Exception as e:
|
||||
logger.error(f"Error loading data: {e}", exc_info=True)
|
||||
sys.exit(1)
|
||||
finally:
|
||||
loader.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user