From d7b31e1d421f4820f91f46c8e2ce2ea1a75878e9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Joakim=20Hells=C3=A9n?= Date: Thu, 1 May 2025 02:41:25 +0200 Subject: [PATCH] Refactor models --- .pre-commit-config.yaml | 12 +- .vscode/launch.json | 12 + core/admin.py | 4 +- core/import_json.py | 187 ------- core/management/__init__.py | 0 core/management/commands/__init__.py | 0 .../commands/import_twitch_drops.py | 293 ++++++++++ core/migrations/0001_initial.py | 400 ++++++-------- core/migrations/0002_alter_user_options.py | 24 - .../0003_alter_benefit_created_at_and_more.py | 48 -- core/migrations/0004_alter_game_created_at.py | 25 - core/models.py | 502 +++--------------- core/models_utils.py | 2 +- core/templates/index.html | 205 +++---- core/urls.py | 5 +- core/views.py | 103 ++-- tests/models_test.py | 274 ---------- 17 files changed, 704 insertions(+), 1392 deletions(-) delete mode 100644 core/import_json.py create mode 100644 core/management/__init__.py create mode 100644 core/management/commands/__init__.py create mode 100644 core/management/commands/import_twitch_drops.py delete mode 100644 core/migrations/0002_alter_user_options.py delete mode 100644 core/migrations/0003_alter_benefit_created_at_and_more.py delete mode 100644 core/migrations/0004_alter_game_created_at.py delete mode 100644 tests/models_test.py diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index deacf58..9956900 100644 --- a/.pre-commit-config.yaml +++ b/.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 diff --git a/.vscode/launch.json b/.vscode/launch.json index cbe6967..c72b4b0 100644 --- a/.vscode/launch.json +++ b/.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 } ] } diff --git a/core/admin.py b/core/admin.py index 10b9200..a1f9cb1 100644 --- a/core/admin.py +++ b/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) diff --git a/core/import_json.py b/core/import_json.py deleted file mode 100644 index c22e3ff..0000000 --- a/core/import_json.py +++ /dev/null @@ -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 diff --git a/core/management/__init__.py b/core/management/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/core/management/commands/__init__.py b/core/management/commands/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/core/management/commands/import_twitch_drops.py b/core/management/commands/import_twitch_drops.py new file mode 100644 index 0000000..96aac08 --- /dev/null +++ b/core/management/commands/import_twitch_drops.py @@ -0,0 +1,293 @@ +from __future__ import annotations + +import json +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 + + +class Command(BaseCommand): + """Imports Twitch Drop campaign data from a specified JSON file into the database. + + This command reads a JSON file containing Twitch Drop campaign information, + typically obtained from Twitch's GQL endpoint or similar sources. It parses + the JSON, expecting a structure containing details about the campaign, + associated game, owner organization, time-based drops, and their benefits. + + The command uses a database transaction to ensure atomicity, meaning either + all data is imported successfully, or no changes are made to the database + if an error occurs during the import process. + + It performs the following steps: + 1. Parses the command-line argument for the path to the JSON file. + 2. Opens and reads the JSON file, handling potential file-related errors. + 3. Navigates the JSON structure to find the core 'dropCampaign' data and + the associated Twitch User ID. + 4. Within a transaction: + a. Processes Game data: Creates or updates a `Game` record. + b. Processes Owner Organization data: Creates or updates an `Organization` record. + c. Processes Drop Campaign data: Creates or updates a `DropCampaign` record, + linking it to the Game and Organization. Parses and stores start/end times. + d. Processes Time Based Drops: Iterates through `timeBasedDrops` within the campaign. + i. For each drop: Creates or updates a `TimeBasedDrop` record, linking it + to the `DropCampaign`. Parses and stores start/end times and required watch time. + ii. Processes Benefits within each drop: Iterates through `benefitEdges`. + - Creates or updates `Benefit` records associated with the drop. + - Handles potential missing related Game or Organization for the benefit + by creating minimal placeholder records if necessary. + - Associates the created/updated `Benefit` objects with the current + `TimeBasedDrop` via a ManyToMany relationship. + 5. Provides feedback to the console about created or updated records and any warnings + or errors encountered. + 6. Raises `CommandError` for critical issues like file not found, JSON errors, + missing essential data (e.g., campaign ID), or unexpected exceptions during processing. + + Args: + json_file (str): The filesystem path to the JSON file containing the + Twitch Drop campaign data. + Example Usage: + python manage.py import_twitch_drops /path/to/your/twitch_drops.json + """ + + help = "Imports Twitch Drop campaign data from a specific JSON file structure." + + def add_arguments(self, parser: CommandParser) -> None: # noqa: D102 + parser.add_argument("json_file", type=str, help="Path to the Twitch Drop JSON file") + + def handle(self, *args: Any, **options) -> None: # noqa: ANN003, ANN401, ARG002, C901, PLR0912, PLR0915 + """Import data to DB. + + Args: + *args: Positional arguments. + **options: Keyword arguments containing the command-line options. + + Raises: + CommandError: If the JSON file is not found, cannot be decoded, + or if there are issues with the JSON structure. + """ + json_file_path = options["json_file"] + self.stdout.write(f"Starting import from {json_file_path}...") + + try: + with Path(json_file_path).open(encoding="utf-8") as f: + raw_data = json.load(f) + except FileNotFoundError as e: + msg = f"Error: File not found at {json_file_path}" + raise CommandError(msg) from e + except json.JSONDecodeError as e: + msg = f"Error: Could not decode JSON from {json_file_path}" + raise CommandError(msg) from e + except Exception as e: + msg = f"Error reading file: {e}" + 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 = "Error: 'dropCampaign' key not found or is null in the JSON data." + raise CommandError(msg) + + except AttributeError as e: + msg = "Error: Unexpected JSON structure. Could not find 'data' or 'user'." + raise CommandError(msg) from e + + try: + # 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}")) + else: + self.stdout.write(f"Updated/Found Game: {game_obj.display_name}") + else: + self.stdout.write(self.style.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}")) + else: + self.stdout.write(f"Updated/Found Organization: {owner_obj.name}") + else: + self.stdout.write(self.style.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." + 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.")) + # 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}")) + else: + self.stdout.write(f"Updated/Found Campaign: {campaign_obj.name}") + + # --- 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.")) + + 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.")) + 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.")) + 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}")) + else: + self.stdout.write(f" Updated/Found Time Drop: {drop_obj.name}") + + # --- 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.")) + 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: + self.stdout.write( + self.style.WARNING( + f" Game {benefit_game_data.get('name')} for benefit {benefit_id} not found. Creating minimally.", # noqa: E501 + ), + ) + # Optionally create a minimal game entry here if desired + benefit_game_obj, _ = Game.objects.update_or_create( + game_id=benefit_game_data["id"], + defaults={"display_name": benefit_game_data.get("name", "Unknown Game")}, + ) + + 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: + self.stdout.write( + self.style.WARNING( + f" Organization {benefit_owner_data.get('name')} for benefit {benefit_id} not found. Creating minimally.", # noqa: E501 + ), + ) + benefit_owner_obj, _ = Organization.objects.update_or_create( + org_id=benefit_owner_data["id"], + defaults={"name": benefit_owner_data.get("name", "Unknown Org")}, + ) + + 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"), + "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}")) + else: + self.stdout.write(f" Updated/Found Benefit: {benefit_obj.name}") + benefit_objs.append(benefit_obj) + + # Set the ManyToMany relationship for the drop + if benefit_objs: + drop_obj.benefits.set(benefit_objs) + self.stdout.write(f" Associated {len(benefit_objs)} benefits with drop {drop_obj.name}.") + + except KeyError as e: + msg = f"Error: Missing expected key in JSON data - {e}" + 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}" + raise CommandError(msg) from e + + self.stdout.write(self.style.SUCCESS("Import process completed successfully.")) diff --git a/core/migrations/0001_initial.py b/core/migrations/0001_initial.py index 938523c..7907944 100644 --- a/core/migrations/0001_initial.py +++ b/core/migrations/0001_initial.py @@ -1,8 +1,6 @@ -# Generated by Django 5.1.4 on 2024-12-11 04:58 +# Generated by Django 5.2 on 2025-05-01 00:02 from __future__ import annotations -from typing import TYPE_CHECKING - import auto_prefetch import django.contrib.auth.models import django.contrib.auth.validators @@ -11,30 +9,70 @@ 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"], + "abstract": False, + "base_manager_name": "prefetch_manager", + }, + managers=[ + ("objects", django.db.models.manager.Manager()), + ("prefetch_manager", django.db.models.manager.Manager()), + ], + ), + 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"], + "abstract": False, + "base_manager_name": "prefetch_manager", + }, + managers=[ + ("objects", django.db.models.manager.Manager()), + ("prefetch_manager", django.db.models.manager.Manager()), + ], + ), migrations.CreateModel( name="ScrapedJson", fields=[ @@ -94,10 +132,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 +141,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,129 +162,126 @@ 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( + 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", + "owner", auto_prefetch.ForeignKey( - help_text="Reference to the JSON data from the Twitch API.", + 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"], + "ordering": ["end_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="Benefit", + fields=[ + ( + "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), + ), + ( + "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(help_text="If the benefit is farmable on iOS.", null=True)), + ("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( + null=True, + on_delete=django.db.models.deletion.SET_NULL, + related_name="benefits", + to="core.game", + ), + ), + ( + "owner_organization", + auto_prefetch.ForeignKey( + null=True, + on_delete=django.db.models.deletion.SET_NULL, + related_name="benefits", + to="core.organization", + ), + ), + ], + options={ + "ordering": ["-twitch_created_at"], "abstract": False, "base_manager_name": "prefetch_manager", }, @@ -265,32 +294,45 @@ class Migration(migrations.Migration): name="TimeBasedDrop", fields=[ ( - "created_at", - models.DateTimeField(auto_created=True, help_text="When the drop was first added to the database."), + "drop_id", + models.TextField( + help_text="The Twitch ID of the drop.", + primary_key=True, + serialize=False, + unique=True, + ), ), - ( - "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.")), + ("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), ), - ("starts_at", models.DateTimeField(help_text="When the drop starts.", null=True)), + ("start_at", models.DateTimeField(help_text="Drop start time")), + ("end_at", models.DateTimeField(help_text="Drop end time")), ( - "drop_campaign", + "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", auto_prefetch.ForeignKey( - help_text="The drop campaign this drop is part of.", - null=True, + help_text="The campaign this drop belongs to", on_delete=django.db.models.deletion.CASCADE, - related_name="drops", + related_name="time_based_drops", to="core.dropcampaign", ), ), @@ -305,116 +347,4 @@ class Migration(migrations.Migration): ("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)), - ( - "twitch_created_at", - models.DateTimeField(help_text="When the benefit was created on Twitch.", null=True), - ), - ( - "entitlement_limit", - models.PositiveBigIntegerField( - help_text="The number of times the benefit can be claimed.", - null=True, - ), - ), - ("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.")), - ("distribution_type", models.TextField(blank=True, help_text="The distribution type of the benefit.")), - ( - "game", - auto_prefetch.ForeignKey( - null=True, - on_delete=django.db.models.deletion.CASCADE, - related_name="benefits", - to="core.game", - ), - ), - ( - "owner_organization", - auto_prefetch.ForeignKey( - null=True, - on_delete=django.db.models.deletion.CASCADE, - 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", - ), - ), - ], - 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.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"), - ), ] diff --git a/core/migrations/0002_alter_user_options.py b/core/migrations/0002_alter_user_options.py deleted file mode 100644 index fb1eb94..0000000 --- a/core/migrations/0002_alter_user_options.py +++ /dev/null @@ -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"]}, - ), - ] diff --git a/core/migrations/0003_alter_benefit_created_at_and_more.py b/core/migrations/0003_alter_benefit_created_at_and_more.py deleted file mode 100644 index 96b41fd..0000000 --- a/core/migrations/0003_alter_benefit_created_at_and_more.py +++ /dev/null @@ -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."), - ), - ] diff --git a/core/migrations/0004_alter_game_created_at.py b/core/migrations/0004_alter_game_created_at.py deleted file mode 100644 index f6f81cc..0000000 --- a/core/migrations/0004_alter_game_created_at.py +++ /dev/null @@ -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."), - ), - ] diff --git a/core/models.py b/core/models.py index 239edfe..57eee26 100644 --- a/core/models.py +++ b/core/models.py @@ -1,17 +1,12 @@ 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__) @@ -45,481 +40,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(auto_prefetch.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): 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. + """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): 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): """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 = auto_prefetch.ForeignKey( + to=Game, + help_text="The game associated with this campaign", + null=True, + on_delete=models.SET_NULL, + related_name="drop_campaigns", + ) + owner = auto_prefetch.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"), - ] + 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}" + return f"{self.name or self.campaign_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. +class Benefit(auto_prefetch.Model): + """Represents a specific reward/benefit within a Drop.""" - Returns: - Self: The updated drop campaign. - """ - if wrong_typename(data, "DropCampaign"): - return self + 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 = auto_prefetch.ForeignKey(Game, on_delete=models.SET_NULL, related_name="benefits", null=True) + owner_organization = auto_prefetch.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.") - 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", - } + created_at = models.DateTimeField(auto_now_add=True) + modified_at = models.DateTimeField(auto_now=True) - updated: int = update_fields(instance=self, data=data, field_mapping=field_mapping) - if updated > 0: - logger.info("Updated %s fields for %s", updated, self) + class Meta(auto_prefetch.Model.Meta): + ordering: ClassVar[list[str]] = ["-twitch_created_at"] - 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 + 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(auto_prefetch.Model): - """This is the drop we will see on the front end. + """Represents a time-based drop within a Campaign.""" - 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" - } - """ - - # 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.") - - # Twitch fields - # "1" + 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 = auto_prefetch.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", + ) + + 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(auto_prefetch.Model.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}" diff --git a/core/models_utils.py b/core/models_utils.py index 89e981b..4bc2297 100644 --- a/core/models_utils.py +++ b/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 diff --git a/core/templates/index.html b/core/templates/index.html index 584cb1a..2bf9270 100644 --- a/core/templates/index.html +++ b/core/templates/index.html @@ -1,106 +1,111 @@ {% extends "base.html" %} {% load custom_filters static time_filters %} {% block content %} -
- {% include "partials/info_box.html" %} - -
-

- Drop Campaigns - - {{ games.count }} game{{ games.count|pluralize }} -

- - {% for game in games %} -
-
- -
- {{ game.name|default:'Game name unknown' }} box art -
- -
-
-

- {{ game.name|default:'Unknown' }} - - - Twitch -

- - {% for campaign in game.drop_campaigns.all %} -
-

{{ campaign.name }}

- Details - {% if campaign.details_url != campaign.account_link_url %} - | Link Account - {% endif %} -

- Ends in: - - {{ campaign.ends_at|timeuntil }} - -

- -
- - - - - - - - - - {% for drop in campaign.drops.all %} - {% if drop.benefits.exists %} - {% for benefit in drop.benefits.all %} - - - - - - {% endfor %} - {% else %} - - - - - - {% endif %} - {% endfor %} - -
Benefit ImageBenefit NameRequired Minutes Watched
- {{ benefit.name|default:'Unknown' }} - - - {{ benefit.name|default:'Unknown' }} - - {{ drop.required_minutes_watched|minutes_to_hours }}
- {{ drop.name|default:'Unknown' }} - {{ drop.name|default:'Unknown' }}N/A
-
-
- {% endfor %} -
-
+
+ {% include "partials/info_box.html" %} + +
+

+ Drop Campaigns - + {{ grouped_drops|length }} game{{ grouped_drops|length|pluralize }} +

+ {% if grouped_drops %} + {% for game, drops_list in grouped_drops.items %} + {# Retain card structure for layout, replace table with list #} +
+
+ +
+ {% if game and game.box_art_url %} + {{ game.display_name|default:'Game name unknown' }} box art + {% else %} + {% if game %}{{ game.display_name }}{% else %}Unknown Game{% endif %} box art + {% endif %} +
+ +
+
+

+ {% if game %} + {{ game.display_name|default:'Unknown Game' }} + {% if game.slug %} + - Twitch + {% endif %} + {% else %} + Drops without an associated Game + {% endif %} +

+ + {% if drops_list %} + +
    + {% for drop in drops_list %} +
  • {# Add margin between list items #} + {{ drop.name|default:'Unknown Drop' }} + (Requires {{ drop.required_minutes_watched|minutes_to_hours }}) +
    + Campaign: + + {{ drop.campaign.name|truncatechars:40|default:'N/A' }} {# Adjusted truncate #} + + {% if drop.campaign.details_url != drop.campaign.account_link_url and drop.campaign.account_link_url %} + | Link + {% endif %} + +
    + Ends in: + {{ drop.campaign.end_at|timeuntil }} + + + {% if drop.benefits.exists %} +
    + Benefits: +
      + {% for benefit in drop.benefits.all %} +
    • + + {% if benefit.image_asset_url %} + {{ benefit.name|default:'Unknown Benefit' }} + {% endif %} + {{ benefit.name|truncatechars:25|default:'Unknown Benefit' }} + {# Wrap text in small #} + +
      • + {% endfor %} +
    + {% endif %} + {# Removed hr, using li margin instead #} +
    • + {% endfor %} +
+ {% else %} +

No active drops found for this game currently.

+ {% endif %}
- {% endfor %} -
-
+
+
+ {% empty %} + + {% endfor %} + {% else %} + + {% endif %} +
+
{% endblock content %} diff --git a/core/urls.py b/core/urls.py index 62d722b..7fd97db 100644 --- a/core/urls.py +++ b/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//", view=get_game, name="game"), + path(route="game//", 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(), ] diff --git a/core/views.py b/core/views.py index f84bde8..85c9095 100644 --- a/core/views.py +++ b/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,41 @@ 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.now() + grouped_drops = defaultdict(list) - context: dict[str, Any] = {"games": games} - return TemplateResponse(request, "index.html", context) + # Query for active drops, efficiently fetching related campaign and game + # Also prefetch benefits if you need them in the template + current_drops_qs = ( + TimeBasedDrop.objects.filter(start_at__lte=now, end_at__gte=now) + .select_related( + "campaign__game", # Follows ForeignKey relationships campaign -> game + ) + .prefetch_related( + "benefits", # Efficiently fetches ManyToMany benefits + ) + .order_by( + "campaign__game__display_name", # Order by game name first + "name", # Then by drop name + ) + ) + + # Group the drops by game in Python + for drop in current_drops_qs: + # Check if the drop has an associated campaign and game + if drop.campaign and drop.campaign.game: + game = drop.campaign.game + grouped_drops[game].append(drop) + else: + # Handle drops without a game (optional, based on your data integrity) + # You could group them under a 'None' key or log a warning + # grouped_drops[None].append(drop) + pass # Or ignore them + + context = { + "grouped_drops": dict(grouped_drops), # Convert defaultdict back to dict for template if preferred + } + return render(request, "index.html", context) @require_http_methods(request_method_list=["GET", "HEAD"]) @@ -117,25 +114,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) diff --git a/tests/models_test.py b/tests/models_test.py deleted file mode 100644 index 4c3d93f..0000000 --- a/tests/models_test.py +++ /dev/null @@ -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()