from __future__ import annotations import logging from datetime import datetime from io import BytesIO from typing import TYPE_CHECKING, Any, ClassVar, Literal, Self, cast import requests from asgiref.sync import sync_to_async from django.contrib.auth.models import AbstractUser from django.core.files import File from django.core.files.base import ContentFile from django.db import models from PIL import Image if TYPE_CHECKING: from django.db.models.fields.files import ImageFieldFile logger: logging.Logger = logging.getLogger(__name__) # The image file format to save images as. image_file_format: Literal["webp"] = "webp" async def update_scraped_json(model_instance: models.Model, new_json_data: dict) -> None: """Update the JSON data for the drop campaign. Args: model_instance (models.Model): The Django model instance which must have a 'scraped_json' attribute. new_json_data (dict): The new JSON data to be updated. """ if await sync_to_async(hasattr)(model_instance, "scraped_json"): if model_instance.scraped_json: # type: ignore[attr-defined] model_instance.scraped_json.json_data = new_json_data # type: ignore[attr-defined] await model_instance.scraped_json.asave() # type: ignore[attr-defined] else: scraped_json_instance: ScrapedJson = await ScrapedJson.objects.acreate(json_data=new_json_data) model_instance.scraped_json = scraped_json_instance # type: ignore[attr-defined] await model_instance.asave() else: logger.error("The model instance does not have a 'scraped_json' attribute.") model_instance.save() def wrong_typename(data: dict, expected: str) -> bool: """Check if the data is the expected type. # TODO(TheLovinator): Double check this. # noqa: TD003 Type name examples: - Game - DropCampaign - TimeBasedDrop - DropBenefit - RewardCampaign - Reward Args: data (dict): The data to check. expected (str): The expected type. Returns: bool: True if the data is not the expected type. """ is_unexpected_type: bool = data.get("__typename", "") != expected if is_unexpected_type: logger.error("Not a %s? %s", expected, data) return is_unexpected_type def update_field(instance: models.Model, django_field_name: str, new_value: str | datetime | None) -> int: """Update a field on an instance if the new value is different from the current value. Args: instance (models.Model): The Django model instance. django_field_name (str): The name of the field to update. new_value (str | datetime | None): The new value to update the field with. Returns: int: If the field was updated, returns 1. Otherwise, returns 0. """ # Get the current value of the field. try: current_value = getattr(instance, django_field_name) except AttributeError: logger.exception("Field %s does not exist on %s", django_field_name, instance) return 0 # Only update the field if the new value is different from the current value. if new_value and new_value != current_value: setattr(instance, django_field_name, new_value) return 1 # 0 fields updated. return 0 def get_value(data: dict, key: str) -> datetime | str | None: """Get a value from a dictionary. We have this function so we can handle values that we need to convert to a different type. For example, we might need to convert a string to a datetime object. Args: data (dict): The dictionary to get the value from. key (str): The key to get the value for. Returns: datetime | str | None: The value from the dictionary """ data_key: Any | None = data.get(key) if not data_key: return None # Dates are in the format "2024-08-12T05:59:59.999Z" dates: list[str] = ["endAt", "endsAt,", "startAt", "startsAt", "createdAt", "earnableUntil"] if key in dates: return datetime.fromisoformat(data_key.replace("Z", "+00:00")) return data_key async def update_fields(instance: models.Model, data: dict, field_mapping: dict[str, str]) -> int: """Update multiple fields on an instance using a mapping from external field names to model field names. Args: instance (models.Model): The Django model instance. data (dict): The new data to update the fields with. field_mapping (dict[str, str]): A dictionary mapping external field names to model field names. Returns: int: The number of fields updated. Used for only saving the instance if there were changes. """ dirty = 0 for json_field, django_field_name in field_mapping.items(): data_key: datetime | str | None = get_value(data, json_field) dirty += update_field(instance=instance, django_field_name=django_field_name, new_value=data_key) if dirty > 0: await instance.asave() return dirty def convert_image_to_webp(data: bytes | None) -> File | None: """Convert an image to a webp format. Args: data (bytes | None): The image data to convert. Returns: ImageFile | None: The image converted to a webp format. """ if not data: return None try: with BytesIO(data) as input_buffer, Image.open(input_buffer) as image: output_buffer = BytesIO() image.save(output_buffer, format=image_file_format) output_buffer.seek(0) return File(file=Image.open(output_buffer)) except Exception: logger.exception("Failed to convert image to webp.") return File(file=Image.open(fp=ContentFile(data)).convert("RGB")) def fetch_image(image_url: str) -> bytes | None: """Fetch an image from a URL and return the response. Args: image_url (str): The URL of the image to fetch. Returns: requests.Response | None: The response if the image was fetched, otherwise None. """ response: requests.Response = requests.get(image_url, timeout=5, stream=True) response.raise_for_status() if response.ok: if response.raw: logging.debug("Fetched image from %s", image_url) return response.raw.read() logging.error("Response raw is None for %s", image_url) return None logging.error("Failed to retrieve content. Status code: %s", response.status_code) return None class User(AbstractUser): ... class ScrapedJson(models.Model): json_data = models.JSONField(unique=True, help_text="The JSON data from the Twitch API.") created_at = models.DateTimeField(auto_now_add=True) class Meta: ordering: ClassVar[list[str]] = ["-created_at"] def __str__(self) -> str: return f"{self.pk}" class Owner(models.Model): """The company or person that owns the game. Drops will be grouped by the owner. Users can also subscribe to owners. """ # "ad299ac0-f1a5-417d-881d-952c9aed00e9" twitch_id = models.TextField(primary_key=True) # When the owner was first added to the database. created_at = models.DateTimeField(null=True, auto_created=True) # When the owner was last modified. modified_at = models.DateTimeField(null=True, auto_now=True) # "Microsoft" name = models.TextField(null=True) class Meta: ordering: ClassVar[list[str]] = ["name"] def __str__(self) -> str: return self.name or self.twitch_id async def aimport_json(self, data: dict) -> Self: if wrong_typename(data, "Organization"): return self field_mapping: dict[str, str] = {"name": "name"} updated: int = await update_fields(instance=self, data=data, field_mapping=field_mapping) if updated > 0: logger.info("Updated %s fields for %s", updated, self) return self def get_game_image_path(instance: models.Model, filename: str) -> str: """Get the path for the game image. Args: instance (models.Model): The instance of the model. Is a Game. filename (str): The filename of the image. Returns: str: The path to the image. """ instance = cast(Game, instance) # Example: game/509658.png image_path: str = f"game/{filename}" logger.debug("Saved image to %s", image_path) return image_path class Game(models.Model): """The game the drop campaign is for. Note that some reward campaigns are not tied to a game.""" # "509658" twitch_id = models.TextField(primary_key=True) # When the game was first added to the database. created_at = models.DateTimeField(null=True, auto_created=True) # When the game was last modified. modified_at = models.DateTimeField(null=True, auto_now=True) # "https://www.twitch.tv/directory/category/halo-infinite" game_url = models.URLField(null=True) # "Halo Infinite" name = models.TextField(null=True) # "https://static-cdn.jtvnw.net/ttv-boxart/Halo%20Infinite.jpg" box_art_url = models.URLField(null=True) image = models.ImageField(null=True, upload_to=get_game_image_path) # "halo-infinite" slug = models.TextField(null=True, unique=True) org = models.ForeignKey(Owner, on_delete=models.CASCADE, related_name="games", null=True) class Meta: ordering: ClassVar[list[str]] = ["name"] def __str__(self) -> str: return self.name or self.twitch_id def download_image(self) -> ImageFieldFile | None: """Download the image for the game. Returns: ImageFieldFile | None: The image file or None if it doesn't exist. """ # We don't want to re-download the image if it already exists. # TODO(TheLovinator): Check if there is a different image available. # noqa: TD003 if self.image: return self.image if not self.box_art_url: return None response: bytes | None = fetch_image(image_url=self.box_art_url) image: File | None = convert_image_to_webp(response) if image: self.image.save(name=f"{self.twitch_id}.{image_file_format}", content=image, save=True) logger.info("Downloaded image for %s to %s", self, self.image.url) return None async def aimport_json(self, data: dict, owner: Owner | None) -> Self: if wrong_typename(data, "Game"): return self # Map the fields from the JSON data to the Django model fields. field_mapping: dict[str, str] = { "displayName": "name", "boxArtURL": "box_art_url", # TODO(TheLovinator): Should download the image. # noqa: TD003 "slug": "slug", } updated: int = await update_fields(instance=self, data=data, field_mapping=field_mapping) if updated > 0: logger.info("Updated %s fields for %s", updated, self) # Handle the owner of the game. if owner and await sync_to_async(lambda: self not in owner.games.all())(): # type: ignore[attr-defined] await owner.games.aadd(self) # type: ignore # noqa: PGH003 await self.asave() logger.info("Added game %s for %s", self, owner) return self def get_drop_campaign_image_path(instance: models.Model, filename: str) -> str: """Get the path for the drop campaign image. Args: instance (models.Model): The instance of the model. Is a DropCampaign. filename (str): The filename of the image. Returns: str: The path to the image. """ instance = cast(DropCampaign, instance) # Example: drop_campaigns/509658/509658.png image_path: str = f"drop_campaign/{filename}" logger.debug("Saved image to %s", image_path) return image_path class DropCampaign(models.Model): """This is the drop campaign we will see on the front end.""" # "f257ce6e-502a-11ef-816e-0a58a9feac02" twitch_id = models.TextField(primary_key=True) # When the drop campaign was first added to the database. created_at = models.DateTimeField(null=True, auto_created=True) # When the drop campaign was last modified. modified_at = models.DateTimeField(null=True, auto_now=True) # "https://www.halowaypoint.com/settings/linked-accounts" account_link_url = models.URLField(null=True) # "Tune into this HCS Grassroots event to earn Halo Infinite in-game content!" description = models.TextField(null=True) # "https://www.halowaypoint.com" details_url = models.URLField(null=True) # "2024-08-12T05:59:59.999Z" ends_at = models.DateTimeField(null=True) # "2024-08-11T11:00:00Z"" starts_at = models.DateTimeField(null=True) # "https://static-cdn.jtvnw.net/twitch-quests-assets/CAMPAIGN/c8e02666-8b86-471f-bf38-7ece29a758e4.png" image_url = models.URLField(null=True) image = models.ImageField(null=True, upload_to=get_drop_campaign_image_path) # "HCS Open Series - Week 1 - DAY 2 - AUG11" name = models.TextField(null=True) # "ACTIVE" status = models.TextField(null=True) game = models.ForeignKey(Game, on_delete=models.CASCADE, related_name="drop_campaigns", null=True) scraped_json = models.ForeignKey( ScrapedJson, null=True, on_delete=models.SET_NULL, help_text="Reference to the JSON data from the Twitch API.", ) class Meta: ordering: ClassVar[list[str]] = ["ends_at"] def __str__(self) -> str: return self.name or self.twitch_id def download_image(self) -> ImageFieldFile | None: """Download the image for the drop campaign. Returns: ImageFieldFile | None: The image file or None if it doesn't exist. """ # We don't want to re-download the image if it already exists. # TODO(TheLovinator): Check if there is a different image available. # noqa: TD003 if self.image: return self.image if not self.image_url: return None response: bytes | None = fetch_image(image_url=self.image_url) image: File | None = convert_image_to_webp(response) if image: self.image.save(name=f"{self.twitch_id}.{image_file_format}", content=image, save=True) logger.info("Downloaded image for %s to %s", self, self.image.url) return None async def aimport_json(self, data: dict, game: Game | None, *, scraping_local_files: bool = False) -> Self: """Import the data from the Twitch API. Args: data (dict | None): The data from the Twitch API. game (Game | None): The game this drop campaign is for. scraping_local_files (bool, optional): If this was scraped from local data. Defaults to True. Returns: Self: The updated drop campaign. """ if wrong_typename(data, "DropCampaign"): return self # Save the JSON data for debugging purposes. await update_scraped_json(model_instance=self, new_json_data=data) field_mapping: dict[str, str] = { "name": "name", "accountLinkURL": "account_link_url", # TODO(TheLovinator): Should archive site. # noqa: TD003 "description": "description", "endAt": "ends_at", "startAt": "starts_at", "detailsURL": "details_url", # TODO(TheLovinator): Should archive site. # noqa: TD003 "imageURL": "image_url", # TODO(TheLovinator): Should download the image. # noqa: TD003 } updated: int = await update_fields(instance=self, data=data, field_mapping=field_mapping) if updated > 0: logger.info("Updated %s fields for %s", updated, self) # Update the drop campaign's status if the new status is different. # When scraping local files: # - Only update if the status changes from "ACTIVE" to "EXPIRED". # When scraping from the Twitch API: # - Always update the status regardless of its value. status = data.get("status") if status and status != self.status: # Check if scraping local files and status changes from ACTIVE to EXPIRED should_update = scraping_local_files and status == "EXPIRED" and self.status == "ACTIVE" # Always update if not scraping local files if not scraping_local_files or should_update: self.status = status await self.asave() # Update the game if the game is different or not set. if game and await sync_to_async(lambda: game != self.game)(): self.game = game logger.info("Updated game %s for %s", game, self) await self.asave() return self class TimeBasedDrop(models.Model): """This is the drop we will see on the front end.""" # "d5cdf372-502b-11ef-bafd-0a58a9feac02" twitch_id = models.TextField(primary_key=True) # When the drop was first added to the database. created_at = models.DateTimeField(null=True, auto_created=True) # When the drop was last modified. modified_at = models.DateTimeField(null=True, auto_now=True) # "1" required_subs = models.PositiveBigIntegerField(null=True) # "2024-08-12T05:59:59.999Z" ends_at = models.DateTimeField(null=True) # "Cosmic Nexus Chimera" name = models.TextField(null=True) # "120" required_minutes_watched = models.PositiveBigIntegerField(null=True) # "2024-08-11T11:00:00Z" starts_at = models.DateTimeField(null=True) drop_campaign = models.ForeignKey(DropCampaign, on_delete=models.CASCADE, related_name="drops", null=True) class Meta: ordering: ClassVar[list[str]] = ["required_minutes_watched"] def __str__(self) -> str: return self.name or self.twitch_id async def aimport_json(self, data: dict, drop_campaign: DropCampaign | None) -> Self: if wrong_typename(data, "TimeBasedDrop"): return self field_mapping: dict[str, str] = { "name": "name", "requiredSubs": "required_subs", "requiredMinutesWatched": "required_minutes_watched", "startAt": "starts_at", "endAt": "ends_at", } updated: int = await update_fields(instance=self, data=data, field_mapping=field_mapping) if updated > 0: logger.info("Updated %s fields for %s", updated, self) if drop_campaign and await sync_to_async(lambda: drop_campaign != self.drop_campaign)(): self.drop_campaign = drop_campaign logger.info("Updated drop campaign %s for %s", drop_campaign, self) await self.asave() return self def get_benefit_image_path(instance: models.Model, filename: str) -> str: """Get the path for the benefit image. Args: instance (models.Model): The instance of the model. Is a Benefit. filename (str): The filename of the image. Returns: str: The path to the image. """ instance = cast(Benefit, instance) # Example: benefit_images/509658.png image_path: str = f"benefit/{filename}" logger.debug("Saved image to %s", image_path) return image_path class Benefit(models.Model): """Benefits are the rewards for the drops.""" # "d5cdf372-502b-11ef-bafd-0a58a9feac02" twitch_id = models.TextField(primary_key=True) # When the benefit was first added to the database. created_at = models.DateTimeField(null=True, auto_created=True) # When the benefit was last modified. modified_at = models.DateTimeField(null=True, auto_now=True) # Note: This is Twitch's created_at from the API. # "2023-11-09T01:18:00.126Z" twitch_created_at = models.DateTimeField(null=True) # "1" entitlement_limit = models.PositiveBigIntegerField(null=True) # "https://static-cdn.jtvnw.net/twitch-quests-assets/REWARD/e58ad175-73f6-4392-80b8-fb0223163733.png" image_url = models.URLField(null=True) image = models.ImageField(null=True, upload_to=get_benefit_image_path) # "True" or "False". None if unknown. is_ios_available = models.BooleanField(null=True) # "Cosmic Nexus Chimera" name = models.TextField(null=True) time_based_drop = models.ForeignKey( TimeBasedDrop, on_delete=models.CASCADE, related_name="benefits", null=True, ) class Meta: ordering: ClassVar[list[str]] = ["-twitch_created_at"] def __str__(self) -> str: return self.name or self.twitch_id def download_image(self) -> ImageFieldFile | None: """Download the image for the benefit. Returns: ImageFieldFile | None: The image file or None if it doesn't exist. """ # TODO(TheLovinator): Check if the image on Twitch is different. # noqa: TD003 if self.image: logger.debug("Image already exists for %s", self) return self.image if not self.image_url: logger.error("No image URL for %s", self) return None response: bytes | None = fetch_image(image_url=self.image_url) image: File | None = convert_image_to_webp(response) if image: self.image.save(name=f"{self.twitch_id}.{image_file_format}", content=image, save=True) logger.info("Downloaded image for %s to %s", self, self.image.url) return None async def aimport_json(self, data: dict, time_based_drop: TimeBasedDrop | None) -> Self: if wrong_typename(data, "DropBenefit"): return self field_mapping: dict[str, str] = { "name": "name", "imageAssetURL": "image_url", # TODO(TheLovinator): Should download the image. # noqa: TD003 "entitlementLimit": "entitlement_limit", "isIOSAvailable": "is_ios_available", "createdAt": "twitch_created_at", } updated: int = await update_fields(instance=self, data=data, field_mapping=field_mapping) if updated > 0: logger.info("Updated %s fields for %s", updated, self) if time_based_drop and await sync_to_async(lambda: self not in time_based_drop.benefits.all())(): # type: ignore[attr-defined] await time_based_drop.benefits.aadd(self) # type: ignore # noqa: PGH003 await time_based_drop.asave() logger.info("Added benefit %s for %s", self, time_based_drop) return self def get_reward_image_path(instance: models.Model, filename: str) -> str: """Get the path for the reward image. Args: instance (models.Model): The instance of the model. Is a Reward. filename (str): The filename of the image. Returns: str: The path to the image. """ instance = cast(Reward, instance) # Example: reward/509658.png image_path: str = f"reward/{filename}" logger.debug("Saved image to %s", image_path) return image_path class RewardCampaign(models.Model): """Buy subscriptions to earn rewards.""" # "dc4ff0b4-4de0-11ef-9ec3-621fb0811846" twitch_id = models.TextField(primary_key=True) # When the reward campaign was first added to the database. created_at = models.DateTimeField(null=True, auto_created=True) # When the reward campaign was last modified. modified_at = models.DateTimeField(null=True, auto_now=True) # "Buy 1 new sub, get 3 months of Apple TV+" name = models.TextField(null=True) # "Apple TV+" brand = models.TextField(null=True) # "2024-08-11T11:00:00Z" starts_at = models.DateTimeField(null=True) # "2024-08-12T05:59:59.999Z" ends_at = models.DateTimeField(null=True) # "UNKNOWN" status = models.TextField(null=True) # "Get 3 months of Apple TV+ with the purchase of a new sub" summary = models.TextField(null=True) # "Buy a new sub to get 3 months of Apple TV+" instructions = models.TextField(null=True) # "" reward_value_url_param = models.TextField(null=True) # "https://tv.apple.com/includes/commerce/redeem/code-entry" external_url = models.URLField(null=True) # "https://blog.twitch.tv/2024/07/26/sub-and-get-apple-tv/" about_url = models.URLField(null=True) # "True" or "False". None if unknown. is_site_wide = models.BooleanField(null=True) # "1" subs_goal = models.PositiveBigIntegerField(null=True) # "0" minute_watched_goal = models.PositiveBigIntegerField(null=True) # "https://static-cdn.jtvnw.net/twitch-quests-assets/CAMPAIGN/quests_appletv_q3_2024/apple_150x200.png" image_url = models.URLField(null=True) image = models.ImageField(null=True, upload_to=get_reward_image_path) game = models.ForeignKey(Game, on_delete=models.CASCADE, related_name="reward_campaigns", null=True) scraped_json = models.ForeignKey( ScrapedJson, null=True, on_delete=models.SET_NULL, help_text="Reference to the JSON data from the Twitch API.", ) class Meta: ordering: ClassVar[list[str]] = ["-starts_at"] def __str__(self) -> str: return self.name or self.twitch_id def download_image(self) -> ImageFieldFile | None: """Download the image for the drop campaign. Returns: ImageFieldFile | None: The image file or None if it doesn't exist. """ # We don't want to re-download the image if it already exists. # TODO(TheLovinator): Check if there is a different image available. # noqa: TD003 if self.image: return self.image if not self.image_url: return None response: bytes | None = fetch_image(image_url=self.image_url) image: File | None = convert_image_to_webp(response) if image: file_name: str = f"{self.twitch_id}.{image_file_format}" self.image.save(name=file_name, content=image, save=True) logger.info("Downloaded image for %s to %s", self, self.image.url) return self.image return None async def aimport_json(self, data: dict) -> Self: if wrong_typename(data, "RewardCampaign"): return self # Save the JSON data for debugging purposes. await update_scraped_json(model_instance=self, new_json_data=data) field_mapping: dict[str, str] = { "name": "name", "brand": "brand", "startsAt": "starts_at", "endsAt": "ends_at", "status": "status", "summary": "summary", "instructions": "instructions", "rewardValueURLParam": "reward_value_url_param", # wtf is this? "externalURL": "external_url", "aboutURL": "about_url", "isSitewide": "is_site_wide", } updated: int = await update_fields(instance=self, data=data, field_mapping=field_mapping) if updated > 0: logger.info("Updated %s fields for %s", updated, self) if data.get("unlockRequirements", {}): subs_goal = data["unlockRequirements"].get("subsGoal") if subs_goal and subs_goal != self.subs_goal: self.subs_goal = subs_goal await self.asave() minutes_watched_goal = data["unlockRequirements"].get("minuteWatchedGoal") if minutes_watched_goal and minutes_watched_goal != self.minute_watched_goal: self.minute_watched_goal = minutes_watched_goal await self.asave() image_url = data.get("image", {}).get("image1xURL") if image_url and image_url != self.image_url: # await sync_to_async(self.download_image)() # TODO(TheLovinator): Download the image. # noqa: TD003 self.image_url = image_url await self.asave() if data.get("game") and data["game"].get("id"): game, _ = await Game.objects.aget_or_create(twitch_id=data["game"]["id"]) await game.reward_campaigns.aadd(self) # type: ignore # noqa: PGH003 await self.asave() if "rewards" in data: for reward in data["rewards"]: reward_instance, created = await Reward.objects.aupdate_or_create(twitch_id=reward["id"]) await reward_instance.aimport_json(reward, self) if created: logger.info("Added reward %s to %s", reward_instance, self) return self def get_reward_banner_image_path(instance: models.Model, filename: str) -> str: """Get the path for the reward banner image. Args: instance (models.Model): The instance of the model. Is a Reward. filename (str): The filename of the image. Returns: str: The path to the image. """ instance = cast(Reward, instance) # Example: reward/banner_509658.png image_path: str = f"reward/banner_{filename}" logger.debug("Saved image to %s", image_path) return image_path def get_reward_thumbnail_image_path(instance: models.Model, filename: str) -> str: """Get the path for the reward thumbnail image. Args: instance (models.Model): The instance of the model. Is a Reward. filename (str): The filename of the image. Returns: str: The path to the image. """ instance = cast(Reward, instance) # Example: reward/thumb_509658.png image_path: str = f"reward/thumb_{filename}" logger.debug("Saved image to %s", image_path) return image_path class Reward(models.Model): """This from the RewardCampaign.""" # "dc2e9810-4de0-11ef-9ec3-621fb0811846" twitch_id = models.TextField(primary_key=True) # When the reward was first added to the database. created_at = models.DateTimeField(null=True, auto_created=True) # When the reward was last modified. modified_at = models.DateTimeField(null=True, auto_now=True) # "3 months of Apple TV+" name = models.TextField(null=True) # "https://static-cdn.jtvnw.net/twitch-quests-assets/CAMPAIGN/quests_appletv_q3_2024/apple_200x200.png" banner_image_url = models.URLField(null=True) banner_image = models.ImageField(null=True, upload_to=get_reward_banner_image_path) # "https://static-cdn.jtvnw.net/twitch-quests-assets/CAMPAIGN/quests_appletv_q3_2024/apple_200x200.png" thumbnail_image_url = models.URLField(null=True) thumbnail_image = models.ImageField(null=True, upload_to=get_reward_thumbnail_image_path) # "2024-08-19T19:00:00Z" earnable_until = models.DateTimeField(null=True) # "" redemption_instructions = models.TextField(null=True) # "https://tv.apple.com/includes/commerce/redeem/code-entry" redemption_url = models.URLField(null=True) campaign = models.ForeignKey(RewardCampaign, on_delete=models.CASCADE, related_name="rewards", null=True) class Meta: ordering: ClassVar[list[str]] = ["-earnable_until"] def __str__(self) -> str: return self.name or "Reward name unknown" def download_banner_image(self) -> ImageFieldFile | None: """Download the banner image for the reward. Returns: ImageFieldFile | None: The image file or None if it doesn't exist. """ # We don't want to re-download the image if it already exists. # TODO(TheLovinator): Check if there is a different image available. # noqa: TD003 if self.banner_image: return self.banner_image if not self.banner_image_url: return None if not self.banner_image and self.banner_image_url: response: bytes | None = fetch_image(image_url=self.banner_image_url) image: File | None = convert_image_to_webp(response) if image: self.banner_image.save(name=f"{self.twitch_id}.{image_file_format}", content=image, save=True) logger.info("Downloaded image for %s to %s", self, self.banner_image.url) return None def download_thumbnail_image(self) -> ImageFieldFile | None: """Download the thumbnail image for the reward. Returns: ImageFieldFile | None: The image file or None if it doesn't exist. """ # We don't want to re-download the image if it already exists. # TODO(TheLovinator): Check if there is a different image available. # noqa: TD003 if self.thumbnail_image: return self.thumbnail_image if not self.thumbnail_image_url: return None if not self.thumbnail_image and self.thumbnail_image_url: response: bytes | None = fetch_image(image_url=self.thumbnail_image_url) image: File | None = convert_image_to_webp(response) if image: self.thumbnail_image.save(name=f"{self.twitch_id}.{image_file_format}", content=image, save=True) logger.info("Downloaded image for %s to %s", self, self.thumbnail_image.url) return None async def aimport_json(self, data: dict, reward_campaign: RewardCampaign | None) -> Self: if wrong_typename(data, "Reward"): return self field_mapping: dict[str, str] = { "name": "name", "earnableUntil": "earnable_until", "redemptionInstructions": "redemption_instructions", "redemptionURL": "redemption_url", } updated: int = await update_fields(instance=self, data=data, field_mapping=field_mapping) if updated > 0: logger.info("Updated %s fields for %s", updated, self) banner_image_url = data.get("bannerImage", {}).get("image1xURL") if banner_image_url and banner_image_url != self.banner_image_url: # await sync_to_async(self.download_banner_image)() # TODO(TheLovinator): Download the image. # noqa: TD003 self.banner_image_url = banner_image_url await self.asave() thumbnail_image_url = data.get("thumbnailImage", {}).get("image1xURL") if thumbnail_image_url and thumbnail_image_url != self.thumbnail_image_url: # await sync_to_async(self.download_thumbnail_image)() # TODO(TheLovinator): Download the image. # noqa: TD003 self.thumbnail_image_url = thumbnail_image_url await self.asave() if reward_campaign and await sync_to_async(lambda: reward_campaign != self.campaign)(): self.campaign = reward_campaign await self.asave() return self class Webhook(models.Model): """Discord webhook.""" id = models.TextField(primary_key=True) avatar = models.TextField(null=True) channel_id = models.TextField(null=True) guild_id = models.TextField(null=True) name = models.TextField(null=True) type = models.TextField(null=True) token = models.TextField() url = models.TextField() # When the webhook was first added to the database. created_at = models.DateTimeField(null=True, auto_created=True) # When the webhook was last modified. modified_at = models.DateTimeField(null=True, auto_now=True) # Get notified when the site finds a new game. subscribed_new_games = models.ManyToManyField(Game, related_name="subscribed_new_games") # Get notified when a drop goes live. subscribed_live_games = models.ManyToManyField(Game, related_name="subscribed_live_games") # Get notified when the site finds a new drop campaign for a specific organization. subscribed_new_owners = models.ManyToManyField(Owner, related_name="subscribed_new_owners") # Get notified when a drop goes live for a specific organization. subscribed_live_owners = models.ManyToManyField(Owner, related_name="subscribed_live_owners") # So we don't spam the same drop campaign over and over. seen_drops = models.ManyToManyField(DropCampaign, related_name="seen_drops") class Meta: unique_together: ClassVar[list[str]] = ["id", "token"] def __str__(self) -> str: return f"{self.name} - {self.get_webhook_url()}" def get_webhook_url(self) -> str: try: return f"https://discord.com/api/webhooks/{self.id}/{self.token}" except AttributeError: return ""