diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 69361ac..050a0d8 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -27,7 +27,7 @@ repos: args: [--target-version, "6.0"] - repo: https://github.com/astral-sh/ruff-pre-commit - rev: v0.15.12 + rev: v0.15.9 hooks: - id: ruff-check args: ["--fix", "--exit-non-zero-on-fix"] diff --git a/pyproject.toml b/pyproject.toml index ea57c1d..74d15fa 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,9 +12,7 @@ dependencies = [ "django-celery-beat", "django-celery-results", "django-debug-toolbar", - "django-ninja", "django-silk", - "django-zeal", "django", "flower", "gunicorn", @@ -33,6 +31,7 @@ dependencies = [ "setproctitle", "sitemap-parser", "tqdm", + "django-zeal>=2.1.0", ] diff --git a/templates/base.html b/templates/base.html index d558273..be2daed 100644 --- a/templates/base.html +++ b/templates/base.html @@ -220,7 +220,6 @@ Channels | Badges | Emotes | - API Docs | Inventory
Kick diff --git a/templates/core/dashboard.html b/templates/core/dashboard.html index 317eaf7..c563b8c 100644 --- a/templates/core/dashboard.html +++ b/templates/core/dashboard.html @@ -48,8 +48,6 @@ title="Atom feed for Twitch campaigns">[atom] [discord] - [api] [explain] diff --git a/templates/core/docs_rss.html b/templates/core/docs_rss.html index 69ed1de..13de544 100644 --- a/templates/core/docs_rss.html +++ b/templates/core/docs_rss.html @@ -16,15 +16,6 @@ that include Discord relative timestamps (e.g., <t:1773450272:R>) for dates, making them ideal for Discord bots and integrations. Future enhancements may include Discord-specific formatting or content.

-

- Twitch JSON API documentation is available at - /twitch/api/v1/docs. -

-

- Twitch campaign feeds accept ?limit=50 to change item count and - ?hide_paid=1 to hide subscription-gated drops and skip campaigns - with no free drops. -

Global RSS Feeds

diff --git a/templates/twitch/campaign_detail.html b/templates/twitch/campaign_detail.html index 08dd892..6251304 100644 --- a/templates/twitch/campaign_detail.html +++ b/templates/twitch/campaign_detail.html @@ -103,8 +103,6 @@ title="Atom feed for {{ campaign.game.display_name }} campaigns">[atom][discord] - [api][explain] {% endif %} diff --git a/templates/twitch/campaign_list.html b/templates/twitch/campaign_list.html index 9fbe54d..219217a 100644 --- a/templates/twitch/campaign_list.html +++ b/templates/twitch/campaign_list.html @@ -33,8 +33,6 @@ title="Atom feed for all campaigns">[atom][discord] - [api][explain][csv] diff --git a/templates/twitch/dashboard.html b/templates/twitch/dashboard.html index 7051e2d..5e6f7cd 100644 --- a/templates/twitch/dashboard.html +++ b/templates/twitch/dashboard.html @@ -36,8 +36,6 @@ title="Atom feed for campaigns">[atom][discord] - [api][explain]
diff --git a/templates/twitch/game_detail.html b/templates/twitch/game_detail.html index f48e13f..ed24fbb 100644 --- a/templates/twitch/game_detail.html +++ b/templates/twitch/game_detail.html @@ -65,8 +65,6 @@ title="Atom feed for {{ game.display_name }} campaigns">[atom][discord] - [api][explain] diff --git a/templates/twitch/games_grid.html b/templates/twitch/games_grid.html index b85eb34..7facdde 100644 --- a/templates/twitch/games_grid.html +++ b/templates/twitch/games_grid.html @@ -32,8 +32,6 @@ title="Atom feed for all games">[atom][discord] - [api][explain][csv] diff --git a/templates/twitch/games_list.html b/templates/twitch/games_list.html index 130a4ef..3497856 100644 --- a/templates/twitch/games_list.html +++ b/templates/twitch/games_list.html @@ -30,8 +30,6 @@ title="Atom feed for all games">[atom][discord] - [api][explain][csv] diff --git a/templates/twitch/org_list.html b/templates/twitch/org_list.html index 43e1fde..911e08a 100644 --- a/templates/twitch/org_list.html +++ b/templates/twitch/org_list.html @@ -15,8 +15,6 @@ title="Atom feed for all organizations">[atom][discord] - [api][explain][csv] diff --git a/templates/twitch/reward_campaign_detail.html b/templates/twitch/reward_campaign_detail.html index bdca15a..40cb357 100644 --- a/templates/twitch/reward_campaign_detail.html +++ b/templates/twitch/reward_campaign_detail.html @@ -110,8 +110,6 @@ title="Atom feed for all reward campaigns">[atom][discord] - [api][explain] {% if reward_campaign.external_url %} [atom][discord] - [api][explain] {% if reward_campaigns %} diff --git a/twitch/api.py b/twitch/api.py deleted file mode 100644 index a657e31..0000000 --- a/twitch/api.py +++ /dev/null @@ -1,822 +0,0 @@ -from __future__ import annotations - -import datetime # noqa: TC003 -from dataclasses import dataclass -from typing import TYPE_CHECKING -from typing import Literal - -from django.db import models -from django.http import Http404 -from django.shortcuts import get_object_or_404 -from django.utils import timezone -from ninja import NinjaAPI -from ninja import Schema - -from twitch.models import Channel -from twitch.models import ChatBadgeSet -from twitch.models import DropCampaign -from twitch.models import Game -from twitch.models import Organization -from twitch.models import RewardCampaign -from twitch.utils import normalize_twitch_box_art_url - -if TYPE_CHECKING: - from django.db.models import QuerySet - from django.http import HttpRequest - - from twitch.models import ChatBadge - from twitch.models import DropBenefit - from twitch.models import TimeBasedDrop - -V1StatusFilter = Literal["active", "upcoming", "expired"] - -DEFAULT_PAGE_SIZE = 100 -MAX_PAGE_SIZE = 500 - - -@dataclass(frozen=True) -class V1Page[ModelT: models.Model]: - """Typed slice metadata for v1 list responses.""" - - items: list[ModelT] - total: int - page: int - page_size: int - - -class V1PaginationSchema(Schema): - """Common pagination fields for v1 list responses.""" - - total: int - page: int - page_size: int - - -class V1OrganizationSummarySchema(Schema): - """Compact Twitch organization representation.""" - - twitch_id: str - name: str - - -class V1GameSummarySchema(Schema): - """Compact Twitch game representation.""" - - twitch_id: str - slug: str - name: str - display_name: str - box_art_url: str - organizations: list[V1OrganizationSummarySchema] - campaign_count: int - active_campaign_count: int - - -class V1OrganizationSchema(V1OrganizationSummarySchema): - """Twitch organization response.""" - - added_at: datetime.datetime - updated_at: datetime.datetime - - -class V1GameSchema(V1GameSummarySchema): - """Twitch game response.""" - - added_at: datetime.datetime - updated_at: datetime.datetime - - -class V1ChannelSummarySchema(Schema): - """Compact Twitch channel representation.""" - - twitch_id: str - name: str - display_name: str - - -class V1ChannelSchema(V1ChannelSummarySchema): - """Twitch channel response.""" - - allowed_campaign_count: int - added_at: datetime.datetime - updated_at: datetime.datetime - - -class V1DropBenefitSchema(Schema): - """Twitch drop benefit response.""" - - twitch_id: str - name: str - image_url: str - distribution_type: str - created_at: datetime.datetime | None - entitlement_limit: int - is_ios_available: bool - - -class V1TimeBasedDropSchema(Schema): - """Twitch time-based drop response.""" - - twitch_id: str - name: str - required_minutes_watched: int | None - required_subs: int - start_at: datetime.datetime | None - end_at: datetime.datetime | None - benefits: list[V1DropBenefitSchema] - - -class V1DropCampaignSummarySchema(Schema): - """Compact Twitch drop campaign representation.""" - - twitch_id: str - name: str - description: str - status: str - image_url: str - details_url: str - account_link_url: str - start_at: datetime.datetime | None - end_at: datetime.datetime | None - game: V1GameSummarySchema - allow_is_enabled: bool - is_fully_imported: bool - added_at: datetime.datetime - updated_at: datetime.datetime - - -class V1DropCampaignDetailSchema(V1DropCampaignSummarySchema): - """Twitch drop campaign detail response.""" - - operation_names: list[str] - allowed_channels: list[V1ChannelSummarySchema] - drops: list[V1TimeBasedDropSchema] - - -class V1OrganizationDetailSchema(V1OrganizationSchema): - """Twitch organization detail response.""" - - games: list[V1GameSummarySchema] - campaigns: list[V1DropCampaignDetailSchema] - - -class V1GameDetailSchema(V1GameSchema): - """Twitch game detail response.""" - - campaigns: list[V1DropCampaignSummarySchema] - - -class V1ChannelDetailSchema(V1ChannelSchema): - """Twitch channel detail response.""" - - campaigns: list[V1DropCampaignSummarySchema] - - -class V1RewardCampaignSchema(Schema): - """Twitch reward campaign response.""" - - twitch_id: str - name: str - brand: str - status: str - computed_status: str - summary: str - instructions: str - external_url: str - reward_value_url_param: str - about_url: str - image_url: str - is_sitewide: bool - starts_at: datetime.datetime | None - ends_at: datetime.datetime | None - game: V1GameSummarySchema | None - added_at: datetime.datetime - updated_at: datetime.datetime - - -class V1ChatBadgeSchema(Schema): - """Twitch chat badge response.""" - - badge_id: str - image_url_1x: str - image_url_2x: str - image_url_4x: str - title: str - description: str - click_action: str | None - click_url: str | None - - -class V1ChatBadgeSetSchema(Schema): - """Twitch chat badge set response.""" - - set_id: str - badges: list[V1ChatBadgeSchema] - added_at: datetime.datetime - updated_at: datetime.datetime - - -class V1DropCampaignListSchema(V1PaginationSchema): - """Paginated Twitch drop campaign list.""" - - items: list[V1DropCampaignSummarySchema] - - -class V1GameListSchema(V1PaginationSchema): - """Paginated Twitch game list.""" - - items: list[V1GameSchema] - - -class V1OrganizationListSchema(V1PaginationSchema): - """Paginated Twitch organization list.""" - - items: list[V1OrganizationSchema] - - -class V1ChannelListSchema(V1PaginationSchema): - """Paginated Twitch channel list.""" - - items: list[V1ChannelSchema] - - -class V1RewardCampaignListSchema(V1PaginationSchema): - """Paginated Twitch reward campaign list.""" - - items: list[V1RewardCampaignSchema] - - -class V1ChatBadgeSetListSchema(V1PaginationSchema): - """Paginated Twitch chat badge set list.""" - - items: list[V1ChatBadgeSetSchema] - - -api = NinjaAPI( - title="TTVDrops Twitch API", - version="1.0.0", - urls_namespace="twitch:twitch-api-v1", -) - - -def _paginate[ModelT: models.Model]( - queryset: QuerySet[ModelT, ModelT], - *, - page: int, - page_size: int, -) -> V1Page[ModelT]: - page = max(page, 1) - page_size = min(max(page_size, 1), MAX_PAGE_SIZE) - offset = (page - 1) * page_size - return V1Page( - items=list(queryset[offset : offset + page_size]), - page=page, - page_size=page_size, - total=queryset.count(), - ) - - -def _campaign_status( - start_at: datetime.datetime | None, - end_at: datetime.datetime | None, - now: datetime.datetime, -) -> str: - if start_at and end_at: - if start_at <= now <= end_at: - return "active" - if start_at > now: - return "upcoming" - return "expired" - return "unknown" - - -def _apply_status_filter[ModelT: models.Model]( - queryset: QuerySet[ModelT, ModelT], - status: V1StatusFilter | None, - now: datetime.datetime, - *, - start_field: str, - end_field: str, -) -> QuerySet[ModelT, ModelT]: - if status == "active": - return queryset.filter(**{f"{start_field}__lte": now, f"{end_field}__gte": now}) - if status == "upcoming": - return queryset.filter(**{f"{start_field}__gt": now}) - if status == "expired": - return queryset.filter(**{f"{end_field}__lt": now}) - return queryset - - -def _serialize_organization_summary(org: Organization) -> V1OrganizationSummarySchema: - return V1OrganizationSummarySchema(twitch_id=org.twitch_id, name=org.name) - - -def _serialize_organization(org: Organization) -> V1OrganizationSchema: - return V1OrganizationSchema( - twitch_id=org.twitch_id, - name=org.name, - added_at=org.added_at, - updated_at=org.updated_at, - ) - - -def _game_campaign_count(game: Game) -> int: - return DropCampaign.objects.filter(game=game).count() - - -def _game_active_campaign_count(game: Game, now: datetime.datetime) -> int: - return DropCampaign.objects.filter( - game=game, - start_at__lte=now, - end_at__gte=now, - ).count() - - -def _game_box_art_url(game: Game) -> str: - deferred_fields: set[str] = game.get_deferred_fields() - local_image_fields = {"box_art_file", "box_art_width", "box_art_height"} - if deferred_fields.isdisjoint(local_image_fields): - return game.box_art_best_url - if "box_art" not in deferred_fields: - return normalize_twitch_box_art_url(game.box_art or "") - return "" - - -def _campaign_summary_queryset( - queryset: QuerySet[DropCampaign, DropCampaign], -) -> QuerySet[DropCampaign, DropCampaign]: - return ( - queryset - .select_related("game") - .only( - "twitch_id", - "name", - "description", - "details_url", - "account_link_url", - "image_url", - "image_file", - "image_width", - "image_height", - "start_at", - "end_at", - "allow_is_enabled", - "is_fully_imported", - "added_at", - "updated_at", - "game", - "game__twitch_id", - "game__name", - "game__display_name", - "game__slug", - "game__box_art", - "game__box_art_file", - "game__box_art_width", - "game__box_art_height", - ) - .prefetch_related("game__owners") - ) - - -def _channel_list_queryset(search_query: str | None) -> QuerySet[Channel, Channel]: - queryset = Channel.objects.only( - "twitch_id", - "name", - "display_name", - "allowed_campaign_count", - "added_at", - "updated_at", - ) - normalized_query = (search_query or "").strip() - if normalized_query: - queryset = queryset.filter( - models.Q(name__icontains=normalized_query) - | models.Q(display_name__icontains=normalized_query), - ) - return queryset.annotate( - campaign_count=models.F("allowed_campaign_count"), - ).order_by("-campaign_count", "name") - - -def _serialize_game_summary( - game: Game, - now: datetime.datetime, -) -> V1GameSummarySchema: - return V1GameSummarySchema( - twitch_id=game.twitch_id, - slug=game.slug, - name=game.name, - display_name=game.display_name, - box_art_url=_game_box_art_url(game), - organizations=[ - _serialize_organization_summary(org) for org in game.owners.all() - ], - campaign_count=_game_campaign_count(game), - active_campaign_count=_game_active_campaign_count(game, now), - ) - - -def _serialize_game(game: Game, now: datetime.datetime) -> V1GameSchema: - return V1GameSchema( - twitch_id=game.twitch_id, - slug=game.slug, - name=game.name, - display_name=game.display_name, - box_art_url=_game_box_art_url(game), - organizations=[ - _serialize_organization_summary(org) for org in game.owners.all() - ], - campaign_count=_game_campaign_count(game), - active_campaign_count=_game_active_campaign_count(game, now), - added_at=game.added_at, - updated_at=game.updated_at, - ) - - -def _serialize_channel_summary(channel: Channel) -> V1ChannelSummarySchema: - return V1ChannelSummarySchema( - twitch_id=channel.twitch_id, - name=channel.name, - display_name=channel.display_name, - ) - - -def _serialize_channel(channel: Channel) -> V1ChannelSchema: - return V1ChannelSchema( - twitch_id=channel.twitch_id, - name=channel.name, - display_name=channel.display_name, - allowed_campaign_count=channel.allowed_campaign_count, - added_at=channel.added_at, - updated_at=channel.updated_at, - ) - - -def _serialize_benefit(benefit: DropBenefit) -> V1DropBenefitSchema: - return V1DropBenefitSchema( - twitch_id=benefit.twitch_id, - name=benefit.name, - image_url=benefit.image_best_url, - distribution_type=benefit.distribution_type, - created_at=benefit.created_at, - entitlement_limit=benefit.entitlement_limit, - is_ios_available=benefit.is_ios_available, - ) - - -def _serialize_drop(drop: TimeBasedDrop) -> V1TimeBasedDropSchema: - return V1TimeBasedDropSchema( - twitch_id=drop.twitch_id, - name=drop.name, - required_minutes_watched=drop.required_minutes_watched, - required_subs=drop.required_subs, - start_at=drop.start_at, - end_at=drop.end_at, - benefits=[_serialize_benefit(benefit) for benefit in drop.benefits.all()], - ) - - -def _serialize_campaign_summary( - campaign: DropCampaign, - now: datetime.datetime, -) -> V1DropCampaignSummarySchema: - return V1DropCampaignSummarySchema( - twitch_id=campaign.twitch_id, - name=campaign.name, - description=campaign.description, - status=_campaign_status(campaign.start_at, campaign.end_at, now), - image_url=campaign.listing_image_url, - details_url=campaign.details_url, - account_link_url=campaign.account_link_url, - start_at=campaign.start_at, - end_at=campaign.end_at, - game=_serialize_game_summary(campaign.game, now), - allow_is_enabled=campaign.allow_is_enabled, - is_fully_imported=campaign.is_fully_imported, - added_at=campaign.added_at, - updated_at=campaign.updated_at, - ) - - -def _serialize_campaign_detail( - campaign: DropCampaign, - now: datetime.datetime, -) -> V1DropCampaignDetailSchema: - return V1DropCampaignDetailSchema( - twitch_id=campaign.twitch_id, - name=campaign.name, - description=campaign.description, - status=_campaign_status(campaign.start_at, campaign.end_at, now), - image_url=campaign.listing_image_url, - details_url=campaign.details_url, - account_link_url=campaign.account_link_url, - start_at=campaign.start_at, - end_at=campaign.end_at, - game=_serialize_game_summary(campaign.game, now), - allow_is_enabled=campaign.allow_is_enabled, - is_fully_imported=campaign.is_fully_imported, - added_at=campaign.added_at, - updated_at=campaign.updated_at, - operation_names=campaign.operation_names, - allowed_channels=[ - _serialize_channel_summary(channel) - for channel in campaign.allow_channels.all() - ], - drops=[_serialize_drop(drop) for drop in campaign.time_based_drops.all()], # pyright: ignore[reportAttributeAccessIssue] - ) - - -def _serialize_reward_campaign( - campaign: RewardCampaign, - now: datetime.datetime, -) -> V1RewardCampaignSchema: - return V1RewardCampaignSchema( - twitch_id=campaign.twitch_id, - name=campaign.name, - brand=campaign.brand, - status=campaign.status, - computed_status=_campaign_status(campaign.starts_at, campaign.ends_at, now), - summary=campaign.summary, - instructions=campaign.instructions, - external_url=campaign.external_url, - reward_value_url_param=campaign.reward_value_url_param, - about_url=campaign.about_url, - image_url=campaign.image_best_url, - is_sitewide=campaign.is_sitewide, - starts_at=campaign.starts_at, - ends_at=campaign.ends_at, - game=_serialize_game_summary(campaign.game, now) if campaign.game else None, - added_at=campaign.added_at, - updated_at=campaign.updated_at, - ) - - -def _serialize_badge(badge: ChatBadge) -> V1ChatBadgeSchema: - return V1ChatBadgeSchema( - badge_id=badge.badge_id, - image_url_1x=badge.image_url_1x, - image_url_2x=badge.image_url_2x, - image_url_4x=badge.image_url_4x, - title=badge.title, - description=badge.description, - click_action=badge.click_action, - click_url=badge.click_url, - ) - - -def _serialize_badge_set(badge_set: ChatBadgeSet) -> V1ChatBadgeSetSchema: - return V1ChatBadgeSetSchema( - set_id=badge_set.set_id, - badges=[_serialize_badge(badge) for badge in badge_set.badges.all()], # pyright: ignore[reportAttributeAccessIssue] - added_at=badge_set.added_at, - updated_at=badge_set.updated_at, - ) - - -@api.get("/campaigns/", response=V1DropCampaignListSchema) -def list_campaigns( - request: HttpRequest, - page: int = 1, - page_size: int = DEFAULT_PAGE_SIZE, - game: str | None = None, - status: V1StatusFilter | None = None, -) -> V1DropCampaignListSchema: - """Return paginated Twitch drop campaigns.""" - now = timezone.now() - queryset = DropCampaign.for_campaign_list(now, game_twitch_id=game, status=status) - page_data = _paginate(queryset, page=page, page_size=page_size) - return V1DropCampaignListSchema( - total=page_data.total, - page=page_data.page, - page_size=page_data.page_size, - items=[ - _serialize_campaign_summary(campaign, now) for campaign in page_data.items - ], - ) - - -@api.get("/campaigns/{twitch_id}/", response=V1DropCampaignDetailSchema) -def get_campaign(request: HttpRequest, twitch_id: str) -> V1DropCampaignDetailSchema: - """Return one Twitch drop campaign. - - Raises: - Http404: if not found - """ - try: - campaign = DropCampaign.for_detail_view(twitch_id) - except DropCampaign.DoesNotExist as exc: - msg = "Campaign not found" - raise Http404(msg) from exc - return _serialize_campaign_detail(campaign, timezone.now()) - - -@api.get("/games/", response=V1GameListSchema) -def list_games( - request: HttpRequest, - page: int = 1, - page_size: int = DEFAULT_PAGE_SIZE, -) -> V1GameListSchema: - """Return paginated Twitch games.""" - now = timezone.now() - queryset = Game.with_campaign_counts(now).order_by("display_name") - page_data = _paginate(queryset, page=page, page_size=page_size) - return V1GameListSchema( - total=page_data.total, - page=page_data.page, - page_size=page_data.page_size, - items=[_serialize_game(game, now) for game in page_data.items], - ) - - -@api.get("/games/{twitch_id}/", response=V1GameDetailSchema) -def get_game(request: HttpRequest, twitch_id: str) -> V1GameDetailSchema: - """Return one Twitch game.""" - game = get_object_or_404(Game.for_detail_view(), twitch_id=twitch_id) - campaigns = _campaign_summary_queryset( - DropCampaign.objects.filter(game=game), - ).order_by("-end_at") - now = timezone.now() - return V1GameDetailSchema( - twitch_id=game.twitch_id, - slug=game.slug, - name=game.name, - display_name=game.display_name, - box_art_url=_game_box_art_url(game), - organizations=[ - _serialize_organization_summary(org) for org in game.owners.all() - ], - campaign_count=_game_campaign_count(game), - active_campaign_count=_game_active_campaign_count(game, now), - added_at=game.added_at, - updated_at=game.updated_at, - campaigns=[ - _serialize_campaign_summary(campaign, now) for campaign in campaigns - ], - ) - - -@api.get("/organizations/", response=V1OrganizationListSchema) -def list_organizations( - request: HttpRequest, - page: int = 1, - page_size: int = DEFAULT_PAGE_SIZE, -) -> V1OrganizationListSchema: - """Return paginated Twitch organizations.""" - page_data = _paginate( - Organization.objects.all().order_by("name"), - page=page, - page_size=page_size, - ) - return V1OrganizationListSchema( - total=page_data.total, - page=page_data.page, - page_size=page_data.page_size, - items=[_serialize_organization(org) for org in page_data.items], - ) - - -@api.get("/organizations/{twitch_id}/", response=V1OrganizationDetailSchema) -def get_organization( - request: HttpRequest, - twitch_id: str, -) -> V1OrganizationDetailSchema: - """Return one Twitch organization.""" - org = get_object_or_404(Organization.for_detail_view(), twitch_id=twitch_id) - now = timezone.now() - campaigns = ( - DropCampaign.objects - .filter(game__owners=org) - .select_related("game") - .prefetch_related( - "game__owners", - "allow_channels", - "time_based_drops__benefits", - ) - .order_by("-start_at") - .distinct() - ) - return V1OrganizationDetailSchema( - twitch_id=org.twitch_id, - name=org.name, - added_at=org.added_at, - updated_at=org.updated_at, - games=[_serialize_game_summary(game, now) for game in org.games.all()], # pyright: ignore[reportAttributeAccessIssue] - campaigns=[_serialize_campaign_detail(campaign, now) for campaign in campaigns], - ) - - -@api.get("/channels/", response=V1ChannelListSchema) -def list_channels( - request: HttpRequest, - page: int = 1, - page_size: int = DEFAULT_PAGE_SIZE, - search: str | None = None, -) -> V1ChannelListSchema: - """Return paginated Twitch channels.""" - page_data = _paginate( - _channel_list_queryset(search), - page=page, - page_size=page_size, - ) - return V1ChannelListSchema( - total=page_data.total, - page=page_data.page, - page_size=page_data.page_size, - items=[_serialize_channel(channel) for channel in page_data.items], - ) - - -@api.get("/channels/{twitch_id}/", response=V1ChannelDetailSchema) -def get_channel(request: HttpRequest, twitch_id: str) -> V1ChannelDetailSchema: - """Return one Twitch channel.""" - channel = get_object_or_404(Channel.for_detail_view(), twitch_id=twitch_id) - campaigns = _campaign_summary_queryset( - DropCampaign.objects.filter(allow_channels=channel), - ).order_by("-start_at") - now = timezone.now() - return V1ChannelDetailSchema( - twitch_id=channel.twitch_id, - name=channel.name, - display_name=channel.display_name, - allowed_campaign_count=channel.allowed_campaign_count, - added_at=channel.added_at, - updated_at=channel.updated_at, - campaigns=[ - _serialize_campaign_summary(campaign, now) for campaign in campaigns - ], - ) - - -@api.get("/reward-campaigns/", response=V1RewardCampaignListSchema) -def list_reward_campaigns( - request: HttpRequest, - page: int = 1, - page_size: int = DEFAULT_PAGE_SIZE, - game: str | None = None, - status: V1StatusFilter | None = None, -) -> V1RewardCampaignListSchema: - """Return paginated Twitch reward campaigns.""" - now = timezone.now() - queryset = RewardCampaign.objects.select_related("game").prefetch_related( - "game__owners", - ) - if game: - queryset = queryset.filter(game__twitch_id=game) - queryset = _apply_status_filter( - queryset.order_by("-starts_at"), - status, - now, - start_field="starts_at", - end_field="ends_at", - ) - page_data = _paginate(queryset, page=page, page_size=page_size) - return V1RewardCampaignListSchema( - total=page_data.total, - page=page_data.page, - page_size=page_data.page_size, - items=[ - _serialize_reward_campaign(campaign, now) for campaign in page_data.items - ], - ) - - -@api.get("/reward-campaigns/{twitch_id}/", response=V1RewardCampaignSchema) -def get_reward_campaign( - request: HttpRequest, - twitch_id: str, -) -> V1RewardCampaignSchema: - """Return one Twitch reward campaign.""" - campaign = get_object_or_404( - RewardCampaign.objects.select_related("game").prefetch_related( - "game__owners", - ), - twitch_id=twitch_id, - ) - return _serialize_reward_campaign(campaign, timezone.now()) - - -@api.get("/badges/", response=V1ChatBadgeSetListSchema) -def list_badges( - request: HttpRequest, - page: int = 1, - page_size: int = DEFAULT_PAGE_SIZE, -) -> V1ChatBadgeSetListSchema: - """Return paginated Twitch chat badge sets.""" - page_data = _paginate( - ChatBadgeSet.for_list_view(), - page=page, - page_size=page_size, - ) - return V1ChatBadgeSetListSchema( - total=page_data.total, - page=page_data.page, - page_size=page_data.page_size, - items=[_serialize_badge_set(badge_set) for badge_set in page_data.items], - ) - - -@api.get("/badges/{set_id}/", response=V1ChatBadgeSetSchema) -def get_badge_set(request: HttpRequest, set_id: str) -> V1ChatBadgeSetSchema: - """Return one Twitch chat badge set.""" - badge_set = get_object_or_404(ChatBadgeSet.for_list_view(), set_id=set_id) - return _serialize_badge_set(badge_set) diff --git a/twitch/feeds.py b/twitch/feeds.py index c1c07d4..b235c58 100644 --- a/twitch/feeds.py +++ b/twitch/feeds.py @@ -9,8 +9,6 @@ from django.conf import settings from django.contrib.humanize.templatetags.humanize import naturaltime from django.contrib.sites.shortcuts import get_current_site from django.contrib.syndication.views import Feed -from django.db.models import Exists -from django.db.models import OuterRef from django.db.models import Prefetch from django.db.models.query import QuerySet from django.http.request import HttpRequest @@ -49,23 +47,6 @@ if TYPE_CHECKING: logger: logging.Logger = logging.getLogger("ttvdrops") RSS_STYLESHEETS: list[str] = [static("rss_styles.xslt")] -TRUE_QUERY_VALUES: frozenset[str] = frozenset({"1", "true", "yes", "on"}) - - -def _query_bool(request: HttpRequest, name: str) -> bool: - """Return True when a query parameter is present with a truthy value.""" - return request.GET.get(name, "").strip().casefold() in TRUE_QUERY_VALUES - - -def _query_limit(request: HttpRequest) -> int | None: - """Return an integer ?limit value, or None when it is missing/invalid.""" - value: str | None = request.GET.get("limit") - if not value: - return None - try: - return int(value) - except TypeError, ValueError: - return None def discord_timestamp(dt: datetime.datetime | None) -> SafeText: @@ -275,21 +256,6 @@ def _active_drop_campaigns(queryset: QuerySet[DropCampaign]) -> QuerySet[DropCam return queryset.filter(start_at__lte=now, end_at__gte=now) -def _campaigns_with_free_drops( - queryset: QuerySet[DropCampaign], -) -> QuerySet[DropCampaign]: - """Keep campaigns that contain at least one drop without a sub requirement. - - Returns: - QuerySet[DropCampaign]: Campaigns that have at least one free drop. - """ - free_drops: QuerySet[TimeBasedDrop] = TimeBasedDrop.objects.filter( - campaign_id=OuterRef("pk"), - required_subs=0, - ) - return queryset.filter(Exists(free_drops)) - - def _active_reward_campaigns( queryset: QuerySet[RewardCampaign], ) -> QuerySet[RewardCampaign]: @@ -484,16 +450,11 @@ def generate_discord_date_html(item: Model) -> list[SafeText]: return parts -def generate_drops_summary_html( - item: DropCampaign, - *, - hide_paid: bool = False, -) -> list[SafeString]: +def generate_drops_summary_html(item: DropCampaign) -> list[SafeString]: """Generate HTML summary for drops and append to parts list. Args: item (DropCampaign): The drop campaign item containing the drops to summarize. - hide_paid: Exclude drops that require paid subscriptions from the summary. Returns: list[SafeString]: A list of SafeText elements summarizing the drops, or empty if no drops. @@ -506,7 +467,7 @@ def generate_drops_summary_html( drops: QuerySet[TimeBasedDrop] | None = getattr(item, "time_based_drops", None) if drops: - drops_data = _build_drops_data(drops.all(), hide_paid=hide_paid) + drops_data = _build_drops_data(drops.all()) if drops_data: parts.append( @@ -519,14 +480,13 @@ def generate_drops_summary_html( return parts -def generate_channels_html(item: Model, *, hide_paid: bool = False) -> list[SafeText]: +def generate_channels_html(item: Model) -> list[SafeText]: """Generate HTML for the list of channels associated with a drop campaign, if applicable. Only generates channel links if the drop is not subscription-only. If it is subscription-only, it will skip channel links to avoid confusion. Args: item (Model): The campaign item which may have an 'is_subscription_only' attribute. - hide_paid: Treat paid drops as hidden when deciding whether to show channels. Returns: list[SafeText]: A list containing the HTML for the channels section, or empty if subscription-only or no channels. @@ -538,7 +498,7 @@ def generate_channels_html(item: Model, *, hide_paid: bool = False) -> list[Safe if not channels: return parts - if not hide_paid and getattr(item, "is_subscription_only", False): + if getattr(item, "is_subscription_only", False): return parts game: Game | None = getattr(item, "game", None) @@ -639,11 +599,7 @@ def create_channel_list_html( parts.append(format_html("

Channels with this drop:

{}", html)) -def _build_drops_data( - drops_qs: QuerySet[TimeBasedDrop], - *, - hide_paid: bool = False, -) -> list[dict]: +def _build_drops_data(drops_qs: QuerySet[TimeBasedDrop]) -> list[dict]: """Build a simplified data structure for rendering drops in a template. Returns: @@ -655,8 +611,6 @@ def _build_drops_data( requirements: str = "" required_minutes: int | None = getattr(drop, "required_minutes_watched", None) required_subs: int = getattr(drop, "required_subs", 0) or 0 - if hide_paid and required_subs > 0: - continue if required_minutes: requirements = f"{required_minutes} minutes watched" if required_subs > 0: @@ -1017,7 +971,6 @@ class DropCampaignFeed(TTVDropsBaseFeed): item_guid_is_permalink = True _limit: int | None = None - _hide_paid: bool = False def __call__( self, @@ -1025,28 +978,29 @@ class DropCampaignFeed(TTVDropsBaseFeed): *args: str | int, **kwargs: str | int, ) -> HttpResponse: - """Override to capture supported feed query parameters from request. + """Override to capture limit parameter from request. Args: - request (HttpRequest): The incoming HTTP request, potentially containing 'limit' and 'hide_paid' query parameters. + request (HttpRequest): The incoming HTTP request, potentially containing a 'limit' query parameter. *args: Additional positional arguments. **kwargs: Additional keyword arguments. Returns: HttpResponse: The HTTP response generated by the parent Feed class after processing the request. """ - self._limit = _query_limit(request) - self._hide_paid = _query_bool(request, "hide_paid") + if request.GET.get("limit"): + try: + self._limit = int(request.GET.get("limit", 200)) + except ValueError, TypeError: + self._limit = None return super().__call__(request, *args, **kwargs) def items(self) -> list[DropCampaign]: - """Return latest active drop campaigns.""" + """Return the latest drop campaigns ordered by most recent start date (default 200, or limited by ?limit query param).""" limit: int = self._limit if self._limit is not None else 200 queryset: QuerySet[DropCampaign] = _active_drop_campaigns( DropCampaign.objects.order_by("-start_at"), ) - if self._hide_paid: - queryset = _campaigns_with_free_drops(queryset) return list(_with_campaign_related(queryset)[:limit]) def item_title(self, item: DropCampaign) -> SafeText: @@ -1061,8 +1015,8 @@ class DropCampaignFeed(TTVDropsBaseFeed): parts.extend(generate_item_image(item)) parts.extend(generate_description_html(item=item)) parts.extend(generate_date_html(item=item)) - parts.extend(generate_drops_summary_html(item=item, hide_paid=self._hide_paid)) - parts.extend(generate_channels_html(item, hide_paid=self._hide_paid)) + parts.extend(generate_drops_summary_html(item=item)) + parts.extend(generate_channels_html(item)) parts.extend(generate_details_link_html(item)) return SafeText("".join(str(p) for p in parts)) @@ -1156,7 +1110,6 @@ class GameCampaignFeed(TTVDropsBaseFeed): item_guid_is_permalink = True _limit: int | None = None - _hide_paid: bool = False def __call__( self, @@ -1164,18 +1117,21 @@ class GameCampaignFeed(TTVDropsBaseFeed): *args: str | int, **kwargs: str | int, ) -> HttpResponse: - """Override to capture supported feed query parameters from request. + """Override to capture limit parameter from request. Args: - request (HttpRequest): The incoming HTTP request, potentially containing 'limit' and 'hide_paid' query parameters. + request (HttpRequest): The incoming HTTP request, potentially containing a 'limit' query parameter *args: Additional positional arguments. **kwargs: Additional keyword arguments. Returns: HttpResponse: The HTTP response generated by the parent Feed class after processing the request. """ - self._limit = _query_limit(request) - self._hide_paid = _query_bool(request, "hide_paid") + if request.GET.get("limit"): + try: + self._limit = int(request.GET.get("limit", 200)) + except ValueError, TypeError: + self._limit = None return super().__call__(request, *args, **kwargs) def get_object(self, request: HttpRequest, twitch_id: str) -> Game: # noqa: ARG002 @@ -1203,15 +1159,13 @@ class GameCampaignFeed(TTVDropsBaseFeed): return f"Latest drop campaigns for {obj.display_name}" def items(self, obj: Game) -> list[DropCampaign]: - """Return latest active drop campaigns for this game.""" + """Return the latest drop campaigns for this game, ordered by most recent start date (default 200, or limited by ?limit query param).""" limit: int = self._limit if self._limit is not None else 200 queryset: QuerySet[DropCampaign] = _active_drop_campaigns( DropCampaign.objects.filter( game=obj, ).order_by("-start_at"), ) - if self._hide_paid: - queryset = _campaigns_with_free_drops(queryset) return list(_with_campaign_related(queryset)[:limit]) def item_title(self, item: DropCampaign) -> SafeText: @@ -1226,8 +1180,8 @@ class GameCampaignFeed(TTVDropsBaseFeed): parts.extend(generate_item_image_tag(item)) parts.extend(generate_details_link(item)) parts.extend(generate_date_html(item)) - parts.extend(generate_drops_summary_html(item, hide_paid=self._hide_paid)) - parts.extend(generate_channels_html(item, hide_paid=self._hide_paid)) + parts.extend(generate_drops_summary_html(item)) + parts.extend(generate_channels_html(item)) return SafeText("".join(str(p) for p in parts)) @@ -1600,8 +1554,8 @@ class DropCampaignDiscordFeed(TTVDropsAtomBaseFeed, DropCampaignFeed): parts.extend(generate_item_image(item)) parts.extend(generate_description_html(item=item)) parts.extend(generate_discord_date_html(item=item)) - parts.extend(generate_drops_summary_html(item=item, hide_paid=self._hide_paid)) - parts.extend(generate_channels_html(item, hide_paid=self._hide_paid)) + parts.extend(generate_drops_summary_html(item=item)) + parts.extend(generate_channels_html(item)) parts.extend(generate_details_link_html(item)) return SafeText("".join(str(p) for p in parts)) @@ -1621,8 +1575,8 @@ class GameCampaignDiscordFeed(TTVDropsAtomBaseFeed, GameCampaignFeed): parts.extend(generate_item_image_tag(item)) parts.extend(generate_details_link(item)) parts.extend(generate_discord_date_html(item)) - parts.extend(generate_drops_summary_html(item, hide_paid=self._hide_paid)) - parts.extend(generate_channels_html(item, hide_paid=self._hide_paid)) + parts.extend(generate_drops_summary_html(item)) + parts.extend(generate_channels_html(item)) return SafeText("".join(str(p) for p in parts)) diff --git a/twitch/models.py b/twitch/models.py index 943e520..4b93c99 100644 --- a/twitch/models.py +++ b/twitch/models.py @@ -91,10 +91,6 @@ class Organization(auto_prefetch.Model): "name", "display_name", "slug", - "box_art", - "box_art_file", - "box_art_width", - "box_art_height", ).order_by("display_name"), to_attr="games_for_detail", ), @@ -344,8 +340,6 @@ class Game(auto_prefetch.Model): "box_art_file", "box_art_width", "box_art_height", - "added_at", - "updated_at", ) .prefetch_related( Prefetch( @@ -809,8 +803,6 @@ class DropCampaign(auto_prefetch.Model): "start_at", "end_at", "game", - "game__twitch_id", - "game__name", "game__display_name", ) .prefetch_related( @@ -862,19 +854,12 @@ class DropCampaign(auto_prefetch.Model): "image_height", "start_at", "end_at", - "allow_is_enabled", - "operation_names", "added_at", "updated_at", - "is_fully_imported", "game__twitch_id", "game__name", "game__display_name", "game__slug", - "game__box_art", - "game__box_art_file", - "game__box_art_width", - "game__box_art_height", ) .prefetch_related( Prefetch( diff --git a/twitch/tests/test_api.py b/twitch/tests/test_api.py deleted file mode 100644 index 40edecc..0000000 --- a/twitch/tests/test_api.py +++ /dev/null @@ -1,438 +0,0 @@ -from datetime import timedelta - -from django.db import connection -from django.test import Client -from django.test import TestCase -from django.test.utils import CaptureQueriesContext -from django.urls import reverse -from django.utils import timezone - -from twitch import api as twitch_api -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 -from twitch.models import Organization -from twitch.models import RewardCampaign -from twitch.models import TimeBasedDrop - - -class TwitchApiV1TestCase(TestCase): - """Tests for the versioned Twitch API.""" - - def setUp(self) -> None: - """Create representative Twitch API fixture data.""" - self.client = Client() - now = timezone.now() - - self.org = Organization.objects.create( - twitch_id="org123", - name="Test Organization", - ) - self.game = Game.objects.create( - twitch_id="game123", - slug="test-game", - name="Test Game", - display_name="Test Game", - box_art="https://example.com/game.png", - ) - self.game.owners.add(self.org) - - self.channel = Channel.objects.create( - twitch_id="channel123", - name="testchannel", - display_name="TestChannel", - allowed_campaign_count=1, - ) - - self.campaign = DropCampaign.objects.create( - twitch_id="campaign123", - name="Test Campaign", - description="A test campaign", - details_url="https://example.com/details", - account_link_url="https://example.com/link", - image_url="https://example.com/campaign.png", - game=self.game, - start_at=now - timedelta(days=1), - end_at=now + timedelta(days=1), - operation_names=["DropCampaignDetails"], - is_fully_imported=True, - ) - self.campaign.allow_channels.add(self.channel) - - self.drop = TimeBasedDrop.objects.create( - twitch_id="drop123", - name="Test Drop", - campaign=self.campaign, - required_minutes_watched=30, - start_at=now - timedelta(days=1), - end_at=now + timedelta(days=1), - ) - self.benefit = DropBenefit.objects.create( - twitch_id="benefit123", - name="Test Benefit", - image_asset_url="https://example.com/benefit.png", - distribution_type="ITEM", - ) - self.drop.benefits.add(self.benefit) - - self.reward_campaign = RewardCampaign.objects.create( - twitch_id="reward123", - name="Test Reward", - brand="Test Brand", - starts_at=now - timedelta(days=1), - ends_at=now + timedelta(days=1), - status="ACTIVE", - summary="Reward summary", - external_url="https://example.com/reward", - game=self.game, - ) - - self.badge_set = ChatBadgeSet.objects.create(set_id="test-badge-set") - ChatBadge.objects.create( - badge_set=self.badge_set, - badge_id="1", - image_url_1x="https://example.com/badge-1x.png", - image_url_2x="https://example.com/badge-2x.png", - image_url_4x="https://example.com/badge-4x.png", - title="Test Badge", - description="Test badge description", - ) - - def _create_secondary_api_fixture(self) -> None: - now = timezone.now() - org = Organization.objects.create( - twitch_id="org456", - name="Second Organization", - ) - game = Game.objects.create( - twitch_id="game456", - slug="second-game", - name="Second Game", - display_name="Second Game", - box_art="https://example.com/second-game.png", - ) - game.owners.add(org) - - channel = Channel.objects.create( - twitch_id="channel456", - name="secondchannel", - display_name="SecondChannel", - allowed_campaign_count=1, - ) - - campaign = DropCampaign.objects.create( - twitch_id="campaign456", - name="Second Campaign", - description="Another test campaign", - details_url="https://example.com/second-details", - account_link_url="https://example.com/second-link", - image_url="https://example.com/second-campaign.png", - game=game, - start_at=now - timedelta(days=2), - end_at=now + timedelta(days=2), - operation_names=["DropCampaignDetails"], - is_fully_imported=True, - ) - campaign.allow_channels.add(channel) - - drop = TimeBasedDrop.objects.create( - twitch_id="drop456", - name="Second Drop", - campaign=campaign, - required_minutes_watched=60, - start_at=now - timedelta(days=2), - end_at=now + timedelta(days=2), - ) - benefit = DropBenefit.objects.create( - twitch_id="benefit456", - name="Second Benefit", - image_asset_url="https://example.com/second-benefit.png", - distribution_type="ITEM", - ) - drop.benefits.add(benefit) - - RewardCampaign.objects.create( - twitch_id="reward456", - name="Second Reward", - brand="Second Brand", - starts_at=now - timedelta(days=2), - ends_at=now + timedelta(days=2), - status="ACTIVE", - summary="Second reward summary", - external_url="https://example.com/second-reward", - game=game, - ) - - badge_set = ChatBadgeSet.objects.create(set_id="second-badge-set") - ChatBadge.objects.create( - badge_set=badge_set, - badge_id="1", - image_url_1x="https://example.com/second-badge-1x.png", - image_url_2x="https://example.com/second-badge-2x.png", - image_url_4x="https://example.com/second-badge-4x.png", - title="Second Badge", - description="Second badge description", - ) - - def test_v1_campaign_list(self) -> None: - """Return active campaigns from the v1 list endpoint.""" - response = self.client.get("/twitch/api/v1/campaigns/?status=active") - - assert response.status_code == 200 - assert "Content-Disposition" not in response - data = response.json() - assert data["total"] == 1 - assert data["page"] == 1 - assert data["items"][0]["twitch_id"] == "campaign123" - assert data["items"][0]["status"] == "active" - assert data["items"][0]["game"]["twitch_id"] == "game123" - - def test_v1_campaign_detail(self) -> None: - """Return nested campaign detail data from the v1 endpoint.""" - response = self.client.get("/twitch/api/v1/campaigns/campaign123/") - - assert response.status_code == 200 - data = response.json() - assert data["operation_names"] == ["DropCampaignDetails"] - assert data["game"]["box_art_url"] == "https://example.com/game.png" - assert data["allowed_channels"][0]["twitch_id"] == "channel123" - assert data["drops"][0]["benefits"][0]["twitch_id"] == "benefit123" - - def test_v1_campaign_detail_game_box_art_does_not_load_deferred_file(self) -> None: - """Serialize campaign game box art without lazy-loading ImageField data.""" - campaign = DropCampaign.for_detail_view("campaign123") - - image_fields = {"box_art_file", "box_art_width", "box_art_height"} - assert campaign.game.get_deferred_fields().isdisjoint(image_fields) - with CaptureQueriesContext(connection) as capture: - box_art_url = twitch_api._game_box_art_url(campaign.game) - - assert box_art_url == "https://example.com/game.png" - assert len(capture) == 0 - - def test_v1_campaign_detail_uses_local_game_box_art(self) -> None: - """Return locally cached game box art from campaign detail responses.""" - self.game.box_art_file = "games/box_art/local.png" - self.game.box_art_width = 285 - self.game.box_art_height = 380 - self.game.save( - update_fields=["box_art_file", "box_art_width", "box_art_height"], - ) - - response = self.client.get("/twitch/api/v1/campaigns/campaign123/") - - assert response.status_code == 200 - data = response.json() - assert data["game"]["box_art_url"] == self.game.box_art_file.url - - def test_v1_all_endpoints_handle_multiple_rows(self) -> None: - """Exercise all v1 routes with enough rows to catch deferred loads.""" - self._create_secondary_api_fixture() - list_urls: list[tuple[str, int]] = [ - ("/twitch/api/v1/campaigns/?page_size=50", 2), - ("/twitch/api/v1/games/?page_size=50", 2), - ("/twitch/api/v1/organizations/?page_size=50", 2), - ("/twitch/api/v1/channels/?page_size=50", 2), - ("/twitch/api/v1/reward-campaigns/?page_size=50", 2), - ("/twitch/api/v1/badges/?page_size=50", 2), - ] - detail_urls = [ - "/twitch/api/v1/campaigns/campaign123/", - "/twitch/api/v1/campaigns/campaign456/", - "/twitch/api/v1/games/game123/", - "/twitch/api/v1/games/game456/", - "/twitch/api/v1/organizations/org123/", - "/twitch/api/v1/organizations/org456/", - "/twitch/api/v1/channels/channel123/", - "/twitch/api/v1/channels/channel456/", - "/twitch/api/v1/reward-campaigns/reward123/", - "/twitch/api/v1/reward-campaigns/reward456/", - "/twitch/api/v1/badges/test-badge-set/", - "/twitch/api/v1/badges/second-badge-set/", - ] - - for url, expected_total in list_urls: - response = self.client.get(url) - assert response.status_code == 200, url - data = response.json() - assert data["total"] == expected_total - assert len(data["items"]) == expected_total - - for url in detail_urls: - response = self.client.get(url) - assert response.status_code == 200, url - assert response.json() - - schema_response = self.client.get(reverse("twitch:twitch-api-v1:openapi-json")) - assert schema_response.status_code == 200 - assert schema_response.json() - - docs_response = self.client.get(reverse("twitch:twitch-api-v1:openapi-view")) - assert docs_response.status_code == 200 - - def test_v1_collection_endpoints(self) -> None: - """Return v1 list responses for all Twitch API collections.""" - checks = [ - ("/twitch/api/v1/games/", "game123"), - ("/twitch/api/v1/organizations/", "org123"), - ("/twitch/api/v1/channels/", "channel123"), - ("/twitch/api/v1/reward-campaigns/", "reward123"), - ("/twitch/api/v1/badges/", "test-badge-set"), - ] - - for url, expected_id in checks: - response = self.client.get(url) - assert response.status_code == 200 - data = response.json() - actual_id = data["items"][0].get( - "twitch_id", - data["items"][0].get("set_id"), - ) - assert actual_id == expected_id - - games_response = self.client.get("/twitch/api/v1/games/") - games_data = games_response.json() - assert games_data["items"][0]["campaign_count"] == 1 - assert games_data["items"][0]["active_campaign_count"] == 1 - - def test_v1_organization_detail_includes_games_and_campaigns(self) -> None: - """Return concrete game counts and detailed organization campaigns.""" - response = self.client.get("/twitch/api/v1/organizations/org123/") - - assert response.status_code == 200 - data = response.json() - assert data["games"][0]["twitch_id"] == "game123" - assert data["games"][0]["campaign_count"] == 1 - assert data["games"][0]["active_campaign_count"] == 1 - assert data["campaigns"][0]["twitch_id"] == "campaign123" - assert data["campaigns"][0]["operation_names"] == ["DropCampaignDetails"] - assert data["campaigns"][0]["allowed_channels"][0]["twitch_id"] == "channel123" - assert data["campaigns"][0]["drops"][0]["twitch_id"] == "drop123" - assert ( - data["campaigns"][0]["drops"][0]["benefits"][0]["twitch_id"] == "benefit123" - ) - - def test_v1_game_and_channel_detail_include_campaign_data(self) -> None: - """Return campaign API fields on game and channel detail responses.""" - checks = [ - "/twitch/api/v1/games/game123/", - "/twitch/api/v1/channels/channel123/", - ] - - for url in checks: - response = self.client.get(url) - assert response.status_code == 200 - data = response.json() - campaign = data["campaigns"][0] - assert campaign["description"] == "A test campaign" - assert campaign["details_url"] == "https://example.com/details" - assert campaign["account_link_url"] == "https://example.com/link" - assert campaign["image_url"] == "https://example.com/campaign.png" - - def test_v1_detail_not_found(self) -> None: - """Return 404 for missing v1 campaign detail records.""" - response = self.client.get("/twitch/api/v1/campaigns/missing/") - - assert response.status_code == 404 - - def test_v1_docs_endpoint(self) -> None: - """Render the versioned Twitch API documentation page.""" - response = self.client.get("/twitch/api/v1/docs") - - assert response.status_code == 200 - assert reverse("twitch:twitch-api-v1:openapi-json") in response.content.decode() - - def test_v1_docs_links_render_on_twitch_pages(self) -> None: - """Expose API docs in nav and resource API links in feed link groups.""" - checks = [ - ( - reverse("core:docs_rss"), - "API Docs", - "/twitch/api/v1/docs", - reverse("twitch:twitch-api-v1:openapi-view"), - ), - ( - reverse("twitch:dashboard"), - "API Docs", - "[api]", - reverse("twitch:twitch-api-v1:list_campaigns"), - ), - ( - reverse("twitch:campaign_list"), - "API Docs", - "[api]", - reverse("twitch:twitch-api-v1:list_campaigns"), - ), - ( - reverse("twitch:campaign_detail", args=[self.campaign.twitch_id]), - "API Docs", - "[api]", - reverse( - "twitch:twitch-api-v1:get_campaign", - args=[self.campaign.twitch_id], - ), - ), - ( - reverse("twitch:game_detail", args=[self.game.twitch_id]), - "API Docs", - "[api]", - reverse("twitch:twitch-api-v1:get_game", args=[self.game.twitch_id]), - ), - ( - reverse("twitch:games_grid"), - "API Docs", - "[api]", - reverse("twitch:twitch-api-v1:list_games"), - ), - ( - reverse("twitch:org_list"), - "API Docs", - "[api]", - reverse("twitch:twitch-api-v1:list_organizations"), - ), - ( - reverse("twitch:reward_campaign_list"), - "API Docs", - "[api]", - reverse("twitch:twitch-api-v1:list_reward_campaigns"), - ), - ( - reverse( - "twitch:reward_campaign_detail", - args=[self.reward_campaign.twitch_id], - ), - "API Docs", - "[api]", - reverse( - "twitch:twitch-api-v1:get_reward_campaign", - args=[self.reward_campaign.twitch_id], - ), - ), - ] - - for url, nav_text, feed_text, api_href in checks: - response = self.client.get(url) - assert response.status_code == 200 - content = response.content.decode() - assert reverse("twitch:twitch-api-v1:openapi-view") in content - assert api_href in content - assert nav_text in content - assert feed_text in content - - def test_campaign_detail_api_link_targets_campaign_endpoint(self) -> None: - """Link campaign detail [api] directly to that campaign JSON endpoint.""" - response = self.client.get( - reverse("twitch:campaign_detail", args=[self.campaign.twitch_id]), - ) - - assert response.status_code == 200 - content = response.content.decode() - campaign_api_url = reverse( - "twitch:twitch-api-v1:get_campaign", - args=[self.campaign.twitch_id], - ) - assert f'href="{campaign_api_url}"' in content - assert 'title="Twitch campaign API">[api]' in content diff --git a/twitch/tests/test_feeds.py b/twitch/tests/test_feeds.py index 19d5f71..afd7a83 100644 --- a/twitch/tests/test_feeds.py +++ b/twitch/tests/test_feeds.py @@ -104,90 +104,6 @@ class RSSFeedTestCase(TestCase): game=self.game, ) - def _create_mixed_paid_and_free_campaign(self) -> DropCampaign: - """Create an active campaign containing both free and subscription-gated drops. - - Returns: - DropCampaign: The mixed campaign fixture. - """ - campaign: DropCampaign = DropCampaign.objects.create( - twitch_id="mixed-campaign-123", - name="Mixed Watch And Subscription Campaign", - game=self.game, - start_at=timezone.now(), - end_at=timezone.now() + timedelta(days=7), - operation_names=["DropCampaignDetails"], - ) - channel: Channel = Channel.objects.create( - twitch_id="mixed-channel-123", - name="mixedchannel", - display_name="MixedChannel", - ) - campaign.allow_channels.add(channel) - - free_drop: TimeBasedDrop = TimeBasedDrop.objects.create( - twitch_id="free-drop-123", - name="Watch Drop", - campaign=campaign, - required_minutes_watched=30, - required_subs=0, - start_at=timezone.now(), - end_at=timezone.now() + timedelta(hours=1), - ) - paid_drop: TimeBasedDrop = TimeBasedDrop.objects.create( - twitch_id="paid-drop-123", - name="Subscription Required Drop", - campaign=campaign, - required_minutes_watched=0, - required_subs=1, - start_at=timezone.now(), - end_at=timezone.now() + timedelta(hours=1), - ) - free_benefit: DropBenefit = DropBenefit.objects.create( - twitch_id="free-benefit-123", - name="Free Benefit", - distribution_type="ITEM", - ) - paid_benefit: DropBenefit = DropBenefit.objects.create( - twitch_id="paid-benefit-123", - name="Paid Benefit", - distribution_type="ITEM", - ) - free_drop.benefits.add(free_benefit) - paid_drop.benefits.add(paid_benefit) - return campaign - - def _create_paid_only_campaign(self) -> DropCampaign: - """Create an active campaign where every drop requires a subscription. - - Returns: - DropCampaign: The paid-only campaign fixture. - """ - campaign: DropCampaign = DropCampaign.objects.create( - twitch_id="paid-only-campaign-123", - name="Paid Only Campaign", - game=self.game, - start_at=timezone.now(), - end_at=timezone.now() + timedelta(days=7), - operation_names=["DropCampaignDetails"], - ) - drop: TimeBasedDrop = TimeBasedDrop.objects.create( - twitch_id="paid-only-drop-123", - name="Paid Only Drop", - campaign=campaign, - required_minutes_watched=0, - required_subs=1, - start_at=timezone.now(), - end_at=timezone.now() + timedelta(hours=1), - ) - benefit: DropBenefit = DropBenefit.objects.create( - twitch_id="paid-only-benefit-123", - name="Paid Only Benefit", - distribution_type="ITEM", - ) - drop.benefits.add(benefit) - return campaign - def test_organization_feed(self) -> None: """Test organization feed returns 200.""" url: str = reverse("core:organization_feed") @@ -578,43 +494,6 @@ class RSSFeedTestCase(TestCase): assert "Past Campaign" not in content assert "Upcoming Campaign" not in content - def test_campaign_feeds_can_hide_paid_drops(self) -> None: - """Campaign feeds should support ?hide_paid=1 for subscription-gated drops.""" - mixed_campaign: DropCampaign = self._create_mixed_paid_and_free_campaign() - paid_only_campaign: DropCampaign = self._create_paid_only_campaign() - - feed_urls: list[str] = [ - reverse("core:campaign_feed"), - reverse("core:game_campaign_feed", args=[self.game.twitch_id]), - reverse("core:campaign_feed_atom"), - reverse("core:game_campaign_feed_atom", args=[self.game.twitch_id]), - reverse("core:campaign_feed_discord"), - reverse("core:game_campaign_feed_discord", args=[self.game.twitch_id]), - ] - - for url in feed_urls: - hidden_response: _MonkeyPatchedWSGIResponse = self.client.get( - url, - {"hide_paid": "1"}, - ) - assert hidden_response.status_code == 200 - hidden_content: str = hidden_response.content.decode("utf-8") - assert mixed_campaign.name in hidden_content - assert paid_only_campaign.name not in hidden_content - assert "Free Benefit" in hidden_content - assert "Paid Only Benefit" not in hidden_content - assert "30 minutes watched" in hidden_content - assert "1 sub required" not in hidden_content - assert "MixedChannel" in hidden_content - - default_response: _MonkeyPatchedWSGIResponse = self.client.get(url) - assert default_response.status_code == 200 - default_content: str = default_response.content.decode("utf-8") - assert mixed_campaign.name in default_content - assert paid_only_campaign.name in default_content - assert "Paid Only Benefit" in default_content - assert "1 sub required" in default_content - def test_campaign_feed_enclosure_helpers(self) -> None: """Helper methods for campaigns should respect new fields.""" feed = DropCampaignFeed() diff --git a/twitch/tests/test_views.py b/twitch/tests/test_views.py index 4beb39d..62cf1d8 100644 --- a/twitch/tests/test_views.py +++ b/twitch/tests/test_views.py @@ -2504,227 +2504,6 @@ class TestChannelListView: assert "g-rss" in response.content.decode("utf-8") -@pytest.mark.django_db -class TestRewardCampaignViews: - """Tests for Twitch reward campaign list and detail views.""" - - def _create_game(self, twitch_id: str, display_name: str) -> Game: - game: Game = Game.objects.create( - twitch_id=twitch_id, - slug=twitch_id, - name=display_name, - display_name=display_name, - box_art=f"https://example.com/{twitch_id}.png", - ) - org: Organization = Organization.objects.create( - twitch_id=f"{twitch_id}-org", - name=f"{display_name} Org", - ) - game.owners.add(org) - return game - - def _create_reward_campaign( # noqa: PLR0913 - self, - twitch_id: str, - *, - brand: str, - name: str, - game: Game | None, - starts_delta: timedelta, - ends_delta: timedelta, - ) -> RewardCampaign: - now: datetime.datetime = timezone.now() - return RewardCampaign.objects.create( - twitch_id=twitch_id, - brand=brand, - name=name, - summary=f"{name} summary", - instructions=f"{name} instructions", - external_url=f"https://example.com/{twitch_id}/external", - about_url=f"https://example.com/{twitch_id}/about", - image_url=f"https://example.com/{twitch_id}.png", - starts_at=now + starts_delta, - ends_at=now + ends_delta, - status="ACTIVE", - is_sitewide=game is None, - game=game, - ) - - def test_reward_campaign_list_renders_expired_campaigns( - self, - client: Client, - ) -> None: - """Render expired reward campaigns with feed and API links.""" - game: Game = self._create_game("reward-list-game", "Reward List Game") - expired = self._create_reward_campaign( - "reward-list-expired", - brand="Expired Brand", - name="Expired Reward", - game=game, - starts_delta=-timedelta(days=4), - ends_delta=-timedelta(days=1), - ) - self._create_reward_campaign( - "reward-list-active", - brand="Active Brand", - name="Active Reward", - game=game, - starts_delta=-timedelta(days=1), - ends_delta=timedelta(days=1), - ) - - response: _MonkeyPatchedWSGIResponse = client.get( - reverse("twitch:reward_campaign_list"), - ) - - assert response.status_code == 200 - content: str = response.content.decode() - assert "Expired Brand: Expired Reward" in content - assert ( - reverse("twitch:reward_campaign_detail", args=[expired.twitch_id]) - in content - ) - assert reverse("twitch:game_detail", args=[game.twitch_id]) in content - assert reverse("core:reward_campaign_feed") in content - assert reverse("twitch:twitch-api-v1:list_reward_campaigns") in content - assert response.context["reward_campaigns"].paginator.count == 2 - - def test_reward_campaign_list_filters_status_and_game( - self, - client: Client, - ) -> None: - """Filter reward campaign context by status and game.""" - selected_game: Game = self._create_game("reward-filter-game", "Reward Filter") - other_game: Game = self._create_game("reward-filter-other", "Reward Other") - active = self._create_reward_campaign( - "reward-filter-active", - brand="Filter Brand", - name="Active Filter Reward", - game=selected_game, - starts_delta=-timedelta(days=1), - ends_delta=timedelta(days=1), - ) - self._create_reward_campaign( - "reward-filter-other-active", - brand="Other Brand", - name="Other Active Reward", - game=other_game, - starts_delta=-timedelta(days=1), - ends_delta=timedelta(days=1), - ) - self._create_reward_campaign( - "reward-filter-expired", - brand="Expired Brand", - name="Expired Filter Reward", - game=selected_game, - starts_delta=-timedelta(days=4), - ends_delta=-timedelta(days=1), - ) - - response: _MonkeyPatchedWSGIResponse = client.get( - reverse("twitch:reward_campaign_list") - + f"?status=active&game={selected_game.twitch_id}", - ) - - assert response.status_code == 200 - campaigns = list(response.context["reward_campaigns"]) - assert campaigns == [active] - assert response.context["selected_status"] == "active" - assert response.context["selected_game"] == selected_game.twitch_id - - def test_reward_campaign_detail_renders_campaign_data( - self, - client: Client, - ) -> None: - """Render reward campaign detail fields and resource links.""" - game: Game = self._create_game("reward-detail-game", "Reward Detail Game") - reward = self._create_reward_campaign( - "reward-detail", - brand="Detail Brand", - name="Detail Reward", - game=game, - starts_delta=-timedelta(days=1), - ends_delta=timedelta(days=1), - ) - - response: _MonkeyPatchedWSGIResponse = client.get( - reverse("twitch:reward_campaign_detail", args=[reward.twitch_id]), - ) - - assert response.status_code == 200 - content: str = response.content.decode() - assert "Detail Brand: Detail Reward" in content - assert "Detail Reward summary" in content - assert "Detail Reward instructions" in content - assert reverse("twitch:game_detail", args=[game.twitch_id]) in content - assert ( - reverse( - "twitch:twitch-api-v1:get_reward_campaign", - args=[reward.twitch_id], - ) - in content - ) - assert reward.external_url in content - assert reward.about_url in content - assert response.context["is_active"] is True - - def test_reward_campaign_detail_404_for_missing_campaign( - self, - client: Client, - ) -> None: - """Return 404 for missing reward campaign detail pages.""" - response: _MonkeyPatchedWSGIResponse = client.get( - reverse("twitch:reward_campaign_detail", args=["missing-reward"]), - ) - - assert response.status_code == 404 - - def test_reward_campaign_list_query_count_stays_flat( - self, - client: Client, - ) -> None: - """Reward campaign list should not issue N+1 queries as rows grow.""" - game: Game = self._create_game("reward-flat-game", "Reward Flat Game") - - def _select_count() -> int: - with CaptureQueriesContext(connection) as ctx: - response: _MonkeyPatchedWSGIResponse = client.get( - reverse("twitch:reward_campaign_list"), - ) - assert response.status_code == 200 - return sum( - 1 - for query in ctx.captured_queries - if query["sql"].lstrip().upper().startswith("SELECT") - ) - - self._create_reward_campaign( - "reward-flat-base", - brand="Flat Brand", - name="Flat Base Reward", - game=game, - starts_delta=-timedelta(days=4), - ends_delta=-timedelta(days=1), - ) - baseline: int = _select_count() - - for index in range(10): - self._create_reward_campaign( - f"reward-flat-extra-{index}", - brand="Flat Brand", - name=f"Flat Extra Reward {index}", - game=game, - starts_delta=-timedelta(days=4), - ends_delta=-timedelta(days=1), - ) - - scaled: int = _select_count() - assert scaled <= baseline + 2, ( - "Reward campaign list SELECT count grew; possible N+1. " - f"baseline={baseline}, scaled={scaled}" - ) - - @pytest.mark.django_db class TestSEOHelperFunctions: """Tests for SEO helper functions.""" diff --git a/twitch/urls.py b/twitch/urls.py index 8acb346..0ca95d1 100644 --- a/twitch/urls.py +++ b/twitch/urls.py @@ -3,7 +3,6 @@ from typing import TYPE_CHECKING from django.urls import path from twitch import views -from twitch.api import api as twitch_api_v1 if TYPE_CHECKING: from django.urls.resolvers import URLPattern @@ -13,8 +12,6 @@ app_name = "twitch" urlpatterns: list[URLPattern | URLResolver] = [ - # /twitch/api/v1/ - path("api/v1/", twitch_api_v1.urls), # /twitch/ path("", views.dashboard, name="dashboard"), # /twitch/badges/