diff --git a/.env.example b/.env.example index 4ebfc42..ea1c6e9 100644 --- a/.env.example +++ b/.env.example @@ -6,6 +6,12 @@ DEBUG=True # Generate a new secret key for production: python -c 'from django.core.management.utils import get_random_secret_key; print(get_random_secret_key())' DJANGO_SECRET_KEY=your-secret-key-here +# Twitch API Configuration +# Get your Client ID and Client Secret from https://dev.twitch.tv/console +# You can use either an app access token or user access token +TWITCH_CLIENT_ID=your-twitch-client-id +TWITCH_CLIENT_SECRET=your-twitch-client-secret + # Email Configuration # SMTP Host (examples below) EMAIL_HOST=smtp.gmail.com diff --git a/README.md b/README.md index 3fd9d87..10b915a 100644 --- a/README.md +++ b/README.md @@ -18,3 +18,13 @@ uv run pytest ```bash uv run python manage.py better_import_drops [--recursive] [--verbose] [--crash-on-error] [--skip-broken-moves] ``` + +## Import Chat Badges + +Import Twitch's global chat badges for archival and reference: + +```bash +uv run python manage.py import_chat_badges +``` + +Requires `TWITCH_CLIENT_ID` and `TWITCH_CLIENT_SECRET` environment variables to be set. diff --git a/templates/base.html b/templates/base.html index 1864394..492ff0b 100644 --- a/templates/base.html +++ b/templates/base.html @@ -162,6 +162,7 @@ Games | Orgs | Channels | + Badges | Emotes
RSS | Debug diff --git a/templates/twitch/badge_list.html b/templates/twitch/badge_list.html new file mode 100644 index 0000000..9c31657 --- /dev/null +++ b/templates/twitch/badge_list.html @@ -0,0 +1,48 @@ +{% extends "base.html" %} +{% block title %} + Chat Badges - ttvdrops +{% endblock title %} +{% block content %} +

Twitch Chat Badges

+ 
+These are the global chat badges available on Twitch.
+
+ {% if badge_sets %} +

total badge sets: {{ badge_sets.count }}

+ {% for data in badge_data %} +
+

+ [{{ data.set.set_id }}] +

+ {% for badge in data.badges %} + + + + + +
+ + {{ badge.title }} + + + {{ badge.title }} +
+
+ {{ badge.description }} +
+ {% endfor %} +
+ {% if data.badges|length > 1 %}versions: {{ data.badges|length }}{% endif %} + {% endfor %} +
+ {% else %} +

no badge sets found.

+

+ run: uv run python manage.py import_chat_badges +

+ {% endif %} +{% endblock content %} diff --git a/templates/twitch/badge_set_detail.html b/templates/twitch/badge_set_detail.html new file mode 100644 index 0000000..a7fd956 --- /dev/null +++ b/templates/twitch/badge_set_detail.html @@ -0,0 +1,79 @@ +{% extends "base.html" %} +{% block title %} + {{ badge_set.set_id }} Badges - ttvdrops +{% endblock title %} +{% block content %} +

+ Badge Set: {{ badge_set.set_id }} +

+

+ Back to all badges +

+ + {% if badges %} +

Badge Versions ({{ badges.count }})

+ + + + + + + + + + + + + {% for badge in badges %} + + + + + + + + + {% endfor %} + +
IDPreviewTitleDescriptionImagesAction
+ {{ badge.badge_id }} + + {{ badge.title }} + + {{ badge.title }} + {{ badge.description }} + 18px | + 36px | + 72px + + {% if badge.click_url %} + {{ badge.click_action|default:"visit_url" }} + {% else %} + None + {% endif %} +
+ {% else %} +

No badges found in this set.

+ {% endif %} + {{ set_data|safe }} +{% endblock content %} diff --git a/templates/twitch/search_results.html b/templates/twitch/search_results.html index f0ce28b..dc5ed61 100644 --- a/templates/twitch/search_results.html +++ b/templates/twitch/search_results.html @@ -5,7 +5,7 @@ {% block content %}

Search Results for "{{ query }}"

- {% if not results.organizations and not results.games and not results.campaigns and not results.drops and not results.benefits and not results.reward_campaigns %} + {% if not results.organizations and not results.games and not results.campaigns and not results.drops and not results.benefits and not results.reward_campaigns and not results.badge_sets and not results.badges %}

No results found.

{% else %} {% if results.organizations %} @@ -82,6 +82,27 @@ {% endfor %} {% endif %} + {% if results.badge_sets %} +

Badge Sets

+ + {% endif %} + {% if results.badges %} +

Chat Badges

+ + {% endif %} {% endif %}
{% endblock content %} diff --git a/twitch/management/commands/import_chat_badges.py b/twitch/management/commands/import_chat_badges.py new file mode 100644 index 0000000..405554d --- /dev/null +++ b/twitch/management/commands/import_chat_badges.py @@ -0,0 +1,273 @@ +"""Management command to import Twitch global chat badges.""" + +from __future__ import annotations + +import logging +import os +from typing import TYPE_CHECKING +from typing import Any + +import httpx +from colorama import Fore +from colorama import Style +from colorama import init as colorama_init +from django.core.management.base import BaseCommand +from django.core.management.base import CommandError +from django.core.management.base import CommandParser +from pydantic import ValidationError + +from twitch.models import ChatBadge +from twitch.models import ChatBadgeSet +from twitch.schemas import ChatBadgeSetSchema +from twitch.schemas import GlobalChatBadgesResponse + +if TYPE_CHECKING: + from twitch.schemas import ChatBadgeVersionSchema + +logger: logging.Logger = logging.getLogger("ttvdrops") + + +class Command(BaseCommand): + """Import Twitch global chat badges from the Twitch Helix API.""" + + help = "Import Twitch global chat badges from the Twitch Helix API" + requires_migrations_checks = True + + def add_arguments(self, parser: CommandParser) -> None: + """Add command arguments.""" + parser.add_argument( + "--client-id", + type=str, + help="Twitch Client ID (or set TWITCH_CLIENT_ID environment variable)", + ) + parser.add_argument( + "--client-secret", + type=str, + help="Twitch Client Secret (or set TWITCH_CLIENT_SECRET environment variable)", + ) + parser.add_argument( + "--access-token", + type=str, + help="Twitch Access Token (optional - will be obtained automatically if not provided)", + ) + + def handle(self, *args, **options) -> None: # noqa: ARG002 + """Main entry point for the command. + + Raises: + CommandError: If required arguments are missing or API calls fail + """ + colorama_init(autoreset=True) + + # Get credentials from arguments or environment + client_id: str | None = options.get("client_id") or os.getenv("TWITCH_CLIENT_ID") + client_secret: str | None = options.get("client_secret") or os.getenv("TWITCH_CLIENT_SECRET") + access_token: str | None = options.get("access_token") or os.getenv("TWITCH_ACCESS_TOKEN") + + if not client_id: + msg = ( + "Twitch Client ID is required. " + "Provide it via --client-id argument or set TWITCH_CLIENT_ID environment variable." + ) + raise CommandError(msg) + + # If access token is not provided, obtain it automatically using client credentials + if not access_token: + if not client_secret: + msg = ( + "Either --access-token or --client-secret must be provided. " + "Set TWITCH_ACCESS_TOKEN or TWITCH_CLIENT_SECRET environment variable, " + "or provide them via command arguments." + ) + raise CommandError(msg) + + self.stdout.write("Obtaining access token from Twitch...") + try: + access_token = self._get_app_access_token(client_id, client_secret) + self.stdout.write(self.style.SUCCESS("✓ Access token obtained successfully")) + except httpx.HTTPError as e: + msg = f"Failed to obtain access token: {e}" + raise CommandError(msg) from e + + self.stdout.write("Fetching global chat badges from Twitch API...") + + try: + badges_data: GlobalChatBadgesResponse = self._fetch_global_chat_badges( + client_id=client_id, + access_token=access_token, + ) + except httpx.HTTPError as e: + msg: str = f"Failed to fetch chat badges from Twitch API: {e}" + raise CommandError(msg) from e + except ValidationError as e: + msg: str = f"Failed to validate chat badges response: {e}" + raise CommandError(msg) from e + + self.stdout.write(f"Received {len(badges_data.data)} badge sets") + + # Process and store badge data + created_sets = 0 + updated_sets = 0 + created_badges = 0 + updated_badges = 0 + + for badge_set_schema in badges_data.data: + badge_set_obj, set_created = self._process_badge_set(badge_set_schema) + + if set_created: + created_sets += 1 + else: + updated_sets += 1 + + # Process each badge version in the set + for version_schema in badge_set_schema.versions: + badge_created: bool = self._process_badge_version( + badge_set_obj=badge_set_obj, + version_schema=version_schema, + ) + + if badge_created: + created_badges += 1 + else: + updated_badges += 1 + + # Print summary + self.stdout.write("\n" + "=" * 50) + self.stdout.write( + self.style.SUCCESS( + f"✓ Created {created_sets} new badge sets, updated {updated_sets} existing", + ), + ) + self.stdout.write( + self.style.SUCCESS( + f"✓ Created {created_badges} new badges, updated {updated_badges} existing", + ), + ) + self.stdout.write(f"Total badge sets: {created_sets + updated_sets}") + self.stdout.write(f"Total badges: {created_badges + updated_badges}") + self.stdout.write("=" * 50) + + def _fetch_global_chat_badges( + self, + client_id: str, + access_token: str, + ) -> GlobalChatBadgesResponse: + """Fetch global chat badges from Twitch Helix API. + + Args: + client_id: Twitch Client ID + access_token: Twitch Access Token (app or user token) + + Returns: + Validated GlobalChatBadgesResponse + """ + url = "https://api.twitch.tv/helix/chat/badges/global" + headers: dict[str, str] = { + "Authorization": f"Bearer {access_token}", + "Client-Id": client_id, + } + + with httpx.Client() as client: + response: httpx.Response = client.get(url, headers=headers, timeout=30.0) + response.raise_for_status() + + data: dict[str, Any] = response.json() + logger.debug("Received chat badges response: %s", data) + + # Validate response with Pydantic + return GlobalChatBadgesResponse.model_validate(data) + + def _process_badge_set( + self, + badge_set_schema: ChatBadgeSetSchema, + ) -> tuple[ChatBadgeSet, bool]: + """Get or create a ChatBadgeSet from the schema. + + Args: + badge_set_schema: Validated badge set schema + + Returns: + Tuple of (ChatBadgeSet instance, created flag) + """ + badge_set_obj, created = ChatBadgeSet.objects.get_or_create( + set_id=badge_set_schema.set_id, + ) + + if created: + self.stdout.write( + f"{Fore.GREEN}✓{Style.RESET_ALL} Created new badge set: {badge_set_schema.set_id}", + ) + else: + self.stdout.write( + f"{Fore.YELLOW}→{Style.RESET_ALL} Badge set already exists: {badge_set_schema.set_id}", + ) + + return badge_set_obj, created + + def _get_app_access_token(self, client_id: str, client_secret: str) -> str: + """Get an app access token from Twitch. + + Args: + client_id: Twitch Client ID + client_secret: Twitch Client Secret + + Returns: + Access token string + """ + url = "https://id.twitch.tv/oauth2/token" + params: dict[str, str] = { + "client_id": client_id, + "client_secret": client_secret, + "grant_type": "client_credentials", + } + + with httpx.Client() as client: + response: httpx.Response = client.post(url, params=params, timeout=30.0) + response.raise_for_status() + data: dict[str, str] = response.json() + return data["access_token"] + + def _process_badge_version( + self, + badge_set_obj: ChatBadgeSet, + version_schema: ChatBadgeVersionSchema, + ) -> bool: + """Get or create a ChatBadge from the schema. + + Args: + badge_set_obj: Parent ChatBadgeSet instance + version_schema: Validated badge version schema + + Returns: + True if created, False if updated + """ + defaults: dict[str, str | None] = { + "image_url_1x": version_schema.image_url_1x, + "image_url_2x": version_schema.image_url_2x, + "image_url_4x": version_schema.image_url_4x, + "title": version_schema.title, + "description": version_schema.description, + "click_action": version_schema.click_action, + "click_url": version_schema.click_url, + } + + _badge_obj, created = ChatBadge.objects.update_or_create( + badge_set=badge_set_obj, + badge_id=version_schema.badge_id, + defaults=defaults, + ) + + if created: + msg: str = ( + f"{Fore.GREEN}✓{Style.RESET_ALL} Created badge: " + f"{badge_set_obj.set_id}/{version_schema.badge_id} - {version_schema.title}" + ) + self.stdout.write(msg) + else: + msg: str = ( + f"{Fore.YELLOW}→{Style.RESET_ALL} Updated badge: " + f"{badge_set_obj.set_id}/{version_schema.badge_id} - {version_schema.title}" + ) + self.stdout.write(msg) + + return created diff --git a/twitch/migrations/0006_add_chat_badges.py b/twitch/migrations/0006_add_chat_badges.py new file mode 100644 index 0000000..6181519 --- /dev/null +++ b/twitch/migrations/0006_add_chat_badges.py @@ -0,0 +1,153 @@ +# Generated by Django 6.0.1 on 2026-01-15 21:57 +from __future__ import annotations + +import django.db.models.deletion +from django.db import migrations +from django.db import models + + +class Migration(migrations.Migration): + """Add ChatBadgeSet and ChatBadge models for Twitch chat badges.""" + + dependencies = [ + ("twitch", "0005_add_reward_campaign"), + ] + + operations = [ + migrations.CreateModel( + name="ChatBadgeSet", + fields=[ + ("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")), + ( + "set_id", + models.TextField( + help_text="Identifier for this badge set (e.g., 'vip', 'subscriber', 'bits').", + unique=True, + verbose_name="Set ID", + ), + ), + ( + "added_at", + models.DateTimeField( + auto_now_add=True, + help_text="Timestamp when this badge set record was created.", + verbose_name="Added At", + ), + ), + ( + "updated_at", + models.DateTimeField( + auto_now=True, + help_text="Timestamp when this badge set record was last updated.", + verbose_name="Updated At", + ), + ), + ], + options={ + "ordering": ["set_id"], + "indexes": [ + models.Index(fields=["set_id"], name="twitch_chat_set_id_9319f2_idx"), + models.Index(fields=["added_at"], name="twitch_chat_added_a_b0023a_idx"), + models.Index(fields=["updated_at"], name="twitch_chat_updated_90afed_idx"), + ], + }, + ), + migrations.CreateModel( + name="ChatBadge", + fields=[ + ("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")), + ( + "badge_id", + models.TextField( + help_text="Version identifier for this badge (e.g., '1', 'Alliance', '10000').", + verbose_name="Badge ID", + ), + ), + ( + "image_url_1x", + models.URLField( + help_text="URL to the small version (18px x 18px) of the badge.", + max_length=500, + verbose_name="Image URL (18px)", + ), + ), + ( + "image_url_2x", + models.URLField( + help_text="URL to the medium version (36px x 36px) of the badge.", + max_length=500, + verbose_name="Image URL (36px)", + ), + ), + ( + "image_url_4x", + models.URLField( + help_text="URL to the large version (72px x 72px) of the badge.", + max_length=500, + verbose_name="Image URL (72px)", + ), + ), + ("title", models.TextField(help_text="The title of the badge (e.g., 'VIP').", verbose_name="Title")), + ( + "description", + models.TextField(help_text="The description of the badge.", verbose_name="Description"), + ), + ( + "click_action", + models.TextField( + blank=True, + help_text="The action to take when clicking on the badge (e.g., 'visit_url').", + null=True, + verbose_name="Click Action", + ), + ), + ( + "click_url", + models.URLField( + blank=True, + help_text="The URL to navigate to when clicking on the badge.", + max_length=500, + null=True, + verbose_name="Click URL", + ), + ), + ( + "added_at", + models.DateTimeField( + auto_now_add=True, + help_text="Timestamp when this badge record was created.", + verbose_name="Added At", + ), + ), + ( + "updated_at", + models.DateTimeField( + auto_now=True, + help_text="Timestamp when this badge record was last updated.", + verbose_name="Updated At", + ), + ), + ( + "badge_set", + models.ForeignKey( + help_text="The badge set this badge belongs to.", + on_delete=django.db.models.deletion.CASCADE, + related_name="badges", + to="twitch.chatbadgeset", + verbose_name="Badge Set", + ), + ), + ], + options={ + "ordering": ["badge_set", "badge_id"], + "indexes": [ + models.Index(fields=["badge_set"], name="twitch_chat_badge_s_54f225_idx"), + models.Index(fields=["badge_id"], name="twitch_chat_badge_i_58a68a_idx"), + models.Index(fields=["title"], name="twitch_chat_title_0f42d2_idx"), + models.Index(fields=["added_at"], name="twitch_chat_added_a_9ba7dd_idx"), + models.Index(fields=["updated_at"], name="twitch_chat_updated_568ad1_idx"), + ], + "constraints": [models.UniqueConstraint(fields=("badge_set", "badge_id"), name="unique_badge_set_id")], + }, + ), + ] diff --git a/twitch/models.py b/twitch/models.py index a25c36d..d411289 100644 --- a/twitch/models.py +++ b/twitch/models.py @@ -764,3 +764,125 @@ class RewardCampaign(models.Model): if self.starts_at is None or self.ends_at is None: return False return self.starts_at <= now <= self.ends_at + + +# MARK: ChatBadgeSet +class ChatBadgeSet(models.Model): + """Represents a set of Twitch global chat badges (e.g., VIP, Subscriber, Bits).""" + + set_id = models.TextField( + unique=True, + verbose_name="Set ID", + help_text="Identifier for this badge set (e.g., 'vip', 'subscriber', 'bits').", + ) + + added_at = models.DateTimeField( + auto_now_add=True, + verbose_name="Added At", + editable=False, + help_text="Timestamp when this badge set record was created.", + ) + updated_at = models.DateTimeField( + auto_now=True, + verbose_name="Updated At", + editable=False, + help_text="Timestamp when this badge set record was last updated.", + ) + + class Meta: + ordering = ["set_id"] + indexes = [ + models.Index(fields=["set_id"]), + models.Index(fields=["added_at"]), + models.Index(fields=["updated_at"]), + ] + + def __str__(self) -> str: + """Return a string representation of the badge set.""" + return self.set_id + + +# MARK: ChatBadge +class ChatBadge(models.Model): + """Represents a specific version of a Twitch global chat badge.""" + + badge_set = models.ForeignKey( + ChatBadgeSet, + on_delete=models.CASCADE, + related_name="badges", + verbose_name="Badge Set", + help_text="The badge set this badge belongs to.", + ) + badge_id = models.TextField( + verbose_name="Badge ID", + help_text="Version identifier for this badge (e.g., '1', 'Alliance', '10000').", + ) + image_url_1x = models.URLField( + max_length=500, + verbose_name="Image URL (18px)", + help_text="URL to the small version (18px x 18px) of the badge.", + ) + image_url_2x = models.URLField( + max_length=500, + verbose_name="Image URL (36px)", + help_text="URL to the medium version (36px x 36px) of the badge.", + ) + image_url_4x = models.URLField( + max_length=500, + verbose_name="Image URL (72px)", + help_text="URL to the large version (72px x 72px) of the badge.", + ) + title = models.TextField( + verbose_name="Title", + help_text="The title of the badge (e.g., 'VIP').", + ) + description = models.TextField( + verbose_name="Description", + help_text="The description of the badge.", + ) + click_action = models.TextField( # noqa: DJ001 + blank=True, + null=True, + verbose_name="Click Action", + help_text="The action to take when clicking on the badge (e.g., 'visit_url').", + ) + click_url = models.URLField( # noqa: DJ001 + max_length=500, + blank=True, + null=True, + verbose_name="Click URL", + help_text="The URL to navigate to when clicking on the badge.", + ) + + added_at = models.DateTimeField( + auto_now_add=True, + verbose_name="Added At", + editable=False, + help_text="Timestamp when this badge record was created.", + ) + updated_at = models.DateTimeField( + auto_now=True, + verbose_name="Updated At", + editable=False, + help_text="Timestamp when this badge record was last updated.", + ) + + class Meta: + ordering = ["badge_set", "badge_id"] + constraints = [ + models.UniqueConstraint( + fields=["badge_set", "badge_id"], + name="unique_badge_set_id", + ), + ] + indexes = [ + models.Index(fields=["badge_set"]), + models.Index(fields=["badge_id"]), + models.Index(fields=["title"]), + models.Index(fields=["added_at"]), + models.Index(fields=["updated_at"]), + ] + + def __str__(self) -> str: + """Return a string representation of the badge.""" + return f"{self.badge_set.set_id}/{self.badge_id}: {self.title}" diff --git a/twitch/schemas.py b/twitch/schemas.py index 79022c7..2339c36 100644 --- a/twitch/schemas.py +++ b/twitch/schemas.py @@ -548,3 +548,51 @@ class BatchedGraphQLResponse(BaseModel): "strict": True, "populate_by_name": True, } + + +# MARK: ChatBadge Schemas +class ChatBadgeVersionSchema(BaseModel): + """Schema for a single chat badge version.""" + + badge_id: str = Field(alias="id") + image_url_1x: str = Field(alias="image_url_1x") + image_url_2x: str = Field(alias="image_url_2x") + image_url_4x: str = Field(alias="image_url_4x") + title: str + description: str + click_action: str | None = Field(default=None, alias="click_action") + click_url: str | None = Field(default=None, alias="click_url") + + model_config = { + "extra": "forbid", + "validate_assignment": True, + "strict": True, + "populate_by_name": True, + } + + +class ChatBadgeSetSchema(BaseModel): + """Schema for a chat badge set containing multiple versions.""" + + set_id: str = Field(alias="set_id") + versions: list[ChatBadgeVersionSchema] + + model_config = { + "extra": "forbid", + "validate_assignment": True, + "strict": True, + "populate_by_name": True, + } + + +class GlobalChatBadgesResponse(BaseModel): + """Schema for the Twitch global chat badges API response.""" + + data: list[ChatBadgeSetSchema] + + model_config = { + "extra": "forbid", + "validate_assignment": True, + "strict": True, + "populate_by_name": True, + } diff --git a/twitch/tests/test_badge_views.py b/twitch/tests/test_badge_views.py new file mode 100644 index 0000000..a80401b --- /dev/null +++ b/twitch/tests/test_badge_views.py @@ -0,0 +1,199 @@ +"""Tests for chat badge views.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest +from django.urls import reverse + +from twitch.models import ChatBadge +from twitch.models import ChatBadgeSet + +if TYPE_CHECKING: + from django.test import Client + + +@pytest.mark.django_db +class TestBadgeListView: + """Tests for the badge list view.""" + + def test_badge_list_empty(self, client: Client) -> None: + """Test badge list view with no badges.""" + response = client.get(reverse("twitch:badge_list")) + assert response.status_code == 200 + assert "No badge sets found" in response.content.decode() + + def test_badge_list_displays_sets(self, client: Client) -> None: + """Test that badge sets are displayed.""" + badge_set1 = ChatBadgeSet.objects.create(set_id="vip") + badge_set2 = ChatBadgeSet.objects.create(set_id="subscriber") + + response = client.get(reverse("twitch:badge_list")) + assert response.status_code == 200 + content = response.content.decode() + + assert badge_set1.set_id in content + assert badge_set2.set_id in content + + def test_badge_list_displays_badge_count(self, client: Client) -> None: + """Test that badge version count is displayed.""" + badge_set = ChatBadgeSet.objects.create(set_id="bits") + ChatBadge.objects.create( + badge_set=badge_set, + badge_id="1", + image_url_1x="https://example.com/1x.png", + image_url_2x="https://example.com/2x.png", + image_url_4x="https://example.com/4x.png", + title="Bits 1", + description="1 Bit", + ) + ChatBadge.objects.create( + badge_set=badge_set, + badge_id="100", + image_url_1x="https://example.com/1x.png", + image_url_2x="https://example.com/2x.png", + image_url_4x="https://example.com/4x.png", + title="Bits 100", + description="100 Bits", + ) + + response = client.get(reverse("twitch:badge_list")) + assert response.status_code == 200 + content = response.content.decode() + + # Should show version count (the template uses "versions" not "version") + assert "2" in content + assert "versions" in content + + +@pytest.mark.django_db +class TestBadgeSetDetailView: + """Tests for the badge set detail view.""" + + def test_badge_set_detail_not_found(self, client: Client) -> None: + """Test 404 when badge set doesn't exist.""" + response = client.get(reverse("twitch:badge_set_detail", kwargs={"set_id": "nonexistent"})) + assert response.status_code == 404 + + def test_badge_set_detail_displays_badges(self, client: Client) -> None: + """Test that badge versions are displayed.""" + badge_set = ChatBadgeSet.objects.create(set_id="moderator") + badge = ChatBadge.objects.create( + badge_set=badge_set, + badge_id="1", + image_url_1x="https://example.com/1x.png", + image_url_2x="https://example.com/2x.png", + image_url_4x="https://example.com/4x.png", + title="Moderator", + description="Channel Moderator", + click_action="visit_url", + click_url="https://help.twitch.tv", + ) + + response = client.get(reverse("twitch:badge_set_detail", kwargs={"set_id": "moderator"})) + assert response.status_code == 200 + content = response.content.decode() + + assert badge.title in content + assert badge.description in content + assert badge.badge_id in content + assert badge.image_url_4x in content + + def test_badge_set_detail_displays_metadata(self, client: Client) -> None: + """Test that badge set metadata is displayed.""" + badge_set = ChatBadgeSet.objects.create(set_id="vip") + ChatBadge.objects.create( + badge_set=badge_set, + badge_id="1", + image_url_1x="https://example.com/1x.png", + image_url_2x="https://example.com/2x.png", + image_url_4x="https://example.com/4x.png", + title="VIP", + description="VIP Badge", + ) + + response = client.get(reverse("twitch:badge_set_detail", kwargs={"set_id": "vip"})) + assert response.status_code == 200 + content = response.content.decode() + + assert "vip" in content + assert "Total Versions:" in content + assert "1" in content + + def test_badge_set_detail_json_data(self, client: Client) -> None: + """Test that JSON data is displayed.""" + badge_set = ChatBadgeSet.objects.create(set_id="test_set") + ChatBadge.objects.create( + badge_set=badge_set, + badge_id="1", + image_url_1x="https://example.com/1x.png", + image_url_2x="https://example.com/2x.png", + image_url_4x="https://example.com/4x.png", + title="Test", + description="Test Badge", + ) + + response = client.get(reverse("twitch:badge_set_detail", kwargs={"set_id": "test_set"})) + assert response.status_code == 200 + content = response.content.decode() + + # Should include JSON data section + assert "Badge Set Data (JSON)" in content + assert "test_set" in content + + +@pytest.mark.django_db +class TestBadgeSearch: + """Tests for badge search functionality.""" + + def test_search_finds_badge_sets(self, client: Client) -> None: + """Test that search finds badge sets.""" + ChatBadgeSet.objects.create(set_id="vip") + ChatBadgeSet.objects.create(set_id="subscriber") + + response = client.get(reverse("twitch:search"), {"q": "vip"}) + assert response.status_code == 200 + content = response.content.decode() + + assert "Badge Sets" in content + assert "vip" in content + + def test_search_finds_badges_by_title(self, client: Client) -> None: + """Test that search finds badges by title.""" + badge_set = ChatBadgeSet.objects.create(set_id="test") + ChatBadge.objects.create( + badge_set=badge_set, + badge_id="1", + image_url_1x="https://example.com/1x.png", + image_url_2x="https://example.com/2x.png", + image_url_4x="https://example.com/4x.png", + title="Moderator Badge", + description="Test description", + ) + + response = client.get(reverse("twitch:search"), {"q": "Moderator"}) + assert response.status_code == 200 + content = response.content.decode() + + assert "Chat Badges" in content + assert "Moderator Badge" in content + + def test_search_finds_badges_by_description(self, client: Client) -> None: + """Test that search finds badges by description.""" + badge_set = ChatBadgeSet.objects.create(set_id="test") + ChatBadge.objects.create( + badge_set=badge_set, + badge_id="1", + image_url_1x="https://example.com/1x.png", + image_url_2x="https://example.com/2x.png", + image_url_4x="https://example.com/4x.png", + title="Test Badge", + description="Unique description text", + ) + + response = client.get(reverse("twitch:search"), {"q": "Unique description"}) + assert response.status_code == 200 + content = response.content.decode() + + assert "Chat Badges" in content or "Test Badge" in content diff --git a/twitch/tests/test_chat_badges.py b/twitch/tests/test_chat_badges.py new file mode 100644 index 0000000..494c34c --- /dev/null +++ b/twitch/tests/test_chat_badges.py @@ -0,0 +1,314 @@ +"""Tests for chat badge models and functionality.""" + +from __future__ import annotations + +import pytest +from django.db import IntegrityError +from pydantic import ValidationError + +from twitch.models import ChatBadge +from twitch.models import ChatBadgeSet +from twitch.schemas import ChatBadgeSetSchema +from twitch.schemas import ChatBadgeVersionSchema +from twitch.schemas import GlobalChatBadgesResponse + + +@pytest.mark.django_db +class TestChatBadgeSetModel: + """Tests for the ChatBadgeSet model.""" + + def test_create_badge_set(self) -> None: + """Test creating a new badge set.""" + badge_set = ChatBadgeSet.objects.create(set_id="vip") + assert badge_set.set_id == "vip" + assert badge_set.added_at is not None + assert badge_set.updated_at is not None + assert str(badge_set) == "vip" + + def test_unique_set_id(self) -> None: + """Test that set_id must be unique.""" + ChatBadgeSet.objects.create(set_id="vip") + with pytest.raises(IntegrityError): + ChatBadgeSet.objects.create(set_id="vip") + + def test_badge_set_ordering(self) -> None: + """Test that badge sets are ordered by set_id.""" + ChatBadgeSet.objects.create(set_id="subscriber") + ChatBadgeSet.objects.create(set_id="bits") + ChatBadgeSet.objects.create(set_id="vip") + + badge_sets = list(ChatBadgeSet.objects.all()) + assert badge_sets[0].set_id == "bits" + assert badge_sets[1].set_id == "subscriber" + assert badge_sets[2].set_id == "vip" + + +@pytest.mark.django_db +class TestChatBadgeModel: + """Tests for the ChatBadge model.""" + + def test_create_badge(self) -> None: + """Test creating a new badge.""" + badge_set = ChatBadgeSet.objects.create(set_id="vip") + badge = ChatBadge.objects.create( + badge_set=badge_set, + badge_id="1", + image_url_1x="https://example.com/1x.png", + image_url_2x="https://example.com/2x.png", + image_url_4x="https://example.com/4x.png", + title="VIP", + description="VIP Badge", + click_action="visit_url", + click_url="https://help.twitch.tv", + ) + + assert badge.badge_set == badge_set + assert badge.badge_id == "1" + assert badge.title == "VIP" + assert badge.description == "VIP Badge" + assert badge.click_action == "visit_url" + assert badge.click_url == "https://help.twitch.tv" + assert str(badge) == "vip/1: VIP" + + def test_unique_badge_set_and_id(self) -> None: + """Test that badge_set and badge_id combination must be unique.""" + badge_set = ChatBadgeSet.objects.create(set_id="vip") + ChatBadge.objects.create( + badge_set=badge_set, + badge_id="1", + image_url_1x="https://example.com/1x.png", + image_url_2x="https://example.com/2x.png", + image_url_4x="https://example.com/4x.png", + title="VIP", + description="VIP Badge", + ) + + with pytest.raises(IntegrityError): + ChatBadge.objects.create( + badge_set=badge_set, + badge_id="1", + image_url_1x="https://example.com/1x.png", + image_url_2x="https://example.com/2x.png", + image_url_4x="https://example.com/4x.png", + title="VIP Duplicate", + description="Duplicate Badge", + ) + + def test_different_badge_ids_same_set(self) -> None: + """Test that different badge_ids can exist in the same set.""" + badge_set = ChatBadgeSet.objects.create(set_id="bits") + badge1 = ChatBadge.objects.create( + badge_set=badge_set, + badge_id="1", + image_url_1x="https://example.com/1x.png", + image_url_2x="https://example.com/2x.png", + image_url_4x="https://example.com/4x.png", + title="Bits 1", + description="1 Bit", + ) + badge2 = ChatBadge.objects.create( + badge_set=badge_set, + badge_id="100", + image_url_1x="https://example.com/1x.png", + image_url_2x="https://example.com/2x.png", + image_url_4x="https://example.com/4x.png", + title="Bits 100", + description="100 Bits", + ) + + assert badge1.badge_id == "1" + assert badge2.badge_id == "100" + assert badge1.badge_set == badge2.badge_set + + def test_nullable_click_fields(self) -> None: + """Test that click_action and click_url can be null.""" + badge_set = ChatBadgeSet.objects.create(set_id="moderator") + badge = ChatBadge.objects.create( + badge_set=badge_set, + badge_id="1", + image_url_1x="https://example.com/1x.png", + image_url_2x="https://example.com/2x.png", + image_url_4x="https://example.com/4x.png", + title="Moderator", + description="Channel Moderator", + click_action=None, + click_url=None, + ) + + assert badge.click_action is None + assert badge.click_url is None + + def test_badge_cascade_delete(self) -> None: + """Test that badges are deleted when their badge set is deleted.""" + badge_set = ChatBadgeSet.objects.create(set_id="test_set") + ChatBadge.objects.create( + badge_set=badge_set, + badge_id="1", + image_url_1x="https://example.com/1x.png", + image_url_2x="https://example.com/2x.png", + image_url_4x="https://example.com/4x.png", + title="Test Badge", + description="Test Description", + ) + + assert ChatBadge.objects.filter(badge_set=badge_set).count() == 1 + badge_set.delete() + assert ChatBadge.objects.filter(badge_set__set_id="test_set").count() == 0 + + +class TestChatBadgeSchemas: + """Tests for chat badge Pydantic schemas.""" + + def test_chat_badge_version_schema_valid(self) -> None: + """Test that ChatBadgeVersionSchema validates correct data.""" + data = { + "id": "1", + "image_url_1x": "https://static-cdn.jtvnw.net/badges/v1/example/1", + "image_url_2x": "https://static-cdn.jtvnw.net/badges/v1/example/2", + "image_url_4x": "https://static-cdn.jtvnw.net/badges/v1/example/3", + "title": "VIP", + "description": "VIP Badge", + "click_action": "visit_url", + "click_url": "https://help.twitch.tv", + } + + schema = ChatBadgeVersionSchema.model_validate(data) + assert schema.badge_id == "1" + assert schema.title == "VIP" + assert schema.click_action == "visit_url" + + def test_chat_badge_version_schema_nullable_fields(self) -> None: + """Test that nullable fields in ChatBadgeVersionSchema work correctly.""" + data = { + "id": "1", + "image_url_1x": "https://static-cdn.jtvnw.net/badges/v1/example/1", + "image_url_2x": "https://static-cdn.jtvnw.net/badges/v1/example/2", + "image_url_4x": "https://static-cdn.jtvnw.net/badges/v1/example/3", + "title": "VIP", + "description": "VIP Badge", + "click_action": None, + "click_url": None, + } + + schema = ChatBadgeVersionSchema.model_validate(data) + assert schema.click_action is None + assert schema.click_url is None + + def test_chat_badge_version_schema_missing_required(self) -> None: + """Test that ChatBadgeVersionSchema raises error on missing required fields.""" + data = { + "id": "1", + "title": "VIP", + # Missing required image URLs and description + } + + with pytest.raises(ValidationError): + ChatBadgeVersionSchema.model_validate(data) + + def test_chat_badge_set_schema_valid(self) -> None: + """Test that ChatBadgeSetSchema validates correct data.""" + data = { + "set_id": "vip", + "versions": [ + { + "id": "1", + "image_url_1x": "https://static-cdn.jtvnw.net/badges/v1/example/1", + "image_url_2x": "https://static-cdn.jtvnw.net/badges/v1/example/2", + "image_url_4x": "https://static-cdn.jtvnw.net/badges/v1/example/3", + "title": "VIP", + "description": "VIP Badge", + "click_action": "visit_url", + "click_url": "https://help.twitch.tv", + }, + ], + } + + schema = ChatBadgeSetSchema.model_validate(data) + assert schema.set_id == "vip" + assert len(schema.versions) == 1 + assert schema.versions[0].badge_id == "1" + + def test_chat_badge_set_schema_multiple_versions(self) -> None: + """Test that ChatBadgeSetSchema handles multiple badge versions.""" + data = { + "set_id": "bits", + "versions": [ + { + "id": "1", + "image_url_1x": "https://example.com/1/1x.png", + "image_url_2x": "https://example.com/1/2x.png", + "image_url_4x": "https://example.com/1/4x.png", + "title": "Bits 1", + "description": "1 Bit", + "click_action": None, + "click_url": None, + }, + { + "id": "100", + "image_url_1x": "https://example.com/100/1x.png", + "image_url_2x": "https://example.com/100/2x.png", + "image_url_4x": "https://example.com/100/4x.png", + "title": "Bits 100", + "description": "100 Bits", + "click_action": None, + "click_url": None, + }, + ], + } + + schema = ChatBadgeSetSchema.model_validate(data) + assert schema.set_id == "bits" + assert len(schema.versions) == 2 + assert schema.versions[0].badge_id == "1" + assert schema.versions[1].badge_id == "100" + + def test_global_chat_badges_response_valid(self) -> None: + """Test that GlobalChatBadgesResponse validates correct API response.""" + data = { + "data": [ + { + "set_id": "vip", + "versions": [ + { + "id": "1", + "image_url_1x": "https://static-cdn.jtvnw.net/badges/v1/example/1", + "image_url_2x": "https://static-cdn.jtvnw.net/badges/v1/example/2", + "image_url_4x": "https://static-cdn.jtvnw.net/badges/v1/example/3", + "title": "VIP", + "description": "VIP Badge", + "click_action": "visit_url", + "click_url": "https://help.twitch.tv", + }, + ], + }, + ], + } + + response = GlobalChatBadgesResponse.model_validate(data) + assert len(response.data) == 1 + assert response.data[0].set_id == "vip" + + def test_global_chat_badges_response_empty(self) -> None: + """Test that GlobalChatBadgesResponse validates empty response.""" + data = {"data": []} + + response = GlobalChatBadgesResponse.model_validate(data) + assert len(response.data) == 0 + + def test_chat_badge_schema_extra_forbidden(self) -> None: + """Test that extra fields are forbidden in schemas.""" + data = { + "id": "1", + "image_url_1x": "https://example.com/1x.png", + "image_url_2x": "https://example.com/2x.png", + "image_url_4x": "https://example.com/4x.png", + "title": "VIP", + "description": "VIP Badge", + "extra_field": "should_fail", # This should cause validation to fail + } + + with pytest.raises(ValidationError) as exc_info: + ChatBadgeVersionSchema.model_validate(data) + + # Check that the error is about the extra field + assert "extra_field" in str(exc_info.value) diff --git a/twitch/urls.py b/twitch/urls.py index e7bf406..bc60dbb 100644 --- a/twitch/urls.py +++ b/twitch/urls.py @@ -19,6 +19,8 @@ app_name = "twitch" urlpatterns: list[URLPattern] = [ path("", views.dashboard, name="dashboard"), + path("badges/", views.badge_list_view, name="badge_list"), + path("badges//", views.badge_set_detail_view, name="badge_set_detail"), path("campaigns/", views.drop_campaign_list_view, name="campaign_list"), path("campaigns//", views.drop_campaign_detail_view, name="campaign_detail"), path("channels/", views.ChannelListView.as_view(), name="channel_list"), diff --git a/twitch/views.py b/twitch/views.py index ee907dd..aa0eeb1 100644 --- a/twitch/views.py +++ b/twitch/views.py @@ -32,6 +32,8 @@ from pygments.formatters import HtmlFormatter from pygments.lexers.data import JsonLexer from twitch.models import Channel +from twitch.models import ChatBadge +from twitch.models import ChatBadgeSet from twitch.models import DropBenefit from twitch.models import DropCampaign from twitch.models import Game @@ -113,6 +115,10 @@ def search_view(request: HttpRequest) -> HttpResponse: results["reward_campaigns"] = RewardCampaign.objects.filter( Q(name__istartswith=query) | Q(brand__istartswith=query) | Q(summary__icontains=query), ).select_related("game") + results["badge_sets"] = ChatBadgeSet.objects.filter(set_id__istartswith=query) + results["badges"] = ChatBadge.objects.filter( + Q(title__istartswith=query) | Q(description__icontains=query), + ).select_related("badge_set") else: # SQLite-compatible text search using icontains results["organizations"] = Organization.objects.filter( @@ -133,6 +139,10 @@ def search_view(request: HttpRequest) -> HttpResponse: results["reward_campaigns"] = RewardCampaign.objects.filter( Q(name__icontains=query) | Q(brand__icontains=query) | Q(summary__icontains=query), ).select_related("game") + results["badge_sets"] = ChatBadgeSet.objects.filter(set_id__icontains=query) + results["badges"] = ChatBadge.objects.filter( + Q(title__icontains=query) | Q(description__icontains=query), + ).select_related("badge_set") return render( request, @@ -1143,3 +1153,110 @@ class ChannelDetailView(DetailView): ) return context + + +# MARK: /badges/ +def badge_list_view(request: HttpRequest) -> HttpResponse: + """List view for chat badge sets. + + Args: + request: The HTTP request. + + Returns: + HttpResponse: The rendered badge list page. + """ + badge_sets: QuerySet[ChatBadgeSet] = ( + ChatBadgeSet.objects + .all() + .prefetch_related( + Prefetch( + "badges", + queryset=ChatBadge.objects.order_by("badge_id"), + ), + ) + .order_by("set_id") + ) + + # Group badges by set for easier display + badge_data: list[dict[str, Any]] = [ + { + "set": badge_set, + "badges": list(badge_set.badges.all()), # pyright: ignore[reportAttributeAccessIssue] + } + for badge_set in badge_sets + ] + + context: dict[str, Any] = { + "badge_sets": badge_sets, + "badge_data": badge_data, + } + + return render(request, "twitch/badge_list.html", context) + + +# MARK: /badges// +def badge_set_detail_view(request: HttpRequest, set_id: str) -> HttpResponse: + """Detail view for a specific badge set. + + Args: + request: The HTTP request. + set_id: The ID of the badge set. + + Returns: + HttpResponse: The rendered badge set detail page. + + Raises: + Http404: If the badge set is not found. + """ + try: + badge_set: ChatBadgeSet = ChatBadgeSet.objects.prefetch_related( + Prefetch( + "badges", + queryset=ChatBadge.objects.order_by("badge_id"), + ), + ).get(set_id=set_id) + except ChatBadgeSet.DoesNotExist as exc: + msg = "No badge set found matching the query" + raise Http404(msg) from exc + + badges: QuerySet[ChatBadge] = badge_set.badges.all() # pyright: ignore[reportAttributeAccessIssue] + + # Serialize for JSON display + serialized_set = serialize( + "json", + [badge_set], + fields=( + "set_id", + "added_at", + "updated_at", + ), + ) + set_data: list[dict[str, Any]] = json.loads(serialized_set) + + if badges.exists(): + serialized_badges = serialize( + "json", + badges, + fields=( + "badge_id", + "image_url_1x", + "image_url_2x", + "image_url_4x", + "title", + "description", + "click_action", + "click_url", + "added_at", + "updated_at", + ), + ) + badges_data: list[dict[str, Any]] = json.loads(serialized_badges) + set_data[0]["fields"]["badges"] = badges_data + + context: dict[str, Any] = { + "badge_set": badge_set, + "badges": badges, + "set_data": format_and_color_json(set_data[0]), + } + + return render(request, "twitch/badge_set_detail.html", context)