feat(mtg): Add model MTGSets

requirements.txt

@@ -1,5 +1,6 @@
-curl_cffi==0.7.2
+curl_cffi==0.7.3
 requests==2.32.3
+bs4==0.0.2
 
 Django==5.1.2
 django-money==3.5.3
@@ -8,3 +9,13 @@ django-unfold==0.40.0
 
 ptpython==3.0.29
 pytz==2024.2
+
+celery~=5.4.0
+semgrep~=1.90.0
+PyYAML~=6.0.2
+
+# dev
+black==24.10.0
+isort==5.13.2
+bandit==1.7.10
+semgrep==1.90.0

src/cm_prices/settings.py

@@ -128,6 +128,7 @@
     # ('module.submodule2', 'function3'),
     # ('module.submodule3', '*'),
     # 'module.submodule4',
+    ('prices.services', '*'),
     ('lib.utils', '*'),
 )
 

src/lib/utils.py

@@ -1,222 +1,78 @@
-import hashlib
-import json
 import logging
 from datetime import datetime
+from pathlib import Path
 
 import pytz
-import requests
 
-from prices.models import Catalog, MTGCard, MTGCardPrice
+from prices.services import update_cm_prices
 
 logging.basicConfig(level=logging.INFO)  # temporary
 logger = logging.getLogger(__name__)
 germany_tz = pytz.timezone('Europe/Berlin')
 
 
-def update_mtg():
-    """Fetch new cards, new prices and save them in the local models."""
+def simple_trend(price_dates, price_values):
+    """Calculate the trend (slope) of a card's price history using basic linear regression."""
 
-    new_cards = update_cm_products()
-    updated_prices = update_cm_prices()
-    logger.info('Added new %d cards and updated %d prices', new_cards, updated_prices)
+    num_values = len(price_dates)
 
+    # Convert datetime objects to numeric values (e.g., days since the first date)
+    time_values = [(date - price_dates[0]).days for date in price_dates]
 
-def update_cm_products():
-    """Fetch and store product data for MTG cards."""
+    # Calculate the sums
+    sum_time = sum(time_values)
+    sum_price = sum(price_values)
+    sum_time_price = sum(t * p for t, p in zip(time_values, price_values))
+    sum_time_squared = sum(t**2 for t in time_values)
 
-    # Lists for bulk create/update
-    insert_cards = []
+    # Calculate slope (trend)
+    numerator = num_values * sum_time_price - sum_time * sum_price
+    denominator = num_values * sum_time_squared - sum_time**2
+    trend_slope = numerator / denominator if denominator != 0 else 0
 
-    url = 'https://downloads.s3.cardmarket.com/productCatalog/productList/products_singles_1.json'
+    return trend_slope
 
-    response = requests.get(url, timeout=10)
-    if response.ok:
-        # skip catalog if already downloaded
-        md5sum = hashlib.md5(response.text.encode('utf-8'), usedforsecurity=False).hexdigest()  # nosemgrep
-        if Catalog.objects.filter(md5sum=md5sum, catalog_type=Catalog.PRODUCTS).exists():
-            catalog_date = Catalog.objects.get(md5sum=md5sum, catalog_type=Catalog.PRODUCTS).catalog_date
-            logger.info('Products already up to date since %s (%s)', catalog_date, md5sum)
-            return 0
-
-        data = response.json()
-
-        catalog_date = data['createdAt']
-        catalog_date = datetime.strptime(catalog_date, '%Y-%m-%dT%H:%M:%S%z')
-        Catalog.objects.create(catalog_date=catalog_date, md5sum=md5sum, catalog_type=Catalog.PRODUCTS)
-
-        # List all existing cards within the current JSON
-        all_cm_ids = [item['idProduct'] for item in data['products']]
-        existing_cards = MTGCard.objects.filter(cm_id__in=all_cm_ids).in_bulk(field_name='cm_id')
-
-        for product_item in data['products']:
-            cm_id = product_item['idProduct']
-            name = product_item.get('name', None)
-            # slug = product_item.get('website', None).replace("/en/", "") if product_item.get('website') else None
-
-            # Check if the card already exists
-            card = existing_cards.get(cm_id)
-            if not card:
-
-                date_added = product_item.get('dateAdded')
-                if date_added:
-                    date_added = datetime.strptime(date_added, '%Y-%m-%d %H:%M:%S')  # convert str to date
-                    date_added = date_added.replace(tzinfo=germany_tz)
-
-                card = MTGCard(
-                    cm_id=cm_id,
-                    name=name,
-                    category_id=product_item.get('idCategory', None),
-                    expansion_id=product_item.get('idExpansion', None),
-                    metacard_id=product_item.get('idMetacard', None),
-                    cm_date_added=date_added,
-                )
-                insert_cards.append(card)
 
-        # Bulk create new cards
-        if insert_cards:
-            MTGCard.objects.bulk_create(insert_cards)
-            logger.info('%d new cards inserted.', len(insert_cards))
+def trend(card, days=None):
+    """Find the trend of x days of a card pricing."""
 
-    return len(insert_cards)
-
-
-def update_cm_prices():
-    """Fetch and store catalog prices for MTG cards."""
-
-    # Lists to be used in bulk_create
-    insert_prices = []
-
-    url = 'https://downloads.s3.cardmarket.com/productCatalog/priceGuide/price_guide_1.json'
-
-    response = requests.get(url, timeout=10)
-    if response.ok:
-        # skip catalog if already downloaded
-        md5sum = hashlib.md5(response.text.encode('utf-8'), usedforsecurity=False).hexdigest()  # nosemgrep
-        if Catalog.objects.filter(md5sum=md5sum, catalog_type=Catalog.PRICES).exists():
-            catalog_date = Catalog.objects.get(md5sum=md5sum, catalog_type=Catalog.PRICES).catalog_date
-            logger.info('Prices already up to date since %s (%s)', catalog_date, md5sum)
+    if days:
+        days_ago = datetime.datetime.now() - datetime.timedelta(days=days)
+        if card.prices.filter(catalog_date__gte=days_ago).count() <= 1:
+            logger.warning('error getting trend from %d days', days_ago.days)
             return 0
-
-        data = response.json()
-        if data['version'] == 1:
-
-            catalog_date = data['createdAt']
-            catalog_date = datetime.strptime(catalog_date, '%Y-%m-%dT%H:%M:%S%z')
-            Catalog.objects.create(catalog_date=catalog_date, md5sum=md5sum, catalog_type=Catalog.PRICES)
-
-            # List all existing cards within the current JSON
-            all_cm_ids = [item['idProduct'] for item in data['priceGuides']]
-            existing_cards = MTGCard.objects.filter(cm_id__in=all_cm_ids).in_bulk(field_name='cm_id')
-
-            for price_item in data['priceGuides']:
-                cm_id = price_item['idProduct']
-
-                # Check if the card already exists
-                # products function should handle this
-                card = existing_cards.get(cm_id)
-                if not card:
-                    # card = MTGCard(cm_id=cm_id)
-                    # insert_cards.append(card)
-                    logger.warning('Card with idProduct %s not found in MTGCard.', cm_id)
-                    continue
-
-                mtg_card_price = MTGCardPrice(
-                    catalog_date=catalog_date,
-                    card=card,
-                    cm_id=cm_id,
-                    avg=price_item.get('avg', None),
-                    low=price_item.get('low', None),
-                    trend=price_item.get('trend', None),
-                    avg1=price_item.get('avg1', None),
-                    avg7=price_item.get('avg7', None),
-                    avg30=price_item.get('avg30', None),
-                    avg_foil=price_item.get('avg-foil', None),
-                    low_foil=price_item.get('low-foil', None),
-                    trend_foil=price_item.get('trend-foil', None),
-                    avg1_foil=price_item.get('avg1-foil', None),
-                    avg7_foil=price_item.get('avg7-foil', None),
-                    avg30_foil=price_item.get('avg30-foil', None),
-                )
-
-                insert_prices.append(mtg_card_price)
-
-            # Bulk create all prices
-            if insert_prices:
-                MTGCardPrice.objects.bulk_create(insert_prices)
-                logger.info('%d new prices inserted.', len(insert_prices))
-
-    return len(insert_prices)
-
-
-def save_local_prices(content):
-    """User previously saved jsons to save its prices info in the MTGCardPrice model."""
-
-    # from pathlib import Path
-    # directory = Path('../local/catalogs')
-    # for file in directory.glob("202*json"):
-    #     with open(file, 'r') as f:
-    #         content = f.read()
-    #         try:
-    #             save_local_prices(content)
-    #         except Exception as err:
-    #             print(err)
-
-    # Lists to be used in bulk_create
-    insert_prices = []
-
-    # skip catalog if already downloaded
-    md5sum = hashlib.md5(content.encode('utf-8'), usedforsecurity=False).hexdigest()  # nosemgrep
-    if Catalog.objects.filter(md5sum=md5sum, catalog_type=Catalog.PRICES).exists():
-        catalog_date = Catalog.objects.get(md5sum=md5sum, catalog_type=Catalog.PRICES).catalog_date
-        logger.info('Prices already up to date since %s (%s)', catalog_date, md5sum)
+        prices = card.prices.filter(catalog_date__gte=days_ago)
+    else:
+        prices = card.prices.all()
+    prices = prices.exclude(trend__isnull=True)
+    if prices.count() <= 1:
         return 0
-
-    data = json.loads(content)
-    if data['version'] == 1:
-
-        catalog_date = data['createdAt']
-        catalog_date = datetime.strptime(catalog_date, '%Y-%m-%dT%H:%M:%S%z')
-        Catalog.objects.create(catalog_date=catalog_date, md5sum=md5sum, catalog_type=Catalog.PRICES)
-
-        # List all existing cards within the current JSON
-        all_cm_ids = [item['idProduct'] for item in data['priceGuides']]
-        existing_cards = MTGCard.objects.filter(cm_id__in=all_cm_ids).in_bulk(field_name='cm_id')
-
-        for price_item in data['priceGuides']:
-            cm_id = price_item['idProduct']
-
-            # Check if the card already exists
-            # products function should handle this
-            card = existing_cards.get(cm_id)
-            if not card:
-                # card = MTGCard(cm_id=cm_id)
-                # insert_cards.append(card)
-                logger.warning('Card with idProduct %d not found in MTGCard.', cm_id)
-                continue
-
-            mtg_card_price = MTGCardPrice(
-                catalog_date=catalog_date,
-                card=card,
-                cm_id=cm_id,
-                avg=price_item.get('avg', None),
-                low=price_item.get('low', None),
-                trend=price_item.get('trend', None),
-                avg1=price_item.get('avg1', None),
-                avg7=price_item.get('avg7', None),
-                avg30=price_item.get('avg30', None),
-                avg_foil=price_item.get('avg-foil', None),
-                low_foil=price_item.get('low-foil', None),
-                trend_foil=price_item.get('trend-foil', None),
-                avg1_foil=price_item.get('avg1-foil', None),
-                avg7_foil=price_item.get('avg7-foil', None),
-                avg30_foil=price_item.get('avg30-foil', None),
-            )
-
-            insert_prices.append(mtg_card_price)
-
-        # Bulk create all prices
-        if insert_prices:
-            MTGCardPrice.objects.bulk_create(insert_prices)
-            logger.info('%d new prices inserted.', len(insert_prices))
-
-    return len(insert_prices)
+    price_labels = list(prices.values_list('catalog_date', flat=True))
+    price_values = list(prices.values_list('trend', flat=True))
+    price_values = [0 if val is None else val for val in price_values]  # replace None with 0
+
+    # price_labels = list(self.charted_prices.values_list('price_date', flat=True))[-days:]
+    # price_values = list(self.charted_prices.values_list('price_value', flat=True))[-days:]
+
+    # x_values = np.linspace(0, 1, len(price_labels))
+    # y_values = [float(x) for x in price_values]
+    # price_trend = np.polyfit(x_values, y_values, 1)[-2]
+    price_trend = simple_trend(price_labels, price_values)
+
+    return price_trend
+
+
+def update_from_local_files():
+    """Update prices from local json files."""
+
+    directory = Path("../local/catalogs")
+    files = sorted(directory.glob("202*json"), key=lambda f: f.name)
+    for json in files:
+        # noset
+        with open(json, "r", encoding='utf-8') as file:
+            # content = json.load(f)
+            content = file.read()
+            try:
+                update_cm_prices(local_content=content)
+            except Exception as err:  # NOQA
+                print(err)

src/prices/admin.py

@@ -1,3 +1,16 @@
-# from django.contrib import admin
+from django.contrib import admin
+from unfold.admin import ModelAdmin
 
-# Register your models here.
+from prices.models import MTGCard
+from prices.services import update_mtg
+
+
+@admin.register(MTGCard)
+class CustomAdminClass(ModelAdmin):
+    """Update all cardmarket prices from admin."""
+
+    actions = ['update_all']
+
+    def update_all(self):
+        """Update all cardmarket prices from admin."""
+        update_mtg()

src/prices/constants.py

@@ -0,0 +1,24 @@
+LEGAL_STANDARD_SETS = [
+    5072,  # Dominaria United
+    5073,  # Dominaria United: Extras
+    5164,  # The Brothers' War
+    5165,  # The Brothers' War: Extras
+    5184,  # Phyrexia: All Will Be One
+    5191,  # Phyrexia: All Will Be One: Extras
+    5227,  # March of the Machine
+    5278,  # March of the Machine: Extras
+    5320,  # March of the Machine: The Aftermath
+    5208,  # March of the Machine: The Aftermath: Extras
+    5359,  # Wilds of Eldraine
+    5428,  # Wilds of Eldraine: Extras
+    5490,  # The Lost Caverns of Ixalan
+    5491,  # The Lost Caverns of Ixalan: Extras
+    5561,  # Murders at Karlov Manor
+    5599,  # Murders at Karlov Manor: Extras
+    5647,  # Outlaws of Thunder Junction
+    5662,  # Outlaws of Thunder Junction: Extras
+    5658,  # Bloomburrow
+    5659,  # Bloomburrow: Extras
+    5806,  # Duskmourn: House of Horror
+    5807,  # Duskmourn: House of Horror: Extras
+]