diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 93ba618..414816f 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -43,7 +43,7 @@ repos: # An extremely fast Python linter and formatter. - repo: https://github.com/astral-sh/ruff-pre-commit - rev: v0.5.5 + rev: v0.5.7 hooks: - id: ruff-format - id: ruff diff --git a/.vscode/settings.json b/.vscode/settings.json index 8df0627..8f4e84b 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -17,6 +17,7 @@ "Stresss", "ttvdrops", "ulimits", + "Valair", "xdefiant" ] } diff --git a/config/urls.py b/config/urls.py deleted file mode 100644 index 3a1ca6f..0000000 --- a/config/urls.py +++ /dev/null @@ -1,14 +0,0 @@ -import logging - -from debug_toolbar.toolbar import debug_toolbar_urls -from django.urls import URLPattern, include, path -from django.urls.resolvers import URLResolver - -logger: logging.Logger = logging.getLogger(__name__) - -app_name: str = "config" - -urlpatterns: list[URLPattern | URLResolver] = [ - path(route="", view=include(arg="core.urls")), - *debug_toolbar_urls(), -] diff --git a/core/apps.py b/core/apps.py index c0ce093..8b90062 100644 --- a/core/apps.py +++ b/core/apps.py @@ -2,5 +2,5 @@ from django.apps import AppConfig class CoreConfig(AppConfig): - default_auto_field = "django.db.models.BigAutoField" + default_auto_field: str = "django.db.models.BigAutoField" name = "core" diff --git a/config/__init__.py b/core/management/__init__.py similarity index 100% rename from config/__init__.py rename to core/management/__init__.py diff --git a/twitch_app/__init__.py b/core/management/commands/__init__.py similarity index 100% rename from twitch_app/__init__.py rename to core/management/commands/__init__.py diff --git a/twitch_app/management/commands/scrape_twitch.py b/core/management/commands/scrape_twitch.py similarity index 77% rename from twitch_app/management/commands/scrape_twitch.py rename to core/management/commands/scrape_twitch.py index 75a578b..d3ed386 100644 --- a/twitch_app/management/commands/scrape_twitch.py +++ b/core/management/commands/scrape_twitch.py @@ -1,4 +1,5 @@ import asyncio +import json import logging import typing from datetime import datetime @@ -11,7 +12,7 @@ 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 ( +from core.models import ( Allow, Benefit, BenefitEdge, @@ -28,37 +29,38 @@ from twitch_app.models import ( if TYPE_CHECKING: from playwright.async_api._generated import BrowserContext, Page -import json -# Where to store the Chrome profile -data_dir = Path( - user_data_dir( - appname="TTVDrops", - appauthor="TheLovinator", - roaming=True, - ensure_exists=True, - ), -) - -if not data_dir: - msg = "DATA_DIR is not set in settings.py" - raise ValueError(msg) logger: logging.Logger = logging.getLogger(__name__) -async def add_or_get_game(json_data: dict, name: str) -> tuple[Game | None, bool]: +def get_data_dir() -> Path: + """Get the data directory. + + Returns: + Path: The data directory. + """ + return Path( + user_data_dir( + appname="TTVDrops", + appauthor="TheLovinator", + roaming=True, + ensure_exists=True, + ), + ) + + +async def add_or_get_game(json_data: dict | None) -> tuple[Game | None, bool]: """Add or get Game from JSON data. 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) + logger.warning("Couldn't find game data, probably a reward campaign?") return None, False game, created = await Game.objects.aupdate_or_create( @@ -71,21 +73,23 @@ async def add_or_get_game(json_data: dict, name: str) -> tuple[Game | None, bool }, ) + if created: + logger.info("Found new game: %s", game.display_name or "Unknown Game") + return game, created -async def add_or_get_owner(json_data: dict, name: str) -> tuple[Owner | None, bool]: +async def add_or_get_owner(json_data: dict | None) -> 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) + logger.warning("No owner data provided") return None, False owner, created = await Owner.objects.aupdate_or_create( @@ -99,18 +103,17 @@ async def add_or_get_owner(json_data: dict, name: str) -> tuple[Owner | None, bo return owner, created -async def add_or_get_allow(json_data: dict, name: str) -> tuple[Allow | None, bool]: +async def add_or_get_allow(json_data: dict | None) -> 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) + logger.warning("No allow data provided") return None, False allow, created = await Allow.objects.aupdate_or_create( @@ -133,14 +136,15 @@ async def add_or_get_time_based_drops( 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 [] + logger.warning("No time based drops data provided") + return time_based_drops for time_based_drop_data in time_based_drops_data: time_based_drop, _ = await TimeBasedDrop.objects.aupdate_or_create( @@ -188,7 +192,7 @@ async def add_or_get_time_based_drops( async def add_or_get_drop_campaign( - drop_campaign_data: dict, + drop_campaign_data: dict | None, game: Game | None, owner: Owner | None, ) -> tuple[DropCampaign | None, bool]: @@ -199,11 +203,12 @@ async def add_or_get_drop_campaign( 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") + logger.warning("No drop campaign data provided") return None, False if drop_campaign_data.get("__typename") != "Game": @@ -232,17 +237,18 @@ async def add_or_get_drop_campaign( return drop_campaign, True -async def add_or_get_channel(json_data: dict) -> tuple[Channel | None, bool]: +async def add_or_get_channel(json_data: dict | None) -> 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") + logger.warning("No channel data provided") return None, False channel, created = await Channel.objects.aupdate_or_create( @@ -257,29 +263,31 @@ async def add_or_get_channel(json_data: dict) -> tuple[Channel | None, bool]: return channel, created -async def add_drop_campaign(json_data: dict) -> None: - """Add data from JSON to the database.""" +async def add_drop_campaign(json_data: dict | None) -> None: + """Add data from JSON to the database. + + Args: + json_data (dict): JSON data to add to the database. + """ + if not json_data: + logger.warning("No JSON data provided") + return + # 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")) + game, _ = await add_or_get_game(json_data=game_data) # 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"), - ) + owner, _ = await add_or_get_owner(json_data=owner_data) # 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"), - ) + allow, _ = await add_or_get_allow(json_data=allow_data) # Add channels to Allow if allow: @@ -309,7 +317,7 @@ async def add_drop_campaign(json_data: dict) -> None: 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]: +async def add_or_get_image(json_data: dict | None) -> tuple[Image | None, bool]: """Add or get Image from JSON data. Args: @@ -337,7 +345,7 @@ async def add_or_get_image(json_data: dict) -> tuple[Image | None, bool]: return image, created -async def add_or_get_rewards(json_data: dict) -> list[Reward]: +async def add_or_get_rewards(json_data: dict | None) -> list[Reward]: """Add or get Rewards from JSON data. Args: @@ -397,17 +405,19 @@ async def add_or_get_rewards(json_data: dict) -> list[Reward]: return rewards -async def add_or_get_unlock_requirements(json_data: dict) -> tuple[UnlockRequirements | None, bool]: +async def add_or_get_unlock_requirements(json_data: dict | None) -> 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") + logger.warning("No unlock requirements data provided") return None, False unlock_requirements, created = await UnlockRequirements.objects.aget_or_create( @@ -421,20 +431,21 @@ async def add_or_get_unlock_requirements(json_data: dict) -> tuple[UnlockRequire return unlock_requirements, created -async def add_reward_campaign(json_data: dict) -> None: +async def add_reward_campaign(json_data: dict | None) -> None: """Add data from JSON to the database. Args: json_data (dict): JSON data to add to the database. - - Returns: - None: No return value. """ + if not json_data: + logger.warning("No JSON data provided") + return + 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")) + game, _ = await add_or_get_game(json_data=game_data) # Add or get Image image_data: dict = campaign.get("image", {}) @@ -477,30 +488,81 @@ async def add_reward_campaign(json_data: dict) -> None: await reward_campaign.asave() +def get_profile_dir() -> Path: + """Get the profile directory for the browser. + + Returns: + Path: The profile directory. + """ + profile_dir: Path = Path(get_data_dir() / "chrome-profile") + profile_dir.mkdir(parents=True, exist_ok=True) + logger.debug("Launching Chrome browser with user data directory: %s", profile_dir) + return profile_dir + + +def save_json(campaign: dict, dir_name: str) -> None: + """Save JSON data to a file. + + Args: + campaign (dict): The JSON data to save. + dir_name (Path): The directory to save the JSON data to. + """ + save_dir: Path = Path(dir_name) + save_dir.mkdir(parents=True, exist_ok=True) + + # File name is the hash of the JSON data + file_name: str = f"{hash(json.dumps(campaign))}.json" + + with Path(save_dir / file_name).open(mode="w", encoding="utf-8") as f: + json.dump(campaign, f, indent=4) + + +async def process_json_data(num: int, campaign: dict | None, json_data: list[dict] | None) -> None: + """Process JSON data. + + Args: + num (int): The number of the JSON data. + campaign (dict): The JSON data to process. + json_data (list[dict]): The list of JSON + """ + if not json_data: + logger.warning("No JSON data provided") + return + + logger.info("Processing JSON %d of %d", num, len(json_data)) + if not isinstance(campaign, dict): + logger.warning("Campaign is not a dictionary") + return + + if "rewardCampaignsAvailableToUser" in campaign["data"]: + save_json(campaign, "reward_campaigns") + await add_reward_campaign(campaign) + + if "dropCampaign" in campaign.get("data", {}).get("user", {}): + if not campaign["data"]["user"]["dropCampaign"]: + logger.warning("No drop campaign found") + return + + save_json(campaign, "drop_campaign") + await add_drop_campaign(campaign) + + if "dropCampaigns" in campaign.get("data", {}).get("user", {}): + for drop_campaign in campaign["data"]["user"]["dropCampaigns"]: + save_json(campaign, "drop_campaigns") + await add_drop_campaign(drop_campaign) + + class Command(BaseCommand): help = "Scrape Twitch Drops Campaigns with login using Firefox" - async def run( # noqa: PLR6301, C901 - self, - playwright: Playwright, - ) -> list[dict[str, typing.Any]]: - args: list[str] = [] - - # disable navigator.webdriver:true flag - args.append("--disable-blink-features=AutomationControlled") - - profile_dir: Path = Path(data_dir / "chrome-profile") - profile_dir.mkdir(parents=True, exist_ok=True) - logger.debug( - "Launching Chrome browser with user data directory: %s", - profile_dir, - ) - + @staticmethod + async def run(playwright: Playwright) -> list[dict[str, typing.Any]]: + profile_dir: Path = get_profile_dir() browser: BrowserContext = await playwright.chromium.launch_persistent_context( channel="chrome", user_data_dir=profile_dir, headless=False, - args=args, + args=["--disable-blink-features=AutomationControlled"], ) logger.debug("Launched Chrome browser") @@ -540,47 +602,10 @@ class Command(BaseCommand): await page.wait_for_load_state("networkidle") logger.debug("Page loaded. Scraping data...") - # Wait 5 seconds for the page to load - # await asyncio.sleep(5) - await browser.close() for num, campaign in enumerate(json_data, start=1): - logger.info("Processing JSON %d of %d", num, len(json_data)) - if not isinstance(campaign, dict): - continue - - if "rewardCampaignsAvailableToUser" in campaign["data"]: - # Save to folder named "reward_campaigns" - dir_name: Path = Path("reward_campaigns") - dir_name.mkdir(parents=True, exist_ok=True) - with open(file=Path(dir_name / f"reward_campaign_{num}.json"), mode="w", encoding="utf-8") as f: - json.dump(campaign, f, indent=4) - - await add_reward_campaign(campaign) - - if "dropCampaign" in campaign.get("data", {}).get("user", {}): - if not campaign["data"]["user"]["dropCampaign"]: - logger.warning("No drop campaign found") - continue - - # Save to folder named "drop_campaign" - dir_name: Path = Path("drop_campaign") - dir_name.mkdir(parents=True, exist_ok=True) - with open(file=Path(dir_name / f"drop_campaign_{num}.json"), mode="w", encoding="utf-8") as f: - json.dump(campaign, f, indent=4) - - await add_drop_campaign(campaign) - - if "dropCampaigns" in campaign.get("data", {}).get("user", {}): - for drop_campaign in campaign["data"]["user"]["dropCampaigns"]: - # Save to folder named "drop_campaigns" - dir_name: Path = Path("drop_campaigns") - dir_name.mkdir(parents=True, exist_ok=True) - with open(file=Path(dir_name / f"drop_campaign_{num}.json"), mode="w", encoding="utf-8") as f: - json.dump(drop_campaign, f, indent=4) - - await add_drop_campaign(drop_campaign) + await process_json_data(num, campaign, json_data) return json_data diff --git a/core/migrations/0001_initial.py b/core/migrations/0001_initial.py new file mode 100644 index 0000000..d1cd773 --- /dev/null +++ b/core/migrations/0001_initial.py @@ -0,0 +1,482 @@ +# Generated by Django 5.1 on 2024-08-09 02:49 + +import auto_prefetch +import django.db.models.deletion +import django.db.models.manager +from django.db import migrations, models +from django.db.migrations.operations.base import Operation + + +class Migration(migrations.Migration): + initial = True + + dependencies: list[tuple[str, str]] = [] + + operations: list[Operation] = [ + migrations.CreateModel( + name="Benefit", + fields=[ + ("id", models.TextField(primary_key=True, serialize=False)), + ("created_at", models.DateTimeField(null=True)), + ("entitlement_limit", models.TextField(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)), + ], + options={ + "abstract": False, + "base_manager_name": "prefetch_manager", + }, + managers=[ + ("objects", django.db.models.manager.Manager()), + ("prefetch_manager", django.db.models.manager.Manager()), + ], + ), + migrations.CreateModel( + name="Channel", + fields=[ + ("id", models.TextField(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)), + ], + options={ + "abstract": False, + "base_manager_name": "prefetch_manager", + }, + managers=[ + ("objects", django.db.models.manager.Manager()), + ("prefetch_manager", django.db.models.manager.Manager()), + ], + ), + migrations.CreateModel( + name="FrontEndChannel", + fields=[ + ("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")), + ("name", models.TextField(blank=True, null=True)), + ("twitch_url", models.URLField(blank=True, null=True)), + ("live", models.BooleanField(default=False)), + ], + options={ + "abstract": False, + "base_manager_name": "prefetch_manager", + }, + managers=[ + ("objects", django.db.models.manager.Manager()), + ("prefetch_manager", django.db.models.manager.Manager()), + ], + ), + migrations.CreateModel( + name="FrontEndGame", + fields=[ + ("twitch_id", models.TextField(primary_key=True, serialize=False)), + ("game_url", models.URLField(blank=True, null=True)), + ("display_name", models.TextField(blank=True, null=True)), + ], + options={ + "abstract": False, + "base_manager_name": "prefetch_manager", + }, + managers=[ + ("objects", django.db.models.manager.Manager()), + ("prefetch_manager", django.db.models.manager.Manager()), + ], + ), + migrations.CreateModel( + name="FrontEndOrg", + fields=[ + ("id", models.TextField(primary_key=True, serialize=False)), + ("name", models.TextField(blank=True, null=True)), + ("url", models.TextField(blank=True, null=True)), + ], + options={ + "abstract": False, + "base_manager_name": "prefetch_manager", + }, + managers=[ + ("objects", django.db.models.manager.Manager()), + ("prefetch_manager", django.db.models.manager.Manager()), + ], + ), + migrations.CreateModel( + name="Game", + fields=[ + ("id", models.AutoField(primary_key=True, serialize=False)), + ("slug", models.TextField(blank=True, null=True)), + ("display_name", models.TextField(blank=True, null=True)), + ("box_art_url", models.URLField(blank=True, null=True)), + ("typename", models.TextField(blank=True, null=True)), + ], + options={ + "abstract": False, + "base_manager_name": "prefetch_manager", + }, + managers=[ + ("objects", django.db.models.manager.Manager()), + ("prefetch_manager", django.db.models.manager.Manager()), + ], + ), + migrations.CreateModel( + name="Image", + fields=[ + ("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")), + ("image1_x_url", models.URLField(blank=True, null=True)), + ("typename", models.TextField(blank=True, null=True)), + ], + options={ + "abstract": False, + "base_manager_name": "prefetch_manager", + }, + managers=[ + ("objects", django.db.models.manager.Manager()), + ("prefetch_manager", django.db.models.manager.Manager()), + ], + ), + migrations.CreateModel( + name="Owner", + fields=[ + ("id", models.TextField(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)), + ], + options={ + "abstract": False, + "base_manager_name": "prefetch_manager", + }, + managers=[ + ("objects", django.db.models.manager.Manager()), + ("prefetch_manager", django.db.models.manager.Manager()), + ], + ), + migrations.CreateModel( + name="UnlockRequirements", + fields=[ + ("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")), + ("subs_goal", models.TextField(null=True)), + ("minute_watched_goal", models.TextField(null=True)), + ("typename", models.TextField(blank=True, null=True)), + ], + options={ + "abstract": False, + "base_manager_name": "prefetch_manager", + }, + managers=[ + ("objects", django.db.models.manager.Manager()), + ("prefetch_manager", django.db.models.manager.Manager()), + ], + ), + migrations.CreateModel( + name="BenefitEdge", + fields=[ + ("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")), + ("entitlement_limit", models.TextField(null=True)), + ("typename", models.TextField(blank=True, null=True)), + ( + "benefit", + auto_prefetch.ForeignKey( + null=True, + on_delete=django.db.models.deletion.CASCADE, + related_name="benefit_edges", + to="core.benefit", + ), + ), + ], + options={ + "abstract": False, + "base_manager_name": "prefetch_manager", + }, + managers=[ + ("objects", django.db.models.manager.Manager()), + ("prefetch_manager", django.db.models.manager.Manager()), + ], + ), + 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="core.channel")), + ], + options={ + "abstract": False, + "base_manager_name": "prefetch_manager", + }, + managers=[ + ("objects", django.db.models.manager.Manager()), + ("prefetch_manager", django.db.models.manager.Manager()), + ], + ), + migrations.CreateModel( + name="FrontEndDropCampaign", + fields=[ + ("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")), + ("account_link_url", models.URLField(blank=True, null=True)), + ("about_url", models.URLField(blank=True, null=True)), + ("ends_at", models.DateTimeField(null=True)), + ("starts_at", models.DateTimeField(null=True)), + ("channels", models.ManyToManyField(related_name="drop_campaigns", to="core.frontendchannel")), + ( + "game", + models.ForeignKey( + null=True, + on_delete=django.db.models.deletion.CASCADE, + related_name="drop_campaigns", + to="core.frontendgame", + ), + ), + ], + options={ + "abstract": False, + "base_manager_name": "prefetch_manager", + }, + managers=[ + ("objects", django.db.models.manager.Manager()), + ("prefetch_manager", django.db.models.manager.Manager()), + ], + ), + migrations.CreateModel( + name="FrontEndDrop", + fields=[ + ("id", models.TextField(primary_key=True, serialize=False)), + ("created_at", models.DateTimeField(null=True)), + ("name", models.TextField(blank=True, null=True)), + ("image_url", models.URLField(blank=True, null=True)), + ("limit", models.PositiveBigIntegerField(null=True)), + ("is_ios_available", models.BooleanField(null=True)), + ("minutes_watched", models.PositiveBigIntegerField(null=True)), + ( + "drop_campaign", + models.ForeignKey( + null=True, + on_delete=django.db.models.deletion.CASCADE, + related_name="drops", + to="core.frontenddropcampaign", + ), + ), + ], + options={ + "abstract": False, + "base_manager_name": "prefetch_manager", + }, + managers=[ + ("objects", django.db.models.manager.Manager()), + ("prefetch_manager", django.db.models.manager.Manager()), + ], + ), + migrations.AddField( + model_name="frontendgame", + name="org", + field=models.ForeignKey( + null=True, + on_delete=django.db.models.deletion.CASCADE, + related_name="games", + to="core.frontendorg", + ), + ), + migrations.AddField( + model_name="benefit", + name="game", + field=auto_prefetch.ForeignKey( + null=True, + on_delete=django.db.models.deletion.CASCADE, + related_name="benefits", + to="core.game", + ), + ), + migrations.AddField( + model_name="benefit", + name="owner_organization", + field=auto_prefetch.ForeignKey( + null=True, + on_delete=django.db.models.deletion.CASCADE, + related_name="benefits", + to="core.owner", + ), + ), + migrations.CreateModel( + name="Reward", + fields=[ + ("id", models.TextField(primary_key=True, serialize=False)), + ("name", models.TextField(blank=True, null=True)), + ("earnable_until", models.DateTimeField(null=True)), + ("redemption_instructions", models.TextField(blank=True, null=True)), + ("redemption_url", models.URLField(blank=True, null=True)), + ("typename", models.TextField(blank=True, null=True)), + ( + "banner_image", + auto_prefetch.ForeignKey( + null=True, + on_delete=django.db.models.deletion.CASCADE, + related_name="banner_rewards", + to="core.image", + ), + ), + ( + "thumbnail_image", + auto_prefetch.ForeignKey( + null=True, + on_delete=django.db.models.deletion.CASCADE, + related_name="thumbnail_rewards", + to="core.image", + ), + ), + ], + options={ + "abstract": False, + "base_manager_name": "prefetch_manager", + }, + managers=[ + ("objects", django.db.models.manager.Manager()), + ("prefetch_manager", django.db.models.manager.Manager()), + ], + ), + migrations.CreateModel( + name="TimeBasedDrop", + fields=[ + ("id", models.TextField(primary_key=True, serialize=False)), + ("created_at", models.DateTimeField(null=True)), + ("entitlement_limit", models.TextField(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", + auto_prefetch.ForeignKey( + null=True, + on_delete=django.db.models.deletion.CASCADE, + related_name="time_based_drops", + to="core.game", + ), + ), + ( + "owner_organization", + auto_prefetch.ForeignKey( + null=True, + on_delete=django.db.models.deletion.CASCADE, + related_name="time_based_drops", + to="core.owner", + ), + ), + ], + options={ + "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=[ + ("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, choices=[("ACTIVE", "Active"), ("EXPIRED", "Expired")], null=True), + ), + ("typename", models.TextField(blank=True, null=True)), + ( + "allow", + auto_prefetch.ForeignKey( + null=True, + on_delete=django.db.models.deletion.CASCADE, + related_name="drop_campaigns", + to="core.allow", + ), + ), + ( + "game", + auto_prefetch.ForeignKey( + null=True, + on_delete=django.db.models.deletion.CASCADE, + related_name="drop_campaigns", + to="core.game", + ), + ), + ( + "owner", + auto_prefetch.ForeignKey( + null=True, + on_delete=django.db.models.deletion.CASCADE, + related_name="drop_campaigns", + to="core.owner", + ), + ), + ("time_based_drops", models.ManyToManyField(related_name="drop_campaigns", to="core.timebaseddrop")), + ], + options={ + "abstract": False, + "base_manager_name": "prefetch_manager", + }, + managers=[ + ("objects", django.db.models.manager.Manager()), + ("prefetch_manager", django.db.models.manager.Manager()), + ], + ), + migrations.CreateModel( + name="RewardCampaign", + fields=[ + ("id", models.TextField(primary_key=True, serialize=False)), + ("name", models.TextField(blank=True, null=True)), + ("brand", models.TextField(blank=True, null=True)), + ("starts_at", models.DateTimeField(null=True)), + ("ends_at", models.DateTimeField(null=True)), + ("status", models.TextField(blank=True, null=True)), + ("summary", models.TextField(blank=True, null=True)), + ("instructions", models.TextField(blank=True, null=True)), + ("external_url", models.URLField(blank=True, null=True)), + ("reward_value_url_param", models.TextField(blank=True, null=True)), + ("about_url", models.URLField(blank=True, null=True)), + ("is_sitewide", models.BooleanField(null=True)), + ("typename", models.TextField(blank=True, null=True)), + ( + "game", + auto_prefetch.ForeignKey( + null=True, + on_delete=django.db.models.deletion.CASCADE, + related_name="reward_campaigns", + to="core.game", + ), + ), + ( + "image", + auto_prefetch.ForeignKey( + null=True, + on_delete=django.db.models.deletion.CASCADE, + related_name="reward_campaigns", + to="core.image", + ), + ), + ("rewards", models.ManyToManyField(related_name="reward_campaigns", to="core.reward")), + ( + "unlock_requirements", + auto_prefetch.ForeignKey( + null=True, + on_delete=django.db.models.deletion.CASCADE, + related_name="reward_campaigns", + to="core.unlockrequirements", + ), + ), + ], + options={ + "abstract": False, + "base_manager_name": "prefetch_manager", + }, + managers=[ + ("objects", django.db.models.manager.Manager()), + ("prefetch_manager", django.db.models.manager.Manager()), + ], + ), + ] diff --git a/twitch_app/models.py b/core/models.py similarity index 85% rename from twitch_app/models.py rename to core/models.py index 92daf42..9e4bde1 100644 --- a/twitch_app/models.py +++ b/core/models.py @@ -1,39 +1,71 @@ +from __future__ import annotations + +import logging import typing import auto_prefetch from django.db import models +logger: logging.Logger = logging.getLogger(__name__) + class Game(auto_prefetch.Model): """The game that the reward is for. - Used for reward campaigns (buy subs) and drop campaigns (watch games). + Optional for reward campaigns. Required for drop campaigns. Attributes: - id (int): The primary key of the game. + id (str): The primary key of the game. slug (str): The slug identifier of the game. display_name (str): The display name of the game. typename (str): The type name of the object, typically "Game". - JSON example: + Example JSON data: { - "id": "780302568", - "slug": "xdefiant", - "displayName": "XDefiant", - "__typename": "Game" + "data": { + "currentUser": { + "dropCampaigns": [ + { + "game": { + "id": "263490", + "slug": "rust", + "displayName": "Rust", + "__typename": "Game" + } + } + ] + } + } } """ - id = models.AutoField(primary_key=True) - slug = models.TextField(null=True, blank=True) - display_name = models.TextField(null=True, blank=True) - box_art_url = models.URLField(null=True, blank=True) - typename = models.TextField(null=True, blank=True) + id = models.TextField(primary_key=True, unique=True, help_text="The game ID.", verbose_name="Game ID") + slug = models.TextField(null=True, blank=True, help_text="Slug used for building URL where all the streams are.") + display_name = models.TextField( + null=True, + blank=True, + help_text="Game name.", + default="Unknown Game", + verbose_name="Game Name", + ) + typename = models.TextField(null=True, blank=True, help_text="Always 'Game'.", verbose_name="Type Name") + + # Only used for reward campaigns? + box_art_url = models.URLField( + null=True, + blank=True, + help_text="URL to the box art of the game.", + default="https://static-cdn.jtvnw.net/ttv-static/404_boxart.jpg", + verbose_name="Box Art URL", + ) def __str__(self) -> str: return self.display_name or "Unknown" def get_twitch_url(self) -> str: + if not self.slug: + logger.error("Game %s has no slug", self.display_name) + return "https://www.twitch.tv/" return f"https://www.twitch.tv/directory/game/{self.slug}" @@ -240,24 +272,38 @@ class Channel(auto_prefetch.Model): name (str): The name of the channel. typename (str): The type name of the object, typically "Channel". - JSON example: + Example JSON data: { - "id": "25254906", - "displayName": "Stresss", - "name": "stresss", - "__typename": "Channel" + "data": { + "user": { + "dropCampaign": { + "allow": { + "channels": [ + { + "id": "464161875", + "displayName": "Valair", + "name": "valair", + "__typename": "Channel" + } + ] + } + } + } + } } """ + # Used in Drop Campaigns id = models.TextField(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" + return self.display_name or "Unknown Channel" def get_twitch_url(self) -> str: + # TODO(TheLovinator): Use a field instead # noqa: TD003 return f"https://www.twitch.tv/{self.name}" @@ -269,19 +315,20 @@ class Allow(auto_prefetch.Model): 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" + Example JSON data: + { + "data": { + "user": { + "dropCampaign": { + "allow": { + "channels": [], + "isEnabled": true, + "__typename": "DropCampaignACL" + } + } } - ], - "isEnabled": false, - "__typename": "DropCampaignACL" - }, + } + } """ channels = models.ManyToManyField(Channel, related_name="allow") @@ -295,6 +342,9 @@ class Allow(auto_prefetch.Model): class Owner(auto_prefetch.Model): """Represents the owner of the reward campaign. + Used for: + - Reward campaigns + Attributes: id (int): The primary key of the owner. slug (str): The slug identifier of the owner. @@ -302,23 +352,27 @@ class Owner(auto_prefetch.Model): typename (str): The type name of the object, typically "Organization". JSON example: - "game": { - "id": "491487", # Can also be a string like 'c57a089c-088f-4402-b02d-c13281b3397e' - "slug": "dead-by-daylight", - "displayName": "Dead by Daylight", - "__typename": "Game" - }," + "owner": { + "id": "a1a51d5a-233d-41c3-9acd-a03bdab35159", + "name": "Out of the Park Developments", + "__typename": "Organization" + }, """ - id = models.TextField(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) + id = models.TextField(primary_key=True, unique=True, help_text="The owner ID.") + name = models.TextField(null=True, blank=True, help_text="Owner name.") + slug = models.TextField(null=True, blank=True, help_text="Slug used for building URL where all the streams are.") + display_name = models.TextField(null=True, blank=True, help_text="Owner name.") + typename = models.TextField(null=True, blank=True, help_text="Always 'Organization'.") def __str__(self) -> str: return self.display_name or "Unknown" def get_twitch_url(self) -> str: + if not self.slug: + logger.error("Owner %s has no slug", self.display_name) + return "https://www.twitch.tv/" + return f"https://www.twitch.tv/{self.slug}" diff --git a/config/settings.py b/core/settings.py similarity index 97% rename from config/settings.py rename to core/settings.py index 0d6f219..be9baa7 100644 --- a/config/settings.py +++ b/core/settings.py @@ -30,14 +30,14 @@ if not DEBUG: BASE_DIR: Path = Path(__file__).resolve().parent.parent ADMINS: list[tuple[str, str]] = [("Joakim Hellsén", "tlovinator@gmail.com")] -WSGI_APPLICATION = "config.wsgi.application" +WSGI_APPLICATION = "core.wsgi.application" SECRET_KEY: str = os.getenv("DJANGO_SECRET_KEY", default="") TIME_ZONE = "Europe/Stockholm" USE_TZ = True LANGUAGE_CODE = "en-us" DECIMAL_SEPARATOR = "," THOUSAND_SEPARATOR = " " -ROOT_URLCONF = "config.urls" +ROOT_URLCONF = "core.urls" STATIC_URL = "static/" DEFAULT_AUTO_FIELD = "django.db.models.BigAutoField" STATICFILES_DIRS: list[Path] = [BASE_DIR / "static"] @@ -67,7 +67,6 @@ DISCORD_WEBHOOK_URL: str = os.getenv(key="DISCORD_WEBHOOK_URL", default="") INSTALLED_APPS: list[str] = [ "core.apps.CoreConfig", - "twitch_app.apps.TwitchConfig", "whitenoise.runserver_nostatic", "django.contrib.contenttypes", "django.contrib.sessions", diff --git a/core/templates/partials/header.html b/core/templates/partials/header.html index cce3236..89d0a2e 100644 --- a/core/templates/partials/header.html +++ b/core/templates/partials/header.html @@ -1,14 +1,14 @@

- Twitch drops + Twitch drops