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

WIP for moving everything into models

Browse Source
This commit is contained in:
Joakim Hellsén
2024-09-01 23:08:15 +02:00
parent 8f68b38116
commit 717adb2df9
5 changed files with 210 additions and 284 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
core/admin.py
View File

@ -1,12 +1,11 @@
from django.contrib import admin
from .models import Benefit, Channel, DropCampaign, Game, Owner, Reward, RewardCampaign, TimeBasedDrop
from .models import Benefit, DropCampaign, Game, Owner, Reward, RewardCampaign, TimeBasedDrop
admin.site.register(Game)
admin.site.register(Owner)
admin.site.register(RewardCampaign)
admin.site.register(DropCampaign)
admin.site.register(Channel)
admin.site.register(TimeBasedDrop)
admin.site.register(Benefit)
admin.site.register(Reward)

280
core/management/commands/scrape_twitch.py
View File

@ -10,7 +10,7 @@ from platformdirs import user_data_dir
from playwright.async_api import Playwright, async_playwright
from playwright.async_api._generated import Response
from core.models import Benefit, Channel, DropCampaign, Game, Owner, Reward, RewardCampaign, TimeBasedDrop
from core.models import Benefit, DropCampaign, Game, Owner, Reward, RewardCampaign, TimeBasedDrop
if TYPE_CHECKING:
from playwright.async_api._generated import BrowserContext, Page
@ -78,197 +78,17 @@ async def add_reward_campaign(campaign: dict | None) -> None:
return
if "data" in campaign and "rewardCampaignsAvailableToUser" in campaign["data"]:
for reward_campaign in campaign["data"]["rewardCampaignsAvailableToUser"]:
our_reward_campaign: RewardCampaign = await handle_reward_campaign(reward_campaign)
our_reward_campaign, created = await RewardCampaign.objects.aupdate_or_create(id=reward_campaign["id"])
await our_reward_campaign.import_json(reward_campaign)
if created:
logger.info("Added reward campaign %s", our_reward_campaign)
if "rewards" in reward_campaign:
for reward in reward_campaign["rewards"]:
await handle_rewards(reward, our_reward_campaign)
async def handle_rewards(reward: dict, reward_campaign: RewardCampaign | None) -> None:
"""Add or update a reward in the database.
Args:
reward (dict): The JSON from Twitch.
reward_campaign (RewardCampaign | None): The reward campaign the reward belongs to.
"""
mappings: dict[str, str] = {
"name": "name",
"earnableUntil": "earnable_until",
"redemptionInstructions": "redemption_instructions",
"redemptionURL": "redemption_url",
}
defaults: dict = {new_key: reward[key] for key, new_key in mappings.items() if reward.get(key)}
if reward_campaign:
defaults["campaign"] = reward_campaign
if reward.get("bannerImage"):
defaults["banner_image_url"] = reward["bannerImage"]["image1xURL"]
if reward.get("thumbnailImage"):
defaults["thumbnail_image_url"] = reward["thumbnailImage"]["image1xURL"]
reward_instance, created = await Reward.objects.aupdate_or_create(id=reward["id"], defaults=defaults)
if created:
logger.info("Added reward %s", reward_instance.id)
async def handle_reward_campaign(reward_campaign: dict) -> RewardCampaign:
"""Add or update a reward campaign in the database.
Args:
reward_campaign (dict): The reward campaign JSON from Twitch.
Returns:
RewardCampaign: The reward campaign that was added or updated.
"""
mappings: dict[str, str] = {
"name": "name",
"brand": "brand",
"createdAt": "created_at",
"startsAt": "starts_at",
"endsAt": "ends_at",
"status": "status",
"summary": "summary",
"instructions": "instructions",
"rewardValueURLParam": "reward_value_url_param",
"externalURL": "external_url",
"aboutURL": "about_url",
"isSitewide": "is_site_wide",
}
defaults: dict = {new_key: reward_campaign[key] for key, new_key in mappings.items() if reward_campaign.get(key)}
unlock_requirements: dict = reward_campaign.get("unlockRequirements", {})
if unlock_requirements.get("subsGoal"):
defaults["sub_goal"] = unlock_requirements["subsGoal"]
if unlock_requirements.get("minuteWatchedGoal"):
defaults["minute_watched_goal"] = unlock_requirements["minuteWatchedGoal"]
if reward_campaign.get("image"):
defaults["image_url"] = reward_campaign["image"]["image1xURL"]
reward_campaign_instance, created = await RewardCampaign.objects.aupdate_or_create(
id=reward_campaign["id"],
defaults=defaults,
)
if created:
logger.info("Added reward campaign %s", reward_campaign_instance.id)
if reward_campaign["game"]:
# TODO(TheLovinator): Add game to reward campaign # noqa: TD003
# TODO(TheLovinator): Send JSON to Discord # noqa: TD003
logger.error("Not implemented: Add game to reward campaign, JSON: %s", reward_campaign["game"])
return reward_campaign_instance
async def add_or_update_game(game_json: dict | None, owner: Owner | None) -> Game | None:
"""Add or update a game in the database.
Args:
game_json (dict): The game to add or update.
owner (Owner): The owner of the game.
Returns:
Game: The game that was added or updated.
"""
if not game_json:
return None
mappings: dict[str, str] = {
"slug": "slug",
"displayName": "name",
"boxArtURL": "box_art_url",
}
defaults: dict = {new_key: game_json[key] for key, new_key in mappings.items() if game_json.get(key)}
if game_json.get("slug"):
defaults["game_url"] = f"https://www.twitch.tv/directory/game/{game_json["slug"]}"
our_game, created = await Game.objects.aupdate_or_create(twitch_id=game_json["id"], defaults=defaults)
if created:
logger.info("Added game %s", our_game.twitch_id)
if owner:
await owner.games.aadd(our_game) # type: ignore # noqa: PGH003
return our_game
async def add_or_update_owner(owner_json: dict | None) -> Owner | None:
"""Add or update an owner in the database.
Args:
owner_json (dict): The owner to add or update.
Returns:
Owner: The owner that was added or updated.
"""
if not owner_json:
return None
defaults: dict[str, str] = {}
if owner_json.get("name"):
defaults["name"] = owner_json["name"]
our_owner, created = await Owner.objects.aupdate_or_create(id=owner_json.get("id"), defaults=defaults)
if created:
logger.info("Added owner %s", our_owner.id)
return our_owner
async def add_or_update_channels(channels_json: list[dict]) -> list[Channel] | None:
"""Add or update channels in the database.
Args:
channels_json (list[dict]): The channels to add or update.
Returns:
list[Channel]: The channels that were added or updated.
"""
if not channels_json:
return None
channels: list[Channel] = []
for channel_json in channels_json:
defaults: dict[str, str] = {}
if channel_json.get("displayName"):
defaults["display_name"] = channel_json["displayName"]
if channel_json.get("name"):
defaults["name"] = channel_json["name"]
defaults["twitch_url"] = f'https://www.twitch.tv/{channel_json["name"]}'
our_channel, created = await Channel.objects.aupdate_or_create(twitch_id=channel_json["id"], defaults=defaults)
if created:
logger.info("Added channel %s", our_channel.twitch_id)
channels.append(our_channel)
return channels
async def add_benefit(benefit: dict, time_based_drop: TimeBasedDrop) -> None:
"""Add a benefit to the database.
Args:
benefit (dict): The benefit to add.
time_based_drop (TimeBasedDrop): The time-based drop the benefit belongs to.
"""
mappings: dict[str, str] = {
"createdAt": "twitch_created_at",
"entitlementLimit": "entitlement_limit",
"imageAssetURL": "image_url",
"isIosAvailable": "is_ios_available",
"name": "name",
}
defaults: dict[str, str] = {new_key: benefit[key] for key, new_key in mappings.items() if benefit.get(key)}
our_benefit, created = await Benefit.objects.aupdate_or_create(id=benefit["id"], defaults=defaults)
if created:
logger.info("Added benefit %s", our_benefit.id)
if time_based_drop:
await time_based_drop.benefits.aadd(our_benefit) # type: ignore # noqa: PGH003
reward_instance, created = await Reward.objects.aupdate_or_create(id=reward["id"])
await reward_instance.import_json(reward, our_reward_campaign)
if created:
logger.info("Added reward %s", reward_instance)
async def add_drop_campaign(drop_campaign: dict | None) -> None:
@ -280,60 +100,24 @@ async def add_drop_campaign(drop_campaign: dict | None) -> None:
if not drop_campaign:
return
defaults: dict[str, str | Game | Owner] = {}
owner: Owner | None = await get_owner(drop_campaign)
if drop_campaign.get("game"):
game: Game | None = await add_or_update_game(drop_campaign["game"], owner)
if game:
defaults["game"] = game
owner, created = await Owner.objects.aupdate_or_create(id=drop_campaign["owner"]["id"])
owner.import_json(drop_campaign["owner"])
mappings: dict[str, str] = {
"accountLinkURL": "account_link_url",
"description": "description",
"detailsURL": "details_url",
"endAt": "ends_at",
"startAt": "starts_at",
"imageURL": "image_url",
"name": "name",
"status": "status",
}
for key, new_key in mappings.items():
if drop_campaign.get(key):
defaults[new_key] = drop_campaign[key]
game, created = await Game.objects.aupdate_or_create(id=drop_campaign["game"]["id"])
await game.import_json(drop_campaign["game"], owner)
if created:
logger.info("Added game %s", game)
our_drop_campaign, created = await DropCampaign.objects.aupdate_or_create(id=drop_campaign["id"])
await our_drop_campaign.import_json(drop_campaign, game)
our_drop_campaign, created = await DropCampaign.objects.aupdate_or_create(
id=drop_campaign["id"],
defaults=defaults,
)
if created:
logger.info("Added drop campaign %s", our_drop_campaign.id)
if drop_campaign.get("allow") and drop_campaign["allow"].get("channels"):
channels: list[Channel] | None = await add_or_update_channels(drop_campaign["allow"]["channels"])
if channels:
for channel in channels:
await channel.drop_campaigns.aadd(our_drop_campaign)
await add_time_based_drops(drop_campaign, our_drop_campaign)
async def get_owner(drop_campaign: dict) -> Owner | None:
"""Get the owner of the drop campaign.
Args:
drop_campaign (dict): The drop campaign containing the owner.
Returns:
Owner: The owner of the drop campaign.
"""
owner = None
if drop_campaign.get("owner"):
owner: Owner | None = await add_or_update_owner(drop_campaign["owner"])
return owner
async def add_time_based_drops(drop_campaign: dict, our_drop_campaign: DropCampaign) -> None:
"""Add time-based drops to the database.
@ -346,31 +130,21 @@ async def add_time_based_drops(drop_campaign: dict, our_drop_campaign: DropCampa
if time_based_drop.get("preconditionDrops"):
# TODO(TheLovinator): Add precondition drops to time-based drop # noqa: TD003
# TODO(TheLovinator): Send JSON to Discord # noqa: TD003
logger.error("Not implemented: Add precondition drops to time-based drop, JSON: %s", time_based_drop)
msg = "Not implemented: Add precondition drops to time-based drop"
raise NotImplementedError(msg)
mappings: dict[str, str] = {
"requiredSubs": "required_subs",
"endAt": "ends_at",
"name": "name",
"requiredMinutesWatched": "required_minutes_watched",
"startAt": "starts_at",
}
defaults: dict[str, str | DropCampaign] = {
new_key: time_based_drop[key] for key, new_key in mappings.items() if time_based_drop.get(key)
}
if our_drop_campaign:
defaults["drop_campaign"] = our_drop_campaign
our_time_based_drop, created = await TimeBasedDrop.objects.aupdate_or_create(id=time_based_drop["id"])
await our_time_based_drop.import_json(time_based_drop, our_drop_campaign)
our_time_based_drop, created = await TimeBasedDrop.objects.aupdate_or_create(
id=time_based_drop["id"],
defaults=defaults,
)
if created:
logger.info("Added time-based drop %s", our_time_based_drop.id)
if time_based_drop.get("benefitEdges") and our_time_based_drop:
if our_time_based_drop and time_based_drop.get("benefitEdges"):
for benefit_edge in time_based_drop["benefitEdges"]:
await add_benefit(benefit_edge["benefit"], our_time_based_drop)
benefit, created = await Benefit.objects.aupdate_or_create(id=benefit_edge["benefit"])
await benefit.import_json(benefit_edge["benefit"], our_time_based_drop)
if created:
logger.info("Added benefit %s", benefit.id)
async def process_json_data(num: int, campaign: dict | None) -> None:

15
core/migrations/0002_delete_channel.py Normal file
View File

@ -0,0 +1,15 @@
# Generated by Django 5.1 on 2024-09-01 02:59
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
("core", "0001_initial"),
]
operations = [
migrations.DeleteModel(
name="Channel",
),
]

191
core/models.py
View File

@ -1,7 +1,7 @@
from __future__ import annotations
import logging
from typing import ClassVar
from typing import ClassVar, Self
from django.contrib.auth.models import AbstractUser
from django.db import models
@ -24,6 +24,15 @@ class Owner(models.Model):
def __str__(self) -> str:
return self.name or "Owner name unknown"
def import_json(self, data: dict | None) -> Self:
if not data:
return self
self.name = data.get("name", self.name)
self.save()
return self
class Game(models.Model):
"""This is the game we will see on the front end."""
@ -32,59 +41,97 @@ class Game(models.Model):
# "https://www.twitch.tv/directory/category/halo-infinite"
game_url = models.URLField(null=True, default="https://www.twitch.tv/")
name = models.TextField(null=True, default="Game name unknown") # "Halo Infinite"
# "Halo Infinite"
name = models.TextField(null=True, default="Game name unknown")
# "https://static-cdn.jtvnw.net/ttv-boxart/Halo%20Infinite.jpg"
box_art_url = models.URLField(null=True, default="https://static-cdn.jtvnw.net/ttv-static/404_boxart.jpg")
slug = models.TextField(null=True) # "halo-infinite"
# "halo-infinite"
slug = models.TextField(null=True)
org = models.ForeignKey(Owner, on_delete=models.CASCADE, related_name="games", null=True)
def __str__(self) -> str:
return self.name or "Game name unknown"
return self.name or self.twitch_id
async def import_json(self, data: dict | None, owner: Owner | None) -> Self:
if not data:
logger.error("No data provided for %s.", self)
return self
self.name = data.get("name", self.name)
self.box_art_url = data.get("boxArtURL", self.box_art_url)
self.slug = data.get("slug", self.slug)
if data.get("slug"):
self.game_url = f"https://www.twitch.tv/directory/game/{data["slug"]}"
if owner:
await owner.games.aadd(self) # type: ignore # noqa: PGH003
self.save()
return self
class DropCampaign(models.Model):
"""This is the drop campaign we will see on the front end."""
id = models.TextField(primary_key=True) # "f257ce6e-502a-11ef-816e-0a58a9feac02"
created_at = models.DateTimeField(null=True, auto_created=True) # "2024-08-11T00:00:00Z"
modified_at = models.DateTimeField(null=True, auto_now=True) # "2024-08-12T00:00:00Z"
# "f257ce6e-502a-11ef-816e-0a58a9feac02"
id = models.TextField(primary_key=True)
created_at = models.DateTimeField(null=True, auto_created=True)
modified_at = models.DateTimeField(null=True, auto_now=True)
account_link_url = models.URLField(null=True) # "https://www.halowaypoint.com/settings/linked-accounts"
# "https://www.halowaypoint.com/settings/linked-accounts"
account_link_url = models.URLField(null=True)
# "Tune into this HCS Grassroots event to earn Halo Infinite in-game content!"
description = models.TextField(null=True)
details_url = models.URLField(null=True) # "https://www.halowaypoint.com"
ends_at = models.DateTimeField(null=True) # "2024-08-12T05:59:59.999Z"
starts_at = models.DateTimeField(null=True) # "2024-08-11T11:00:00Z""
# "https://www.halowaypoint.com"
details_url = models.URLField(null=True)
# "2024-08-12T05:59:59.999Z"
ends_at = models.DateTimeField(null=True)
# "2024-08-11T11:00:00Z""
starts_at = models.DateTimeField(null=True)
game = models.ForeignKey(Game, on_delete=models.CASCADE, related_name="drop_campaigns", null=True)
# "https://static-cdn.jtvnw.net/twitch-quests-assets/CAMPAIGN/c8e02666-8b86-471f-bf38-7ece29a758e4.png"
image_url = models.URLField(null=True)
name = models.TextField(null=True) # "HCS Open Series - Week 1 - DAY 2 - AUG11"
status = models.TextField(null=True) # "ACTIVE"
# "HCS Open Series - Week 1 - DAY 2 - AUG11"
name = models.TextField(null=True, default="Unknown")
# "ACTIVE"
status = models.TextField(null=True)
def __str__(self) -> str:
return self.name or "Drop campaign name unknown"
return self.name or self.id
async def import_json(self, data: dict | None, game: Game) -> Self:
if not data:
logger.error("No data provided for %s.", self)
return self
class Channel(models.Model):
"""This is the channel we will see on the front end."""
self.name = data.get("name", self.name)
self.account_link_url = data.get("accountLinkURL", self.account_link_url)
self.description = data.get("description", self.description)
self.details_url = data.get("detailsURL", self.details_url)
self.ends_at = data.get("endAt", self.ends_at)
self.starts_at = data.get("startAt", self.starts_at)
self.status = data.get("status", self.status)
self.image_url = data.get("imageURL", self.image_url)
twitch_id = models.TextField(primary_key=True) # "222719079"
display_name = models.TextField(null=True, default="Channel name unknown") # "LVTHalo"
name = models.TextField(null=True) # "lvthalo"
twitch_url = models.URLField(null=True, default="https://www.twitch.tv/") # "https://www.twitch.tv/lvthalo"
live = models.BooleanField(default=False) # "True"
if game:
self.game = game
drop_campaigns = models.ManyToManyField(DropCampaign, related_name="channels")
self.save()
def __str__(self) -> str:
return self.display_name or "Channel name unknown"
return self
class TimeBasedDrop(models.Model):
@ -105,6 +152,24 @@ class TimeBasedDrop(models.Model):
def __str__(self) -> str:
return self.name or "Drop name unknown"
async def import_json(self, data: dict | None, drop_campaign: DropCampaign) -> Self:
if not data:
logger.error("No data provided for %s.", self)
return self
self.name = data.get("name", self.name)
self.required_subs = data.get("requiredSubs", self.required_subs)
self.required_minutes_watched = data.get("requiredMinutesWatched", self.required_minutes_watched)
self.starts_at = data.get("startAt", self.starts_at)
self.ends_at = data.get("endAt", self.ends_at)
if drop_campaign:
self.drop_campaign = drop_campaign
self.save()
return self
class Benefit(models.Model):
"""Benefits are the rewards for the drops."""
@ -134,6 +199,24 @@ class Benefit(models.Model):
def __str__(self) -> str:
return self.name or "Benefit name unknown"
async def import_json(self, data: dict | None, time_based_drop: TimeBasedDrop) -> Self:
if not data:
logger.error("No data provided for %s.", self)
return self
self.name = data.get("name", self.name)
self.entitlement_limit = data.get("entitlementLimit", self.entitlement_limit)
self.is_ios_available = data.get("isIosAvailable", self.is_ios_available)
self.image_url = data.get("imageAssetURL", self.image_url)
self.twitch_created_at = data.get("createdAt", self.twitch_created_at)
if time_based_drop:
await time_based_drop.benefits.aadd(self) # type: ignore # noqa: PGH003
self.save()
return self
class RewardCampaign(models.Model):
"""Buy subscriptions to earn rewards."""
@ -164,6 +247,41 @@ class RewardCampaign(models.Model):
def __str__(self) -> str:
return self.name or "Reward campaign name unknown"
async def import_json(self, data: dict | None) -> Self:
if not data:
logger.error("No data provided for %s.", self)
return self
self.name = data.get("name", self.name)
self.brand = data.get("brand", self.brand)
self.starts_at = data.get("startsAt", self.starts_at)
self.ends_at = data.get("endsAt", self.ends_at)
self.status = data.get("status", self.status)
self.summary = data.get("summary", self.summary)
self.instructions = data.get("instructions", self.instructions)
self.reward_value_url_param = data.get("rewardValueURLParam", self.reward_value_url_param)
self.external_url = data.get("externalURL", self.external_url)
self.about_url = data.get("aboutURL", self.about_url)
self.is_site_wide = data.get("isSiteWide", self.is_site_wide)
unlock_requirements: dict = data.get("unlockRequirements", {})
if unlock_requirements:
self.sub_goal = unlock_requirements.get("subsGoal", self.sub_goal)
self.minute_watched_goal = unlock_requirements.get("minuteWatchedGoal", self.minute_watched_goal)
image = data.get("image", {})
if image:
self.image_url = image.get("image1xURL", self.image_url)
if data.get("game"):
game: Game | None = Game.objects.filter(twitch_id=data["game"]["id"]).first()
if game:
await game.reward_campaigns.aadd(self) # type: ignore # noqa: PGH003
self.save()
return self
class Reward(models.Model):
"""This from the RewardCampaign."""
@ -185,6 +303,31 @@ class Reward(models.Model):
def __str__(self) -> str:
return self.name or "Reward name unknown"
async def import_json(self, data: dict | None, reward_campaign: RewardCampaign) -> Self:
if not data:
logger.error("No data provided for %s.", self)
return self
self.name = data.get("name", self.name)
self.earnable_until = data.get("earnableUntil", self.earnable_until)
self.redemption_instructions = data.get("redemptionInstructions", self.redemption_instructions)
self.redemption_url = data.get("redemptionURL", self.redemption_url)
banner_image = data.get("bannerImage", {})
if banner_image:
self.banner_image_url = banner_image.get("image1xURL", self.banner_image_url)
thumbnail_image = data.get("thumbnailImage", {})
if thumbnail_image:
self.thumbnail_image_url = thumbnail_image.get("image1xURL", self.thumbnail_image_url)
if reward_campaign:
self.campaign = reward_campaign
self.save()
return self
class Webhook(models.Model):
"""Discord webhook."""

5
core/templates/game.html
View File

@ -63,8 +63,6 @@
<hr>
<div>
<h3 class="mb-2">{{ drop.name }}</h3>
{% if drop.benefits.all %}
{% for benefit in drop.benefits.all %}
<img src="{{ benefit.image_url }}" alt="{{ benefit.name }} image"
class="img-fluid rounded mb-2">
@ -79,9 +77,6 @@
{% empty %}
<div>No benefits available for this drop.</div>
{% endfor %}
{% else %}
<p>No benefits available for this drop.</p>
{% endif %}
</div>
{% empty %}
<div>No time-based drops available for this campaign.</div>