diff --git a/core/management/commands/scrape_twitch.py b/core/management/commands/scrape_twitch.py index 06269db..acb0f92 100644 --- a/core/management/commands/scrape_twitch.py +++ b/core/management/commands/scrape_twitch.py @@ -47,13 +47,16 @@ def get_profile_dir() -> Path: return profile_dir -def save_json(campaign: dict, dir_name: str) -> None: +def save_json(campaign: dict | None, 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. """ + if not campaign: + return + save_dir: Path = Path(dir_name) save_dir.mkdir(parents=True, exist_ok=True) @@ -64,38 +67,47 @@ def save_json(campaign: dict, dir_name: str) -> None: json.dump(campaign, f, indent=4) -async def add_reward_campaign(campaign: dict) -> None: +async def add_reward_campaign(campaign: dict | None) -> None: # noqa: C901 """Add a reward campaign to the database. Args: campaign (dict): The reward campaign to add. """ + if not campaign: + return + logger.info("Adding reward campaign to database") for reward_campaign in campaign["data"]["rewardCampaignsAvailableToUser"]: - our_reward_campaign, created = await RewardCampaign.objects.aget_or_create( + mappings: dict[str, str] = { + "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 = {new_key: reward_campaign[key] for key, new_key in mappings.items() if reward_campaign.get(key)} + + if reward_campaign.get("unlockRequirements", {}).get("subsGoal"): + defaults["sub_goal"] = reward_campaign["unlockRequirements"]["subsGoal"] + + if reward_campaign.get("unlockRequirements", {}).get("minuteWatchedGoal"): + defaults["minute_watched_goal"] = reward_campaign["unlockRequirements"]["minuteWatchedGoal"] + + if reward_campaign.get("image"): + defaults["image_url"] = reward_campaign["image"]["image1xURL"] + + our_reward_campaign, created = await RewardCampaign.objects.aupdate_or_create( id=reward_campaign["id"], - defaults={ - "name": reward_campaign["name"], - "brand": reward_campaign["brand"], - "starts_at": reward_campaign["startsAt"], - "ends_at": reward_campaign["endsAt"], - "status": reward_campaign["status"], - "summary": reward_campaign["summary"], - "instructions": reward_campaign["instructions"], - "reward_value_url_param": reward_campaign["rewardValueURLParam"], - "external_url": reward_campaign["externalURL"], - "about_url": reward_campaign["aboutURL"], - "is_site_wide": reward_campaign["isSitewide"], - "sub_goal": reward_campaign["unlockRequirements"]["subsGoal"], - "minute_watched_goal": reward_campaign["unlockRequirements"]["minuteWatchedGoal"], - "image_url": reward_campaign["image"]["image1xURL"], - # "game" # To be implemented - }, + defaults=defaults, ) if created: logger.info("Added reward campaign %s", our_reward_campaign.id) - else: - logger.info("Updated reward campaign %s", our_reward_campaign.id) if reward_campaign["game"]: # TODO(TheLovinator): Add game to reward campaign # noqa: TD003 @@ -104,17 +116,21 @@ async def add_reward_campaign(campaign: dict) -> None: # Add rewards for reward in reward_campaign["rewards"]: - our_reward, created = await Reward.objects.aget_or_create( + mappings: dict[str, str] = { + "name": "name", + "bannerImage": "banner_image_url", + "thumbnailImage": "thumbnail_image_url", + "earnableUntil": "earnable_until", + "redemptionInstructions": "redemption_instructions", + "redemptionURL": "redemption_url", + } + defaults = {new_key: reward[key] for key, new_key in mappings.items() if reward.get(key)} + + if our_reward_campaign: + defaults["campaign"] = our_reward_campaign + our_reward, created = await Reward.objects.aupdate_or_create( id=reward["id"], - defaults={ - "name": reward["name"], - "banner_image_url": reward["bannerImage"]["image1xURL"], - "thumbnail_image_url": reward["thumbnailImage"]["image1xURL"], - "earnable_until": reward["earnableUntil"], - "redemption_instructions": reward["redemptionInstructions"], - "redemption_url": reward["redemptionURL"], - "campaign": our_reward_campaign, - }, + defaults=defaults, ) if created: logger.info("Added reward %s", our_reward.id) @@ -122,39 +138,41 @@ async def add_reward_campaign(campaign: dict) -> None: logger.info("Updated reward %s", our_reward.id) -async def add_or_update_game(game_json: dict, owner: Owner | None) -> Game | None: +async def add_or_update_game(game_json: dict | 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 game_json: - game_url: str | None = ( - f"https://www.twitch.tv/directory/game/{game_json["slug"]}" if game_json["slug"] else None - ) - our_game, created = await Game.objects.aget_or_create( - twitch_id=game_json["id"], - defaults={ - "slug": game_json["slug"], - "name": game_json["displayName"], - "game_url": game_url, - "org": owner, - # TODO(TheLovinator): Add box_art_url to game # noqa: TD003 - }, - ) - if created: - logger.info("Added game %s", our_game.twitch_id) - else: - logger.info("Updated game %s", our_game.twitch_id) - return our_game - return None + if not game_json: + return None + + mappings: dict[str, str] = { + "slug": "slug", + "displayName": "name", + "boxArtURL": "box_art_url", + } + defaults = {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"]}" + + if game_json.get("owner"): + owner: Owner | None = await add_or_update_owner(game_json["owner"]) + if owner: + defaults["org"] = owner + + 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) + + return our_game -async def add_or_update_owner(owner_json: dict) -> Owner | None: +async def add_or_update_owner(owner_json: dict | None) -> Owner | None: """Add or update an owner in the database. Args: @@ -163,17 +181,19 @@ async def add_or_update_owner(owner_json: dict) -> Owner | None: Returns: Owner: The owner that was added or updated. """ - if owner_json: - our_owner, created = await Owner.objects.aget_or_create( - id=owner_json["id"], - defaults={"name": owner_json["name"]}, - ) - if created: - logger.info("Added owner %s", our_owner.id) - else: - logger.info("Updated owner %s", our_owner.id) - return our_owner - return None + if not owner_json: + return None + + defaults = {} + + if owner_json.get("name"): + defaults["name"] = owner_json["name"] + + our_owner, created = await Owner.objects.aupdate_or_create(id=owner_json["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: @@ -190,20 +210,18 @@ async def add_or_update_channels(channels_json: list[dict]) -> list[Channel] | N channels: list[Channel] = [] for channel_json in channels_json: - twitch_url: str | None = f"https://www.twitch.tv/{channel_json["name"]}" if channel_json["name"] else None - our_channel, created = await Channel.objects.aget_or_create( - twitch_id=channel_json["id"], - defaults={ - "name": channel_json["name"], - "display_name": channel_json["displayName"], - "twitch_url": twitch_url, - "live": False, # Toggle this later - }, - ) + defaults = {} + 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) - else: - logger.info("Updated channel %s", our_channel.twitch_id) + channels.append(our_channel) return channels @@ -216,82 +234,103 @@ async def add_benefit(benefit: dict, time_based_drop: TimeBasedDrop) -> None: benefit (dict): The benefit to add. time_based_drop (TimeBasedDrop): The time-based drop the benefit belongs to. """ - our_benefit, created = await Benefit.objects.aget_or_create( - id=benefit["id"], - defaults={ - "twitch_created_at": benefit["createdAt"], - "entitlement_limit": benefit["entitlementLimit"], - "image_url": benefit["imageAssetURL"], - "is_ios_available": benefit["isIosAvailable"], - "name": benefit["name"], - "time_based_drop": time_based_drop, - }, - ) + logger.info("Adding benefit to database") + + mappings = { + "createdAt": "twitch_created_at", + "entitlementLimit": "entitlement_limit", + "imageAssetURL": "image_url", + "isIosAvailable": "is_ios_available", + "name": "name", + } + + defaults = {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) - else: - logger.info("Updated benefit %s", our_benefit.id) + + if time_based_drop: + await time_based_drop.benefits.aadd(our_benefit) # type: ignore # noqa: PGH003 -async def add_drop_campaign(drop_campaign: dict) -> None: +async def add_drop_campaign(drop_campaign: dict | None) -> None: """Add a drop campaign to the database. Args: drop_campaign (dict): The drop campaign to add. """ - logger.info("Adding drop campaign to database") - owner: Owner | None = await add_or_update_owner(drop_campaign["owner"]) - game: Game | None = await add_or_update_game(drop_campaign["game"], owner) - channels: list[Channel] | None = await add_or_update_channels(drop_campaign["allow"]["channels"]) + if not drop_campaign: + return - our_drop_campaign, created = await DropCampaign.objects.aget_or_create( + defaults = {} + if drop_campaign.get("game"): + game: Game | None = await add_or_update_game(drop_campaign["game"]) + defaults["game"] = game + + 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] + + our_drop_campaign, created = await DropCampaign.objects.aupdate_or_create( id=drop_campaign["id"], - defaults={ - "account_link_url": drop_campaign["accountLinkURL"], - "description": drop_campaign["description"], - "details_url": drop_campaign["detailsURL"], - "ends_at": drop_campaign["endAt"], - "starts_at": drop_campaign["startAt"], - "image_url": drop_campaign["imageURL"], - "name": drop_campaign["name"], - "status": drop_campaign["status"], - "game": game, - }, + defaults=defaults, ) if created: logger.info("Added drop campaign %s", our_drop_campaign.id) - else: - logger.info("Updated drop campaign %s", our_drop_campaign.id) - if channels: - our_drop_campaign.channels.aset(channels) # type: ignore # noqa: PGH003 + 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) - # Add time-based drops - for time_based_drop in drop_campaign["timeBasedDrops"]: - if time_based_drop["preconditionDrops"]: + await add_time_based_drops(drop_campaign, our_drop_campaign) + + +async def add_time_based_drops(drop_campaign: dict, our_drop_campaign: DropCampaign) -> None: + """Add time-based drops to the database. + + Args: + drop_campaign (dict): The drop campaign containing time-based drops. + our_drop_campaign (DropCampaign): The drop campaign object in the database. + """ + for time_based_drop in drop_campaign.get("timeBasedDrops", []): + 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 + # TODO(TheLovinator): Send JSON to Discord # noqa: TD003 logger.error("Not implemented: Add precondition drops to time-based drop, JSON: %s", time_based_drop) - our_time_based_drop, created = await TimeBasedDrop.objects.aget_or_create( + mappings = { + "requiredSubs": "required_subs", + "endAt": "ends_at", + "name": "name", + "requiredMinutesWatched": "required_minutes_watched", + "startAt": "starts_at", + } + defaults = {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"], - defaults={ - "required_subs": time_based_drop["requiredSubs"], - "ends_at": time_based_drop["endAt"], - "name": time_based_drop["name"], - "required_minutes_watched": time_based_drop["requiredMinutesWatched"], - "starts_at": time_based_drop["startAt"], - "drop_campaign": our_drop_campaign, - }, + defaults=defaults, ) if created: logger.info("Added time-based drop %s", our_time_based_drop.id) - else: - logger.info("Updated time-based drop %s", our_time_based_drop.id) - # Add benefits - for benefit_edge in time_based_drop["benefitEdges"]: - await add_benefit(benefit_edge["benefit"], our_time_based_drop) + if time_based_drop.get("benefitEdges") and our_time_based_drop: + for benefit_edge in time_based_drop["benefitEdges"]: + await add_benefit(benefit_edge["benefit"], our_time_based_drop) async def process_json_data(num: int, campaign: dict | None) -> None: @@ -311,7 +350,7 @@ async def process_json_data(num: int, campaign: dict | None) -> None: return # This is a Reward Campaign - if "rewardCampaignsAvailableToUser" in campaign["data"]: + if "rewardCampaignsAvailableToUser" in campaign.get("data", {}): save_json(campaign, "reward_campaigns") await add_reward_campaign(campaign) @@ -321,9 +360,11 @@ async def process_json_data(num: int, campaign: dict | None) -> None: return save_json(campaign, "drop_campaign") - await add_drop_campaign(campaign["data"]["user"]["dropCampaign"]) - if "dropCampaigns" in campaign.get("data", {}).get("user", {}): + if campaign.get("data", {}).get("user", {}).get("dropCampaign"): + await add_drop_campaign(campaign["data"]["user"]["dropCampaign"]) + + if "dropCampaigns" in campaign.get("data", {}).get("currentUser", {}): for drop_campaign in campaign["data"]["currentUser"]["dropCampaigns"]: save_json(campaign, "drop_campaigns") await add_drop_campaign(drop_campaign) diff --git a/core/migrations/0004_alter_channel_display_name_alter_channel_twitch_url_and_more.py b/core/migrations/0004_alter_channel_display_name_alter_channel_twitch_url_and_more.py new file mode 100644 index 0000000..4944620 --- /dev/null +++ b/core/migrations/0004_alter_channel_display_name_alter_channel_twitch_url_and_more.py @@ -0,0 +1,37 @@ +# Generated by Django 5.1 on 2024-08-13 18:17 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + dependencies = [ + ("core", "0001_squashed_0003_alter_benefit_options_alter_channel_options_and_more"), + ] + + operations = [ + migrations.AlterField( + model_name="channel", + name="display_name", + field=models.TextField(default="Channel name unknown", null=True), + ), + migrations.AlterField( + model_name="channel", + name="twitch_url", + field=models.URLField(default="https://www.twitch.tv/", null=True), + ), + migrations.AlterField( + model_name="game", + name="box_art_url", + field=models.URLField(default="https://static-cdn.jtvnw.net/ttv-static/404_boxart.jpg", null=True), + ), + migrations.AlterField( + model_name="game", + name="game_url", + field=models.URLField(default="https://www.twitch.tv/", null=True), + ), + migrations.AlterField( + model_name="game", + name="name", + field=models.TextField(default="Game name unknown", null=True), + ), + ] diff --git a/core/models/twitch.py b/core/models/twitch.py index f486b96..453b3f1 100644 --- a/core/models/twitch.py +++ b/core/models/twitch.py @@ -24,9 +24,13 @@ class Game(models.Model): """This is the game we will see on the front end.""" twitch_id = models.TextField(primary_key=True) # "509658" - game_url = models.URLField(null=True) # "https://www.twitch.tv/directory/category/halo-infinite" - name = models.TextField(null=True) # "Halo Infinite" - box_art_url = models.URLField(null=True) # "https://static-cdn.jtvnw.net/ttv-boxart/Halo%20Infinite.jpg" + + # "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" + + # "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" org = models.ForeignKey(Owner, on_delete=models.CASCADE, related_name="games", null=True) @@ -67,9 +71,9 @@ class Channel(models.Model): """This is the channel we will see on the front end.""" twitch_id = models.TextField(primary_key=True) # "222719079" - display_name = models.TextField(null=True) # "LVTHalo" + display_name = models.TextField(null=True, default="Channel name unknown") # "LVTHalo" name = models.TextField(null=True) # "lvthalo" - twitch_url = models.URLField(null=True) # "https://www.twitch.tv/lvthalo" + twitch_url = models.URLField(null=True, default="https://www.twitch.tv/") # "https://www.twitch.tv/lvthalo" live = models.BooleanField(default=False) # "True" drop_campaigns = models.ManyToManyField(DropCampaign, related_name="channels") diff --git a/core/templates/partials/game_card.html b/core/templates/partials/game_card.html index b9cb29c..c627590 100644 --- a/core/templates/partials/game_card.html +++ b/core/templates/partials/game_card.html @@ -1,7 +1,7 @@
-