Use Gitea for workflows; remove SECURITY.md

.github/SECURITY.md

@@ -1,11 +0,0 @@
-# Reporting a Vulnerability
-
-tl;dr: [open a draft security advisory](https://github.com/TheLovinator1/twitch-drop-notifier/security/advisories/new).
-
----
-
-You can also email me at [tlovinator@gmail.com](mailto:tlovinator@gmail.com).
-
-I am also available on Discord at `TheLovinator#9276`.
-
-Thanks :-)

.pre-commit-config.yaml

@@ -3,7 +3,7 @@ default_language_version:
 repos:
   # A Django template formatter.
   - repo: https://github.com/adamchainz/djade-pre-commit
-    rev: "1.3.2"
+    rev: "1.4.0"
     hooks:
       - id: djade
         args: [--target-version, "5.1"]
@@ -43,7 +43,7 @@ repos:
 
   # Automatically upgrade your Django project code
   - repo: https://github.com/adamchainz/django-upgrade
-    rev: "1.22.2"
+    rev: "1.24.0"
     hooks:
       - id: django-upgrade
         args: [--target-version, "5.1"]
@@ -57,14 +57,8 @@ repos:
 
   # An extremely fast Python linter and formatter.
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.8.4
+    rev: v0.11.7
     hooks:
       - id: ruff-format
       - id: ruff
         args: [--fix]
-
-  # Static checker for GitHub Actions workflow files.
-  - repo: https://github.com/rhysd/actionlint
-    rev: v1.7.4
-    hooks:
-      - id: actionlint

.vscode/launch.json

@@ -19,6 +19,18 @@
       ],
       "django": true,
       "justMyCode": true
+    },
+    {
+      "name": "Django: Import Twitch Drops",
+      "type": "debugpy",
+      "request": "launch",
+      "program": "${workspaceFolder}/manage.py",
+      "args": [
+        "import_twitch_drops",
+        "${file}"
+      ],
+      "django": true,
+      "justMyCode": true
     }
   ]
 }

.vscode/settings.json

@@ -16,6 +16,7 @@
     "createcachetable",
     "createsuperuser",
     "djade",
+    "djlint",
     "docstrings",
     "dotenv",
     "dropcampaign",
@@ -43,6 +44,7 @@
     "psycopg",
     "PUID",
     "pydocstyle",
+    "pytest",
     "pyupgrade",
     "requirepass",
     "rewardcampaign",

README.md

@@ -7,38 +7,15 @@ Get notified when a new drop is available on Twitch
 - [Python 3.13](https://www.python.org/downloads/) is required to run this project.
 
 ```bash
-# Create and activate a virtual environment.
-python -m venv .venv
-source .venv/bin/activate
-
-# Remember to run `source .venv/bin/activate` before running the following commands:
-# You will need to run this command every time you open a new terminal.
-# VSCode will automatically activate the virtual environment if you have the Python extension installed.
-
-# Install dependencies.
-pip install -r requirements.txt
-pip install -r requirements-dev.txt
+uv sync
 
 # Run the following to get passwords for the .env file.
-python -c 'from django.core.management.utils import get_random_secret_key; print(get_random_secret_key())'
+uv run python -c 'from django.core.management.utils import get_random_secret_key; print(get_random_secret_key())'
 
-# Rename .env.example to .env and fill in the required values.
-# Only DJANGO_SECRET_KEY is required to run the server.
-# EMAIL_HOST_USER, EMAIL_HOST_PASSWORD and DISCORD_WEBHOOK_URL can be left empty if not needed.
-mv .env.example .env
-
-# Run the migrations.
-python manage.py migrate
-
-# Create cache table.
-python manage.py createcachetable
-
-# Create a superuser.
-python manage.py createsuperuser
-
-# Run the server.
-python manage.py runserver
-
-# Run the tests.
-pytest
+# Commands
+uv run python manage.py migrate
+uv run python manage.py createcachetable
+uv run python manage.py createsuperuser
+uv run python manage.py runserver
+uv run pytest
 ```

core/admin.py

@@ -2,10 +2,10 @@ from __future__ import annotations
 
 from django.contrib import admin
 
-from core.models import Benefit, DropCampaign, Game, Owner, TimeBasedDrop
+from core.models import Benefit, DropCampaign, Game, Organization, TimeBasedDrop
 
 admin.site.register(Game)
-admin.site.register(Owner)
+admin.site.register(Organization)
 admin.site.register(DropCampaign)
 admin.site.register(TimeBasedDrop)
 admin.site.register(Benefit)

core/import_json.py

@@ -1,187 +0,0 @@
-from __future__ import annotations
-
-import logging
-from typing import Any, Literal
-
-from core.models import Benefit, DropCampaign, Game, Owner, TimeBasedDrop
-
-logger: logging.Logger = logging.getLogger(__name__)
-
-
-type_names = Literal["Organization", "Game", "DropCampaign", "TimeBasedDrop", "DropBenefit", "RewardCampaign", "Reward"]
-
-
-def import_data(data: dict[str, Any]) -> None:
-    """Import the data from the JSON object.
-
-    This looks for '__typename' with the value 'DropCampaign' in the JSON object and imports the data.
-
-    Args:
-        data (dict[str, Any]): The data to import.
-    """
-    drop_campaigns: list[dict[str, Any]] = find_typename_in_json(json_obj=data, typename_to_find="DropCampaign")
-    for drop_campaign_json in drop_campaigns:
-        import_drop_campaigns(drop_campaigns=drop_campaign_json)
-
-
-def import_drop_campaigns(drop_campaigns: dict[str, Any]) -> DropCampaign | None:
-    """Import the drop campaigns from the data.
-
-    Args:
-        drop_campaigns (dict[str, Any]): The drop campaign data.
-
-    Returns:
-        DropCampaign | None: The drop campaign instance if created, otherwise None
-    """
-    twitch_id: str = drop_campaigns.get("id", "")
-    logger.info("\tProcessing drop campaign: %s", twitch_id)
-
-    if not twitch_id:
-        logger.error("\tDrop campaign has no ID: %s", drop_campaigns)
-        return None
-
-    drop_campaign, created = DropCampaign.objects.get_or_create(twitch_id=twitch_id)
-    if created:
-        logger.info("\tCreated drop campaign: %s", drop_campaign)
-
-    owner: Owner = import_owner_data(drop_campaign=drop_campaigns)
-    game: Game = import_game_data(drop_campaign=drop_campaigns, owner=owner)
-    drop_campaign.import_json(data=drop_campaigns, game=game)
-
-    import_time_based_drops(drop_campaigns, drop_campaign)
-
-    return drop_campaign
-
-
-def import_time_based_drops(drop_campaign_json: dict[str, Any], drop_campaign: DropCampaign) -> list[TimeBasedDrop]:
-    """Import the time-based drops from a drop campaign.
-
-    Args:
-        drop_campaign_json (dict[str, Any]): The drop campaign data.
-        drop_campaign (DropCampaign): The drop campaign instance.
-
-    Returns:
-        list[TimeBasedDrop]: The imported time-based drops.
-    """
-    imported_drops: list[TimeBasedDrop] = []
-    time_based_drops: list[dict[str, Any]] = find_typename_in_json(drop_campaign_json, "TimeBasedDrop")
-    for time_based_drop_json in time_based_drops:
-        time_based_drop_id: str = time_based_drop_json.get("id", "")
-        if not time_based_drop_id:
-            logger.error("\tTime-based drop has no ID: %s", time_based_drop_json)
-            continue
-
-        time_based_drop, created = TimeBasedDrop.objects.get_or_create(twitch_id=time_based_drop_id)
-        if created:
-            logger.info("\tCreated time-based drop: %s", time_based_drop)
-
-        time_based_drop.import_json(time_based_drop_json, drop_campaign)
-
-        import_drop_benefits(time_based_drop_json, time_based_drop)
-        imported_drops.append(time_based_drop)
-
-    return imported_drops
-
-
-def import_drop_benefits(time_based_drop_json: dict[str, Any], time_based_drop: TimeBasedDrop) -> list[Benefit]:
-    """Import the drop benefits from a time-based drop.
-
-    Args:
-        time_based_drop_json (dict[str, Any]): The time-based drop data.
-        time_based_drop (TimeBasedDrop): The time-based drop instance.
-
-    Returns:
-        list[Benefit]: The imported drop benefits.
-    """
-    drop_benefits: list[Benefit] = []
-    benefits: list[dict[str, Any]] = find_typename_in_json(time_based_drop_json, "DropBenefit")
-    for benefit_json in benefits:
-        benefit_id: str = benefit_json.get("id", "")
-        if not benefit_id:
-            logger.error("\tBenefit has no ID: %s", benefit_json)
-            continue
-
-        benefit, created = Benefit.objects.get_or_create(twitch_id=benefit_id)
-        if created:
-            logger.info("\tCreated benefit: %s", benefit)
-
-        benefit.import_json(benefit_json, time_based_drop)
-        drop_benefits.append(benefit)
-
-    return drop_benefits
-
-
-def import_owner_data(drop_campaign: dict[str, Any]) -> Owner:
-    """Import the owner data from a drop campaign.
-
-    Args:
-        drop_campaign (dict[str, Any]): The drop campaign data.
-
-    Returns:
-        Owner: The owner instance.
-    """
-    owner_data_list: list[dict[str, Any]] = find_typename_in_json(drop_campaign, "Organization")
-    for owner_data in owner_data_list:
-        owner_id: str = owner_data.get("id", "")
-        if not owner_id:
-            logger.error("\tOwner has no ID: %s", owner_data)
-            continue
-
-        owner, created = Owner.objects.get_or_create(twitch_id=owner_id)
-        if created:
-            logger.info("\tCreated owner: %s", owner)
-
-        owner.import_json(owner_data)
-    return owner
-
-
-def import_game_data(drop_campaign: dict[str, Any], owner: Owner) -> Game:
-    """Import the game data from a drop campaign.
-
-    Args:
-        drop_campaign (dict[str, Any]): The drop campaign data.
-        owner (Owner): The owner of the game.
-
-    Returns:
-        Game: The game instance.
-    """
-    game_data_list: list[dict[str, Any]] = find_typename_in_json(drop_campaign, "Game")
-    for game_data in game_data_list:
-        game_id: str = game_data.get("id", "")
-        if not game_id:
-            logger.error("\tGame has no ID: %s", game_data)
-            continue
-
-        game, created = Game.objects.get_or_create(twitch_id=game_id)
-        if created:
-            logger.info("\tCreated game: %s", game)
-
-        game.import_json(game_data, owner)
-    return game
-
-
-def find_typename_in_json(json_obj: list | dict, typename_to_find: type_names) -> list[dict[str, Any]]:
-    """Recursively search for '__typename' in a JSON object and return dictionaries where '__typename' equals the specified value.
-
-    Args:
-        json_obj (list | dict): The JSON object to search.
-        typename_to_find (str): The '__typename' value to search for.
-
-    Returns:
-        A list of dictionaries where '__typename' equals the specified value.
-    """  # noqa: E501
-    matching_dicts: list[dict[str, Any]] = []
-
-    if isinstance(json_obj, dict):
-        # Check if '__typename' exists and matches the value
-        if json_obj.get("__typename") == typename_to_find:
-            matching_dicts.append(json_obj)
-        # Recurse into the dictionary
-        for value in json_obj.values():
-            matching_dicts.extend(find_typename_in_json(value, typename_to_find))
-    elif isinstance(json_obj, list):
-        # Recurse into each item in the list
-        for item in json_obj:
-            matching_dicts.extend(find_typename_in_json(item, typename_to_find))
-
-    return matching_dicts

core/management/commands/import_twitch_drops.py

@@ -0,0 +1,417 @@
+from __future__ import annotations
+
+import json
+import logging
+from pathlib import Path
+from typing import Any
+
+from django.core.management.base import BaseCommand, CommandError, CommandParser
+from django.db import transaction
+from django.utils import timezone
+from django.utils.dateparse import parse_datetime
+
+from core.models import Benefit, DropCampaign, Game, Organization, TimeBasedDrop
+
+# Configure logger for this module
+logger = logging.getLogger(__name__)
+
+
+class Command(BaseCommand):
+    """Imports Twitch Drop campaign data from a directory of JSON files into the database."""
+
+    help = "Imports Twitch Drop campaign data from a directory of JSON files."
+
+    def add_arguments(self, parser: CommandParser) -> None:  # noqa: PLR6301
+        """Add command line arguments for the command.
+
+        Args:
+            parser (CommandParser): The command parser to add arguments to.
+        """
+        parser.add_argument("json_dir", type=str, help="Path to the directory containing Twitch Drop JSON files")
+        parser.add_argument(
+            "--pattern",
+            type=str,
+            default="*.json",
+            help="Glob pattern to filter JSON files (default: *.json)",
+        )
+
+    def handle(self, *args: tuple[str, ...], **options: dict[str, Any]) -> None:
+        """Import data to DB from multiple JSON files in a directory.
+
+        Args:
+            *args (tuple[str, ...]): Positional arguments passed to the command.
+            **options (dict[str, Any]): Keyword arguments passed to the command.
+
+        Raises:
+            CommandError: If the directory does not exist or is not a directory.
+        """
+        self.stdout.write(self.style.NOTICE("Starting import process..."))
+        self.stdout.write(self.style.NOTICE(f"Arguments: {args}; Options: {options}"))
+
+        json_dir_path = options["json_dir"]
+        pattern = options["pattern"]
+        dir_path = Path(json_dir_path)  # pyright: ignore[reportArgumentType]
+
+        # Check if directory exists
+        if not dir_path.exists():
+            msg = f"Error: Directory not found at {json_dir_path}"
+            logger.error(msg)
+            raise CommandError(msg)
+
+        if not dir_path.is_dir():
+            msg = f"Error: {json_dir_path} is not a directory"
+            logger.error(msg)
+            raise CommandError(msg)
+
+        # Find JSON files in the directory
+        json_files = list(dir_path.glob(pattern))  # pyright: ignore[reportArgumentType]
+
+        if not json_files:
+            msg = f"No JSON files found in {json_dir_path} matching pattern '{pattern}'"
+            self.stdout.write(self.style.WARNING(msg))
+            logger.warning(msg)
+            return
+
+        self.stdout.write(f"Found {len(json_files)} JSON files to process in {json_dir_path}")
+        logger.info("Found %s JSON files to process in %s", len(json_files), json_dir_path)
+
+        # Tracking statistics
+        total_stats = {
+            "files_processed": 0,
+            "files_with_errors": 0,
+            "games_created": 0,
+            "games_updated": 0,
+            "orgs_created": 0,
+            "orgs_updated": 0,
+            "campaigns_created": 0,
+            "campaigns_updated": 0,
+            "drops_created": 0,
+            "drops_updated": 0,
+            "benefits_created": 0,
+            "benefits_updated": 0,
+        }
+
+        # Process each file
+        for json_file in sorted(json_files):
+            try:
+                file_stats = self._process_json_file(json_file)
+                total_stats["files_processed"] += 1
+
+                # Update totals
+                for key, value in file_stats.items():
+                    if key in total_stats:
+                        total_stats[key] += value
+            except CommandError as e:
+                self.stdout.write(self.style.ERROR(str(e)))
+                total_stats["files_with_errors"] += 1
+                logger.exception("Error processing %s", json_file)
+            except Exception as e:
+                msg = f"Unexpected error processing {json_file}: {e}"
+                self.stdout.write(self.style.ERROR(msg))
+                logger.exception(msg)
+                total_stats["files_with_errors"] += 1
+
+        # Print summary
+        self.stdout.write(self.style.SUCCESS("Import process completed."))
+        self.stdout.write(f"Files processed: {total_stats['files_processed']}")
+        self.stdout.write(f"Files with errors: {total_stats['files_with_errors']}")
+        self.stdout.write(f"Games created: {total_stats['games_created']}, updated: {total_stats['games_updated']}")
+        self.stdout.write(
+            f"Organizations created: {total_stats['orgs_created']}, updated: {total_stats['orgs_updated']}",
+        )
+        self.stdout.write(
+            f"Campaigns created: {total_stats['campaigns_created']}, updated: {total_stats['campaigns_updated']}",
+        )
+        self.stdout.write(f"Drops created: {total_stats['drops_created']}, updated: {total_stats['drops_updated']}")
+        self.stdout.write(
+            f"Benefits created: {total_stats['benefits_created']}, updated: {total_stats['benefits_updated']}",
+        )
+
+        logger.info(
+            "Import completed. Processed: %s, Errors: %s",
+            total_stats["files_processed"],
+            total_stats["files_with_errors"],
+        )
+
+    def _process_json_file(self, json_file: Path) -> dict[str, int]:  # noqa: C901, PLR0912, PLR0914, PLR0915
+        """Process a single JSON file and import its data to the database.
+
+        Args:
+            json_file: Path object pointing to the JSON file.
+
+        Returns:
+            Dict with counts of created and updated records.
+
+        Raises:
+            CommandError: If the JSON file cannot be decoded or if there are
+                          issues with the JSON structure.
+        """
+        self.stdout.write(f"Processing file: {json_file}")
+        logger.info("Processing file: %s", json_file)
+
+        # Initialize stats for this file
+        stats = {
+            "games_created": 0,
+            "games_updated": 0,
+            "orgs_created": 0,
+            "orgs_updated": 0,
+            "campaigns_created": 0,
+            "campaigns_updated": 0,
+            "drops_created": 0,
+            "drops_updated": 0,
+            "benefits_created": 0,
+            "benefits_updated": 0,
+        }
+
+        try:
+            with json_file.open(encoding="utf-8") as f:
+                raw_data = json.load(f)
+        except FileNotFoundError as e:
+            msg = f"Error: File not found at {json_file}"
+            logger.exception(msg)
+            raise CommandError(msg) from e
+        except json.JSONDecodeError as e:
+            msg = f"Error: Could not decode JSON from {json_file}"
+            logger.exception(msg)
+            raise CommandError(msg) from e
+        except Exception as e:
+            msg = f"Error reading file: {e}"
+            logger.exception(msg)
+            raise CommandError(msg) from e
+
+        # Navigate to the relevant part of the JSON structure
+        try:
+            user_data = raw_data.get("data", {}).get("user", {})
+            campaign_data = user_data.get("dropCampaign")
+
+            if not campaign_data:
+                msg = f"Error: 'dropCampaign' key not found or is null in the JSON data: {json_file}"
+                logger.error(msg)
+                raise CommandError(msg)
+
+        except AttributeError as e:
+            msg = f"Error: Unexpected JSON structure in {json_file}. Could not find 'data' or 'user'."
+            logger.exception(msg)
+            raise CommandError(msg) from e
+
+        # Process the campaign data within a transaction
+        try:  # noqa: PLR1702
+            # Use a transaction to ensure atomicity
+            with transaction.atomic():
+                # --- 1. Process Game ---
+                game_data = campaign_data.get("game")
+                game_obj = None
+                if game_data and game_data.get("id"):
+                    game_defaults = {
+                        "slug": game_data.get("slug"),
+                        "display_name": game_data.get("displayName"),
+                    }
+                    game_obj, created = Game.objects.update_or_create(
+                        game_id=game_data["id"],
+                        defaults=game_defaults,
+                    )
+                    if created:
+                        self.stdout.write(self.style.SUCCESS(f"Created Game: {game_obj.display_name}"))
+                        stats["games_created"] += 1
+                    else:
+                        self.stdout.write(f"Updated/Found Game: {game_obj.display_name}")
+                        stats["games_updated"] += 1
+                else:
+                    self.stdout.write(self.style.WARNING("No game data found in campaign."))
+                    logger.warning("No game data found in campaign.")
+
+                # --- 2. Process Owner Organization ---
+                owner_data = campaign_data.get("owner")
+                owner_obj = None
+                if owner_data and owner_data.get("id"):
+                    owner_defaults = {"name": owner_data.get("name")}
+                    owner_obj, created = Organization.objects.update_or_create(
+                        org_id=owner_data["id"],
+                        defaults=owner_defaults,
+                    )
+                    if created:
+                        self.stdout.write(self.style.SUCCESS(f"Created Organization: {owner_obj.name}"))
+                        stats["orgs_created"] += 1
+                    else:
+                        self.stdout.write(f"Updated/Found Organization: {owner_obj.name}")
+                        stats["orgs_updated"] += 1
+                else:
+                    self.stdout.write(self.style.WARNING("No owner organization data found in campaign."))
+                    logger.warning("No owner organization data found in campaign.")
+
+                # --- 3. Process Drop Campaign ---
+                campaign_id = campaign_data.get("id")
+                if not campaign_id:
+                    msg = "Error: Campaign ID is missing."
+                    logger.error(msg)
+                    raise CommandError(msg)  # noqa: TRY301
+
+                start_at_str = campaign_data.get("startAt")
+                end_at_str = campaign_data.get("endAt")
+                start_at_dt = parse_datetime(start_at_str) if start_at_str else None
+                end_at_dt = parse_datetime(end_at_str) if end_at_str else None
+
+                if not start_at_dt or not end_at_dt:
+                    self.stdout.write(self.style.WARNING(f"Campaign {campaign_id} missing start or end date."))
+                    logger.warning("Campaign %s missing start or end date.", campaign_id)
+                    # Decide if you want to skip or handle this case differently
+
+                campaign_defaults = {
+                    "name": campaign_data.get("name"),
+                    "description": campaign_data.get("description"),
+                    "status": campaign_data.get("status"),
+                    "start_at": start_at_dt,
+                    "end_at": end_at_dt,
+                    "image_url": campaign_data.get("imageURL"),
+                    "details_url": campaign_data.get("detailsURL"),
+                    "account_link_url": campaign_data.get("accountLinkURL"),
+                    "game": game_obj,
+                    "owner": owner_obj,
+                }
+                campaign_obj, created = DropCampaign.objects.update_or_create(
+                    campaign_id=campaign_id,
+                    defaults=campaign_defaults,
+                )
+                if created:
+                    self.stdout.write(self.style.SUCCESS(f"Created Campaign: {campaign_obj.name}"))
+                    stats["campaigns_created"] += 1
+                else:
+                    self.stdout.write(f"Updated/Found Campaign: {campaign_obj.name}")
+                    stats["campaigns_updated"] += 1
+
+                # --- 4. Process Time Based Drops ---
+                time_drops_data = campaign_data.get("timeBasedDrops", [])
+                if not time_drops_data:
+                    self.stdout.write(self.style.NOTICE("No timeBasedDrops found in campaign data."))
+                    logger.info("No timeBasedDrops found in campaign data.")
+
+                for drop_data in time_drops_data:
+                    drop_id = drop_data.get("id")
+                    if not drop_id:
+                        self.stdout.write(self.style.WARNING("Skipping drop with missing ID."))
+                        logger.warning("Skipping drop with missing ID.")
+                        continue
+
+                    drop_start_str = drop_data.get("startAt")
+                    drop_end_str = drop_data.get("endAt")
+                    drop_start_dt = parse_datetime(drop_start_str) if drop_start_str else None
+                    drop_end_dt = parse_datetime(drop_end_str) if drop_end_str else None
+
+                    if not drop_start_dt or not drop_end_dt:
+                        self.stdout.write(self.style.WARNING(f"Drop {drop_id} missing start or end date. Skipping."))
+                        logger.warning("Drop %s missing start or end date. Skipping.", drop_id)
+                        continue
+
+                    drop_defaults = {
+                        "name": drop_data.get("name"),
+                        "start_at": drop_start_dt,
+                        "end_at": drop_end_dt,
+                        "required_minutes_watched": drop_data.get("requiredMinutesWatched"),
+                        "campaign": campaign_obj,
+                        # Add required_subs if needed: 'required_subs': drop_data.get('requiredSubs', 0)
+                    }
+
+                    drop_obj, created = TimeBasedDrop.objects.update_or_create(drop_id=drop_id, defaults=drop_defaults)
+                    if created:
+                        self.stdout.write(self.style.SUCCESS(f"  Created Time Drop: {drop_obj.name}"))
+                        stats["drops_created"] += 1
+                    else:
+                        self.stdout.write(f"  Updated/Found Time Drop: {drop_obj.name}")
+                        stats["drops_updated"] += 1
+
+                    # --- 5. Process Benefits within the Drop ---
+                    benefits_data = drop_data.get("benefitEdges", [])
+                    benefit_objs = []
+                    for edge in benefits_data:
+                        benefit_info = edge.get("benefit")
+                        if not benefit_info or not benefit_info.get("id"):
+                            self.stdout.write(self.style.WARNING("  Skipping benefit edge with missing data or ID."))
+                            logger.warning("Skipping benefit edge with missing data or ID.")
+                            continue
+
+                        benefit_id = benefit_info["id"]
+                        benefit_game_data = benefit_info.get("game")
+                        benefit_owner_data = benefit_info.get("ownerOrganization")
+                        benefit_created_str = benefit_info.get("createdAt")
+                        benefit_created_dt = parse_datetime(benefit_created_str) if benefit_created_str else None
+
+                        benefit_game_obj = None
+                        if benefit_game_data and benefit_game_data.get("id"):
+                            # Assuming game ID is unique, get it directly
+                            try:
+                                benefit_game_obj = Game.objects.get(game_id=benefit_game_data["id"])
+                            except Game.DoesNotExist:
+                                warning_msg = (
+                                    f"Game {benefit_game_data.get('name')} for benefit {benefit_id} "
+                                    f"not found. Creating minimally."
+                                )
+                                self.stdout.write(self.style.WARNING(f"    {warning_msg}"))
+                                logger.warning(warning_msg)
+                                # Optionally create a minimal game entry here if desired
+                                benefit_game_obj, game_created = Game.objects.update_or_create(
+                                    game_id=benefit_game_data["id"],
+                                    defaults={"display_name": benefit_game_data.get("name", "Unknown Game")},
+                                )
+                                if game_created:
+                                    stats["games_created"] += 1
+
+                        benefit_owner_obj = None
+                        if benefit_owner_data and benefit_owner_data.get("id"):
+                            try:
+                                benefit_owner_obj = Organization.objects.get(org_id=benefit_owner_data["id"])
+                            except Organization.DoesNotExist:
+                                warning_msg = (
+                                    f"Organization {benefit_owner_data.get('name')} for benefit {benefit_id} "
+                                    f"not found. Creating minimally."
+                                )
+                                self.stdout.write(self.style.WARNING(f"    {warning_msg}"))
+                                logger.warning(warning_msg)
+                                benefit_owner_obj, org_created = Organization.objects.update_or_create(
+                                    org_id=benefit_owner_data["id"],
+                                    defaults={"name": benefit_owner_data.get("name", "Unknown Org")},
+                                )
+                                if org_created:
+                                    stats["orgs_created"] += 1
+
+                        benefit_defaults = {
+                            "name": benefit_info.get("name"),
+                            "image_asset_url": benefit_info.get("imageAssetURL"),
+                            "entitlement_limit": edge.get("entitlementLimit", 1),  # Get limit from edge
+                            "distribution_type": benefit_info.get("distributionType"),
+                            "twitch_created_at": benefit_created_dt or timezone.now(),  # Provide a default if missing
+                            "game": benefit_game_obj,
+                            "owner_organization": benefit_owner_obj,
+                            "is_ios_available": benefit_info.get("isIosAvailable", False),
+                        }
+
+                        benefit_obj, b_created = Benefit.objects.update_or_create(
+                            benefit_id=benefit_id,
+                            defaults=benefit_defaults,
+                        )
+                        if b_created:
+                            self.stdout.write(self.style.SUCCESS(f"    Created Benefit: {benefit_obj.name}"))
+                            stats["benefits_created"] += 1
+                        else:
+                            self.stdout.write(f"    Updated/Found Benefit: {benefit_obj.name}")
+                            stats["benefits_updated"] += 1
+                        benefit_objs.append(benefit_obj)
+
+                    # Set the ManyToMany relationship for the drop
+                    if benefit_objs:
+                        drop_obj.benefits.set(benefit_objs)
+                        drops_msg = f"Associated {len(benefit_objs)} benefits with drop {drop_obj.name}."
+                        self.stdout.write(f"  {drops_msg}")
+                        logger.info("Associated %s benefits with drop %s.", len(benefit_objs), drop_obj.name)
+
+        except KeyError as e:
+            msg = f"Error: Missing expected key in JSON data - {e}"
+            logger.exception(msg)
+            raise CommandError(msg) from e
+        except Exception as e:
+            # The transaction will be rolled back automatically on exception
+            msg = f"An error occurred during import: {e}"
+            logger.exception(msg)
+            raise CommandError(msg) from e
+
+        # Return statistics from this file
+        return stats

core/migrations/0001_initial.py

@@ -1,40 +1,64 @@
-# Generated by Django 5.1.4 on 2024-12-11 04:58
+# Generated by Django 5.2 on 2025-05-01 00:45
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
-
-import auto_prefetch
 import django.contrib.auth.models
 import django.contrib.auth.validators
 import django.db.models.deletion
-import django.db.models.manager
 import django.utils.timezone
 from django.db import migrations, models
 
-if TYPE_CHECKING:
-    from django.db.migrations.operations.base import Operation
-
 
 class Migration(migrations.Migration):
-    """Initial migration for the core app.
-
-    This add the following models:
-    - ScrapedJson
-    - User
-    - Owner
-    - Game
-    - DropCampaign
-    - TimeBasedDrop
-    - Benefit
-    """
+    """Initial migration for the core app."""
 
     initial = True
 
-    dependencies: list[tuple[str, str]] = [
+    dependencies = [
         ("auth", "0012_alter_user_first_name_max_length"),
     ]
 
-    operations: list[Operation] = [
+    operations = [
+        migrations.CreateModel(
+            name="Game",
+            fields=[
+                (
+                    "game_id",
+                    models.TextField(help_text="The Twitch ID of the game.", primary_key=True, serialize=False),
+                ),
+                ("game_url", models.URLField(blank=True, help_text="The URL to the game on Twitch.")),
+                ("display_name", models.TextField(blank=True, help_text="The display name of the game.")),
+                ("box_art_url", models.URLField(blank=True, help_text="URL to the box art of the game.")),
+                ("slug", models.SlugField(blank=True, help_text="The slug for the game.")),
+                (
+                    "created_at",
+                    models.DateTimeField(auto_now_add=True, help_text="When the game was first added to the database."),
+                ),
+                ("modified_at", models.DateTimeField(auto_now=True, help_text="When the game was last modified.")),
+            ],
+            options={
+                "ordering": ["display_name"],
+            },
+        ),
+        migrations.CreateModel(
+            name="Organization",
+            fields=[
+                (
+                    "org_id",
+                    models.TextField(
+                        help_text="The Twitch ID of the owner.",
+                        primary_key=True,
+                        serialize=False,
+                        unique=True,
+                    ),
+                ),
+                ("name", models.TextField(blank=True, help_text="The name of the owner.")),
+                ("created_at", models.DateTimeField(auto_now_add=True)),
+                ("modified_at", models.DateTimeField(auto_now=True)),
+            ],
+            options={
+                "ordering": ["name"],
+            },
+        ),
         migrations.CreateModel(
             name="ScrapedJson",
             fields=[
@@ -46,13 +70,7 @@ class Migration(migrations.Migration):
             ],
             options={
                 "ordering": ["-created_at"],
-                "abstract": False,
-                "base_manager_name": "prefetch_manager",
             },
-            managers=[
-                ("objects", django.db.models.manager.Manager()),
-                ("prefetch_manager", django.db.models.manager.Manager()),
-            ],
         ),
         migrations.CreateModel(
             name="User",
@@ -94,10 +112,7 @@ class Migration(migrations.Migration):
                     "is_active",
                     models.BooleanField(
                         default=True,
-                        help_text=(
-                            "Designates whether this user should be treated as active. Unselect this instead of"
-                            " deleting accounts."
-                        ),
+                        help_text="Designates whether this user should be treated as active. Unselect this instead of deleting accounts.",  # noqa: E501
                         verbose_name="active",
                     ),
                 ),
@@ -106,10 +121,7 @@ class Migration(migrations.Migration):
                     "groups",
                     models.ManyToManyField(
                         blank=True,
-                        help_text=(
-                            "The groups this user belongs to. A user will get all permissions granted to each of their"
-                            " groups."
-                        ),
+                        help_text="The groups this user belongs to. A user will get all permissions granted to each of their groups.",  # noqa: E501
                         related_name="user_set",
                         related_query_name="user",
                         to="auth.group",
@@ -130,187 +142,82 @@ class Migration(migrations.Migration):
             ],
             options={
                 "ordering": ["username"],
-                "abstract": False,
-                "base_manager_name": "prefetch_manager",
             },
             managers=[
                 ("objects", django.contrib.auth.models.UserManager()),
             ],
         ),
-        migrations.CreateModel(
-            name="Owner",
-            fields=[
-                ("created_at", models.DateTimeField(auto_created=True)),
-                (
-                    "twitch_id",
-                    models.TextField(help_text="The Twitch ID of the owner.", primary_key=True, serialize=False),
-                ),
-                ("modified_at", models.DateTimeField(auto_now=True)),
-                ("name", models.TextField(blank=True, help_text="The name of the owner.")),
-            ],
-            options={
-                "ordering": ["name"],
-                "abstract": False,
-                "base_manager_name": "prefetch_manager",
-                "indexes": [
-                    models.Index(fields=["name"], name="owner_name_idx"),
-                    models.Index(fields=["created_at"], name="owner_created_at_idx"),
-                ],
-            },
-            managers=[
-                ("objects", django.db.models.manager.Manager()),
-                ("prefetch_manager", django.db.models.manager.Manager()),
-            ],
-        ),
-        migrations.CreateModel(
-            name="Game",
-            fields=[
-                (
-                    "created_at",
-                    models.DateTimeField(auto_created=True, help_text="When the game was first added to the database."),
-                ),
-                (
-                    "twitch_id",
-                    models.TextField(help_text="The Twitch ID of the game.", primary_key=True, serialize=False),
-                ),
-                ("modified_at", models.DateTimeField(auto_now=True, help_text="When the game was last modified.")),
-                ("game_url", models.URLField(blank=True, help_text="The URL to the game on Twitch.")),
-                ("display_name", models.TextField(blank=True, help_text="The display name of the game.")),
-                ("name", models.TextField(blank=True, help_text="The name of the game.")),
-                ("box_art_url", models.URLField(blank=True, help_text="URL to the box art of the game.")),
-                ("slug", models.TextField(blank=True)),
-                (
-                    "org",
-                    auto_prefetch.ForeignKey(
-                        null=True,
-                        on_delete=django.db.models.deletion.CASCADE,
-                        related_name="games",
-                        to="core.owner",
-                    ),
-                ),
-            ],
-            options={
-                "ordering": ["display_name"],
-                "abstract": False,
-                "base_manager_name": "prefetch_manager",
-            },
-            managers=[
-                ("objects", django.db.models.manager.Manager()),
-                ("prefetch_manager", django.db.models.manager.Manager()),
-            ],
-        ),
         migrations.CreateModel(
             name="DropCampaign",
             fields=[
                 (
-                    "created_at",
-                    models.DateTimeField(
-                        auto_created=True,
-                        help_text="When the drop campaign was first added to the database.",
-                    ),
-                ),
-                (
-                    "twitch_id",
+                    "campaign_id",
                     models.TextField(
                         help_text="The Twitch ID of the drop campaign.",
                         primary_key=True,
                         serialize=False,
+                        unique=True,
                     ),
                 ),
-                (
-                    "modified_at",
-                    models.DateTimeField(auto_now=True, help_text="When the drop campaign was last modified."),
-                ),
                 (
                     "account_link_url",
                     models.URLField(blank=True, help_text="The URL to link accounts for the drop campaign."),
                 ),
                 ("description", models.TextField(blank=True, help_text="The description of the drop campaign.")),
                 ("details_url", models.URLField(blank=True, help_text="The URL to the details of the drop campaign.")),
-                ("ends_at", models.DateTimeField(help_text="When the drop campaign ends.", null=True)),
-                ("starts_at", models.DateTimeField(help_text="When the drop campaign starts.", null=True)),
+                ("end_at", models.DateTimeField(help_text="When the drop campaign ends.", null=True)),
+                ("start_at", models.DateTimeField(help_text="When the drop campaign starts.", null=True)),
                 ("image_url", models.URLField(blank=True, help_text="The URL to the image for the drop campaign.")),
                 ("name", models.TextField(blank=True, help_text="The name of the drop campaign.")),
                 ("status", models.TextField(blank=True, help_text="The status of the drop campaign.")),
+                (
+                    "created_at",
+                    models.DateTimeField(
+                        auto_now_add=True,
+                        help_text="When the drop campaign was first added to the database.",
+                    ),
+                ),
+                (
+                    "modified_at",
+                    models.DateTimeField(auto_now=True, help_text="When the drop campaign was last modified."),
+                ),
                 (
                     "game",
-                    auto_prefetch.ForeignKey(
+                    models.ForeignKey(
+                        help_text="The game associated with this campaign",
                         null=True,
-                        on_delete=django.db.models.deletion.CASCADE,
+                        on_delete=django.db.models.deletion.SET_NULL,
                         related_name="drop_campaigns",
                         to="core.game",
                     ),
                 ),
                 (
-                    "scraped_json",
-                    auto_prefetch.ForeignKey(
-                        help_text="Reference to the JSON data from the Twitch API.",
+                    "owner",
+                    models.ForeignKey(
+                        help_text="The organization running this campaign",
                         null=True,
                         on_delete=django.db.models.deletion.SET_NULL,
-                        to="core.scrapedjson",
+                        related_name="drop_campaigns",
+                        to="core.organization",
                     ),
                 ),
             ],
             options={
-                "ordering": ["ends_at"],
-                "abstract": False,
-                "base_manager_name": "prefetch_manager",
+                "ordering": ["end_at"],
             },
-            managers=[
-                ("objects", django.db.models.manager.Manager()),
-                ("prefetch_manager", django.db.models.manager.Manager()),
-            ],
-        ),
-        migrations.CreateModel(
-            name="TimeBasedDrop",
-            fields=[
-                (
-                    "created_at",
-                    models.DateTimeField(auto_created=True, help_text="When the drop was first added to the database."),
-                ),
-                (
-                    "twitch_id",
-                    models.TextField(help_text="The Twitch ID of the drop.", primary_key=True, serialize=False),
-                ),
-                ("modified_at", models.DateTimeField(auto_now=True, help_text="When the drop was last modified.")),
-                (
-                    "required_subs",
-                    models.PositiveBigIntegerField(help_text="The number of subs required for the drop.", null=True),
-                ),
-                ("ends_at", models.DateTimeField(help_text="When the drop ends.", null=True)),
-                ("name", models.TextField(blank=True, help_text="The name of the drop.")),
-                (
-                    "required_minutes_watched",
-                    models.PositiveBigIntegerField(help_text="The number of minutes watched required.", null=True),
-                ),
-                ("starts_at", models.DateTimeField(help_text="When the drop starts.", null=True)),
-                (
-                    "drop_campaign",
-                    auto_prefetch.ForeignKey(
-                        help_text="The drop campaign this drop is part of.",
-                        null=True,
-                        on_delete=django.db.models.deletion.CASCADE,
-                        related_name="drops",
-                        to="core.dropcampaign",
-                    ),
-                ),
-            ],
-            options={
-                "ordering": ["required_minutes_watched"],
-                "abstract": False,
-                "base_manager_name": "prefetch_manager",
-            },
-            managers=[
-                ("objects", django.db.models.manager.Manager()),
-                ("prefetch_manager", django.db.models.manager.Manager()),
-            ],
         ),
         migrations.CreateModel(
             name="Benefit",
             fields=[
-                ("created_at", models.DateTimeField(auto_created=True, null=True)),
-                ("twitch_id", models.TextField(primary_key=True, serialize=False)),
-                ("modified_at", models.DateTimeField(auto_now=True)),
+                (
+                    "benefit_id",
+                    models.TextField(
+                        help_text="Twitch's unique ID for the benefit",
+                        primary_key=True,
+                        serialize=False,
+                        unique=True,
+                    ),
+                ),
                 (
                     "twitch_created_at",
                     models.DateTimeField(help_text="When the benefit was created on Twitch.", null=True),
@@ -318,103 +225,88 @@ class Migration(migrations.Migration):
                 (
                     "entitlement_limit",
                     models.PositiveBigIntegerField(
-                        help_text="The number of times the benefit can be claimed.",
-                        null=True,
+                        default=1,
+                        help_text="How many times this benefit can be claimed per user",
                     ),
                 ),
                 ("image_asset_url", models.URLField(blank=True, help_text="The URL to the image for the benefit.")),
                 ("is_ios_available", models.BooleanField(help_text="If the benefit is farmable on iOS.", null=True)),
-                ("name", models.TextField(blank=True, help_text="The name of the benefit.")),
+                ("name", models.TextField(blank=True, help_text="Name of the benefit/reward")),
                 ("distribution_type", models.TextField(blank=True, help_text="The distribution type of the benefit.")),
+                ("created_at", models.DateTimeField(auto_now_add=True)),
+                ("modified_at", models.DateTimeField(auto_now=True)),
                 (
                     "game",
-                    auto_prefetch.ForeignKey(
+                    models.ForeignKey(
                         null=True,
-                        on_delete=django.db.models.deletion.CASCADE,
+                        on_delete=django.db.models.deletion.SET_NULL,
                         related_name="benefits",
                         to="core.game",
                     ),
                 ),
                 (
                     "owner_organization",
-                    auto_prefetch.ForeignKey(
+                    models.ForeignKey(
                         null=True,
-                        on_delete=django.db.models.deletion.CASCADE,
+                        on_delete=django.db.models.deletion.SET_NULL,
                         related_name="benefits",
-                        to="core.owner",
-                    ),
-                ),
-                (
-                    "time_based_drop",
-                    auto_prefetch.ForeignKey(
-                        help_text="The time based drop this benefit is for.",
-                        null=True,
-                        on_delete=django.db.models.deletion.CASCADE,
-                        related_name="benefits",
-                        to="core.timebaseddrop",
+                        to="core.organization",
                     ),
                 ),
             ],
             options={
                 "ordering": ["-twitch_created_at"],
-                "abstract": False,
-                "base_manager_name": "prefetch_manager",
             },
-            managers=[
-                ("objects", django.db.models.manager.Manager()),
-                ("prefetch_manager", django.db.models.manager.Manager()),
+        ),
+        migrations.CreateModel(
+            name="TimeBasedDrop",
+            fields=[
+                (
+                    "drop_id",
+                    models.TextField(
+                        help_text="The Twitch ID of the drop.",
+                        primary_key=True,
+                        serialize=False,
+                        unique=True,
+                    ),
+                ),
+                (
+                    "required_subs",
+                    models.PositiveBigIntegerField(help_text="The number of subs required for the drop.", null=True),
+                ),
+                ("ends_at", models.DateTimeField(help_text="When the drop ends.", null=True)),
+                ("name", models.TextField(blank=True, help_text="Name of the time-based drop")),
+                (
+                    "required_minutes_watched",
+                    models.PositiveBigIntegerField(help_text="The number of minutes watched required.", null=True),
+                ),
+                ("start_at", models.DateTimeField(help_text="Drop start time")),
+                ("end_at", models.DateTimeField(help_text="Drop end time")),
+                (
+                    "created_at",
+                    models.DateTimeField(auto_now_add=True, help_text="When the drop was first added to the database."),
+                ),
+                ("modified_at", models.DateTimeField(auto_now=True, help_text="When the drop was last modified.")),
+                (
+                    "benefits",
+                    models.ManyToManyField(
+                        help_text="Benefits awarded by this drop",
+                        related_name="time_based_drops",
+                        to="core.benefit",
+                    ),
+                ),
+                (
+                    "campaign",
+                    models.ForeignKey(
+                        help_text="The campaign this drop belongs to",
+                        on_delete=django.db.models.deletion.CASCADE,
+                        related_name="time_based_drops",
+                        to="core.dropcampaign",
+                    ),
+                ),
             ],
-        ),
-        migrations.AddIndex(
-            model_name="game",
-            index=models.Index(fields=["display_name"], name="game_display_name_idx"),
-        ),
-        migrations.AddIndex(
-            model_name="game",
-            index=models.Index(fields=["name"], name="game_name_idx"),
-        ),
-        migrations.AddIndex(
-            model_name="game",
-            index=models.Index(fields=["created_at"], name="game_created_at_idx"),
-        ),
-        migrations.AddIndex(
-            model_name="dropcampaign",
-            index=models.Index(fields=["name"], name="drop_campaign_name_idx"),
-        ),
-        migrations.AddIndex(
-            model_name="dropcampaign",
-            index=models.Index(fields=["starts_at"], name="drop_campaign_starts_at_idx"),
-        ),
-        migrations.AddIndex(
-            model_name="dropcampaign",
-            index=models.Index(fields=["ends_at"], name="drop_campaign_ends_at_idx"),
-        ),
-        migrations.AddIndex(
-            model_name="timebaseddrop",
-            index=models.Index(fields=["name"], name="time_based_drop_name_idx"),
-        ),
-        migrations.AddIndex(
-            model_name="timebaseddrop",
-            index=models.Index(fields=["starts_at"], name="time_based_drop_starts_at_idx"),
-        ),
-        migrations.AddIndex(
-            model_name="timebaseddrop",
-            index=models.Index(fields=["ends_at"], name="time_based_drop_ends_at_idx"),
-        ),
-        migrations.AddIndex(
-            model_name="benefit",
-            index=models.Index(fields=["name"], name="benefit_name_idx"),
-        ),
-        migrations.AddIndex(
-            model_name="benefit",
-            index=models.Index(fields=["twitch_created_at"], name="benefit_twitch_created_at_idx"),
-        ),
-        migrations.AddIndex(
-            model_name="benefit",
-            index=models.Index(fields=["created_at"], name="benefit_created_at_idx"),
-        ),
-        migrations.AddIndex(
-            model_name="benefit",
-            index=models.Index(fields=["is_ios_available"], name="benefit_is_ios_available_idx"),
+            options={
+                "ordering": ["required_minutes_watched"],
+            },
         ),
     ]

core/migrations/0002_alter_user_options.py

@@ -1,24 +0,0 @@
-# Generated by Django 5.1.4 on 2024-12-16 18:28
-from __future__ import annotations
-
-from typing import TYPE_CHECKING
-
-from django.db import migrations
-
-if TYPE_CHECKING:
-    from django.db.migrations.operations.base import Operation
-
-
-class Migration(migrations.Migration):
-    """This migration alters the options of the User model to order by username."""
-
-    dependencies: list[tuple[str, str]] = [
-        ("core", "0001_initial"),
-    ]
-
-    operations: list[Operation] = [
-        migrations.AlterModelOptions(
-            name="user",
-            options={"ordering": ["username"]},
-        ),
-    ]

core/migrations/0003_alter_benefit_created_at_and_more.py

@@ -1,48 +0,0 @@
-# Generated by Django 5.1.4 on 2024-12-16 18:31
-from __future__ import annotations
-
-import datetime
-from typing import TYPE_CHECKING
-
-from django.db import migrations, models
-
-if TYPE_CHECKING:
-    from django.db.migrations.operations.base import Operation
-
-
-class Migration(migrations.Migration):
-    """This migration alters the created_at field on the Benefit, DropCampaign, Owner, and TimeBasedDrop models."""
-
-    dependencies: list[tuple[str, str]] = [
-        ("core", "0002_alter_user_options"),
-    ]
-
-    operations: list[Operation] = [
-        migrations.AlterField(
-            model_name="benefit",
-            name="created_at",
-            field=models.DateTimeField(
-                auto_now_add=True,
-                default=datetime.datetime(2024, 12, 16, 18, 31, 14, 851533, tzinfo=datetime.UTC),
-            ),
-            preserve_default=False,
-        ),
-        migrations.AlterField(
-            model_name="dropcampaign",
-            name="created_at",
-            field=models.DateTimeField(
-                auto_now_add=True,
-                help_text="When the drop campaign was first added to the database.",
-            ),
-        ),
-        migrations.AlterField(
-            model_name="owner",
-            name="created_at",
-            field=models.DateTimeField(auto_now_add=True),
-        ),
-        migrations.AlterField(
-            model_name="timebaseddrop",
-            name="created_at",
-            field=models.DateTimeField(auto_now_add=True, help_text="When the drop was first added to the database."),
-        ),
-    ]

core/migrations/0004_alter_game_created_at.py

@@ -1,25 +0,0 @@
-# Generated by Django 5.1.4 on 2024-12-18 22:35
-from __future__ import annotations
-
-from typing import TYPE_CHECKING
-
-from django.db import migrations, models
-
-if TYPE_CHECKING:
-    from django.db.migrations.operations.base import Operation
-
-
-class Migration(migrations.Migration):
-    """Fix created_at using auto_created instead of auto_now_add."""
-
-    dependencies: list[tuple[str, str]] = [
-        ("core", "0003_alter_benefit_created_at_and_more"),
-    ]
-
-    operations: list[Operation] = [
-        migrations.AlterField(
-            model_name="game",
-            name="created_at",
-            field=models.DateTimeField(auto_now_add=True, help_text="When the game was first added to the database."),
-        ),
-    ]

core/models.py

@@ -1,17 +1,11 @@
 from __future__ import annotations
 
 import logging
-from typing import TYPE_CHECKING, ClassVar, Self
+from typing import ClassVar
 
-import auto_prefetch
 from django.contrib.auth.models import AbstractUser
 from django.db import models
 
-from core.models_utils import update_fields, wrong_typename
-
-if TYPE_CHECKING:
-    from django.db.models import Index
-
 logger: logging.Logger = logging.getLogger(__name__)
 
 
@@ -26,7 +20,7 @@ class User(AbstractUser):
         return self.username
 
 
-class ScrapedJson(auto_prefetch.Model):
+class ScrapedJson(models.Model):
     """The JSON data from the Twitch API.
 
     This data is from https://github.com/TheLovinator1/TwitchDropsMiner.
@@ -37,7 +31,7 @@ class ScrapedJson(auto_prefetch.Model):
     modified_at = models.DateTimeField(auto_now=True)
     imported_at = models.DateTimeField(null=True)
 
-    class Meta(auto_prefetch.Model.Meta):
+    class Meta:
         ordering: ClassVar[list[str]] = ["-created_at"]
 
     def __str__(self) -> str:
@@ -45,481 +39,148 @@ class ScrapedJson(auto_prefetch.Model):
         return f"{'' if self.imported_at else 'Not imported - '}{self.created_at}"
 
 
-class Owner(auto_prefetch.Model):
-    """The company or person that owns the game.
+class Organization(models.Model):
+    """Represents the owner/organization of a Drop Campaign."""
 
-    Drops will be grouped by the owner. Users can also subscribe to owners.
+    org_id = models.TextField(primary_key=True, unique=True, help_text="The Twitch ID of the owner.")
+    name = models.TextField(blank=True, help_text="The name of the owner.")
 
-    JSON:
-        {
-            "data": {
-                "user": {
-                    "dropCampaign": {
-                        "owner": {
-                            "id": "36c4e21d-bdf3-410c-97c3-5a5a4bf1399b",
-                            "name": "The Pok\u00e9mon Company",
-                            "__typename": "Organization"
-                        }
-                    }
-                }
-            }
-        }
-    """
-
-    # Django fields
-    # Example: "36c4e21d-bdf3-410c-97c3-5a5a4bf1399b"
-    twitch_id = models.TextField(primary_key=True, help_text="The Twitch ID of the owner.")
     created_at = models.DateTimeField(auto_now_add=True)
     modified_at = models.DateTimeField(auto_now=True)
 
-    # Twitch fields
-    # Example: "The Pokémon Company"
-    name = models.TextField(blank=True, help_text="The name of the owner.")
-
-    class Meta(auto_prefetch.Model.Meta):
+    class Meta:
         ordering: ClassVar[list[str]] = ["name"]
-        indexes: ClassVar[list[Index]] = [
-            models.Index(fields=["name"], name="owner_name_idx"),
-            models.Index(fields=["created_at"], name="owner_created_at_idx"),
-        ]
 
     def __str__(self) -> str:
         """Return the name of the owner."""
-        return f"{self.name or self.twitch_id} - {self.created_at}"
-
-    def import_json(self, data: dict) -> Self:
-        """Import the data from the Twitch API."""
-        if wrong_typename(data, "Organization"):
-            return self
-
-        field_mapping: dict[str, str] = {"name": "name"}
-        updated: int = update_fields(instance=self, data=data, field_mapping=field_mapping)
-        if updated > 0:
-            logger.info("Updated %s fields for %s", updated, self)
-
-        return self
+        return f"{self.name or self.org_id} - {self.created_at}"
 
 
-class Game(auto_prefetch.Model):
-    """The game the drop campaign is for. Note that some reward campaigns are not tied to a game.
+class Game(models.Model):
+    """The game the drop campaign is for. Note that some reward campaigns are not tied to a game."""
 
-    JSON:
-        "game": {
-            "id": "155409827",
-            "slug": "pokemon-trading-card-game-live",
-            "displayName": "Pok\u00e9mon Trading Card Game Live",
-            "__typename": "Game"
-        }
+    game_id = models.TextField(primary_key=True, help_text="The Twitch ID of the game.")
+    game_url = models.URLField(blank=True, help_text="The URL to the game on Twitch.")
+    display_name = models.TextField(blank=True, help_text="The display name of the game.")
+    box_art_url = models.URLField(blank=True, help_text="URL to the box art of the game.")
+    slug = models.SlugField(blank=True, help_text="The slug for the game.")
 
-
-    Secondary JSON:
-        "game": {
-            "id": "155409827",
-            "displayName": "Pok\u00e9mon Trading Card Game Live",
-            "boxArtURL": "https://static-cdn.jtvnw.net/ttv-boxart/155409827_IGDB-120x160.jpg",
-            "__typename": "Game"
-        }
-
-
-    Tertiary JSON:
-        "game": {
-            "id": "155409827",
-            "name": "Pok\u00e9mon Trading Card Game Live",
-            "__typename": "Game"
-        }
-
-    """
-
-    # Django fields
-    # "155409827"
-    twitch_id = models.TextField(primary_key=True, help_text="The Twitch ID of the game.")
     created_at = models.DateTimeField(auto_now_add=True, help_text="When the game was first added to the database.")
     modified_at = models.DateTimeField(auto_now=True, help_text="When the game was last modified.")
 
-    # Twitch fields
-    # "https://www.twitch.tv/directory/category/pokemon-trading-card-game-live"
-    # This is created when the game is created.
-    game_url = models.URLField(blank=True, help_text="The URL to the game on Twitch.")
-
-    # "Pokémon Trading Card Game Live"
-    display_name = models.TextField(blank=True, help_text="The display name of the game.")
-
-    # "Pokémon Trading Card Game Live"
-    name = models.TextField(blank=True, help_text="The name of the game.")
-
-    # "https://static-cdn.jtvnw.net/ttv-boxart/155409827_IGDB-120x160.jpg"
-    box_art_url = models.URLField(blank=True, help_text="URL to the box art of the game.")
-
-    # "pokemon-trading-card-game-live"
-    slug = models.TextField(blank=True)
-
-    # The owner of the game.
-    # This is optional because some games are not tied to an owner.
-    org = auto_prefetch.ForeignKey(Owner, on_delete=models.CASCADE, related_name="games", null=True)
-
-    class Meta(auto_prefetch.Model.Meta):
+    class Meta:
         ordering: ClassVar[list[str]] = ["display_name"]
-        indexes: ClassVar[list[Index]] = [
-            models.Index(fields=["display_name"], name="game_display_name_idx"),
-            models.Index(fields=["name"], name="game_name_idx"),
-            models.Index(fields=["created_at"], name="game_created_at_idx"),
-        ]
 
     def __str__(self) -> str:
         """Return the name of the game and when it was created."""
-        return f"{self.display_name or self.twitch_id} - {self.created_at}"
-
-    def import_json(self, data: dict, owner: Owner | None) -> Self:
-        """Import the data from the Twitch API."""
-        if wrong_typename(data, "Game"):
-            return self
-
-        if not owner:
-            logger.error("Owner is required for %s: %s", self, data)
-            return self
-
-        # Map the fields from the JSON data to the Django model fields.
-        field_mapping: dict[str, str] = {
-            "displayName": "display_name",
-            "name": "name",
-            "boxArtURL": "box_art_url",
-            "slug": "slug",
-        }
-        updated: int = update_fields(instance=self, data=data, field_mapping=field_mapping)
-
-        if updated > 0:
-            logger.info("Updated %s fields for %s", updated, self)
-
-        # Update the owner if the owner is different or not set.
-        if owner != self.org:
-            self.org = owner
-            logger.info("Updated owner %s for %s", owner, self)
-
-        self.game_url = f"https://www.twitch.tv/directory/category/{self.slug}"
-
-        self.save()
-
-        return self
+        return f"{self.display_name or self.game_id} - {self.created_at}"
 
 
-class DropCampaign(auto_prefetch.Model):
+class DropCampaign(models.Model):
     """This is the drop campaign we will see on the front end."""
 
-    # Django fields
-    # "f257ce6e-502a-11ef-816e-0a58a9feac02"
-    twitch_id = models.TextField(primary_key=True, help_text="The Twitch ID of the drop campaign.")
+    campaign_id = models.TextField(primary_key=True, unique=True, help_text="The Twitch ID of the drop campaign.")
+    account_link_url = models.URLField(blank=True, help_text="The URL to link accounts for the drop campaign.")
+    description = models.TextField(blank=True, help_text="The description of the drop campaign.")
+    details_url = models.URLField(blank=True, help_text="The URL to the details of the drop campaign.")
+    end_at = models.DateTimeField(null=True, help_text="When the drop campaign ends.")
+    start_at = models.DateTimeField(null=True, help_text="When the drop campaign starts.")
+    image_url = models.URLField(blank=True, help_text="The URL to the image for the drop campaign.")
+    name = models.TextField(blank=True, help_text="The name of the drop campaign.")
+    status = models.TextField(blank=True, help_text="The status of the drop campaign.")
+    game = models.ForeignKey(
+        to=Game,
+        help_text="The game associated with this campaign",
+        null=True,
+        on_delete=models.SET_NULL,
+        related_name="drop_campaigns",
+    )
+    owner = models.ForeignKey(
+        Organization,
+        help_text="The organization running this campaign",
+        null=True,
+        on_delete=models.SET_NULL,
+        related_name="drop_campaigns",
+    )
+
     created_at = models.DateTimeField(
         auto_now_add=True,
         help_text="When the drop campaign was first added to the database.",
     )
     modified_at = models.DateTimeField(auto_now=True, help_text="When the drop campaign was last modified.")
 
-    # Twitch fields
-    # "https://www.halowaypoint.com/settings/linked-accounts"
-    account_link_url = models.URLField(blank=True, help_text="The URL to link accounts for the drop campaign.")
-
-    # "Tune into this HCS Grassroots event to earn Halo Infinite in-game content!"
-    description = models.TextField(blank=True, help_text="The description of the drop campaign.")
-
-    # "https://www.halowaypoint.com"
-    details_url = models.URLField(blank=True, help_text="The URL to the details of the drop campaign.")
-
-    # "2024-08-12T05:59:59.999Z"
-    ends_at = models.DateTimeField(null=True, help_text="When the drop campaign ends.")
-
-    # "2024-08-11T11:00:00Z""
-    starts_at = models.DateTimeField(null=True, help_text="When the drop campaign starts.")
-
-    # "https://static-cdn.jtvnw.net/twitch-quests-assets/CAMPAIGN/c8e02666-8b86-471f-bf38-7ece29a758e4.png"
-    image_url = models.URLField(blank=True, help_text="The URL to the image for the drop campaign.")
-
-    # "HCS Open Series - Week 1 - DAY 2 - AUG11"
-    name = models.TextField(blank=True, help_text="The name of the drop campaign.")
-
-    # "ACTIVE"
-    status = models.TextField(blank=True, help_text="The status of the drop campaign.")
-
-    # The game this drop campaign is for.
-    game = auto_prefetch.ForeignKey(to=Game, on_delete=models.CASCADE, related_name="drop_campaigns", null=True)
-
-    # The JSON data from the Twitch API.
-    # We use this to find out where the game came from.
-    scraped_json = auto_prefetch.ForeignKey(
-        to=ScrapedJson,
-        null=True,
-        on_delete=models.SET_NULL,
-        help_text="Reference to the JSON data from the Twitch API.",
-    )
-
-    class Meta(auto_prefetch.Model.Meta):
-        ordering: ClassVar[list[str]] = ["ends_at"]
-        indexes: ClassVar[list[Index]] = [
-            models.Index(fields=["name"], name="drop_campaign_name_idx"),
-            models.Index(fields=["starts_at"], name="drop_campaign_starts_at_idx"),
-            models.Index(fields=["ends_at"], name="drop_campaign_ends_at_idx"),
-        ]
+    class Meta:
+        ordering: ClassVar[list[str]] = ["end_at"]
 
     def __str__(self) -> str:
         """Return the name of the drop campaign and when it was created."""
-        return f"{self.name or self.twitch_id} - {self.created_at}"
-
-    def import_json(self, data: dict, game: Game | None, *, scraping_local_files: bool = False) -> Self:
-        """Import the data from the Twitch API.
-
-        Args:
-            data (dict | None): The data from the Twitch API.
-            game (Game | None): The game this drop campaign is for.
-            scraping_local_files (bool, optional): If this was scraped from local data. Defaults to True.
-
-        Returns:
-            Self: The updated drop campaign.
-        """
-        if wrong_typename(data, "DropCampaign"):
-            return self
-
-        field_mapping: dict[str, str] = {
-            "name": "name",
-            "accountLinkURL": "account_link_url",  # TODO(TheLovinator): Should archive site.  # noqa: TD003
-            "description": "description",
-            "endAt": "ends_at",
-            "startAt": "starts_at",
-            "detailsURL": "details_url",  # TODO(TheLovinator): Should archive site.  # noqa: TD003
-            "imageURL": "image_url",
-        }
-
-        updated: int = update_fields(instance=self, data=data, field_mapping=field_mapping)
-        if updated > 0:
-            logger.info("Updated %s fields for %s", updated, self)
-
-        if not scraping_local_files:
-            status = data.get("status")
-            if status and status != self.status:
-                self.status = status
-                self.save()
-
-        # Update the game if the game is different or not set.
-        if game and game != self.game:
-            self.game = game
-            logger.info("Updated game %s for %s", game, self)
-            self.save()
-
-        return self
+        return f"{self.name or self.campaign_id} - {self.created_at}"
 
 
-class TimeBasedDrop(auto_prefetch.Model):
-    """This is the drop we will see on the front end.
+class Benefit(models.Model):
+    """Represents a specific reward/benefit within a Drop."""
 
-    JSON:
-        {
-            "id": "bd663e10-b297-11ef-a6a3-0a58a9feac02",
-            "requiredSubs": 0,
-            "benefitEdges": [
-                {
-                    "benefit": {
-                        "id": "f751ba67-7c8b-4c41-b6df-bcea0914f3ad_CUSTOM_ID_EnergisingBoltFlaskEffect",
-                        "createdAt": "2024-12-04T23:25:50.995Z",
-                        "entitlementLimit": 1,
-                        "game": {
-                            "id": "1702520304",
-                            "name": "Path of Exile 2",
-                            "__typename": "Game"
-                        },
-                        "imageAssetURL": "https://static-cdn.jtvnw.net/twitch-quests-assets/REWARD/d70e4e75-7237-4730-9a10-b6016aaaa795.png",
-                        "isIosAvailable": false,
-                        "name": "Energising Bolt Flask",
-                        "ownerOrganization": {
-                            "id": "f751ba67-7c8b-4c41-b6df-bcea0914f3ad",
-                            "name": "Grinding Gear Games",
-                            "__typename": "Organization"
-                        },
-                        "distributionType": "DIRECT_ENTITLEMENT",
-                        "__typename": "DropBenefit"
-                    },
-                    "entitlementLimit": 1,
-                    "__typename": "DropBenefitEdge"
-                }
-            ],
-            "endAt": "2024-12-14T07:59:59.996Z",
-            "name": "Early Access Bundle",
-            "preconditionDrops": null,
-            "requiredMinutesWatched": 180,
-            "startAt": "2024-12-06T19:00:00Z",
-            "__typename": "TimeBasedDrop"
-        }
-    """
+    benefit_id = models.TextField(primary_key=True, unique=True, help_text="Twitch's unique ID for the benefit")
+    twitch_created_at = models.DateTimeField(null=True, help_text="When the benefit was created on Twitch.")
+    entitlement_limit = models.PositiveBigIntegerField(
+        default=1,
+        help_text="How many times this benefit can be claimed per user",
+    )
+    image_asset_url = models.URLField(blank=True, help_text="The URL to the image for the benefit.")
+    is_ios_available = models.BooleanField(null=True, help_text="If the benefit is farmable on iOS.")
+    name = models.TextField(blank=True, help_text="Name of the benefit/reward")
+    game = models.ForeignKey(Game, on_delete=models.SET_NULL, related_name="benefits", null=True)
+    owner_organization = models.ForeignKey(
+        Organization,
+        on_delete=models.SET_NULL,
+        related_name="benefits",
+        null=True,
+    )
+    distribution_type = models.TextField(blank=True, help_text="The distribution type of the benefit.")
 
-    # Django fields
-    # "d5cdf372-502b-11ef-bafd-0a58a9feac02"
-    twitch_id = models.TextField(primary_key=True, help_text="The Twitch ID of the drop.")
-    created_at = models.DateTimeField(auto_now_add=True, help_text="When the drop was first added to the database.")
-    modified_at = models.DateTimeField(auto_now=True, help_text="When the drop was last modified.")
+    created_at = models.DateTimeField(auto_now_add=True)
+    modified_at = models.DateTimeField(auto_now=True)
 
-    # Twitch fields
-    # "1"
+    class Meta:
+        ordering: ClassVar[list[str]] = ["-twitch_created_at"]
+
+    def __str__(self) -> str:
+        """Return the name of the benefit and when it was created."""
+        return f"{self.name or self.benefit_id} - {self.twitch_created_at}"
+
+
+class TimeBasedDrop(models.Model):
+    """Represents a time-based drop within a Campaign."""
+
+    drop_id = models.TextField(primary_key=True, unique=True, help_text="The Twitch ID of the drop.")
     required_subs = models.PositiveBigIntegerField(null=True, help_text="The number of subs required for the drop.")
-
-    # "2024-08-12T05:59:59.999Z"
     ends_at = models.DateTimeField(null=True, help_text="When the drop ends.")
-
-    # "Cosmic Nexus Chimera"
-    name = models.TextField(blank=True, help_text="The name of the drop.")
-
-    # "120"
+    name = models.TextField(blank=True, help_text="Name of the time-based drop")
     required_minutes_watched = models.PositiveBigIntegerField(
         null=True,
         help_text="The number of minutes watched required.",
     )
+    start_at = models.DateTimeField(help_text="Drop start time")
+    end_at = models.DateTimeField(help_text="Drop end time")
 
-    # "2024-08-11T11:00:00Z"
-    starts_at = models.DateTimeField(null=True, help_text="When the drop starts.")
-
-    # The drop campaign this drop is part of.
-    drop_campaign = auto_prefetch.ForeignKey(
+    campaign = models.ForeignKey(
         DropCampaign,
+        help_text="The campaign this drop belongs to",
         on_delete=models.CASCADE,
-        related_name="drops",
-        null=True,
-        help_text="The drop campaign this drop is part of.",
+        related_name="time_based_drops",
+    )
+    benefits = models.ManyToManyField(
+        Benefit,
+        related_name="time_based_drops",
+        help_text="Benefits awarded by this drop",
     )
 
-    class Meta(auto_prefetch.Model.Meta):
+    created_at = models.DateTimeField(auto_now_add=True, help_text="When the drop was first added to the database.")
+    modified_at = models.DateTimeField(auto_now=True, help_text="When the drop was last modified.")
+
+    class Meta:
         ordering: ClassVar[list[str]] = ["required_minutes_watched"]
-        indexes: ClassVar[list[Index]] = [
-            models.Index(fields=["name"], name="time_based_drop_name_idx"),
-            models.Index(fields=["starts_at"], name="time_based_drop_starts_at_idx"),
-            models.Index(fields=["ends_at"], name="time_based_drop_ends_at_idx"),
-        ]
 
     def __str__(self) -> str:
         """Return the name of the drop and when it was created."""
-        return f"{self.name or self.twitch_id} - {self.created_at}"
-
-    def import_json(self, data: dict, drop_campaign: DropCampaign | None) -> Self:
-        """Import the data from the Twitch API."""
-        if wrong_typename(data, "TimeBasedDrop"):
-            return self
-
-        # preconditionDrops is null in the JSON. We probably should use it when we know what it is.
-        if data.get("preconditionDrops"):
-            logger.error("preconditionDrops is not None for %s", self)
-
-        field_mapping: dict[str, str] = {
-            "name": "name",
-            "requiredSubs": "required_subs",
-            "requiredMinutesWatched": "required_minutes_watched",
-            "startAt": "starts_at",
-            "endAt": "ends_at",
-        }
-
-        updated: int = update_fields(instance=self, data=data, field_mapping=field_mapping)
-        if updated > 0:
-            logger.info("Updated %s fields for %s", updated, self)
-
-        if drop_campaign and drop_campaign != self.drop_campaign:
-            self.drop_campaign = drop_campaign
-            logger.info("Updated drop campaign %s for %s", drop_campaign, self)
-            self.save()
-
-        return self
-
-
-class Benefit(auto_prefetch.Model):
-    """Benefits are the rewards for the drops."""
-
-    # Django fields
-    # "d5cdf372-502b-11ef-bafd-0a58a9feac02"
-    twitch_id = models.TextField(primary_key=True)
-    created_at = models.DateTimeField(auto_now_add=True)
-    modified_at = models.DateTimeField(auto_now=True)
-
-    # Twitch fields
-    # Note: This is Twitch's created_at from the API and not our created_at.
-    # "2023-11-09T01:18:00.126Z"
-    twitch_created_at = models.DateTimeField(null=True, help_text="When the benefit was created on Twitch.")
-
-    # "1"
-    entitlement_limit = models.PositiveBigIntegerField(
-        null=True,
-        help_text="The number of times the benefit can be claimed.",
-    )
-
-    # "https://static-cdn.jtvnw.net/twitch-quests-assets/REWARD/e58ad175-73f6-4392-80b8-fb0223163733.png"
-    image_asset_url = models.URLField(blank=True, help_text="The URL to the image for the benefit.")
-
-    # "True" or "False". None if unknown.
-    is_ios_available = models.BooleanField(null=True, help_text="If the benefit is farmable on iOS.")
-
-    # "Cosmic Nexus Chimera"
-    name = models.TextField(blank=True, help_text="The name of the benefit.")
-
-    # The game this benefit is for.
-    time_based_drop = auto_prefetch.ForeignKey(
-        TimeBasedDrop,
-        on_delete=models.CASCADE,
-        related_name="benefits",
-        null=True,
-        help_text="The time based drop this benefit is for.",
-    )
-
-    # The game this benefit is for.
-    game = auto_prefetch.ForeignKey(Game, on_delete=models.CASCADE, related_name="benefits", null=True)
-
-    # The owner of the benefit.
-    owner_organization = auto_prefetch.ForeignKey(Owner, on_delete=models.CASCADE, related_name="benefits", null=True)
-
-    # Distribution type.
-    # "DIRECT_ENTITLEMENT"
-    distribution_type = models.TextField(blank=True, help_text="The distribution type of the benefit.")
-
-    class Meta(auto_prefetch.Model.Meta):
-        ordering: ClassVar[list[str]] = ["-twitch_created_at"]
-        indexes: ClassVar[list[Index]] = [
-            models.Index(fields=["name"], name="benefit_name_idx"),
-            models.Index(fields=["twitch_created_at"], name="benefit_twitch_created_at_idx"),
-            models.Index(fields=["created_at"], name="benefit_created_at_idx"),
-            models.Index(fields=["is_ios_available"], name="benefit_is_ios_available_idx"),
-        ]
-
-    def __str__(self) -> str:
-        """Return the name of the benefit and when it was created."""
-        return f"{self.name or self.twitch_id} - {self.twitch_created_at}"
-
-    def import_json(self, data: dict, time_based_drop: TimeBasedDrop | None) -> Self:
-        """Import the data from the Twitch API."""
-        if wrong_typename(data, "DropBenefit"):
-            return self
-
-        field_mapping: dict[str, str] = {
-            "name": "name",
-            "imageAssetURL": "image_asset_url",
-            "entitlementLimit": "entitlement_limit",
-            "isIosAvailable": "is_ios_available",
-            "createdAt": "twitch_created_at",
-            "distributionType": "distribution_type",
-        }
-        updated: int = update_fields(instance=self, data=data, field_mapping=field_mapping)
-        if updated > 0:
-            logger.info("Updated %s fields for %s", updated, self)
-
-        if not time_based_drop:
-            logger.error("TimeBasedDrop is required for %s", self)
-            return self
-
-        if time_based_drop != self.time_based_drop:
-            self.time_based_drop = time_based_drop
-            logger.info("Updated time based drop %s for %s", time_based_drop, self)
-            self.save()
-
-        if data.get("game") and data["game"].get("id"):
-            game_instance, created = Game.objects.update_or_create(twitch_id=data["game"]["id"])
-            game_instance.import_json(data["game"], None)
-            if created:
-                logger.info("Added game %s to %s", game_instance, self)
-
-        if data.get("ownerOrganization") and data["ownerOrganization"].get("id"):
-            owner_instance, created = Owner.objects.update_or_create(twitch_id=data["ownerOrganization"]["id"])
-            owner_instance.import_json(data["ownerOrganization"])
-            if created:
-                logger.info("Added owner %s to %s", owner_instance, self)
-
-        return self
+        return f"{self.name or self.drop_id} - {self.created_at}"

core/models_utils.py

@@ -85,7 +85,7 @@ def get_value(data: dict, key: str) -> datetime | str | None:
     dates: list[str] = ["endAt", "endsAt,", "startAt", "startsAt", "createdAt", "earnableUntil"]
     if key in dates:
         logger.debug("Converting %s to datetime", data_key)
-        return datetime.fromisoformat(data_key.replace("Z", "+00:00"))
+        return datetime.fromisoformat(data_key)
 
     return data_key
 

core/templates/index.html

@@ -1,106 +1,111 @@
 {% extends "base.html" %}
 {% load custom_filters static time_filters %}
 {% block content %}
-    <div class="container mt-4">
+<div class="container mt-4">
     {% include "partials/info_box.html" %}
     <!-- Drop Campaigns Section -->
     <section class="drop-campaigns">
         <h2>
             Drop Campaigns -
-                <span class="d-inline text-muted">{{ games.count }} game{{ games.count|pluralize }}</span>
+            <span class="d-inline text-muted">{{ grouped_drops|length }} game{{ grouped_drops|length|pluralize }}</span>
         </h2>
-            <!-- Loop through games -->
-            {% for game in games %}
+        {% if grouped_drops %}
+        {% for game, drops_list in grouped_drops.items %}
+        {# Retain card structure for layout, replace table with list #}
         <div class="card mb-4 shadow-sm">
             <div class="row g-0">
                 <!-- Game Box Art -->
                 <div class="col-md-2">
-                            <img src="{{ game.box_art_url|default:'https://static-cdn.jtvnw.net/ttv-static/404_boxart.jpg' }}"
-                                 alt="{{ game.name|default:'Game name unknown' }} box art"
-                                 class="img-fluid rounded-start"
-                                 height="283"
-                                 width="212"
-                                 loading="lazy" />
+                    {% if game and game.box_art_url %}
+                    <img src="{{ game.box_art_url }}" alt="{{ game.display_name|default:'Game name unknown' }} box art"
+                        class="img-fluid rounded-start" height="283" width="212" loading="lazy" />
+                    {% else %}
+                    <img src="{% static 'images/404_boxart.jpg' %}"
+                        alt="{% if game %}{{ game.display_name }}{% else %}Unknown Game{% endif %} box art"
+                        class="img-fluid rounded-start" height="283" width="212" loading="lazy" />
+                    {% endif %}
                 </div>
                 <!-- Game Details -->
                 <div class="col-md-10">
                     <div class="card-body">
                         <h2 class="card-title h5">
-                                    <a href="{% url 'game' game.twitch_id %}" class="text-decoration-none">{{ game.name|default:'Unknown' }}</a>
-                                    -
-                                    <a href="https://www.twitch.tv/directory/category/{{ game.slug|default:'game-name-unknown' }}"
+                            {% if game %}
+                            <a href="{% if game.game_id %}{% url 'game_detail' game.game_id %}{% else %}#{% endif %}"
+                                class="text-decoration-none">{{ game.display_name|default:'Unknown Game' }}</a>
+                            {% if game.slug %}
+                            - <a href="https://www.twitch.tv/directory/category/{{ game.slug }}"
                                 class="text-decoration-none text-muted">Twitch</a>
-                                </h2>
-                                <!-- Loop through campaigns for each game -->
-                                {% for campaign in game.drop_campaigns.all %}
-                                    <div class="mt-4">
-                                        <h4 class="h6">{{ campaign.name }}</h4>
-                                        <a href="{{ campaign.details_url }}" class="text-decoration-none">Details</a>
-                                        {% if campaign.details_url != campaign.account_link_url %}
-                                            | <a href="{{ campaign.account_link_url }}" class="text-decoration-none">Link Account</a>
                             {% endif %}
-                                        <p class="mb-2 text-muted">
-                                            Ends in:
-                                            <abbr title="{{ campaign.starts_at|date:'l d F H:i' }} - {{ campaign.ends_at|date:'l d F H:i' }}">
-                                                {{ campaign.ends_at|timeuntil }}
+                            {% else %}
+                            Drops without an associated Game
+                            {% endif %}
+                        </h2>
+
+                        {% if drops_list %}
+                        <!-- NEW Drop List Structure -->
+                        <ul class="list-unstyled mt-3">
+                            {% for drop in drops_list %}
+                            <li class="mb-3"> {# Add margin between list items #}
+                                <strong>{{ drop.name|default:'Unknown Drop' }}</strong>
+                                (Requires {{ drop.required_minutes_watched|minutes_to_hours }})
+                                <br>
+                                <em>Campaign:
+                                    <a href="{{ drop.campaign.details_url|default:'#' }}" class="text-decoration-none"
+                                        title="{{ drop.campaign.name|default:'Unknown Campaign' }}">
+                                        {{ drop.campaign.name|truncatechars:40|default:'N/A' }} {# Adjusted truncate #}
+                                    </a>
+                                    {% if drop.campaign.details_url != drop.campaign.account_link_url and drop.campaign.account_link_url %}
+                                    | <a href="{{ drop.campaign.account_link_url }}" class="text-decoration-none"
+                                        title="Link Account for {{ drop.campaign.name }}">Link</a>
+                                    {% endif %}
+                                </em>
+                                <br>
+                                Ends in: <abbr
+                                    title="{{ drop.campaign.start_at|date:'l d F H:i' }} - {{ drop.campaign.end_at|date:'l d F H:i' }}">
+                                    {{ drop.campaign.end_at|timeuntil }}
                                 </abbr>
-                                        </p>
-                                        <!-- Drop Benefits Table -->
-                                        <div class="table-responsive">
-                                            <table class="table table-striped table-hover align-middle">
-                                                <thead>
-                                                    <tr>
-                                                        <th>Benefit Image</th>
-                                                        <th>Benefit Name</th>
-                                                        <th>Required Minutes Watched</th>
-                                                    </tr>
-                                                </thead>
-                                                <tbody>
-                                                    {% for drop in campaign.drops.all %}
+
                                 {% if drop.benefits.exists %}
+                                <br>
+                                Benefits:
+                                <ul class="list-inline">
                                     {% for benefit in drop.benefits.all %}
-                                                                <tr>
-                                                                    <td>
-                                                                        <img src="{{ benefit.image_url|default:'https://static-cdn.jtvnw.net/ttv-static/404_boxart.jpg' }}"
-                                                                             alt="{{ benefit.name|default:'Unknown' }}"
-                                                                             class="img-fluid rounded"
-                                                                             height="50"
-                                                                             width="50"
-                                                                             loading="lazy" />
-                                                                    </td>
-                                                                    <td>
-                                                                        <abbr title="{{ drop.name|default:'Unknown' }}">
-                                                                            {{ benefit.name|default:'Unknown' }}
+                                    <li class="list-inline-item">
+                                        <abbr title="{{ benefit.name|default:'Unknown Benefit' }}">
+                                            {% if benefit.image_asset_url %}
+                                            <img src="{{ benefit.image_asset_url|default:'https://static-cdn.jtvnw.net/ttv-static/404_boxart.jpg' }}"
+                                                alt="{{ benefit.name|default:'Unknown Benefit' }}"
+                                                class="img-fluid rounded me-1 align-middle" {# Added align-middle #}
+                                                height="20" width="20" loading="lazy" />
+                                            {% endif %}
+                                            <small>{{ benefit.name|truncatechars:25|default:'Unknown Benefit' }}</small>
+                                            {# Wrap text in small #}
                                         </abbr>
-                                                                    </td>
-                                                                    <td>{{ drop.required_minutes_watched|minutes_to_hours }}</td>
-                                                                </tr>
+                                    </li>
+                                    {% endfor %}
+                                </ul>
+                                {% endif %}
+                                {# Removed hr, using li margin instead #}
+                            </li>
+                            {% endfor %}
+                        </ul>
+                        {% else %}
+                        <p class="mt-3 text-muted">No active drops found for this game currently.</p>
+                        {% endif %}
+                    </div>
+                </div>
+            </div>
+        </div>
+        {% empty %}
+        <div class="alert alert-info" role="alert">
+            There are currently no active Twitch drops found matching the criteria.
+        </div>
         {% endfor %}
         {% else %}
-                                                            <tr>
-                                                                <td>
-                                                                    <img src="https://static-cdn.jtvnw.net/ttv-static/404_boxart.jpg"
-                                                                         alt="{{ drop.name|default:'Unknown' }}"
-                                                                         class="img-fluid rounded"
-                                                                         height="50"
-                                                                         width="50"
-                                                                         loading="lazy" />
-                                                                </td>
-                                                                <td>{{ drop.name|default:'Unknown' }}</td>
-                                                                <td>N/A</td>
-                                                            </tr>
+        <div class="alert alert-info" role="alert">
+            There are currently no active Twitch drops found.
+        </div>
         {% endif %}
-                                                    {% endfor %}
-                                                </tbody>
-                                            </table>
-                                        </div>
-                                    </div>
-                                {% endfor %}
-                            </div>
-                        </div>
-                    </div>
-                </div>
-            {% endfor %}
     </section>
-    </div>
+</div>
 {% endblock content %}

core/urls.py

@@ -4,7 +4,7 @@ from debug_toolbar.toolbar import debug_toolbar_urls  # type: ignore[import-unty
 from django.contrib import admin
 from django.urls import URLPattern, URLResolver, path
 
-from core.views import get_game, get_games, get_home, get_import
+from core.views import get_game, get_games, get_home
 
 app_name: str = "core"
 
@@ -33,8 +33,7 @@ app_name: str = "core"
 urlpatterns: list[URLPattern | URLResolver] = [
     path(route="admin/", view=admin.site.urls),
     path(route="", view=get_home, name="index"),
-    path(route="game/<int:twitch_id>/", view=get_game, name="game"),
+    path(route="game/<int:twitch_id>/", view=get_game, name="game_detail"),
     path(route="games/", view=get_games, name="games"),
-    path(route="import/", view=get_import, name="import"),
     *debug_toolbar_urls(),
 ]

core/views.py

@@ -1,17 +1,17 @@
 from __future__ import annotations
 
-import json
 import logging
+from collections import defaultdict
 from typing import TYPE_CHECKING, Any
 
-from django.db.models import F, Prefetch
-from django.http import HttpRequest, HttpResponse, JsonResponse
+from django.db.models import Prefetch
+from django.http import HttpRequest, HttpResponse
+from django.shortcuts import render
 from django.template.response import TemplateResponse
 from django.utils import timezone
 from django.views.decorators.http import require_http_methods
 
-from core.import_json import import_data
-from core.models import Benefit, DropCampaign, Game, TimeBasedDrop
+from core.models import DropCampaign, Game, TimeBasedDrop
 
 if TYPE_CHECKING:
     from django.db.models.query import QuerySet
@@ -20,36 +20,6 @@ if TYPE_CHECKING:
 logger: logging.Logger = logging.getLogger(__name__)
 
 
-def get_games_with_drops() -> QuerySet[Game]:
-    """Get the games with drops, sorted by when the drop campaigns end.
-
-    Returns:
-        QuerySet[Game]: The games with drops.
-    """
-    # Prefetch the benefits for the time-based drops.
-    benefits_prefetch = Prefetch(lookup="benefits", queryset=Benefit.objects.all())
-    active_time_based_drops: QuerySet[TimeBasedDrop] = TimeBasedDrop.objects.filter(
-        ends_at__gte=timezone.now(),
-        starts_at__lte=timezone.now(),
-    ).prefetch_related(benefits_prefetch)
-
-    # Prefetch the active time-based drops for the drop campaigns.
-    drops_prefetch = Prefetch(lookup="drops", queryset=active_time_based_drops)
-    active_campaigns: QuerySet[DropCampaign] = DropCampaign.objects.filter(
-        ends_at__gte=timezone.now(),
-        starts_at__lte=timezone.now(),
-    ).prefetch_related(drops_prefetch)
-
-    return (
-        Game.objects.filter(drop_campaigns__in=active_campaigns)
-        .annotate(drop_campaign_end=F("drop_campaigns__ends_at"))
-        .distinct()
-        .prefetch_related(Prefetch("drop_campaigns", queryset=active_campaigns))
-        .select_related("org")
-        .order_by("drop_campaign_end")
-    )
-
-
 @require_http_methods(request_method_list=["GET", "HEAD"])
 def get_home(request: HttpRequest) -> HttpResponse:
     """Render the index page.
@@ -60,14 +30,25 @@ def get_home(request: HttpRequest) -> HttpResponse:
     Returns:
         HttpResponse: The response object
     """
-    try:
-        games: QuerySet[Game] = get_games_with_drops()
-    except Exception:
-        logger.exception("Error fetching reward campaigns or games.")
-        return HttpResponse(status=500)
+    now: timezone.datetime = timezone.now()
+    grouped_drops = defaultdict(list)
 
-    context: dict[str, Any] = {"games": games}
-    return TemplateResponse(request, "index.html", context)
+    current_drops_qs = (
+        TimeBasedDrop.objects.filter(start_at__lte=now, end_at__gte=now)
+        .select_related("campaign__game")
+        .prefetch_related("benefits")
+        .order_by("campaign__game__display_name", "name")
+    )
+
+    for drop in current_drops_qs:
+        if drop.campaign and drop.campaign.game:
+            game: Game = drop.campaign.game
+            grouped_drops[game].append(drop)
+        else:
+            logger.warning("Drop %s does not have an associated game or campaign.", drop.name)
+
+    context = {"grouped_drops": dict(grouped_drops)}
+    return render(request, "index.html", context)
 
 
 @require_http_methods(request_method_list=["GET", "HEAD"])
@@ -117,25 +98,3 @@ def get_games(request: HttpRequest) -> HttpResponse:
 
     context: dict[str, QuerySet[Game] | str] = {"games": games}
     return TemplateResponse(request=request, template="games.html", context=context)
-
-
-@require_http_methods(request_method_list=["POST"])
-def get_import(request: HttpRequest) -> HttpResponse:
-    """Import data that are sent from Twitch Drop Miner.
-
-    Args:
-        request (HttpRequest): The request object.
-
-    Returns:
-        HttpResponse: The response object.
-    """
-    try:
-        data = json.loads(request.body)
-        logger.info(data)
-
-        # Import the data.
-        import_data(data)
-
-        return JsonResponse({"status": "success"}, status=200)
-    except json.JSONDecodeError as e:
-        return JsonResponse({"status": "error", "message": str(e)}, status=400)

pyproject.toml

@@ -10,27 +10,20 @@ dependencies = [
     "django",
     "platformdirs",
     "python-dotenv",
-    "django-auto-prefetch",
 ]
 
-# You can install development dependencies with `uv install --dev`.
-# Or you can install them with `uv install --dev -r requirements-dev.txt`.
-# uv can be replaced with `pip`if you don't have uv installed.
 [dependency-groups]
-dev = ["pre-commit", "pytest", "pytest-django", "ruff"]
+dev = ["pytest", "pytest-django"]
 
-# https://docs.astral.sh/ruff/settings/
 [tool.ruff]
-# Enable all rules
 lint.select = ["ALL"]
-
-# https://sphinxcontrib-napoleon.readthedocs.io/en/latest/example_google.html
+preview = true
+unsafe-fixes = true
+fix = true
 lint.pydocstyle.convention = "google"
-
-# Add "from __future__ import annotations" to all files
 lint.isort.required-imports = ["from __future__ import annotations"]
+line-length = 120
 
-# Ignore some rules
 lint.ignore = [
     "CPY001", # Checks for the absence of copyright notices within Python files.
     "D100",   # Checks for undocumented public module definitions.
@@ -57,9 +50,6 @@ lint.ignore = [
     "W191",   # Checks for indentation that uses tabs.
 ]
 
-# Default is 88 characters
-line-length = 120
-
 [tool.ruff.lint.per-file-ignores]
 "**/tests/**" = [
     "ARG",     # Unused function args -> fixtures nevertheless are functionally relevant...
@@ -72,51 +62,22 @@ line-length = 120
     "RUF012", # Checks for mutable default values in class attributes.
 ]
 
-# https://www.djlint.com/
+
 [tool.djlint]
-# Set a profile for the template language. The profile will enable linter rules that apply to your template language, and may also change reformatting.
 profile = "django"
-
-# Formatter will attempt to format template syntax inside of tag attributes.
 format_attribute_template_tags = true
-
-# Format contents of style tags using css-beautify
 format_css = true
-
-# Format contents of script tags using js-beautify.
 format_js = true
+ignore = "H006"
 
-# Ignore some rules
-ignore = "H006" # Img tag should have height and width attributes.
-
-# https://pytest-django.readthedocs.io/en/latest/
 [tool.pytest.ini_options]
-# Where our Django settings are located.
 DJANGO_SETTINGS_MODULE = "core.settings"
-
-# Only run tests in files that match this pattern.
 python_files = ["*_test.py"]
-
-# Enable logging in the console.
 log_cli = true
 log_cli_level = "INFO"
 log_cli_format = "%(asctime)s [%(levelname)8s] %(message)s (%(filename)s:%(lineno)s)"
 log_cli_date_format = "%Y-%m-%d %H:%M:%S"
-
-# Only check /tests/ directory for tests.
-# This will speed up the test run significantly. (5.16s -> 0.25s)
 testpaths = ["tests"]
 
-[tool.mypy]
-plugins = ["mypy_django_plugin.main"]
-
 [tool.django-stubs]
 django_settings_module = "core.settings"
-
-[tool.black]
-line-length = 120
-preview = true
-unstable = true
-
-[tool.isort]
-profile = "black"

tests/models_test.py

@@ -1,274 +0,0 @@
-from __future__ import annotations
-
-import json
-from pathlib import Path
-from typing import Any
-from unittest.mock import MagicMock, patch
-
-import pytest
-
-from core.import_json import (
-    find_typename_in_json,
-    import_data,
-    import_drop_benefits,
-    import_drop_campaigns,
-    import_game_data,
-    import_owner_data,
-    import_time_based_drops,
-    type_names,
-)
-from core.models import Benefit, DropCampaign, Game, Owner, TimeBasedDrop
-
-
-def _validate_extraction(json: dict, typename: type_names, no_result_err_msg: str, id_err_msg: str) -> dict[str, Any]:
-    result: dict[str, Any] = find_typename_in_json(json, typename)[0]
-    assert result, no_result_err_msg
-    assert result.get("id"), id_err_msg
-
-    return result
-
-
-def test_find_typename_in_json() -> None:
-    """Test the find_typename_in_json function."""
-    json_file_raw: str = Path("tests/response.json").read_text(encoding="utf-8")
-    json_file: dict = json.loads(json_file_raw)
-
-    result: list[dict[str, Any]] = find_typename_in_json(json_file, typename_to_find="DropCampaign")
-    assert len(result) == 20
-    assert result[0]["__typename"] == "DropCampaign"
-    assert result[0]["id"] == "5b5816c8-a533-11ef-9266-0a58a9feac02"
-
-
-@pytest.mark.django_db
-def test_import_game_data() -> None:
-    """Test the import_game_data function."""
-    json_file_raw: str = Path("tests/response.json").read_text(encoding="utf-8")
-    json_file: dict = json.loads(json_file_raw)
-
-    game_json: dict[str, Any] = _validate_extraction(
-        json=json_file,
-        typename="Game",
-        no_result_err_msg="Game JSON not found",
-        id_err_msg="Game ID not found",
-    )
-    assert game_json.get("id") == "155409827", f"Game ID does not match expected value: {game_json.get('id')}"
-    assert game_json.get("slug") == "pokemon-trading-card-game-live", "Game slug does not match expected value"
-
-    assert_msg: str = f"Game display name does not match expected value: {game_json.get('displayName')}"
-    assert game_json.get("displayName") == "Pokémon Trading Card Game Live", assert_msg
-
-    assert_msg: str = f"Game URL does not match expected value: {game_json.get('gameUrl')}"
-    assert game_json.get("__typename") == "Game", assert_msg
-
-    owner_json: dict[str, Any] = _validate_extraction(
-        json=json_file,
-        typename="Organization",
-        no_result_err_msg="Owner JSON not found",
-        id_err_msg="Owner ID not found",
-    )
-
-    owner, created = Owner.objects.get_or_create(twitch_id=owner_json.get("id"))
-    assert_msg: str = f"Owner was not created: {owner=} != {owner_json.get('id')}. This means the old database was used instead of a new one."  # noqa: E501
-    assert created, assert_msg
-    assert owner
-
-    game: Game = import_game_data(drop_campaign=game_json, owner=owner)
-    assert game, f"Failed to import JSON data into Game model: {game_json=}"
-    assert game.org == owner, f"Owner was not set on the Game model: {game.org=} != {owner=}"
-    assert game.display_name == game_json.get("displayName"), "Game display name was not set on the Game model"
-
-    assert_msg: str = f"Game slug was not set on the Game model: {game.slug=} != {game_json.get('slug')}"
-    assert game.slug == game_json.get("slug"), assert_msg
-
-    assert_msg: str = f"Game ID was not set on the Game model: {game.twitch_id=} != {game_json.get('id')}"
-    assert game.twitch_id == game_json.get("id"), assert_msg
-
-    assert_msg: str = f"Game URL was not set on the Game model: {game.game_url} != https://www.twitch.tv/directory/category/{game.slug}"
-    assert game.game_url == f"https://www.twitch.tv/directory/category/{game.slug}", assert_msg
-
-    assert game.created_at, "Game created_at was not set on the Game model"
-    assert game.modified_at, "Game modified_at was not set on the Game model"
-
-
-@pytest.mark.django_db
-def test_import_owner_data() -> None:
-    """Test the import_owner_data function."""
-    json_file_raw: str = Path("tests/response.json").read_text(encoding="utf-8")
-    json_file: dict = json.loads(json_file_raw)
-
-    owner_json: dict[str, Any] = _validate_extraction(
-        json=json_file,
-        typename="Organization",
-        no_result_err_msg="Owner JSON not found",
-        id_err_msg="Owner ID not found",
-    )
-
-    owner: Owner = import_owner_data(drop_campaign=owner_json)
-    assert owner, f"Failed to import JSON data into Owner model: {owner_json=}"
-
-    assert_msg: str = f"Owner ID was not set on the Owner model: {owner.twitch_id=} != {owner_json.get('id')}"
-    assert owner.twitch_id == owner_json.get("id"), assert_msg
-
-    assert_msg: str = f"Owner name was not set on the Owner model: {owner.name=} != {owner_json.get('name')}"
-    assert owner.name == owner_json.get("name"), assert_msg
-
-    assert owner.created_at, "Owner created_at was not set on the Owner model"
-    assert owner.modified_at, "Owner modified_at was not set on the Owner model"
-
-
-@pytest.mark.django_db
-def test_import_drop_benefits() -> None:
-    """Test the import_drop_benefits function."""
-    json_file_raw: str = Path("tests/response.json").read_text(encoding="utf-8")
-    json_file: dict = json.loads(json_file_raw)
-
-    drop_campaign_json: dict[str, Any] = _validate_extraction(
-        json=json_file,
-        typename="DropCampaign",
-        no_result_err_msg="DropCampaign JSON not found",
-        id_err_msg="DropCampaign ID not found",
-    )
-    assert drop_campaign_json
-
-    drop_campaign: DropCampaign | None = import_drop_campaigns(drop_campaigns=drop_campaign_json)
-    assert drop_campaign, f"Failed to import JSON data into DropCampaign model: {drop_campaign_json=}"
-
-    assert find_typename_in_json(drop_campaign_json, "TimeBasedDrop"), "TimeBasedDrop JSON not found"
-    time_based_drop_json: dict[str, Any] = _validate_extraction(
-        json=drop_campaign_json,
-        typename="TimeBasedDrop",
-        no_result_err_msg="TimeBasedDrop JSON not found",
-        id_err_msg="TimeBasedDrop ID not found",
-    )
-    assert time_based_drop_json
-
-    time_based_drop: list[TimeBasedDrop] = import_time_based_drops(
-        drop_campaign_json=time_based_drop_json,
-        drop_campaign=drop_campaign,
-    )
-    assert time_based_drop, f"Failed to import JSON data into TimeBasedDrop model: {time_based_drop_json=}"
-
-    drop_benefit_json: dict[str, Any] = _validate_extraction(
-        json=drop_campaign_json,
-        typename="DropBenefit",
-        no_result_err_msg="DropBenefit JSON not found",
-        id_err_msg="DropBenefit ID not found",
-    )
-
-    drop_benefit: list[Benefit] = import_drop_benefits(drop_benefit_json, time_based_drop[0])
-    assert drop_benefit, f"Failed to import JSON data into DropBenefit model: {drop_benefit_json=}"
-    assert drop_benefit[0].twitch_id == drop_benefit_json.get("id"), "Benefit ID was not set on the Benefit model"
-
-    assert_msg: str = f"DropBenefit created_at was not set on the Benefit model: {drop_benefit[0].created_at=}"
-    assert drop_benefit[0].created_at, assert_msg
-
-    assert_msg = f"DropBenefit modified_at was not set on the Benefit model: {drop_benefit[0].modified_at=}"
-    assert drop_benefit[0].modified_at, assert_msg
-
-    assert drop_benefit[0].time_based_drop == time_based_drop[0], "TimeBasedDrop was not set on the Benefit model"
-
-
-@pytest.mark.django_db
-def test_import_time_based_drops() -> None:
-    """Test the import_time_based_drops function."""
-    json_file_raw: str = Path("tests/response.json").read_text(encoding="utf-8")
-    json_file: dict = json.loads(json_file_raw)
-
-    drop_campaign_json: dict[str, Any] = _validate_extraction(
-        json=json_file,
-        typename="DropCampaign",
-        no_result_err_msg="DropCampaign JSON not found",
-        id_err_msg="DropCampaign ID not found",
-    )
-    assert drop_campaign_json
-
-    drop_campaign: DropCampaign | None = import_drop_campaigns(drop_campaigns=drop_campaign_json)
-    assert drop_campaign, f"Failed to import JSON data into DropCampaign model: {drop_campaign_json=}"
-
-    time_based_drop_json: dict[str, Any] = _validate_extraction(
-        json=drop_campaign_json,
-        typename="TimeBasedDrop",
-        no_result_err_msg="TimeBasedDrop JSON not found",
-        id_err_msg="TimeBasedDrop ID not found",
-    )
-    assert time_based_drop_json
-
-    time_based_drop: list[TimeBasedDrop] = import_time_based_drops(
-        drop_campaign_json=time_based_drop_json,
-        drop_campaign=drop_campaign,
-    )
-    assert time_based_drop, f"Failed to import JSON data into TimeBasedDrop model: {time_based_drop_json=}"
-
-    assert time_based_drop[0].twitch_id == time_based_drop_json.get(
-        "id",
-    ), "TimeBasedDrop ID was not set on the TimeBasedDrop model"
-
-    assert_msg: str = (
-        f"TimeBasedDrop created_at was not set on the TimeBasedDrop model: {time_based_drop[0].created_at=}"
-    )
-    assert time_based_drop[0].created_at, assert_msg
-
-    assert_msg = f"TimeBasedDrop modified_at was not set on the TimeBasedDrop model: {time_based_drop[0].modified_at=}"
-    assert time_based_drop[0].modified_at, assert_msg
-
-    assert time_based_drop[0].drop_campaign == drop_campaign, "DropCampaign was not set on the TimeBasedDrop model"
-
-
-@pytest.mark.django_db
-def test_import_drop_campaigns() -> None:
-    """Test the import_drop_campaigns function."""
-    json_file_raw: str = Path("tests/response.json").read_text(encoding="utf-8")
-    json_file: dict = json.loads(json_file_raw)
-
-    drop_campaign_json: dict[str, Any] = _validate_extraction(
-        json=json_file,
-        typename="DropCampaign",
-        no_result_err_msg="DropCampaign JSON not found",
-        id_err_msg="DropCampaign ID not found",
-    )
-    assert drop_campaign_json
-
-    drop_campaign: DropCampaign | None = import_drop_campaigns(drop_campaigns=drop_campaign_json)
-    assert drop_campaign, f"Failed to import JSON data into DropCampaign model: {drop_campaign_json=}"
-
-    assert drop_campaign.twitch_id == drop_campaign_json.get(
-        "id",
-    ), "DropCampaign ID was not set on the DropCampaign model"
-
-    assert_msg: str = f"DropCampaign created_at was not set on the DropCampaign model: {drop_campaign.created_at=}"
-    assert drop_campaign.created_at, assert_msg
-
-    assert_msg = f"DropCampaign modified_at was not set on the DropCampaign model: {drop_campaign.modified_at=}"
-    assert drop_campaign.modified_at, assert_msg
-
-
-@pytest.fixture
-def sample_data() -> dict[str, Any]:
-    """Sample data for testing import_data."""
-    return {
-        "__typename": "Root",
-        "data": [
-            {"__typename": "DropCampaign", "id": "campaign1", "name": "Campaign 1"},
-            {"__typename": "DropCampaign", "id": "campaign2", "name": "Campaign 2"},
-        ],
-    }
-
-
-@pytest.fixture
-def empty_data() -> dict[str, Any]:
-    """Empty data for testing import_data."""
-    return {"__typename": "Root", "data": []}
-
-
-@patch("core.import_json.import_drop_campaigns")
-def test_import_data(mock_import_drop_campaigns: MagicMock, sample_data: dict[str, Any]) -> None:
-    """Test the import_data function with valid data."""
-    import_data(sample_data)
-    assert mock_import_drop_campaigns.call_count == 2
-
-
-def test_import_data_no_campaigns(empty_data: dict[str, Any]) -> None:
-    """Test the import_data function with no drop campaigns."""
-    with patch("core.import_json.import_drop_campaigns") as mock_import_drop_campaigns:
-        import_data(empty_data)
-        mock_import_drop_campaigns.assert_not_called()