add: scryfall model

src/cm_prices/settings.py

@@ -38,6 +38,7 @@
     # 'card',
     'prices',
     'lib',
+    'mtg',
 ]
 
 MIDDLEWARE = [
@@ -134,3 +135,4 @@
 
 SCRAPING_RETRIES = 8
 SCRAPING_SLEEP_TIME = 15.5
+SLOPE_THRESHOLD = 0.4

src/lib/utils.py

@@ -7,6 +7,7 @@
 from pathlib import Path
 
 import pytz
+from django.conf import settings
 from django.utils import timezone
 
 from prices.constants import LEGAL_STANDARD_SETS
@@ -15,7 +16,6 @@
 
 logger = logging.getLogger(__name__)
 germany_tz = pytz.timezone('Europe/Berlin')
-SLOPE_THRESHOLD = 0.4  # Move magic values to named constants
 MIN_PRICE_VALUE = 1
 MIN_PERCENTAGE = 1
 
@@ -38,7 +38,7 @@ def show_stats(days=7, cards_qs=None):
             result = future.result()
             if task_type == 'rising' and result:
                 always_rising[card_id] = result
-            elif task_type == 'trending' and result >= SLOPE_THRESHOLD:
+            elif task_type == 'trending' and result >= settings.SLOPE_THRESHOLD:
                 trending_cards[card_id] = result
 
     logger.info('Always Rising:')

src/mtg/constants.py

@@ -0,0 +1,10 @@
+SCRYFALL_BULK_DATA_URL = "https://api.scryfall.com/bulk-data"  # URL to fetch bulk data info
+
+BASIC_TYPES = [
+    'Swamp',
+    'Plains',
+    'Island',
+    'Forest',
+    'Mountain',
+    'Wastes',
+]

src/mtg/migrations/0001_scryfall_data.py

@@ -0,0 +1,57 @@
+# Generated by Django 5.1.2 on 2024-10-29 17:34
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = []
+
+    operations = [
+        migrations.CreateModel(
+            name="ScryfallCard",
+            fields=[
+                (
+                    "date_updated",
+                    models.DateTimeField(auto_now=True, verbose_name="Last update at"),
+                ),
+                (
+                    "date_created",
+                    models.DateTimeField(auto_now_add=True, verbose_name="Created at"),
+                ),
+                ("obs", models.TextField(blank=True, verbose_name="Observations")),
+                ("active", models.BooleanField(default=True, verbose_name="active")),
+                (
+                    "cardmarket_id",
+                    models.PositiveIntegerField(primary_key=True, serialize=False),
+                ),
+                ("oracle_id", models.CharField(max_length=128, null=True)),
+                ("name", models.CharField(max_length=256, null=True)),
+                ("mana_cost", models.CharField(blank=True, max_length=128, null=True)),
+                ("cmc", models.PositiveSmallIntegerField(blank=True, null=True)),
+                ("types", models.CharField(blank=True, max_length=256, null=True)),
+                ("subtypes", models.CharField(blank=True, max_length=256, null=True)),
+                ("colors", models.CharField(blank=True, max_length=128, null=True)),
+                (
+                    "color_identity",
+                    models.CharField(blank=True, max_length=128, null=True),
+                ),
+                (
+                    "oracle_text",
+                    models.CharField(blank=True, max_length=2048, null=True),
+                ),
+                ("legalities", models.CharField(blank=True, max_length=256, null=True)),
+                ("image_small", models.URLField(blank=True, null=True)),
+                ("image_normal", models.URLField(blank=True, null=True)),
+            ],
+            options={
+                "indexes": [
+                    models.Index(
+                        fields=["cardmarket_id"], name="idx_scryfallcard_cm_id"
+                    )
+                ],
+            },
+        ),
+    ]

src/mtg/models.py

@@ -0,0 +1,43 @@
+from django.db import models
+
+from lib.models import BaseAbstractModel
+
+
+class ScryfallCardManager(models.Manager):
+    # Taken from https://github.com/baronvonvaderham/django-mtg-card-catalog
+
+    def get_or_create_card(self, card_data):
+        """Fetch or create a card based on the provided data dictionary."""
+        card, created = self.update_or_create(
+            cardmarket_id=card_data["cardmarket_id"],
+            defaults=card_data,
+        )
+        return created, card
+
+
+class ScryfallCard(BaseAbstractModel):
+    """Class to contain a local version of the scryfall data to limit the need for external API calls."""
+
+    cardmarket_id = models.PositiveIntegerField(blank=False, primary_key=True)
+    oracle_id = models.CharField(max_length=128, null=True)  # NOQA nosemgrep
+    name = models.CharField(max_length=256, null=True)  # nosemgrep
+    mana_cost = models.CharField(max_length=128, blank=True, null=True)  # NOQA nosemgrep
+    cmc = models.PositiveSmallIntegerField(blank=True, null=True)  # NOQA nosemgrep
+    types = models.CharField(max_length=256, blank=True, null=True)  # NOQA nosemgrep
+    subtypes = models.CharField(max_length=256, blank=True, null=True)  # NOQA nosemgrep
+    colors = models.CharField(max_length=128, blank=True, null=True)  # NOQA nosemgrep
+    color_identity = models.CharField(max_length=128, blank=True, null=True)  # NOQA nosemgrep
+    oracle_text = models.CharField(max_length=2048, blank=True, null=True)  # NOQA nosemgrep
+
+    legalities = models.CharField(max_length=256, blank=True, null=True)  # NOQA nosemgrep
+    image_small = models.URLField(blank=True, null=True)  # NOQA nosemgrep
+    image_normal = models.URLField(blank=True, null=True)  # NOQA nosemgrep
+
+    objects = ScryfallCardManager()
+
+    class Meta:
+        indexes = [models.Index(fields=['cardmarket_id'], name='idx_scryfallcard_cm_id')]
+
+    def __str__(self):
+        """Return string representation of ScryfallCard model."""
+        return self.name

src/mtg/services.py

@@ -0,0 +1,146 @@
+import unicodedata
+
+import requests
+
+from .constants import BASIC_TYPES, SCRYFALL_BULK_DATA_URL
+from .models import ScryfallCard
+
+# Inspired on https://github.com/baronvonvaderham/django-mtg-card-catalog
+
+
+def process_card_types(card_data):
+    """Split and clean up types and subtypes for a card."""
+
+    types = card_data.get('type_line', '')
+
+    # If type_line is empty, gather types from card_faces if available
+    if not types and 'card_faces' in card_data:
+        types = ' // '.join(card_face.get('type_line', '') for card_face in card_data['card_faces'])
+
+    types = types.replace('—', '-').split(' // ')
+    card_types, card_subtypes = [], []
+
+    for type_line in types:
+        if not type_line:
+            continue
+
+        # If there is a ' - ', that means we have subtypes to the right, supertypes to the left
+        if ' — ' in type_line:
+            main_types, subtypes = type_line.split(' - ')
+        else:
+            main_types, subtypes = type_line, None
+
+        if subtypes:
+            card_subtypes.extend(subtypes.split())
+
+        card_types.extend(main_types.split())
+        if len(set(types)) == 1:
+            break
+
+    return card_types, card_subtypes
+
+
+def scryfall_download_bulk_data():
+    """Download the bulk data file from Scryfall."""
+    response = requests.get(SCRYFALL_BULK_DATA_URL, timeout=10)
+    response.raise_for_status()  # Raise an error for bad responses
+    url = response.json()
+    url = next(item for item in url['data'] if item['type'] == 'default_cards')
+    url = url['download_uri']
+
+    response = requests.get(url, timeout=10)
+    response.raise_for_status()
+    return response.json()
+
+
+def scryfall_process_data(data):
+    """Parse each process each card entry from data."""
+    for raw_card_data in data:
+        transformed_data = scryfall_transform_card_data(raw_card_data)
+        if transformed_data:
+            scryfall_save_card(transformed_data)
+
+
+def scryfall_transform_card_data(raw_card_data):
+    """Convert raw Scryfall data to model-compatible format, applying constants-based filters and transformations."""
+
+    # Skipping unwanted stuff
+    if not raw_card_data.get('cardmarket_id'):
+        return None
+    if raw_card_data.get('name').split(' ')[0] in BASIC_TYPES:
+        return None
+    if '(' in raw_card_data.get('name'):
+        return None
+
+    card_name = raw_card_data.get('name', '')
+    card_name = ''.join(c for c in unicodedata.normalize('NFD', card_name) if unicodedata.category(c) != 'Mn')
+    card_types, card_subtypes = process_card_types(raw_card_data)
+    mana_cost = []
+    colors = set()  # avoid duplicates
+    oracle_text = []
+    legalities = raw_card_data.get('legalities', None)
+    image_small = None
+    image_normal = None
+    color_identity = raw_card_data.get('color_identity')
+    cardmarket_id = raw_card_data.get('cardmarket_id')
+
+    # Split cards
+    if ' // ' in card_name:
+        for card_face in raw_card_data.get('card_faces', []):
+            mana_cost.append(card_face.get('mana_cost'))
+            colors.update(card_face.get('colors', []))
+            oracle_text.append(card_face.get('oracle_text'))
+
+            names = card_name.split(' // ')
+            if len(set(names)) == 1:  # avoid reversible cards with same name
+                card_name = names[0]
+                break
+        colors = list(colors)
+    else:
+        mana_cost = [raw_card_data.get('mana_cost')]
+        colors = raw_card_data.get('colors', [])
+        oracle_text = [raw_card_data.get('oracle_text')]
+
+    if legalities:
+        legal_card_types = [card_type for card_type, status in legalities.items() if status == 'legal']
+        legalities = ','.join(legal_card_types)
+
+    # Check for image URIs in raw_card_data
+    image_uris = raw_card_data.get('image_uris') or (
+        raw_card_data.get('card_faces', [{}])[0].get('image_uris') if 'card_faces' in raw_card_data else None
+    )
+
+    if image_uris:
+        image_small = image_uris.get('small' if 'image_uris' in raw_card_data else 'image_small', None)
+        image_normal = image_uris.get('image_normal' if 'image_uris' in raw_card_data else 'normal', None)
+
+    transformed_data = {
+        'oracle_id': raw_card_data.get('oracle_id'),
+        'name': card_name,
+        'mana_cost': mana_cost,
+        'cmc': raw_card_data.get('cmc'),
+        'types': card_types,
+        'subtypes': card_subtypes,
+        'colors': list(colors),
+        'color_identity': color_identity,
+        'oracle_text': oracle_text,
+        'cardmarket_id': cardmarket_id,
+        'image_small': image_small,
+        'image_normal': image_normal,
+        'legalities': legalities,
+    }
+    return transformed_data
+
+
+def scryfall_save_card(card_data):
+    """Save a new card or update an existing card in the database."""
+    _, created = ScryfallCard.objects.update_or_create(cardmarket_id=card_data['cardmarket_id'], defaults=card_data)
+    return created
+
+
+def update_scryfall_data():
+    """Update Scryfall data in the local database."""
+    data = scryfall_download_bulk_data()
+
+    # Process and save data
+    scryfall_process_data(data)

src/mtg/tasks.py

@@ -0,0 +1,37 @@
+from celery import group
+from celery.utils.log import get_task_logger
+
+from mtg.models import ScryfallCard
+from mtg.services import scryfall_download_bulk_data, scryfall_transform_card_data, scryfall_save_card
+from cm_prices.celery import app
+from tqdm.auto import tqdm
+
+logger = get_task_logger('tasks.common')
+
+
+@app.task(name='sync_scryfall_task')
+def sync_scryfall(*args, **kwargs):
+    """Run scryfall update bulk task."""
+
+    logger.info('BEGINNING SCRYFALL SYNC TASK')
+    scryfall_data = scryfall_download_bulk_data()
+    if kwargs.get('test'):
+        scryfall_data = scryfall_data[:2]
+    load_tasks = []
+    for raw_card_data in tqdm(scryfall_data, unit='card'):
+        card = scryfall_transform_card_data(raw_card_data)
+        if card:
+            if not ScryfallCard.objects.filter(cardmarket_id=card.get('cardmarket_id')).exists():
+                load_tasks.append(get_or_create_scryfall_card.s(card))
+    task_group = group(load_tasks)
+    task_group.apply()
+    logger.info('SCRYFALL SYNC TASK COMPLETE!')
+
+
+@app.task(name='get_or_create_scryfall_card')
+def get_or_create_scryfall_card(card_data):
+    """Create card in local scryfall model."""
+
+    created, card = scryfall_save_card(card_data)
+    if created:
+        logger.info('Created new Scryfall card: %s', card.name)