|
+
+ {% if campaign.brand %}
+ {{ campaign.brand }}: {{ campaign.name }}
+ {% else %}
+ {{ campaign.name }}
+ {% endif %}
+
+ {% if campaign.summary %}
+ + {{ campaign.summary }} + {% endif %} + |
+ + {% if campaign.game %} + {{ campaign.game.display_name }} + {% elif campaign.is_sitewide %} + Site-wide + {% endif %} + | ++ Ends in {{ campaign.ends_at|timeuntil }} + | +
|
+
+ {% if campaign.brand %}
+ {{ campaign.brand }}: {{ campaign.name }}
+ {% else %}
+ {{ campaign.name }}
+ {% endif %}
+
+ {% if campaign.summary %}
+ + {{ campaign.summary }} + {% endif %} + |
+ + {% if campaign.game %} + {{ campaign.game.display_name }} + {% elif campaign.is_sitewide %} + Site-wide + {% endif %} + | ++ Starts in {{ campaign.starts_at|timeuntil }} + | +
|
+
+ {% if campaign.brand %}
+ {{ campaign.brand }}: {{ campaign.name }}
+ {% else %}
+ {{ campaign.name }}
+ {% endif %}
+
+ {% if campaign.summary %}
+ + {{ campaign.summary }} + {% endif %} + |
+ + {% if campaign.game %} + {{ campaign.game.display_name }} + {% elif campaign.is_sitewide %} + Site-wide + {% endif %} + | ++ {{ campaign.ends_at|timesince }} ago + | +
No reward campaigns found.
+ {% endif %} +{% endblock content %} diff --git a/templates/twitch/search_results.html b/templates/twitch/search_results.html index 82874df..f0ce28b 100644 --- a/templates/twitch/search_results.html +++ b/templates/twitch/search_results.html @@ -5,7 +5,7 @@ {% block content %}No results found.
{% else %} {% if results.organizations %} @@ -68,6 +68,20 @@ {% endfor %} {% endif %} + {% if results.reward_campaigns %} +{}
", summary)) + + # Insert start and end date info (uses starts_at/ends_at instead of start_at/end_at) + ends_at: datetime.datetime | None = getattr(item, "ends_at", None) + starts_at: datetime.datetime | None = getattr(item, "starts_at", None) + + if starts_at or ends_at: + start_part: SafeString = ( + format_html("Starts: {} ({})", starts_at.strftime("%Y-%m-%d %H:%M %Z"), naturaltime(starts_at)) + if starts_at + else SafeText("") + ) + end_part: SafeString = ( + format_html("Ends: {} ({})", ends_at.strftime("%Y-%m-%d %H:%M %Z"), naturaltime(ends_at)) + if ends_at + else SafeText("") + ) + if start_part and end_part: + parts.append(format_html("{}
{}
{}
", start_part)) + elif end_part: + parts.append(format_html("{}
", end_part)) + + is_sitewide: bool = getattr(item, "is_sitewide", False) + if is_sitewide: + parts.append(SafeText("This is a sitewide reward campaign
")) + else: + game: Game | None = getattr(item, "game", None) + if game: + parts.append(format_html("Game: {}
", game.display_name or game.name)) + + about_url: str | None = getattr(item, "about_url", None) + if about_url: + parts.append(format_html('', about_url)) + + external_url: str | None = getattr(item, "external_url", None) + if external_url: + parts.append(format_html('', external_url)) + + return SafeText("".join(str(p) for p in parts)) + + def item_link(self, item: Model) -> str: + """Return the link to the reward campaign (external URL or dashboard).""" + external_url: str | None = getattr(item, "external_url", None) + if external_url: + return external_url + return reverse("twitch:dashboard") + + def item_pubdate(self, item: Model) -> datetime.datetime: + """Returns the publication date to the feed item. + + Fallback to added_at or now if missing. + """ + added_at: datetime.datetime | None = getattr(item, "added_at", None) + if added_at: + return added_at + return timezone.now() + + def item_updateddate(self, item: RewardCampaign) -> datetime.datetime: + """Returns the reward campaign's last update time.""" + return item.updated_at + + def item_categories(self, item: RewardCampaign) -> tuple[str, ...]: + """Returns the associated game's name and brand as categories.""" + categories: list[str] = ["twitch", "rewards", "quests"] + + brand: str | None = getattr(item, "brand", None) + if brand: + categories.append(brand) + + item_game: Game | None = getattr(item, "game", None) + if item_game: + categories.append(item_game.get_game_name) + + return tuple(categories) + + def item_guid(self, item: RewardCampaign) -> str: + """Return a unique identifier for each reward campaign.""" + return item.twitch_id + "@ttvdrops.com" + + def item_author_name(self, item: RewardCampaign) -> str: + """Return the author name for the reward campaign.""" + brand: str | None = getattr(item, "brand", None) + if brand: + return brand + + item_game: Game | None = getattr(item, "game", None) + if item_game and item_game.display_name: + return item_game.display_name + + return "Twitch" diff --git a/twitch/management/commands/better_import_drops.py b/twitch/management/commands/better_import_drops.py index c86b235..40e15b8 100644 --- a/twitch/management/commands/better_import_drops.py +++ b/twitch/management/commands/better_import_drops.py @@ -27,6 +27,7 @@ from twitch.models import DropBenefitEdge from twitch.models import DropCampaign from twitch.models import Game from twitch.models import Organization +from twitch.models import RewardCampaign from twitch.models import TimeBasedDrop from twitch.schemas import ChannelInfoSchema from twitch.schemas import CurrentUserSchema @@ -37,6 +38,7 @@ from twitch.schemas import DropCampaignSchema from twitch.schemas import GameSchema from twitch.schemas import GraphQLResponse from twitch.schemas import OrganizationSchema +from twitch.schemas import RewardCampaign as RewardCampaignSchema from twitch.schemas import TimeBasedDropSchema from twitch.utils import parse_date @@ -852,6 +854,13 @@ class Command(BaseCommand): allow_schema=drop_campaign.allow, ) + # Process reward campaigns if present + if response.data.reward_campaigns_available_to_user: + self._process_reward_campaigns( + reward_campaigns=response.data.reward_campaigns_available_to_user, + options=options, + ) + return True, None def _process_time_based_drops( @@ -989,6 +998,73 @@ class Command(BaseCommand): # Update the M2M relationship with the allowed channels campaign_obj.allow_channels.set(channel_objects) + def _process_reward_campaigns( + self, + reward_campaigns: list[RewardCampaignSchema], + options: dict[str, Any], + ) -> None: + """Process reward campaigns from the API response. + + Args: + reward_campaigns: List of RewardCampaign Pydantic schemas. + options: Command options dictionary. + + Raises: + ValueError: If datetime parsing fails for campaign dates and + crash-on-error is enabled. + """ + for reward_campaign in reward_campaigns: + starts_at_dt: datetime | None = parse_date(reward_campaign.starts_at) + ends_at_dt: datetime | None = parse_date(reward_campaign.ends_at) + + if starts_at_dt is None or ends_at_dt is None: + tqdm.write(f"{Fore.RED}✗{Style.RESET_ALL} Invalid datetime in reward campaign: {reward_campaign.name}") + if options.get("crash_on_error"): + msg: str = f"Failed to parse datetime for reward campaign {reward_campaign.name}" + raise ValueError(msg) + continue + + # Handle game reference if present + game_obj: Game | None = None + if reward_campaign.game: + # The game field in reward campaigns is a dict, not a full GameSchema + # We'll try to find an existing game by twitch_id if available + game_id = reward_campaign.game.get("id") + if game_id: + try: + game_obj = Game.objects.get(twitch_id=game_id) + except Game.DoesNotExist: + if options.get("verbose"): + tqdm.write( + f"{Fore.YELLOW}→{Style.RESET_ALL} Game not found for reward campaign: {game_id}", + ) + + defaults: dict[str, str | datetime | Game | bool | None] = { + "name": reward_campaign.name, + "brand": reward_campaign.brand, + "starts_at": starts_at_dt, + "ends_at": ends_at_dt, + "status": reward_campaign.status, + "summary": reward_campaign.summary, + "instructions": reward_campaign.instructions, + "external_url": reward_campaign.external_url, + "reward_value_url_param": reward_campaign.reward_value_url_param, + "about_url": reward_campaign.about_url, + "is_sitewide": reward_campaign.is_sitewide, + "game": game_obj, + } + + _reward_campaign_obj, created = RewardCampaign.objects.update_or_create( + twitch_id=reward_campaign.twitch_id, + defaults=defaults, + ) + + action: Literal["Imported new", "Updated"] = "Imported new" if created else "Updated" + display_name = ( + f"{reward_campaign.brand}: {reward_campaign.name}" if reward_campaign.brand else reward_campaign.name + ) + tqdm.write(f"{Fore.GREEN}✓{Style.RESET_ALL} {action} reward campaign: {display_name}") + def handle(self, *args, **options) -> None: # noqa: ARG002 """Main entry point for the command. diff --git a/twitch/migrations/0005_add_reward_campaign.py b/twitch/migrations/0005_add_reward_campaign.py new file mode 100644 index 0000000..42add96 --- /dev/null +++ b/twitch/migrations/0005_add_reward_campaign.py @@ -0,0 +1,120 @@ +# Generated by Django 6.0.1 on 2026-01-13 20:31 +from __future__ import annotations + +import django.db.models.deletion +from django.db import migrations +from django.db import models + + +class Migration(migrations.Migration): + """Add RewardCampaign model.""" + + dependencies = [ + ("twitch", "0004_remove_game_twitch_game_owner_i_398fa9_idx_and_more"), + ] + + operations = [ + migrations.CreateModel( + name="RewardCampaign", + fields=[ + ("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")), + ( + "twitch_id", + models.TextField(editable=False, help_text="The Twitch ID for this reward campaign.", unique=True), + ), + ("name", models.TextField(help_text="Name of the reward campaign.")), + ( + "brand", + models.TextField(blank=True, default="", help_text="Brand associated with the reward campaign."), + ), + ( + "starts_at", + models.DateTimeField(blank=True, help_text="Datetime when the reward campaign starts.", null=True), + ), + ( + "ends_at", + models.DateTimeField(blank=True, help_text="Datetime when the reward campaign ends.", null=True), + ), + ( + "status", + models.TextField(default="UNKNOWN", help_text="Status of the reward campaign.", max_length=50), + ), + ( + "summary", + models.TextField(blank=True, default="", help_text="Summary description of the reward campaign."), + ), + ( + "instructions", + models.TextField(blank=True, default="", help_text="Instructions for the reward campaign."), + ), + ( + "external_url", + models.URLField( + blank=True, + default="", + help_text="External URL for the reward campaign.", + max_length=500, + ), + ), + ( + "reward_value_url_param", + models.TextField(blank=True, default="", help_text="URL parameter for reward value."), + ), + ( + "about_url", + models.URLField( + blank=True, + default="", + help_text="About URL for the reward campaign.", + max_length=500, + ), + ), + ( + "is_sitewide", + models.BooleanField(default=False, help_text="Whether the reward campaign is sitewide."), + ), + ( + "added_at", + models.DateTimeField( + auto_now_add=True, + help_text="Timestamp when this reward campaign record was created.", + ), + ), + ( + "updated_at", + models.DateTimeField( + auto_now=True, + help_text="Timestamp when this reward campaign record was last updated.", + ), + ), + ( + "game", + models.ForeignKey( + blank=True, + help_text="Game associated with this reward campaign (if any).", + null=True, + on_delete=django.db.models.deletion.SET_NULL, + related_name="reward_campaigns", + to="twitch.game", + ), + ), + ], + options={ + "ordering": ["-starts_at"], + "indexes": [ + models.Index(fields=["-starts_at"], name="twitch_rewa_starts__4df564_idx"), + models.Index(fields=["ends_at"], name="twitch_rewa_ends_at_354b15_idx"), + models.Index(fields=["twitch_id"], name="twitch_rewa_twitch__797967_idx"), + models.Index(fields=["name"], name="twitch_rewa_name_f1e3dd_idx"), + models.Index(fields=["brand"], name="twitch_rewa_brand_41c321_idx"), + models.Index(fields=["status"], name="twitch_rewa_status_a96d6b_idx"), + models.Index(fields=["is_sitewide"], name="twitch_rewa_is_site_7d2c9f_idx"), + models.Index(fields=["game"], name="twitch_rewa_game_id_678fbb_idx"), + models.Index(fields=["added_at"], name="twitch_rewa_added_a_ae3748_idx"), + models.Index(fields=["updated_at"], name="twitch_rewa_updated_fdf599_idx"), + models.Index(fields=["starts_at", "ends_at"], name="twitch_rewa_starts__dd909d_idx"), + models.Index(fields=["status", "-starts_at"], name="twitch_rewa_status_3641a4_idx"), + ], + }, + ), + ] diff --git a/twitch/models.py b/twitch/models.py index bac55bd..a25c36d 100644 --- a/twitch/models.py +++ b/twitch/models.py @@ -652,3 +652,115 @@ class TimeBasedDrop(models.Model): def __str__(self) -> str: """Return a string representation of the time-based drop.""" return self.name + + +# MARK: RewardCampaign +class RewardCampaign(models.Model): + """Represents a Twitch reward campaign (Quest rewards).""" + + twitch_id = models.TextField( + unique=True, + editable=False, + help_text="The Twitch ID for this reward campaign.", + ) + name = models.TextField( + help_text="Name of the reward campaign.", + ) + brand = models.TextField( + blank=True, + default="", + help_text="Brand associated with the reward campaign.", + ) + starts_at = models.DateTimeField( + null=True, + blank=True, + help_text="Datetime when the reward campaign starts.", + ) + ends_at = models.DateTimeField( + null=True, + blank=True, + help_text="Datetime when the reward campaign ends.", + ) + status = models.TextField( + max_length=50, + default="UNKNOWN", + help_text="Status of the reward campaign.", + ) + summary = models.TextField( + blank=True, + default="", + help_text="Summary description of the reward campaign.", + ) + instructions = models.TextField( + blank=True, + default="", + help_text="Instructions for the reward campaign.", + ) + external_url = models.URLField( + max_length=500, + blank=True, + default="", + help_text="External URL for the reward campaign.", + ) + reward_value_url_param = models.TextField( + blank=True, + default="", + help_text="URL parameter for reward value.", + ) + about_url = models.URLField( + max_length=500, + blank=True, + default="", + help_text="About URL for the reward campaign.", + ) + is_sitewide = models.BooleanField( + default=False, + help_text="Whether the reward campaign is sitewide.", + ) + game = models.ForeignKey( + Game, + on_delete=models.SET_NULL, + null=True, + blank=True, + related_name="reward_campaigns", + help_text="Game associated with this reward campaign (if any).", + ) + + added_at = models.DateTimeField( + auto_now_add=True, + help_text="Timestamp when this reward campaign record was created.", + ) + updated_at = models.DateTimeField( + auto_now=True, + help_text="Timestamp when this reward campaign record was last updated.", + ) + + class Meta: + ordering = ["-starts_at"] + indexes = [ + models.Index(fields=["-starts_at"]), + models.Index(fields=["ends_at"]), + models.Index(fields=["twitch_id"]), + models.Index(fields=["name"]), + models.Index(fields=["brand"]), + models.Index(fields=["status"]), + models.Index(fields=["is_sitewide"]), + models.Index(fields=["game"]), + models.Index(fields=["added_at"]), + models.Index(fields=["updated_at"]), + # Composite indexes for common queries + models.Index(fields=["starts_at", "ends_at"]), + models.Index(fields=["status", "-starts_at"]), + ] + + def __str__(self) -> str: + """Return a string representation of the reward campaign.""" + return f"{self.brand}: {self.name}" if self.brand else self.name + + @property + def is_active(self) -> bool: + """Check if the reward campaign is currently active.""" + now: datetime.datetime = timezone.now() + if self.starts_at is None or self.ends_at is None: + return False + return self.starts_at <= now <= self.ends_at diff --git a/twitch/schemas.py b/twitch/schemas.py index b110432..79022c7 100644 --- a/twitch/schemas.py +++ b/twitch/schemas.py @@ -365,12 +365,16 @@ class DataSchema(BaseModel): """Schema for the data field in Twitch API responses. Handles both currentUser (standard) and user (legacy) field names, - as well as channel-based campaign structures. + as well as channel-based campaign structures and reward campaigns. """ current_user: CurrentUserSchema | None = Field(default=None, alias="currentUser") user: CurrentUserSchema | None = Field(default=None, alias="user") channel: ChannelSchema | None = Field(default=None, alias="channel") + reward_campaigns_available_to_user: list[RewardCampaign] | None = Field( + default=None, + alias="rewardCampaignsAvailableToUser", + ) model_config = { "extra": "forbid", @@ -409,6 +413,84 @@ class DataSchema(BaseModel): return self +class QuestRewardUnlockRequirements(BaseModel): + """Schema for quest reward unlock requirements.""" + + subs_goal: int | None = Field(default=None, alias="subsGoal") + minute_watched_goal: int | None = Field(default=None, alias="minuteWatchedGoal") + type_name: Literal["QuestRewardUnlockRequirements"] = Field(alias="__typename") + + model_config = { + "extra": "forbid", + "validate_assignment": True, + "strict": True, + "populate_by_name": True, + } + + +class RewardCampaignImageSet(BaseModel): + """Schema for reward campaign image sets.""" + + image1x_url: str | None = Field(default=None, alias="image1xURL") + type_name: Literal["RewardCampaignImageSet"] = Field(alias="__typename") + + model_config = { + "extra": "forbid", + "validate_assignment": True, + "strict": True, + "populate_by_name": True, + } + + +class Reward(BaseModel): + """Schema for a reward in a RewardCampaign.""" + + twitch_id: str = Field(alias="id") + name: str + banner_image: RewardCampaignImageSet | None = Field(default=None, alias="bannerImage") + thumbnail_image: RewardCampaignImageSet | None = Field(default=None, alias="thumbnailImage") + earnable_until: str | None = Field(default=None, alias="earnableUntil") + redemption_instructions: str = Field(default="", alias="redemptionInstructions") + redemption_url: str = Field(default="", alias="redemptionURL") + type_name: Literal["Reward"] = Field(alias="__typename") + + model_config = { + "extra": "forbid", + "validate_assignment": True, + "strict": True, + "populate_by_name": True, + } + + +class RewardCampaign(BaseModel): + """Schema for a RewardCampaign from rewardCampaignsAvailableToUser.""" + + twitch_id: str = Field(alias="id") + name: str + brand: str + starts_at: str = Field(alias="startsAt") + ends_at: str = Field(alias="endsAt") + status: str + summary: str = Field(default="") + instructions: str = Field(default="") + external_url: str = Field(default="", alias="externalURL") + reward_value_url_param: str = Field(default="", alias="rewardValueURLParam") + about_url: str = Field(default="", alias="aboutURL") + is_sitewide: bool = Field(default=False, alias="isSitewide") + game: dict | None = None + unlock_requirements: QuestRewardUnlockRequirements | None = Field(default=None, alias="unlockRequirements") + image: RewardCampaignImageSet | None = None + rewards: list[Reward] = Field(default=[]) + type_name: Literal["RewardCampaign"] = Field(alias="__typename") + + model_config = { + "extra": "forbid", + "validate_assignment": True, + "strict": True, + "populate_by_name": True, + } + + class Extensions(BaseModel): """Schema for the extensions field in GraphQL responses.""" diff --git a/twitch/tests/test_schemas.py b/twitch/tests/test_schemas.py index 3dc5059..f9516a0 100644 --- a/twitch/tests/test_schemas.py +++ b/twitch/tests/test_schemas.py @@ -376,3 +376,104 @@ def test_drop_campaign_details_missing_distribution_type() -> None: benefit: DropBenefitSchema = first_drop.benefit_edges[0].benefit assert benefit.name == "13.7 Update: 250 CT" assert benefit.distribution_type is None # This field was missing in the API response + + +def test_reward_campaigns_available_to_user() -> None: + """Test that rewardCampaignsAvailableToUser field validates correctly. + + The ViewerDropsDashboard operation can include reward campaigns (Quest rewards) + alongside drop campaigns. This test verifies that the schema properly handles + this additional data. + """ + payload: dict[str, object] = { + "data": { + "currentUser": { + "id": "58162970", + "login": "lovibot", + "dropCampaigns": [], + "__typename": "User", + }, + "rewardCampaignsAvailableToUser": [ + { + "id": "dc4ff0b4-4de0-11ef-9ec3-621fb0811846", + "name": "Buy 1 new sub, get 3 months of Apple TV+", + "brand": "Apple TV+", + "startsAt": "2024-07-30T19:00:00Z", + "endsAt": "2024-08-19T19:00:00Z", + "status": "UNKNOWN", + "summary": "Get 3 months of Apple TV+", + "instructions": "", + "externalURL": "https://tv.apple.com/includes/commerce/redeem/code-entry", + "rewardValueURLParam": "", + "aboutURL": "https://blog.twitch.tv/2024/07/26/sub-and-get-apple-tv/", + "isSitewide": True, + "game": None, + "unlockRequirements": { + "subsGoal": 1, + "minuteWatchedGoal": 0, + "__typename": "QuestRewardUnlockRequirements", + }, + "image": { + "image1xURL": "https://static-cdn.jtvnw.net/twitch-quests-assets/CAMPAIGN/quests_appletv_q3_2024/apple_150x200.png", + "__typename": "RewardCampaignImageSet", + }, + "rewards": [ + { + "id": "dc2e9810-4de0-11ef-9ec3-621fb0811846", + "name": "3 months of Apple TV+", + "bannerImage": { + "image1xURL": "https://static-cdn.jtvnw.net/twitch-quests-assets/CAMPAIGN/quests_appletv_q3_2024/apple_200x200.png", + "__typename": "RewardCampaignImageSet", + }, + "thumbnailImage": { + "image1xURL": "https://static-cdn.jtvnw.net/twitch-quests-assets/CAMPAIGN/quests_appletv_q3_2024/apple_200x200.png", + "__typename": "RewardCampaignImageSet", + }, + "earnableUntil": "2024-08-19T19:00:00Z", + "redemptionInstructions": "", + "redemptionURL": "https://tv.apple.com/includes/commerce/redeem/code-entry", + "__typename": "Reward", + }, + ], + "__typename": "RewardCampaign", + }, + ], + }, + "extensions": { + "operationName": "ViewerDropsDashboard", + }, + } + + # This should not raise ValidationError + response: GraphQLResponse = GraphQLResponse.model_validate(payload) + + # Verify the reward campaigns were parsed correctly + assert response.data.reward_campaigns_available_to_user is not None + assert len(response.data.reward_campaigns_available_to_user) == 1 + + reward_campaign = response.data.reward_campaigns_available_to_user[0] + assert reward_campaign.twitch_id == "dc4ff0b4-4de0-11ef-9ec3-621fb0811846" + assert reward_campaign.name == "Buy 1 new sub, get 3 months of Apple TV+" + assert reward_campaign.brand == "Apple TV+" + assert reward_campaign.is_sitewide is True + assert reward_campaign.game is None + + # Verify unlock requirements + assert reward_campaign.unlock_requirements is not None + assert reward_campaign.unlock_requirements.subs_goal == 1 + assert reward_campaign.unlock_requirements.minute_watched_goal == 0 + + # Verify image + assert reward_campaign.image is not None + assert ( + reward_campaign.image.image1x_url + == "https://static-cdn.jtvnw.net/twitch-quests-assets/CAMPAIGN/quests_appletv_q3_2024/apple_150x200.png" + ) + + # Verify rewards + assert len(reward_campaign.rewards) == 1 + reward = reward_campaign.rewards[0] + assert reward.twitch_id == "dc2e9810-4de0-11ef-9ec3-621fb0811846" + assert reward.name == "3 months of Apple TV+" + assert reward.banner_image is not None + assert reward.thumbnail_image is not None diff --git a/twitch/urls.py b/twitch/urls.py index 494f51e..e7bf406 100644 --- a/twitch/urls.py +++ b/twitch/urls.py @@ -10,6 +10,7 @@ from twitch.feeds import GameCampaignFeed from twitch.feeds import GameFeed from twitch.feeds import OrganizationCampaignFeed from twitch.feeds import OrganizationFeed +from twitch.feeds import RewardCampaignFeed if TYPE_CHECKING: from django.urls.resolvers import URLPattern @@ -30,10 +31,13 @@ urlpatterns: list[URLPattern] = [ path("games/