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

Add functions for importing data

Browse Source
This commit is contained in:
Joakim Hellsén
2024-12-16 23:16:11 +01:00
parent 7dda6ef3c3
commit db810d907d
7 changed files with 322 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
.vscode/launch.json vendored
View File

@ -1,6 +1,13 @@
{
"version": "0.2.0",
"configurations": [
{
"name": "Python Debugger: Current File",
"type": "debugpy",
"request": "launch",
"program": "${file}",
"console": "integratedTerminal"
},
{
"name": "Django: Runserver",
"type": "debugpy",

163
core/import_json.py
View File

@ -1,11 +1,168 @@
from __future__ import annotations
from typing import Any
import logging
from typing import Any, Literal
from core.models import Benefit, DropCampaign, Game, Owner, TimeBasedDrop
logger: logging.Logger = logging.getLogger(__name__)
def import_data_from_view(data: dict[str, Any]) -> None:
"""Import data that are sent from Twitch Drop Miner.
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]) -> None:
"""Import the drop campaigns from the data.
Args:
drop_campaigns (dict[str, Any]): The drop campaign data.
"""
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
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)
def import_time_based_drops(drop_campaign_json: dict[str, Any], drop_campaign: DropCampaign) -> None:
"""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.
"""
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)
def import_drop_benefits(time_based_drop_json: dict[str, Any], time_based_drop: TimeBasedDrop) -> None:
"""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.
"""
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)
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

24
core/migrations/0002_alter_user_options.py Normal file
View File

@ -0,0 +1,24 @@
# 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 Normal file
View File

@ -0,0 +1,48 @@
# 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."),
),
]

173
core/models.py
View File

@ -69,7 +69,7 @@ class Owner(auto_prefetch.Model):
# 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_created=True)
created_at = models.DateTimeField(auto_now_add=True)
modified_at = models.DateTimeField(auto_now=True)
# Twitch fields
@ -104,68 +104,30 @@ class Game(auto_prefetch.Model):
"""The game the drop campaign is for. Note that some reward campaigns are not tied to a game.
JSON:
{
"data": {
"user": {
"dropCampaign": {
"game": {
"id": "155409827",
"slug": "pokemon-trading-card-game-live",
"displayName": "Pok\u00e9mon Trading Card Game Live",
"__typename": "Game"
}
}
}
}
"game": {
"id": "155409827",
"slug": "pokemon-trading-card-game-live",
"displayName": "Pok\u00e9mon Trading Card Game Live",
"__typename": "Game"
}
Secondary JSON:
{
"data": {
"currentUser": {
"dropCampaigns": [
{
"game": {
"id": "155409827",
"displayName": "Pok\u00e9mon Trading Card Game Live",
"boxArtURL": "https://static-cdn.jtvnw.net/ttv-boxart/155409827_IGDB-120x160.jpg",
"__typename": "Game"
}
}
]
}
}
"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:
[
{
"data": {
"user": {
"dropCampaign": {
"timeBasedDrops": [
{
"benefitEdges": [
{
"benefit": {
"id": "ea74f727-a52f-11ef-811f-0a58a9feac02",
"createdAt": "2024-11-17T22:04:28.735Z",
"entitlementLimit": 1,
"game": {
"id": "155409827",
"name": "Pok\u00e9mon Trading Card Game Live",
"__typename": "Game"
}
}
}
]
}
]
}
}
}
}
]
"game": {
"id": "155409827",
"name": "Pok\u00e9mon Trading Card Game Live",
"__typename": "Game"
}
"""
# Django fields
@ -212,6 +174,10 @@ class Game(auto_prefetch.Model):
if wrong_typename(data, "Game"):
return self
if not owner:
logger.error("Owner is required for %s", self)
return self
# Map the fields from the JSON data to the Django model fields.
field_mapping: dict[str, str] = {
"displayName": "display_name",
@ -224,15 +190,14 @@ class Game(auto_prefetch.Model):
if updated > 0:
logger.info("Updated %s fields for %s", updated, self)
if not owner:
logger.error("Owner is required for %s", self)
return 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.save()
self.game_url = f"https://www.twitch.tv/directory/category/{self.slug}"
self.save()
return self
@ -244,7 +209,7 @@ class DropCampaign(auto_prefetch.Model):
# "f257ce6e-502a-11ef-816e-0a58a9feac02"
twitch_id = models.TextField(primary_key=True, help_text="The Twitch ID of the drop campaign.")
created_at = models.DateTimeField(
auto_created=True,
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.")
@ -346,59 +311,47 @@ class TimeBasedDrop(auto_prefetch.Model):
JSON:
{
"data": {
"user": {
"dropCampaign": {
"timeBasedDrops": [
{
"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"
}
],
"__typename": "DropCampaign"
"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"
},
"__typename": "User"
"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"
}
""" # noqa: E501
"""
# 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_created=True, help_text="When the drop was first added to the database.")
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
@ -476,7 +429,7 @@ class Benefit(auto_prefetch.Model):
# Django fields
# "d5cdf372-502b-11ef-bafd-0a58a9feac02"
twitch_id = models.TextField(primary_key=True)
created_at = models.DateTimeField(null=True, auto_created=True)
created_at = models.DateTimeField(auto_now_add=True)
modified_at = models.DateTimeField(auto_now=True)
# Twitch fields

4
core/views.py
View File

@ -10,7 +10,7 @@ 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_view
from core.import_json import import_data
from core.models import Benefit, DropCampaign, Game, TimeBasedDrop
if TYPE_CHECKING:
@ -134,7 +134,7 @@ def get_import(request: HttpRequest) -> HttpResponse:
logger.info(data)
# Import the data.
import_data_from_view(data)
import_data(data)
return JsonResponse({"status": "success"}, status=200)
except json.JSONDecodeError as e:

18
tests/models_test.py Normal file
View File

@ -0,0 +1,18 @@
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
from core.import_json import find_typename_in_json
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"