TheLovinator/twitch-drop-notifier
1
Fork 0
Code Issues Actions Packages Releases Activity

Refactor models

Browse Source
This commit is contained in:
Joakim Hellsén
2025-05-01 02:41:25 +02:00
parent d137ad61f0
commit d7b31e1d42
17 changed files with 704 additions and 1392 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
.pre-commit-config.yaml
View File

@ -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

12
.vscode/launch.json vendored
View File

@ -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
}
]
}

4
core/admin.py
View File

@ -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)

187
core/import_json.py
View File

@ -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

0
core/management/__init__.py Normal file
View File

0
core/management/commands/__init__.py Normal file
View File

293
core/management/commands/import_twitch_drops.py Normal file
View File

@ -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."))

398
core/migrations/0001_initial.py
View File

@ -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"),
),
]

24
core/migrations/0002_alter_user_options.py
View File

@ -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"]},
),
]

48
core/migrations/0003_alter_benefit_created_at_and_more.py
View File

@ -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."),
),
]

25
core/migrations/0004_alter_game_created_at.py
View File

@ -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."),
),
]

502
core/models.py
View File

@ -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}"

2
core/models_utils.py
View File

@ -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

155
core/templates/index.html
View File

@ -7,100 +7,105 @@
<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>
{% endblock content %}

5
core/urls.py
View File

@ -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(),
]

103
core/views.py
View File

@ -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)

274
tests/models_test.py
View File

@ -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()