Compare commits

...

3 commits

Author SHA1 Message Date
8229b0fe80
Add support for a hide_paid query parameter to campaign feeds
All checks were successful
Deploy to Server / deploy (push) Successful in 28s
2026-05-05 06:05:10 +02:00
e960b09084
Add API for Twitch data 2026-05-05 05:01:48 +02:00
f01b6c9ba1
Bump ruff-pre-commit to v0.15.12 2026-05-04 23:11:51 +02:00
21 changed files with 1728 additions and 31 deletions

View file

@ -27,7 +27,7 @@ repos:
args: [--target-version, "6.0"]
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.15.9
rev: v0.15.12
hooks:
- id: ruff-check
args: ["--fix", "--exit-non-zero-on-fix"]

View file

@ -12,7 +12,9 @@ dependencies = [
"django-celery-beat",
"django-celery-results",
"django-debug-toolbar",
"django-ninja",
"django-silk",
"django-zeal",
"django",
"flower",
"gunicorn",
@ -31,7 +33,6 @@ dependencies = [
"setproctitle",
"sitemap-parser",
"tqdm",
"django-zeal>=2.1.0",
]

View file

@ -220,6 +220,7 @@
<a href="{% url 'twitch:channel_list' %}">Channels</a> |
<a href="{% url 'twitch:badge_list' %}">Badges</a> |
<a href="{% url 'twitch:emote_gallery' %}">Emotes</a> |
<a href="{% url 'twitch:twitch-api-v1:openapi-view' %}">API Docs</a> |
<a href="https://www.twitch.tv/drops/inventory">Inventory</a>
<br />
<strong>Kick</strong>

View file

@ -48,6 +48,8 @@
title="Atom feed for Twitch campaigns">[atom]</a>
<a href="{% url 'core:campaign_feed_discord' %}"
title="Discord feed for Twitch campaigns">[discord]</a>
<a href="{% url 'twitch:twitch-api-v1:list_campaigns' %}"
title="Twitch campaigns API">[api]</a>
<a href="{% url 'core:docs_rss' %}" title="RSS feed documentation">[explain]</a>
</div>
</header>

View file

@ -16,6 +16,15 @@
that include Discord relative timestamps (e.g., <code>&lt;t:1773450272:R&gt;</code>) for dates,
making them ideal for Discord bots and integrations. Future enhancements may include Discord-specific formatting or content.
</p>
<p>
Twitch JSON API documentation is available at
<a href="{% url 'twitch:twitch-api-v1:openapi-view' %}">/twitch/api/v1/docs</a>.
</p>
<p>
Twitch campaign feeds accept <code>?limit=50</code> to change item count and
<code>?hide_paid=1</code> to hide subscription-gated drops and skip campaigns
with no free drops.
</p>
<section>
<h2>Global RSS Feeds</h2>
<table>

View file

@ -103,6 +103,8 @@
title="Atom feed for {{ campaign.game.display_name }} campaigns">[atom]</a>
<a href="{% url 'core:game_campaign_feed_discord' campaign.game.twitch_id %}"
title="Discord feed for {{ campaign.game.display_name }} campaigns">[discord]</a>
<a href="{% url 'twitch:twitch-api-v1:get_campaign' campaign.twitch_id %}"
title="Twitch campaign API">[api]</a>
<a href="{% url 'core:docs_rss' %}" title="RSS feed documentation">[explain]</a>
{% endif %}
</div>

View file

@ -33,6 +33,8 @@
title="Atom feed for all campaigns">[atom]</a>
<a href="{% url 'core:campaign_feed_discord' %}"
title="Discord feed for all campaigns">[discord]</a>
<a href="{% url 'twitch:twitch-api-v1:list_campaigns' %}"
title="Twitch campaigns API">[api]</a>
<a href="{% url 'core:docs_rss' %}" title="RSS feed documentation">[explain]</a>
<a href="{% url 'twitch:export_campaigns_csv' %}{% if request.GET %}?{{ request.GET.urlencode }}{% endif %}"
title="Export campaigns as CSV">[csv]</a>

View file

@ -36,6 +36,8 @@
title="Atom feed for campaigns">[atom]</a>
<a href="{% url 'core:campaign_feed_discord' %}"
title="Discord feed for campaigns">[discord]</a>
<a href="{% url 'twitch:twitch-api-v1:list_campaigns' %}"
title="Twitch campaigns API">[api]</a>
<a href="{% url 'core:docs_rss' %}" title="RSS feed documentation">[explain]</a>
</div>
<hr />

View file

@ -65,6 +65,8 @@
title="Atom feed for {{ game.display_name }} campaigns">[atom]</a>
<a href="{% url 'core:game_campaign_feed_discord' game.twitch_id %}"
title="Discord feed for {{ game.display_name }} campaigns">[discord]</a>
<a href="{% url 'twitch:twitch-api-v1:get_game' game.twitch_id %}"
title="Twitch game API">[api]</a>
<a href="{% url 'core:docs_rss' %}" title="RSS feed documentation">[explain]</a>
</div>
</div>

View file

@ -32,6 +32,8 @@
title="Atom feed for all games">[atom]</a>
<a href="{% url 'core:game_feed_discord' %}"
title="Discord feed for all games">[discord]</a>
<a href="{% url 'twitch:twitch-api-v1:list_games' %}"
title="Twitch games API">[api]</a>
<a href="{% url 'core:docs_rss' %}" title="RSS feed documentation">[explain]</a>
<a href="{% url 'twitch:export_games_csv' %}"
title="Export all games as CSV">[csv]</a>

View file

@ -30,6 +30,8 @@
title="Atom feed for all games">[atom]</a>
<a href="{% url 'core:game_feed_discord' %}"
title="Discord feed for all games">[discord]</a>
<a href="{% url 'twitch:twitch-api-v1:list_games' %}"
title="Twitch games API">[api]</a>
<a href="{% url 'core:docs_rss' %}" title="RSS feed documentation">[explain]</a>
<a href="{% url 'twitch:export_games_csv' %}"
title="Export all games as CSV">[csv]</a>

View file

@ -15,6 +15,8 @@
title="Atom feed for all organizations">[atom]</a>
<a href="{% url 'core:organization_feed_discord' %}"
title="Discord feed for all organizations">[discord]</a>
<a href="{% url 'twitch:twitch-api-v1:list_organizations' %}"
title="Twitch organizations API">[api]</a>
<a href="{% url 'core:docs_rss' %}" title="RSS feed documentation">[explain]</a>
<a href="{% url 'twitch:export_organizations_csv' %}"
title="Export all organizations as CSV">[csv]</a>

View file

@ -110,6 +110,8 @@
title="Atom feed for all reward campaigns">[atom]</a>
<a href="{% url 'core:reward_campaign_feed_discord' %}"
title="Discord feed for all reward campaigns">[discord]</a>
<a href="{% url 'twitch:twitch-api-v1:get_reward_campaign' reward_campaign.twitch_id %}"
title="Twitch reward campaign API">[api]</a>
<a href="{% url 'core:docs_rss' %}" title="RSS feed documentation">[explain]</a>
{% if reward_campaign.external_url %}
<a href="{{ reward_campaign.external_url }}"

View file

@ -36,6 +36,8 @@
title="Atom feed for all reward campaigns">[atom]</a>
<a href="{% url 'core:reward_campaign_feed_discord' %}"
title="Discord feed for all reward campaigns">[discord]</a>
<a href="{% url 'twitch:twitch-api-v1:list_reward_campaigns' %}"
title="Twitch reward campaigns API">[api]</a>
<a href="{% url 'core:docs_rss' %}" title="RSS feed documentation">[explain]</a>
</div>
{% if reward_campaigns %}

822
twitch/api.py Normal file
View file

@ -0,0 +1,822 @@
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)

View file

@ -9,6 +9,8 @@ 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
@ -47,6 +49,23 @@ 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:
@ -256,6 +275,21 @@ 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]:
@ -450,11 +484,16 @@ def generate_discord_date_html(item: Model) -> list[SafeText]:
return parts
def generate_drops_summary_html(item: DropCampaign) -> list[SafeString]:
def generate_drops_summary_html(
item: DropCampaign,
*,
hide_paid: bool = False,
) -> 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.
@ -467,7 +506,7 @@ def generate_drops_summary_html(item: DropCampaign) -> list[SafeString]:
drops: QuerySet[TimeBasedDrop] | None = getattr(item, "time_based_drops", None)
if drops:
drops_data = _build_drops_data(drops.all())
drops_data = _build_drops_data(drops.all(), hide_paid=hide_paid)
if drops_data:
parts.append(
@ -480,13 +519,14 @@ def generate_drops_summary_html(item: DropCampaign) -> list[SafeString]:
return parts
def generate_channels_html(item: Model) -> list[SafeText]:
def generate_channels_html(item: Model, *, hide_paid: bool = False) -> 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.
@ -498,7 +538,7 @@ def generate_channels_html(item: Model) -> list[SafeText]:
if not channels:
return parts
if getattr(item, "is_subscription_only", False):
if not hide_paid and getattr(item, "is_subscription_only", False):
return parts
game: Game | None = getattr(item, "game", None)
@ -599,7 +639,11 @@ def create_channel_list_html(
parts.append(format_html("<p>Channels with this drop:</p>{}", html))
def _build_drops_data(drops_qs: QuerySet[TimeBasedDrop]) -> list[dict]:
def _build_drops_data(
drops_qs: QuerySet[TimeBasedDrop],
*,
hide_paid: bool = False,
) -> list[dict]:
"""Build a simplified data structure for rendering drops in a template.
Returns:
@ -611,6 +655,8 @@ def _build_drops_data(drops_qs: QuerySet[TimeBasedDrop]) -> list[dict]:
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:
@ -971,6 +1017,7 @@ class DropCampaignFeed(TTVDropsBaseFeed):
item_guid_is_permalink = True
_limit: int | None = None
_hide_paid: bool = False
def __call__(
self,
@ -978,29 +1025,28 @@ class DropCampaignFeed(TTVDropsBaseFeed):
*args: str | int,
**kwargs: str | int,
) -> HttpResponse:
"""Override to capture limit parameter from request.
"""Override to capture supported feed query parameters from request.
Args:
request (HttpRequest): The incoming HTTP request, potentially containing a 'limit' query parameter.
request (HttpRequest): The incoming HTTP request, potentially containing 'limit' and 'hide_paid' query parameters.
*args: Additional positional arguments.
**kwargs: Additional keyword arguments.
Returns:
HttpResponse: The HTTP response generated by the parent Feed class after processing the request.
"""
if request.GET.get("limit"):
try:
self._limit = int(request.GET.get("limit", 200))
except ValueError, TypeError:
self._limit = None
self._limit = _query_limit(request)
self._hide_paid = _query_bool(request, "hide_paid")
return super().__call__(request, *args, **kwargs)
def items(self) -> list[DropCampaign]:
"""Return the latest drop campaigns ordered by most recent start date (default 200, or limited by ?limit query param)."""
"""Return latest active drop campaigns."""
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:
@ -1015,8 +1061,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))
parts.extend(generate_channels_html(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_details_link_html(item))
return SafeText("".join(str(p) for p in parts))
@ -1110,6 +1156,7 @@ class GameCampaignFeed(TTVDropsBaseFeed):
item_guid_is_permalink = True
_limit: int | None = None
_hide_paid: bool = False
def __call__(
self,
@ -1117,21 +1164,18 @@ class GameCampaignFeed(TTVDropsBaseFeed):
*args: str | int,
**kwargs: str | int,
) -> HttpResponse:
"""Override to capture limit parameter from request.
"""Override to capture supported feed query parameters from request.
Args:
request (HttpRequest): The incoming HTTP request, potentially containing a 'limit' query parameter
request (HttpRequest): The incoming HTTP request, potentially containing 'limit' and 'hide_paid' query parameters.
*args: Additional positional arguments.
**kwargs: Additional keyword arguments.
Returns:
HttpResponse: The HTTP response generated by the parent Feed class after processing the request.
"""
if request.GET.get("limit"):
try:
self._limit = int(request.GET.get("limit", 200))
except ValueError, TypeError:
self._limit = None
self._limit = _query_limit(request)
self._hide_paid = _query_bool(request, "hide_paid")
return super().__call__(request, *args, **kwargs)
def get_object(self, request: HttpRequest, twitch_id: str) -> Game: # noqa: ARG002
@ -1159,13 +1203,15 @@ class GameCampaignFeed(TTVDropsBaseFeed):
return f"Latest drop campaigns for {obj.display_name}"
def items(self, obj: Game) -> list[DropCampaign]:
"""Return the latest drop campaigns for this game, ordered by most recent start date (default 200, or limited by ?limit query param)."""
"""Return latest active drop campaigns for this game."""
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:
@ -1180,8 +1226,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))
parts.extend(generate_channels_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))
return SafeText("".join(str(p) for p in parts))
@ -1554,8 +1600,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))
parts.extend(generate_channels_html(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_details_link_html(item))
return SafeText("".join(str(p) for p in parts))
@ -1575,8 +1621,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))
parts.extend(generate_channels_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))
return SafeText("".join(str(p) for p in parts))

View file

@ -91,6 +91,10 @@ 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",
),
@ -340,6 +344,8 @@ class Game(auto_prefetch.Model):
"box_art_file",
"box_art_width",
"box_art_height",
"added_at",
"updated_at",
)
.prefetch_related(
Prefetch(
@ -803,6 +809,8 @@ class DropCampaign(auto_prefetch.Model):
"start_at",
"end_at",
"game",
"game__twitch_id",
"game__name",
"game__display_name",
)
.prefetch_related(
@ -854,12 +862,19 @@ 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(

438
twitch/tests/test_api.py Normal file
View file

@ -0,0 +1,438 @@
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]</a>' in content

View file

@ -104,6 +104,90 @@ 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")
@ -494,6 +578,43 @@ 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()

View file

@ -2504,6 +2504,227 @@ 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."""

View file

@ -3,6 +3,7 @@ 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
@ -12,6 +13,8 @@ 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/