diff --git a/.vscode/settings.json b/.vscode/settings.json index eeb8746..8df0627 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -3,6 +3,8 @@ "allauth", "appendonly", "asgiref", + "Behaviour", + "cacd", "forloop", "logdir", "memlock", @@ -12,6 +14,7 @@ "requirepass", "sitewide", "socialaccount", + "Stresss", "ttvdrops", "ulimits", "xdefiant" diff --git a/core/views.py b/core/views.py index 3712fe3..ac236b2 100644 --- a/core/views.py +++ b/core/views.py @@ -5,7 +5,6 @@ from typing import TYPE_CHECKING import hishel from django.conf import settings -from django.db.models.manager import BaseManager from django.template.response import TemplateResponse from django.views.generic import ListView @@ -15,6 +14,7 @@ from twitch_app.models import Game, RewardCampaign if TYPE_CHECKING: from pathlib import Path + from django.db.models.manager import BaseManager from django.http import HttpRequest, HttpResponse from httpx import Response diff --git a/twitch_app/management/commands/scrape_twitch.py b/twitch_app/management/commands/scrape_twitch.py index 8b475a2..9360677 100644 --- a/twitch_app/management/commands/scrape_twitch.py +++ b/twitch_app/management/commands/scrape_twitch.py @@ -11,7 +11,20 @@ from platformdirs import user_data_dir from playwright.async_api import Playwright, async_playwright from playwright.async_api._generated import Response -from twitch_app.models import Game, Image, Reward, RewardCampaign, UnlockRequirements +from twitch_app.models import ( + Allow, + Benefit, + BenefitEdge, + Channel, + DropCampaign, + Game, + Image, + Owner, + Reward, + RewardCampaign, + TimeBasedDrop, + UnlockRequirements, +) if TYPE_CHECKING: from playwright.async_api._generated import BrowserContext, Page @@ -33,92 +46,428 @@ if not data_dir: logger: logging.Logger = logging.getLogger(__name__) -async def add_reward_campaign(json_data: dict) -> None: - """Add data from JSON to the database.""" - for campaign_data in json_data["data"]["rewardCampaignsAvailableToUser"]: - # Add or get Game - game_data = campaign_data["game"] - if game_data: - game, _ = await sync_to_async(Game.objects.get_or_create)( - id=game_data["id"], - slug=game_data["slug"], - defaults={ - "display_name": game_data["displayName"], - "typename": game_data["__typename"], - }, - ) - else: - logger.warning("%s is not for a game?", campaign_data["name"]) - game = None +async def add_or_get_game(json_data: dict, name: str) -> tuple[Game | None, bool]: + """Add or get Game from JSON data. - # Add or get Image - image_data = campaign_data["image"] - image, _ = await sync_to_async(Image.objects.get_or_create)( - image1_x_url=image_data["image1xURL"], - defaults={"typename": image_data["__typename"]}, + Args: + json_data (dict): JSON data to add to the database. + name (str): Name of the drop campaign. + + Returns: + tuple[Game | None, bool]: Game instance and whether it was created. + """ + if not json_data: + logger.warning("%s is not for a game?", name) + return None, False + + game, created = await Game.objects.aupdate_or_create( + id=json_data["id"], + defaults={ + "slug": json_data.get("slug"), + "display_name": json_data.get("displayName"), + "typename": json_data.get("__typename"), + }, + ) + + return game, created + + +async def add_or_get_owner(json_data: dict, name: str) -> tuple[Owner | None, bool]: + """Add or get Owner from JSON data. + + Args: + json_data (dict): JSON data to add to the database. + name (str): Name of the drop campaign. + + Returns: + Owner: Owner instance. + """ + if not json_data: + logger.warning("Owner data is missing for %s", name) + return None, False + + owner, created = await Owner.objects.aupdate_or_create( + id=json_data["id"], + defaults={ + "display_name": json_data.get("name"), + "typename": json_data.get("__typename"), + }, + ) + + return owner, created + + +async def add_or_get_allow(json_data: dict, name: str) -> tuple[Allow | None, bool]: + """Add or get Allow from JSON data. + + Args: + json_data (dict): JSON data to add to the database. + name (str): Name of the drop campaign. + + Returns: + Allow: Allow instance. + """ + if not json_data: + logger.warning("Allow data is missing for %s", name) + return None, False + + allow, created = await Allow.objects.aupdate_or_create( + is_enabled=json_data.get("isEnabled"), + typename=json_data.get("__typename"), + ) + + return allow, created + + +async def add_or_get_time_based_drops( + time_based_drops_data: list[dict] | None, + owner: Owner | None, + game: Game | None, +) -> list[TimeBasedDrop]: + """Handle TimeBasedDrops from JSON data. + + Args: + time_based_drops_data (list[dict]): Time based drops data from JSON. + owner (Owner): Owner instance. + game (Game): Game instance. + + Returns: + list[TimeBasedDrop]: TimeBasedDrop instances. + """ + time_based_drops: list[TimeBasedDrop] = [] + + if not time_based_drops_data: + logger.warning("No time based drops found") + return [] + + for time_based_drop_data in time_based_drops_data: + time_based_drop, _ = await TimeBasedDrop.objects.aupdate_or_create( + id=time_based_drop_data["id"], + defaults={ + "created_at": time_based_drop_data.get("createdAt"), + "entitlement_limit": time_based_drop_data.get("entitlementLimit"), + "image_asset_url": time_based_drop_data.get("imageAssetURL"), + "is_ios_available": time_based_drop_data.get("isIosAvailable"), + "name": time_based_drop_data.get("name"), + "owner_organization": owner, + "game": game, + "typename": time_based_drop_data.get("__typename"), + }, ) - # Create Reward instances - rewards = [] - for reward_data in campaign_data["rewards"]: - banner_image_data = reward_data["bannerImage"] + benefit_edges_data: list[dict] = time_based_drop_data.get("benefitEdges", []) + for benefit_edge_data in benefit_edges_data: + benefit_data: dict = benefit_edge_data.get("benefit", {}) + benefit, _ = await Benefit.objects.aupdate_or_create( + id=benefit_data["id"], + defaults={ + "created_at": benefit_data.get("createdAt"), + "entitlement_limit": benefit_data.get("entitlementLimit"), + "image_asset_url": benefit_data.get("imageAssetURL"), + "is_ios_available": benefit_data.get("isIosAvailable"), + "name": benefit_data.get("name"), + "owner_organization": owner, + "game": game, + "typename": benefit_data.get("__typename"), + }, + ) + + await BenefitEdge.objects.aupdate_or_create( + benefit=benefit, + defaults={ + "entitlement_limit": benefit_edge_data.get("entitlementLimit"), + "typename": benefit_edge_data.get("__typename"), + }, + ) + + time_based_drops.append(time_based_drop) + + return time_based_drops + + +async def add_or_get_drop_campaign( + drop_campaign_data: dict, + game: Game | None, + owner: Owner | None, +) -> tuple[DropCampaign | None, bool]: + """Handle DropCampaign from JSON data. + + Args: + drop_campaign_data (dict): Drop campaign data from JSON. + game (Game): Game instance. + owner (Owner): Owner instance. + + Returns: + tuple[DropCampaign, bool]: DropCampaign instance and whether it was created. + """ + if not drop_campaign_data: + logger.warning("No drop campaign data found") + return None, False + + drop_campaign, _ = await DropCampaign.objects.aupdate_or_create( + id=drop_campaign_data["id"], + defaults={ + # "allow": allow, # We add this later + "account_link_url": drop_campaign_data.get("accountLinkURL"), + "description": drop_campaign_data.get("description"), + "details_url": drop_campaign_data.get("detailsURL"), + "ends_at": drop_campaign_data.get("endAt"), + # event_based_drops = ???? # TODO(TheLovinator): Find out what this is # noqa: TD003 + "game": game, + "image_url": drop_campaign_data.get("imageURL"), + "name": drop_campaign_data.get("name"), + "owner": owner, + "starts_at": drop_campaign_data.get("startAt"), + "status": drop_campaign_data.get("status"), + # "time_based_drops": time_based_drops, # We add this later + "typename": drop_campaign_data.get("__typename"), + }, + ) + + return drop_campaign, True + + +async def add_or_get_channel(json_data: dict) -> tuple[Channel | None, bool]: + """Add or get Channel from JSON data. + + Args: + json_data (dict): JSON data to add to the database. + + Returns: + tuple[Channel | None, bool]: Channel instance and whether it was created. + """ + if not json_data: + logger.warning("Channel data is missing") + return None, False + + channel, created = await Channel.objects.aupdate_or_create( + id=json_data["id"], + defaults={ + "display_name": json_data.get("displayName"), + "name": json_data.get("name"), + "typename": json_data.get("__typename"), + }, + ) + + return channel, created + + +async def add_drop_campaign(json_data: dict) -> None: + """Add data from JSON to the database.""" + # Get the data from the JSON + user_data: dict = json_data.get("data", {}).get("user", {}) + drop_campaign_data: dict = user_data.get("dropCampaign", {}) + + # Add or get Game + game_data: dict = drop_campaign_data.get("game", {}) + game, _ = await add_or_get_game(json_data=game_data, name=drop_campaign_data.get("name", "Unknown Drop Campaign")) + + # Add or get Owner + owner_data: dict = drop_campaign_data.get("owner", {}) + owner, _ = await add_or_get_owner( + json_data=owner_data, + name=drop_campaign_data.get("name", "Unknown Drop Campaign"), + ) + + # Add or get Allow + allow_data: dict = drop_campaign_data.get("allow", {}) + allow, _ = await add_or_get_allow( + json_data=allow_data, + name=drop_campaign_data.get("name", "Unknown Drop Campaign"), + ) + + # Add channels to Allow + if allow: + channel_data: list[dict] = allow_data.get("channels", []) + for json_channel in channel_data: + channel, _ = await add_or_get_channel(json_channel) + if channel: + await allow.channels.aadd(channel) + + # Add or get TimeBasedDrops + time_based_drops_data = drop_campaign_data.get("timeBasedDrops", []) + time_based_drops: list[TimeBasedDrop] = await add_or_get_time_based_drops(time_based_drops_data, owner, game) + + # Add or get DropCampaign + drop_campaign, _ = await add_or_get_drop_campaign( + drop_campaign_data=drop_campaign_data, + game=game, + owner=owner, + ) + if drop_campaign: + drop_campaign.allow = allow + await drop_campaign.time_based_drops.aset(time_based_drops) + await drop_campaign.asave() + + logger.info("Added Drop Campaign: %s", drop_campaign.name or "Unknown Drop Campaign") + + +async def add_or_get_image(json_data: dict) -> tuple[Image | None, bool]: + """Add or get Image from JSON data. + + Args: + json_data (dict): JSON data to add to the database. + + Returns: + tuple[Image | None, bool]: Image instance and whether it was created. + """ + # TODO(TheLovinator): We should download the image and store it locally # noqa: TD003 + if not json_data: + logger.warning("Image data is missing") + return None, False + + if not json_data.get("image1xURL"): + logger.warning("Image URL is missing") + return None, False + + image, created = await Image.objects.aupdate_or_create( + image1_x_url=json_data.get("image1xURL"), + defaults={ + "typename": json_data.get("__typename"), + }, + ) + + return image, created + + +async def add_or_get_rewards(json_data: dict) -> list[Reward]: + """Add or get Rewards from JSON data. + + Args: + json_data (dict): JSON data to add to the database. + + Returns: + list[Reward]: Reward instances + """ + rewards: list[Reward] = [] + + if not json_data: + logger.warning("No rewards found") + return [] + + if "rewards" not in json_data: + logger.warning("No rewards found") + return [] + + rewards_json: list[dict] = json_data.get("rewards", []) + for reward_data in rewards_json: + # Add or get bannerImage + banner_image_data: dict = reward_data.get("bannerImage", {}) + if banner_image_data: banner_image, _ = await sync_to_async(Image.objects.get_or_create)( image1_x_url=banner_image_data["image1xURL"], defaults={"typename": banner_image_data["__typename"]}, ) - thumbnail_image_data = reward_data["thumbnailImage"] + # Add or get thumbnailImage + thumbnail_image_data = reward_data.get("thumbnailImage", {}) + if thumbnail_image_data: thumbnail_image, _ = await sync_to_async(Image.objects.get_or_create)( image1_x_url=thumbnail_image_data["image1xURL"], defaults={"typename": thumbnail_image_data["__typename"]}, ) - reward, _ = await sync_to_async(Reward.objects.get_or_create)( - id=reward_data["id"], - name=reward_data["name"], - banner_image=banner_image, - thumbnail_image=thumbnail_image, - earnable_until=datetime.fromisoformat(reward_data["earnableUntil"].replace("Z", "+00:00")), - redemption_instructions=reward_data["redemptionInstructions"], - redemption_url=reward_data["redemptionURL"], - typename=reward_data["__typename"], - ) - rewards.append(reward) + # Convert earnableUntil to a datetime object + earnable_until: str | None = reward_data.get("earnableUntil") + earnable_until_date: datetime | None = None + if earnable_until: + earnable_until_date = datetime.fromisoformat(earnable_until.replace("Z", "+00:00")) - # Add or get Unlock Requirements - unlock_requirements_data = campaign_data["unlockRequirements"] - _, _ = await sync_to_async(UnlockRequirements.objects.get_or_create)( - subs_goal=unlock_requirements_data["subsGoal"], + reward, _ = await sync_to_async(Reward.objects.get_or_create)( + id=reward_data["id"], defaults={ - "minute_watched_goal": unlock_requirements_data["minuteWatchedGoal"], - "typename": unlock_requirements_data["__typename"], + "name": reward_data.get("name"), + "banner_image": banner_image, + "thumbnail_image": thumbnail_image, + "earnable_until": earnable_until_date, + "redemption_instructions": reward_data.get("redemptionInstructions"), + "redemption_url": reward_data.get("redemptionURL"), + "typename": reward_data.get("__typename"), }, ) + rewards.append(reward) + + return rewards + + +async def add_or_get_unlock_requirements(json_data: dict) -> tuple[UnlockRequirements | None, bool]: + """Add or get UnlockRequirements from JSON data. + + Args: + json_data (dict): JSON data to add to the database. + + Returns: + tuple[UnlockRequirements | None, bool]: UnlockRequirements instance and whether it was created. + """ + if not json_data: + logger.warning("Unlock Requirements data is missing") + return None, False + + unlock_requirements, created = await UnlockRequirements.objects.aget_or_create( + subs_goal=json_data["subsGoal"], + defaults={ + "minute_watched_goal": json_data["minuteWatchedGoal"], + "typename": json_data["__typename"], + }, + ) + + return unlock_requirements, created + + +async def add_reward_campaign(json_data: dict) -> None: + """Add data from JSON to the database. + + Args: + json_data (dict): JSON data to add to the database. + + Returns: + None: No return value. + """ + campaign_data: list[dict] = json_data["data"]["rewardCampaignsAvailableToUser"] + for campaign in campaign_data: + # Add or get Game + game_data: dict = campaign.get("game", {}) + game, _ = await add_or_get_game(json_data=game_data, name=campaign.get("name", "Unknown Reward Campaign")) + + # Add or get Image + image_data: dict = campaign.get("image", {}) + image, _ = await add_or_get_image(json_data=image_data) + + # Add or get Rewards + rewards: list[Reward] = await add_or_get_rewards(campaign) + + # Add or get Unlock Requirements + unlock_requirements_data: dict = campaign["unlockRequirements"] + unlock_requirements, _ = await add_or_get_unlock_requirements(unlock_requirements_data) # Create Reward Campaign - reward_campaign, _ = await sync_to_async(RewardCampaign.objects.get_or_create)( - id=campaign_data["id"], - name=campaign_data["name"], - brand=campaign_data["brand"], - starts_at=datetime.fromisoformat(campaign_data["startsAt"].replace("Z", "+00:00")), - ends_at=datetime.fromisoformat(campaign_data["endsAt"].replace("Z", "+00:00")), - status=campaign_data["status"], - summary=campaign_data["summary"], - instructions=campaign_data["instructions"], - external_url=campaign_data["externalURL"], - reward_value_url_param=campaign_data["rewardValueURLParam"], - about_url=campaign_data["aboutURL"], - is_sitewide=campaign_data["isSitewide"], - game=game, - image=image, - typename=campaign_data["__typename"], + reward_campaign, _ = await RewardCampaign.objects.aget_or_create( + id=campaign["id"], + defaults={ + "name": campaign.get("name"), + "brand": campaign.get("brand"), + "starts_at": campaign.get("startAt"), + "ends_at": campaign.get("endAt"), + "status": campaign.get("status"), + "summary": campaign.get("summary"), + "instructions": campaign.get("instructions"), + "external_url": campaign.get("externalURL"), + "reward_value_url_params": campaign.get("rewardValueURLParams"), + "about_url": campaign.get("aboutURL"), + "is_sitewide": campaign.get("isSitewide"), + "game": game, + "unlock_requirements": unlock_requirements, + "image": image, + # "rewards": rewards, # We add this later + "typename": campaign.get("__typename"), + }, ) # Add Rewards to the Campaign for reward in rewards: - await sync_to_async(reward_campaign.rewards.add)(reward) + await reward_campaign.rewards.aadd(reward) - await sync_to_async(reward_campaign.save)() + await reward_campaign.asave() class Command(BaseCommand): diff --git a/twitch_app/migrations/0002_alter_rewardcampaign_about_url_and_more.py b/twitch_app/migrations/0002_alter_rewardcampaign_about_url_and_more.py index 0740a05..770e283 100644 --- a/twitch_app/migrations/0002_alter_rewardcampaign_about_url_and_more.py +++ b/twitch_app/migrations/0002_alter_rewardcampaign_about_url_and_more.py @@ -2,83 +2,93 @@ import django.db.models.deletion from django.db import migrations, models +from django.db.migrations.operations.base import Operation class Migration(migrations.Migration): - - dependencies = [ - ('twitch_app', '0001_initial'), + dependencies: list[tuple[str, str]] = [ + ("twitch_app", "0001_initial"), ] - operations = [ + operations: list[Operation] = [ migrations.AlterField( - model_name='rewardcampaign', - name='about_url', + model_name="rewardcampaign", + name="about_url", field=models.URLField(blank=True, null=True), ), migrations.AlterField( - model_name='rewardcampaign', - name='brand', + model_name="rewardcampaign", + name="brand", field=models.TextField(blank=True, null=True), ), migrations.AlterField( - model_name='rewardcampaign', - name='ends_at', + model_name="rewardcampaign", + name="ends_at", field=models.DateTimeField(blank=True, null=True), ), migrations.AlterField( - model_name='rewardcampaign', - name='external_url', + model_name="rewardcampaign", + name="external_url", field=models.URLField(blank=True, null=True), ), migrations.AlterField( - model_name='rewardcampaign', - name='game', - field=models.ForeignKey(null=True, on_delete=django.db.models.deletion.CASCADE, related_name='reward_campaigns', to='twitch_app.game'), + model_name="rewardcampaign", + name="game", + field=models.ForeignKey( + null=True, + on_delete=django.db.models.deletion.CASCADE, + related_name="reward_campaigns", + to="twitch_app.game", + ), ), migrations.AlterField( - model_name='rewardcampaign', - name='image', - field=models.ForeignKey(null=True, on_delete=django.db.models.deletion.CASCADE, related_name='reward_campaigns', to='twitch_app.image'), + model_name="rewardcampaign", + name="image", + field=models.ForeignKey( + null=True, + on_delete=django.db.models.deletion.CASCADE, + related_name="reward_campaigns", + to="twitch_app.image", + ), ), migrations.AlterField( - model_name='rewardcampaign', - name='instructions', + model_name="rewardcampaign", + name="instructions", field=models.TextField(blank=True, null=True), ), migrations.AlterField( - model_name='rewardcampaign', - name='is_sitewide', + model_name="rewardcampaign", + name="is_sitewide", field=models.BooleanField(blank=True, null=True), ), migrations.AlterField( - model_name='rewardcampaign', - name='name', + model_name="rewardcampaign", + name="name", field=models.TextField(blank=True, null=True), ), migrations.AlterField( - model_name='rewardcampaign', - name='reward_value_url_param', + model_name="rewardcampaign", + name="reward_value_url_param", field=models.TextField(blank=True, null=True), ), migrations.AlterField( - model_name='rewardcampaign', - name='starts_at', + model_name="rewardcampaign", + name="starts_at", field=models.DateTimeField(blank=True, null=True), ), migrations.AlterField( - model_name='rewardcampaign', - name='status', + model_name="rewardcampaign", + name="status", field=models.TextField(blank=True, null=True), ), migrations.AlterField( - model_name='rewardcampaign', - name='summary', + model_name="rewardcampaign", + name="summary", field=models.TextField(blank=True, null=True), ), migrations.AlterField( - model_name='rewardcampaign', - name='typename', + model_name="rewardcampaign", + name="typename", field=models.TextField(blank=True, null=True), ), ] diff --git a/twitch_app/migrations/0003_alter_game_display_name_alter_game_slug_and_more.py b/twitch_app/migrations/0003_alter_game_display_name_alter_game_slug_and_more.py index 5ea602d..29138f0 100644 --- a/twitch_app/migrations/0003_alter_game_display_name_alter_game_slug_and_more.py +++ b/twitch_app/migrations/0003_alter_game_display_name_alter_game_slug_and_more.py @@ -2,108 +2,118 @@ import django.db.models.deletion from django.db import migrations, models +from django.db.migrations.operations.base import Operation class Migration(migrations.Migration): - - dependencies = [ - ('twitch_app', '0002_alter_rewardcampaign_about_url_and_more'), + dependencies: list[tuple[str, str]] = [ + ("twitch_app", "0002_alter_rewardcampaign_about_url_and_more"), ] - operations = [ + operations: list[Operation] = [ migrations.AlterField( - model_name='game', - name='display_name', + model_name="game", + name="display_name", field=models.TextField(blank=True, null=True), ), migrations.AlterField( - model_name='game', - name='slug', + model_name="game", + name="slug", field=models.TextField(blank=True, null=True), ), migrations.AlterField( - model_name='game', - name='typename', + model_name="game", + name="typename", field=models.TextField(blank=True, null=True), ), migrations.AlterField( - model_name='image', - name='image1_x_url', + model_name="image", + name="image1_x_url", field=models.URLField(blank=True, null=True), ), migrations.AlterField( - model_name='image', - name='typename', + model_name="image", + name="typename", field=models.TextField(blank=True, null=True), ), migrations.AlterField( - model_name='reward', - name='banner_image', - field=models.ForeignKey(null=True, on_delete=django.db.models.deletion.CASCADE, related_name='banner_rewards', to='twitch_app.image'), + model_name="reward", + name="banner_image", + field=models.ForeignKey( + null=True, + on_delete=django.db.models.deletion.CASCADE, + related_name="banner_rewards", + to="twitch_app.image", + ), ), migrations.AlterField( - model_name='reward', - name='earnable_until', + model_name="reward", + name="earnable_until", field=models.DateTimeField(null=True), ), migrations.AlterField( - model_name='reward', - name='name', + model_name="reward", + name="name", field=models.TextField(blank=True, null=True), ), migrations.AlterField( - model_name='reward', - name='redemption_instructions', + model_name="reward", + name="redemption_instructions", field=models.TextField(blank=True, null=True), ), migrations.AlterField( - model_name='reward', - name='redemption_url', + model_name="reward", + name="redemption_url", field=models.URLField(blank=True, null=True), ), migrations.AlterField( - model_name='reward', - name='thumbnail_image', - field=models.ForeignKey(null=True, on_delete=django.db.models.deletion.CASCADE, related_name='thumbnail_rewards', to='twitch_app.image'), + model_name="reward", + name="thumbnail_image", + field=models.ForeignKey( + null=True, + on_delete=django.db.models.deletion.CASCADE, + related_name="thumbnail_rewards", + to="twitch_app.image", + ), ), migrations.AlterField( - model_name='reward', - name='typename', + model_name="reward", + name="typename", field=models.TextField(blank=True, null=True), ), migrations.AlterField( - model_name='rewardcampaign', - name='ends_at', + model_name="rewardcampaign", + name="ends_at", field=models.DateTimeField(null=True), ), migrations.AlterField( - model_name='rewardcampaign', - name='is_sitewide', + model_name="rewardcampaign", + name="is_sitewide", field=models.BooleanField(null=True), ), migrations.AlterField( - model_name='rewardcampaign', - name='rewards', - field=models.ManyToManyField(related_name='reward_campaigns', to='twitch_app.reward'), + model_name="rewardcampaign", + name="rewards", + field=models.ManyToManyField(related_name="reward_campaigns", to="twitch_app.reward"), ), migrations.AlterField( - model_name='rewardcampaign', - name='starts_at', + model_name="rewardcampaign", + name="starts_at", field=models.DateTimeField(null=True), ), migrations.AlterField( - model_name='unlockrequirements', - name='minute_watched_goal', + model_name="unlockrequirements", + name="minute_watched_goal", field=models.IntegerField(null=True), ), migrations.AlterField( - model_name='unlockrequirements', - name='subs_goal', + model_name="unlockrequirements", + name="subs_goal", field=models.IntegerField(null=True), ), migrations.AlterField( - model_name='unlockrequirements', - name='typename', + model_name="unlockrequirements", + name="typename", field=models.TextField(blank=True, null=True), ), ] diff --git a/twitch_app/migrations/0004_alter_reward_id_alter_rewardcampaign_id.py b/twitch_app/migrations/0004_alter_reward_id_alter_rewardcampaign_id.py index 6e4bb69..bfebf6b 100644 --- a/twitch_app/migrations/0004_alter_reward_id_alter_rewardcampaign_id.py +++ b/twitch_app/migrations/0004_alter_reward_id_alter_rewardcampaign_id.py @@ -1,23 +1,23 @@ # Generated by Django 5.1rc1 on 2024-07-30 23:44 from django.db import migrations, models +from django.db.migrations.operations.base import Operation class Migration(migrations.Migration): - - dependencies = [ - ('twitch_app', '0003_alter_game_display_name_alter_game_slug_and_more'), + dependencies: list[tuple[str, str]] = [ + ("twitch_app", "0003_alter_game_display_name_alter_game_slug_and_more"), ] - operations = [ + operations: list[Operation] = [ migrations.AlterField( - model_name='reward', - name='id', + model_name="reward", + name="id", field=models.UUIDField(primary_key=True, serialize=False), ), migrations.AlterField( - model_name='rewardcampaign', - name='id', + model_name="rewardcampaign", + name="id", field=models.UUIDField(primary_key=True, serialize=False), ), ] diff --git a/twitch_app/migrations/0005_channel_owner_and_more.py b/twitch_app/migrations/0005_channel_owner_and_more.py new file mode 100644 index 0000000..8e082b9 --- /dev/null +++ b/twitch_app/migrations/0005_channel_owner_and_more.py @@ -0,0 +1,174 @@ +# Generated by Django 5.1rc1 on 2024-08-01 01:00 + +import django.db.models.deletion +from django.db import migrations, models +from django.db.migrations.operations.base import Operation + + +class Migration(migrations.Migration): + dependencies: list[tuple[str, str]] = [ + ("twitch_app", "0004_alter_reward_id_alter_rewardcampaign_id"), + ] + + operations: list[Operation] = [ + migrations.CreateModel( + name="Channel", + fields=[ + ("id", models.PositiveBigIntegerField(primary_key=True, serialize=False)), + ("display_name", models.TextField(blank=True, null=True)), + ("name", models.TextField(blank=True, null=True)), + ("typename", models.TextField(blank=True, null=True)), + ], + ), + migrations.CreateModel( + name="Owner", + fields=[ + ("id", models.PositiveBigIntegerField(primary_key=True, serialize=False)), + ("slug", models.TextField(blank=True, null=True)), + ("display_name", models.TextField(blank=True, null=True)), + ("typename", models.TextField(blank=True, null=True)), + ], + ), + migrations.AlterField( + model_name="unlockrequirements", + name="minute_watched_goal", + field=models.PositiveBigIntegerField(null=True), + ), + migrations.AlterField( + model_name="unlockrequirements", + name="subs_goal", + field=models.PositiveBigIntegerField(null=True), + ), + migrations.CreateModel( + name="Benefit", + fields=[ + ("id", models.TextField(primary_key=True, serialize=False)), + ("created_at", models.DateTimeField(null=True)), + ("entitlement_limit", models.PositiveBigIntegerField(null=True)), + ("image_asset_url", models.URLField(blank=True, null=True)), + ("is_ios_available", models.BooleanField(null=True)), + ("name", models.TextField(blank=True, null=True)), + ("typename", models.TextField(blank=True, null=True)), + ( + "game", + models.ForeignKey( + null=True, + on_delete=django.db.models.deletion.CASCADE, + related_name="benefits", + to="twitch_app.game", + ), + ), + ( + "owner_organization", + models.ForeignKey( + null=True, + on_delete=django.db.models.deletion.CASCADE, + related_name="benefits", + to="twitch_app.owner", + ), + ), + ], + ), + migrations.CreateModel( + name="BenefitEdge", + fields=[ + ("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")), + ("entitlement_limit", models.PositiveBigIntegerField(null=True)), + ("typename", models.TextField(blank=True, null=True)), + ( + "benefit", + models.ForeignKey( + null=True, + on_delete=django.db.models.deletion.CASCADE, + related_name="benefit_edges", + to="twitch_app.benefit", + ), + ), + ], + ), + migrations.CreateModel( + name="Allow", + fields=[ + ("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")), + ("is_enabled", models.BooleanField(default=True)), + ("typename", models.TextField(blank=True, null=True)), + ("channels", models.ManyToManyField(related_name="allow", to="twitch_app.channel")), + ], + ), + migrations.CreateModel( + name="TimeBasedDrop", + fields=[ + ("id", models.TextField(primary_key=True, serialize=False)), + ("created_at", models.DateTimeField(null=True)), + ("entitlement_limit", models.PositiveBigIntegerField(null=True)), + ("image_asset_url", models.URLField(blank=True, null=True)), + ("is_ios_available", models.BooleanField(null=True)), + ("name", models.TextField(blank=True, null=True)), + ("typename", models.TextField(blank=True, null=True)), + ( + "game", + models.ForeignKey( + null=True, + on_delete=django.db.models.deletion.CASCADE, + related_name="time_based_drops", + to="twitch_app.game", + ), + ), + ( + "owner_organization", + models.ForeignKey( + null=True, + on_delete=django.db.models.deletion.CASCADE, + related_name="time_based_drops", + to="twitch_app.owner", + ), + ), + ], + ), + migrations.CreateModel( + name="DropCampaign", + fields=[ + ("id", models.TextField(primary_key=True, serialize=False)), + ("account_link_url", models.URLField(blank=True, null=True)), + ("description", models.TextField(blank=True, null=True)), + ("details_url", models.URLField(blank=True, null=True)), + ("ends_at", models.DateTimeField(null=True)), + ("image_url", models.URLField(blank=True, null=True)), + ("name", models.TextField(blank=True, null=True)), + ("starts_at", models.DateTimeField(null=True)), + ("status", models.TextField(blank=True, null=True)), + ("typename", models.TextField(blank=True, null=True)), + ( + "allow", + models.ForeignKey( + null=True, + on_delete=django.db.models.deletion.CASCADE, + related_name="drop_campaigns", + to="twitch_app.allow", + ), + ), + ( + "game", + models.ForeignKey( + null=True, + on_delete=django.db.models.deletion.CASCADE, + related_name="drop_campaigns", + to="twitch_app.game", + ), + ), + ( + "owner", + models.ForeignKey( + null=True, + on_delete=django.db.models.deletion.CASCADE, + related_name="drop_campaigns", + to="twitch_app.owner", + ), + ), + ( + "time_based_drops", + models.ManyToManyField(related_name="drop_campaigns", to="twitch_app.timebaseddrop"), + ), + ], + ), + ] diff --git a/twitch_app/models.py b/twitch_app/models.py index 4fcc9a7..4aa3551 100644 --- a/twitch_app/models.py +++ b/twitch_app/models.py @@ -4,6 +4,8 @@ from django.db import models class Game(models.Model): """The game that the reward is for. + Used for reward campaigns (buy subs) and drop campaigns (watch games). + Attributes: id (int): The primary key of the game. slug (str): The slug identifier of the game. @@ -56,7 +58,7 @@ class Reward(models.Model): """The actual reward you get when you complete the requirements. Attributes: - id (UUID): The primary key of the reward. + id (str): The primary key of the reward. name (str): The name of the reward. banner_image (Image): The banner image associated with the reward. thumbnail_image (Image): The thumbnail image associated with the reward. @@ -84,7 +86,7 @@ class Reward(models.Model): } """ - id = models.UUIDField(primary_key=True) + id = models.TextField(primary_key=True) name = models.TextField(null=True, blank=True) banner_image = models.ForeignKey(Image, related_name="banner_rewards", on_delete=models.CASCADE, null=True) thumbnail_image = models.ForeignKey(Image, related_name="thumbnail_rewards", on_delete=models.CASCADE, null=True) @@ -113,8 +115,8 @@ class UnlockRequirements(models.Model): } """ - subs_goal = models.IntegerField(null=True) - minute_watched_goal = models.IntegerField(null=True) + subs_goal = models.PositiveBigIntegerField(null=True) + minute_watched_goal = models.PositiveBigIntegerField(null=True) typename = models.TextField(null=True, blank=True) def __str__(self) -> str: @@ -125,7 +127,7 @@ class RewardCampaign(models.Model): """Represents a reward campaign. Attributes: - id (UUID): The primary key of the reward campaign. + id (str): The primary key of the reward campaign. name (str): The name of the reward campaign. brand (str): The brand associated with the campaign. starts_at (datetime): The start date and time of the campaign. @@ -162,6 +164,11 @@ class RewardCampaign(models.Model): "displayName": "XDefiant", "__typename": "Game" }, + "unlockRequirements": { + "subsGoal": 2, + "minuteWatchedGoal": 0, + "__typename": "QuestRewardUnlockRequirements" + }, "image": { "image1xURL": "https://static-cdn.jtvnw.net/twitch-quests-assets/CAMPAIGN/quests_xdefiant_q3_2024/campaign.png", "__typename": "RewardCampaignImageSet" @@ -188,7 +195,7 @@ class RewardCampaign(models.Model): } """ # noqa: E501 - id = models.UUIDField(primary_key=True) + id = models.TextField(primary_key=True) name = models.TextField(null=True, blank=True) brand = models.TextField(null=True, blank=True) starts_at = models.DateTimeField(null=True) @@ -201,9 +208,304 @@ class RewardCampaign(models.Model): about_url = models.URLField(null=True, blank=True) is_sitewide = models.BooleanField(null=True) game = models.ForeignKey(Game, on_delete=models.CASCADE, related_name="reward_campaigns", null=True) + unlock_requirements = models.ForeignKey( + UnlockRequirements, + on_delete=models.CASCADE, + related_name="reward_campaigns", + null=True, + ) image = models.ForeignKey(Image, on_delete=models.CASCADE, related_name="reward_campaigns", null=True) rewards = models.ManyToManyField(Reward, related_name="reward_campaigns") typename = models.TextField(null=True, blank=True) def __str__(self) -> str: return self.name or "Unknown" + + +class Channel(models.Model): + """Represents a Twitch channel. + + Attributes: + id (int): The primary key of the channel. + display_name (str): The display name of the channel. + name (str): The name of the channel. + typename (str): The type name of the object, typically "Channel". + + JSON example: + { + "id": "25254906", + "displayName": "Stresss", + "name": "stresss", + "__typename": "Channel" + } + """ + + id = models.PositiveBigIntegerField(primary_key=True) + display_name = models.TextField(null=True, blank=True) + name = models.TextField(null=True, blank=True) + typename = models.TextField(null=True, blank=True) + + def __str__(self) -> str: + return self.display_name or "Unknown" + + def get_twitch_url(self) -> str: + return f"https://www.twitch.tv/{self.name}" + + +class Allow(models.Model): + """List of channels that you can watch to earn rewards. + + Attributes: + channels (ManyToManyField): The channels that you can watch to earn rewards. + is_enabled (bool): Indicates if the channel is enabled. + typename (str): The type name of the object, typically "RewardCampaignChannelAllow". + + JSON example: + "allow": { + "channels": [ + { + "id": "25254906", + "displayName": "Stresss", + "name": "stresss", + "__typename": "Channel" + } + ], + "isEnabled": false, + "__typename": "DropCampaignACL" + }, + """ + + channels = models.ManyToManyField(Channel, related_name="allow") + is_enabled = models.BooleanField(default=True) + typename = models.TextField(null=True, blank=True) + + def __str__(self) -> str: + return f"{self.channels.count()} channels" + + +class Owner(models.Model): + """Represents the owner of the reward campaign. + + Attributes: + id (int): The primary key of the owner. + slug (str): The slug identifier of the owner. + display_name (str): The display name of the owner. + typename (str): The type name of the object, typically "Organization". + + JSON example: + "game": { + "id": "491487", + "slug": "dead-by-daylight", + "displayName": "Dead by Daylight", + "__typename": "Game" + }," + """ + + id = models.PositiveBigIntegerField(primary_key=True) + slug = models.TextField(null=True, blank=True) + display_name = models.TextField(null=True, blank=True) + typename = models.TextField(null=True, blank=True) + + def __str__(self) -> str: + return self.display_name or "Unknown" + + def get_twitch_url(self) -> str: + return f"https://www.twitch.tv/{self.slug}" + + +class Benefit(models.Model): + """Represents a benefit that you can earn. + + Attributes: + id (int): The primary key of the benefit. + created_at (datetime): The date and time the benefit was created. + entitlement_limit (int): The limit of entitlement. + game (Game): The game associated with the benefit. + image_asset_url (str): URL to the image asset. + is_ios_available (bool): Indicates if the benefit is available on iOS. + name (str): The name of the benefit. + owner_organization (Owner): The owner organization of the benefit. + typename (str): The type name of the object, typically "Benefit". + + JSON example: + "benefit": { + "id": "6da09649-1fda-4446-a061-cacd8e21b886_CUSTOM_ID_S29_Torso008_01", + "createdAt": "2024-07-09T12:57:31.072Z", + "entitlementLimit": 1, + "game": { + "id": "491487", + "name": "Dead by Daylight", + "__typename": "Game" + }, + "imageAssetURL": "https://static-cdn.jtvnw.net/twitch-quests-assets/REWARD/ed4a7829-cc2b-44d3-90a4-f73ef7d8d636.png", + "isIosAvailable": false, + "name": "Unwanted Attention", + "ownerOrganization": { + "id": "6da09649-1fda-4446-a061-cacd8e21b886", + "name": "Behaviour Interactive Inc.", + "__typename": "Organization" + }, + "__typename": "DropBenefit" + } + """ + + id = models.TextField(primary_key=True) + created_at = models.DateTimeField(null=True) + entitlement_limit = models.PositiveBigIntegerField(null=True) + game = models.ForeignKey(Game, on_delete=models.CASCADE, related_name="benefits", null=True) + image_asset_url = models.URLField(null=True, blank=True) + is_ios_available = models.BooleanField(null=True) + name = models.TextField(null=True, blank=True) + owner_organization = models.ForeignKey(Owner, on_delete=models.CASCADE, related_name="benefits", null=True) + typename = models.TextField(null=True, blank=True) + + def __str__(self) -> str: + return self.name or "Unknown" + + +class BenefitEdge(models.Model): + """Represents a benefit edge. + + Attributes: + benefit (Benefit): The benefit associated with the edge. + entitlement_limit (int): The limit of entitlement. + typename (str): The type name of the object, typically "DropBenefitEdge". + + + JSON example: + "benefitEdges": [ + { + "benefit": { + "id": "6da09649-1fda-4446-a061-cacd8e21b886_CUSTOM_ID_S29_Torso008_01", + "createdAt": "2024-07-09T12:57:31.072Z", + "entitlementLimit": 1, + "game": { + "id": "491487", + "name": "Dead by Daylight", + "__typename": "Game" + }, + "imageAssetURL": "https://static-cdn.jtvnw.net/twitch-quests-assets/REWARD/ed4a7829-cc2b-44d3-90a4-f73ef7d8d636.png", + "isIosAvailable": false, + "name": "Unwanted Attention", + "ownerOrganization": { + "id": "6da09649-1fda-4446-a061-cacd8e21b886", + "name": "Behaviour Interactive Inc.", + "__typename": "Organization" + }, + "__typename": "DropBenefit" + }, + "entitlementLimit": 1, + "__typename": "DropBenefitEdge" + } + ], + """ + + benefit = models.ForeignKey(Benefit, on_delete=models.CASCADE, related_name="benefit_edges", null=True) + entitlement_limit = models.PositiveBigIntegerField(null=True) + typename = models.TextField(null=True, blank=True) + + def __str__(self) -> str: + benefit_name: str | None = self.benefit.name if self.benefit else "Unknown" + return f"{benefit_name} - {self.entitlement_limit}" + + +class TimeBasedDrop(models.Model): + """Represents a time-based drop. + + Attributes: + id (int): The primary key of the time-based drop. + name (str): The name of the time-based drop. + starts_at (datetime): The start date and time of the drop. + ends_at (datetime): The end date and time of the drop. + typename (str): The type name of the object, typically "TimeBasedDrop". + + JSON example: + { + "id": "0ebeff68-3df3-11ef-b15b-0a58a9feac02", + "requiredSubs": 0, + "benefitEdges": [ + { + "benefit": { + "id": "6da09649-1fda-4446-a061-cacd8e21b886_CUSTOM_ID_S29_Legs008_01", + "createdAt": "2024-07-09T12:58:03.654Z", + "entitlementLimit": 1, + "game": { + "id": "491487", + "name": "Dead by Daylight", + "__typename": "Game" + }, + "imageAssetURL": "https://static-cdn.jtvnw.net/twitch-quests-assets/REWARD/f46acdf5-9515-41eb-805e-86956db0a9e9.png", + "isIosAvailable": false, + "name": "Back Home", + "ownerOrganization": { + "id": "6da09649-1fda-4446-a061-cacd8e21b886", + "name": "Behaviour Interactive Inc.", + "__typename": "Organization" + }, + "__typename": "DropBenefit" + }, + "entitlementLimit": 1, + "__typename": "DropBenefitEdge" + } + ], + "endAt": "2024-07-30T14:59:59.999Z", + "name": "Back Home", + "preconditionDrops": null, + "requiredMinutesWatched": 360, + "startAt": "2024-07-16T15:00:00Z", + "__typename": "TimeBasedDrop" + }, + """ + + id = models.TextField(primary_key=True) + created_at = models.DateTimeField(null=True) + entitlement_limit = models.PositiveBigIntegerField(null=True) + game = models.ForeignKey(Game, on_delete=models.CASCADE, related_name="time_based_drops", null=True) + image_asset_url = models.URLField(null=True, blank=True) + is_ios_available = models.BooleanField(null=True) + name = models.TextField(null=True, blank=True) + owner_organization = models.ForeignKey(Owner, on_delete=models.CASCADE, related_name="time_based_drops", null=True) + typename = models.TextField(null=True, blank=True) + + def __str__(self) -> str: + return self.name or "Unknown" + + +class DropCampaign(models.Model): + """Represents a drop campaign. + + Attributes: + id (int): The primary key of the drop campaign. + allow (Allow): The channels that you can watch to earn rewards. + account_link_url (str): URL to link your account. + description (str): The description of the drop campaign. + details_url (str): URL with more details about the drop campaign. + ends_at (datetime): The end date and time of the drop campaign. + game (Game): The game associated with the drop campaign. + image_url (str): URL to the image associated with the drop campaign. + name (str): The name of the drop campaign. + owner (Owner): The owner of the drop campaign. + starts_at (datetime): The start date and time of the drop campaign. + status (str): The status of the drop campaign. + time_based_drops (ManyToManyField): The time-based drops associated with the campaign. + typename (str): The type name of the object, typically "DropCampaign". + """ + + id = models.TextField(primary_key=True) + allow = models.ForeignKey(Allow, on_delete=models.CASCADE, related_name="drop_campaigns", null=True) + account_link_url = models.URLField(null=True, blank=True) + description = models.TextField(null=True, blank=True) + details_url = models.URLField(null=True, blank=True) + ends_at = models.DateTimeField(null=True) + # event_based_drops = ???? + game = models.ForeignKey(Game, on_delete=models.CASCADE, related_name="drop_campaigns", null=True) + image_url = models.URLField(null=True, blank=True) + name = models.TextField(null=True, blank=True) + owner = models.ForeignKey(Owner, on_delete=models.CASCADE, related_name="drop_campaigns", null=True) + starts_at = models.DateTimeField(null=True) + status = models.TextField(null=True, blank=True) + time_based_drops = models.ManyToManyField(TimeBasedDrop, related_name="drop_campaigns", null=True) + typename = models.TextField(null=True, blank=True) + + def __str__(self) -> str: + return self.name or "Unknown"