Compare commits
No commits in common. "8229b0fe80f2b852815874d69dfb15b03f41f73c" and "859e01faf4e19d90e69e56f2ca6518fd65d8e060" have entirely different histories.
8229b0fe80
...
859e01faf4
21 changed files with 31 additions and 1728 deletions
|
|
@ -27,7 +27,7 @@ repos:
|
||||||
args: [--target-version, "6.0"]
|
args: [--target-version, "6.0"]
|
||||||
|
|
||||||
- repo: https://github.com/astral-sh/ruff-pre-commit
|
- repo: https://github.com/astral-sh/ruff-pre-commit
|
||||||
rev: v0.15.12
|
rev: v0.15.9
|
||||||
hooks:
|
hooks:
|
||||||
- id: ruff-check
|
- id: ruff-check
|
||||||
args: ["--fix", "--exit-non-zero-on-fix"]
|
args: ["--fix", "--exit-non-zero-on-fix"]
|
||||||
|
|
|
||||||
|
|
@ -12,9 +12,7 @@ dependencies = [
|
||||||
"django-celery-beat",
|
"django-celery-beat",
|
||||||
"django-celery-results",
|
"django-celery-results",
|
||||||
"django-debug-toolbar",
|
"django-debug-toolbar",
|
||||||
"django-ninja",
|
|
||||||
"django-silk",
|
"django-silk",
|
||||||
"django-zeal",
|
|
||||||
"django",
|
"django",
|
||||||
"flower",
|
"flower",
|
||||||
"gunicorn",
|
"gunicorn",
|
||||||
|
|
@ -33,6 +31,7 @@ dependencies = [
|
||||||
"setproctitle",
|
"setproctitle",
|
||||||
"sitemap-parser",
|
"sitemap-parser",
|
||||||
"tqdm",
|
"tqdm",
|
||||||
|
"django-zeal>=2.1.0",
|
||||||
]
|
]
|
||||||
|
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -220,7 +220,6 @@
|
||||||
<a href="{% url 'twitch:channel_list' %}">Channels</a> |
|
<a href="{% url 'twitch:channel_list' %}">Channels</a> |
|
||||||
<a href="{% url 'twitch:badge_list' %}">Badges</a> |
|
<a href="{% url 'twitch:badge_list' %}">Badges</a> |
|
||||||
<a href="{% url 'twitch:emote_gallery' %}">Emotes</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>
|
<a href="https://www.twitch.tv/drops/inventory">Inventory</a>
|
||||||
<br />
|
<br />
|
||||||
<strong>Kick</strong>
|
<strong>Kick</strong>
|
||||||
|
|
|
||||||
|
|
@ -48,8 +48,6 @@
|
||||||
title="Atom feed for Twitch campaigns">[atom]</a>
|
title="Atom feed for Twitch campaigns">[atom]</a>
|
||||||
<a href="{% url 'core:campaign_feed_discord' %}"
|
<a href="{% url 'core:campaign_feed_discord' %}"
|
||||||
title="Discord feed for Twitch campaigns">[discord]</a>
|
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>
|
<a href="{% url 'core:docs_rss' %}" title="RSS feed documentation">[explain]</a>
|
||||||
</div>
|
</div>
|
||||||
</header>
|
</header>
|
||||||
|
|
|
||||||
|
|
@ -16,15 +16,6 @@
|
||||||
that include Discord relative timestamps (e.g., <code><t:1773450272:R></code>) for dates,
|
that include Discord relative timestamps (e.g., <code><t:1773450272:R></code>) for dates,
|
||||||
making them ideal for Discord bots and integrations. Future enhancements may include Discord-specific formatting or content.
|
making them ideal for Discord bots and integrations. Future enhancements may include Discord-specific formatting or content.
|
||||||
</p>
|
</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>
|
<section>
|
||||||
<h2>Global RSS Feeds</h2>
|
<h2>Global RSS Feeds</h2>
|
||||||
<table>
|
<table>
|
||||||
|
|
|
||||||
|
|
@ -103,8 +103,6 @@
|
||||||
title="Atom feed for {{ campaign.game.display_name }} campaigns">[atom]</a>
|
title="Atom feed for {{ campaign.game.display_name }} campaigns">[atom]</a>
|
||||||
<a href="{% url 'core:game_campaign_feed_discord' campaign.game.twitch_id %}"
|
<a href="{% url 'core:game_campaign_feed_discord' campaign.game.twitch_id %}"
|
||||||
title="Discord feed for {{ campaign.game.display_name }} campaigns">[discord]</a>
|
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>
|
<a href="{% url 'core:docs_rss' %}" title="RSS feed documentation">[explain]</a>
|
||||||
{% endif %}
|
{% endif %}
|
||||||
</div>
|
</div>
|
||||||
|
|
|
||||||
|
|
@ -33,8 +33,6 @@
|
||||||
title="Atom feed for all campaigns">[atom]</a>
|
title="Atom feed for all campaigns">[atom]</a>
|
||||||
<a href="{% url 'core:campaign_feed_discord' %}"
|
<a href="{% url 'core:campaign_feed_discord' %}"
|
||||||
title="Discord feed for all campaigns">[discord]</a>
|
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 'core:docs_rss' %}" title="RSS feed documentation">[explain]</a>
|
||||||
<a href="{% url 'twitch:export_campaigns_csv' %}{% if request.GET %}?{{ request.GET.urlencode }}{% endif %}"
|
<a href="{% url 'twitch:export_campaigns_csv' %}{% if request.GET %}?{{ request.GET.urlencode }}{% endif %}"
|
||||||
title="Export campaigns as CSV">[csv]</a>
|
title="Export campaigns as CSV">[csv]</a>
|
||||||
|
|
|
||||||
|
|
@ -36,8 +36,6 @@
|
||||||
title="Atom feed for campaigns">[atom]</a>
|
title="Atom feed for campaigns">[atom]</a>
|
||||||
<a href="{% url 'core:campaign_feed_discord' %}"
|
<a href="{% url 'core:campaign_feed_discord' %}"
|
||||||
title="Discord feed for campaigns">[discord]</a>
|
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>
|
<a href="{% url 'core:docs_rss' %}" title="RSS feed documentation">[explain]</a>
|
||||||
</div>
|
</div>
|
||||||
<hr />
|
<hr />
|
||||||
|
|
|
||||||
|
|
@ -65,8 +65,6 @@
|
||||||
title="Atom feed for {{ game.display_name }} campaigns">[atom]</a>
|
title="Atom feed for {{ game.display_name }} campaigns">[atom]</a>
|
||||||
<a href="{% url 'core:game_campaign_feed_discord' game.twitch_id %}"
|
<a href="{% url 'core:game_campaign_feed_discord' game.twitch_id %}"
|
||||||
title="Discord feed for {{ game.display_name }} campaigns">[discord]</a>
|
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>
|
<a href="{% url 'core:docs_rss' %}" title="RSS feed documentation">[explain]</a>
|
||||||
</div>
|
</div>
|
||||||
</div>
|
</div>
|
||||||
|
|
|
||||||
|
|
@ -32,8 +32,6 @@
|
||||||
title="Atom feed for all games">[atom]</a>
|
title="Atom feed for all games">[atom]</a>
|
||||||
<a href="{% url 'core:game_feed_discord' %}"
|
<a href="{% url 'core:game_feed_discord' %}"
|
||||||
title="Discord feed for all games">[discord]</a>
|
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 'core:docs_rss' %}" title="RSS feed documentation">[explain]</a>
|
||||||
<a href="{% url 'twitch:export_games_csv' %}"
|
<a href="{% url 'twitch:export_games_csv' %}"
|
||||||
title="Export all games as CSV">[csv]</a>
|
title="Export all games as CSV">[csv]</a>
|
||||||
|
|
|
||||||
|
|
@ -30,8 +30,6 @@
|
||||||
title="Atom feed for all games">[atom]</a>
|
title="Atom feed for all games">[atom]</a>
|
||||||
<a href="{% url 'core:game_feed_discord' %}"
|
<a href="{% url 'core:game_feed_discord' %}"
|
||||||
title="Discord feed for all games">[discord]</a>
|
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 'core:docs_rss' %}" title="RSS feed documentation">[explain]</a>
|
||||||
<a href="{% url 'twitch:export_games_csv' %}"
|
<a href="{% url 'twitch:export_games_csv' %}"
|
||||||
title="Export all games as CSV">[csv]</a>
|
title="Export all games as CSV">[csv]</a>
|
||||||
|
|
|
||||||
|
|
@ -15,8 +15,6 @@
|
||||||
title="Atom feed for all organizations">[atom]</a>
|
title="Atom feed for all organizations">[atom]</a>
|
||||||
<a href="{% url 'core:organization_feed_discord' %}"
|
<a href="{% url 'core:organization_feed_discord' %}"
|
||||||
title="Discord feed for all organizations">[discord]</a>
|
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 'core:docs_rss' %}" title="RSS feed documentation">[explain]</a>
|
||||||
<a href="{% url 'twitch:export_organizations_csv' %}"
|
<a href="{% url 'twitch:export_organizations_csv' %}"
|
||||||
title="Export all organizations as CSV">[csv]</a>
|
title="Export all organizations as CSV">[csv]</a>
|
||||||
|
|
|
||||||
|
|
@ -110,8 +110,6 @@
|
||||||
title="Atom feed for all reward campaigns">[atom]</a>
|
title="Atom feed for all reward campaigns">[atom]</a>
|
||||||
<a href="{% url 'core:reward_campaign_feed_discord' %}"
|
<a href="{% url 'core:reward_campaign_feed_discord' %}"
|
||||||
title="Discord feed for all reward campaigns">[discord]</a>
|
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>
|
<a href="{% url 'core:docs_rss' %}" title="RSS feed documentation">[explain]</a>
|
||||||
{% if reward_campaign.external_url %}
|
{% if reward_campaign.external_url %}
|
||||||
<a href="{{ reward_campaign.external_url }}"
|
<a href="{{ reward_campaign.external_url }}"
|
||||||
|
|
|
||||||
|
|
@ -36,8 +36,6 @@
|
||||||
title="Atom feed for all reward campaigns">[atom]</a>
|
title="Atom feed for all reward campaigns">[atom]</a>
|
||||||
<a href="{% url 'core:reward_campaign_feed_discord' %}"
|
<a href="{% url 'core:reward_campaign_feed_discord' %}"
|
||||||
title="Discord feed for all reward campaigns">[discord]</a>
|
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>
|
<a href="{% url 'core:docs_rss' %}" title="RSS feed documentation">[explain]</a>
|
||||||
</div>
|
</div>
|
||||||
{% if reward_campaigns %}
|
{% if reward_campaigns %}
|
||||||
|
|
|
||||||
822
twitch/api.py
822
twitch/api.py
|
|
@ -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)
|
|
||||||
104
twitch/feeds.py
104
twitch/feeds.py
|
|
@ -9,8 +9,6 @@ from django.conf import settings
|
||||||
from django.contrib.humanize.templatetags.humanize import naturaltime
|
from django.contrib.humanize.templatetags.humanize import naturaltime
|
||||||
from django.contrib.sites.shortcuts import get_current_site
|
from django.contrib.sites.shortcuts import get_current_site
|
||||||
from django.contrib.syndication.views import Feed
|
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 import Prefetch
|
||||||
from django.db.models.query import QuerySet
|
from django.db.models.query import QuerySet
|
||||||
from django.http.request import HttpRequest
|
from django.http.request import HttpRequest
|
||||||
|
|
@ -49,23 +47,6 @@ if TYPE_CHECKING:
|
||||||
|
|
||||||
logger: logging.Logger = logging.getLogger("ttvdrops")
|
logger: logging.Logger = logging.getLogger("ttvdrops")
|
||||||
RSS_STYLESHEETS: list[str] = [static("rss_styles.xslt")]
|
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:
|
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)
|
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(
|
def _active_reward_campaigns(
|
||||||
queryset: QuerySet[RewardCampaign],
|
queryset: QuerySet[RewardCampaign],
|
||||||
) -> QuerySet[RewardCampaign]:
|
) -> QuerySet[RewardCampaign]:
|
||||||
|
|
@ -484,16 +450,11 @@ def generate_discord_date_html(item: Model) -> list[SafeText]:
|
||||||
return parts
|
return parts
|
||||||
|
|
||||||
|
|
||||||
def generate_drops_summary_html(
|
def generate_drops_summary_html(item: DropCampaign) -> list[SafeString]:
|
||||||
item: DropCampaign,
|
|
||||||
*,
|
|
||||||
hide_paid: bool = False,
|
|
||||||
) -> list[SafeString]:
|
|
||||||
"""Generate HTML summary for drops and append to parts list.
|
"""Generate HTML summary for drops and append to parts list.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
item (DropCampaign): The drop campaign item containing the drops to summarize.
|
item (DropCampaign): The drop campaign item containing the drops to summarize.
|
||||||
hide_paid: Exclude drops that require paid subscriptions from the summary.
|
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
list[SafeString]: A list of SafeText elements summarizing the drops, or empty if no drops.
|
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)
|
drops: QuerySet[TimeBasedDrop] | None = getattr(item, "time_based_drops", None)
|
||||||
if drops:
|
if drops:
|
||||||
drops_data = _build_drops_data(drops.all(), hide_paid=hide_paid)
|
drops_data = _build_drops_data(drops.all())
|
||||||
|
|
||||||
if drops_data:
|
if drops_data:
|
||||||
parts.append(
|
parts.append(
|
||||||
|
|
@ -519,14 +480,13 @@ def generate_drops_summary_html(
|
||||||
return parts
|
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.
|
"""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.
|
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:
|
Args:
|
||||||
item (Model): The campaign item which may have an 'is_subscription_only' attribute.
|
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:
|
Returns:
|
||||||
list[SafeText]: A list containing the HTML for the channels section, or empty if subscription-only or no channels.
|
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:
|
if not channels:
|
||||||
return parts
|
return parts
|
||||||
|
|
||||||
if not hide_paid and getattr(item, "is_subscription_only", False):
|
if getattr(item, "is_subscription_only", False):
|
||||||
return parts
|
return parts
|
||||||
|
|
||||||
game: Game | None = getattr(item, "game", None)
|
game: Game | None = getattr(item, "game", None)
|
||||||
|
|
@ -639,11 +599,7 @@ def create_channel_list_html(
|
||||||
parts.append(format_html("<p>Channels with this drop:</p>{}", html))
|
parts.append(format_html("<p>Channels with this drop:</p>{}", html))
|
||||||
|
|
||||||
|
|
||||||
def _build_drops_data(
|
def _build_drops_data(drops_qs: QuerySet[TimeBasedDrop]) -> list[dict]:
|
||||||
drops_qs: QuerySet[TimeBasedDrop],
|
|
||||||
*,
|
|
||||||
hide_paid: bool = False,
|
|
||||||
) -> list[dict]:
|
|
||||||
"""Build a simplified data structure for rendering drops in a template.
|
"""Build a simplified data structure for rendering drops in a template.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
|
|
@ -655,8 +611,6 @@ def _build_drops_data(
|
||||||
requirements: str = ""
|
requirements: str = ""
|
||||||
required_minutes: int | None = getattr(drop, "required_minutes_watched", None)
|
required_minutes: int | None = getattr(drop, "required_minutes_watched", None)
|
||||||
required_subs: int = getattr(drop, "required_subs", 0) or 0
|
required_subs: int = getattr(drop, "required_subs", 0) or 0
|
||||||
if hide_paid and required_subs > 0:
|
|
||||||
continue
|
|
||||||
if required_minutes:
|
if required_minutes:
|
||||||
requirements = f"{required_minutes} minutes watched"
|
requirements = f"{required_minutes} minutes watched"
|
||||||
if required_subs > 0:
|
if required_subs > 0:
|
||||||
|
|
@ -1017,7 +971,6 @@ class DropCampaignFeed(TTVDropsBaseFeed):
|
||||||
item_guid_is_permalink = True
|
item_guid_is_permalink = True
|
||||||
|
|
||||||
_limit: int | None = None
|
_limit: int | None = None
|
||||||
_hide_paid: bool = False
|
|
||||||
|
|
||||||
def __call__(
|
def __call__(
|
||||||
self,
|
self,
|
||||||
|
|
@ -1025,28 +978,29 @@ class DropCampaignFeed(TTVDropsBaseFeed):
|
||||||
*args: str | int,
|
*args: str | int,
|
||||||
**kwargs: str | int,
|
**kwargs: str | int,
|
||||||
) -> HttpResponse:
|
) -> HttpResponse:
|
||||||
"""Override to capture supported feed query parameters from request.
|
"""Override to capture limit parameter from request.
|
||||||
|
|
||||||
Args:
|
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.
|
*args: Additional positional arguments.
|
||||||
**kwargs: Additional keyword arguments.
|
**kwargs: Additional keyword arguments.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
HttpResponse: The HTTP response generated by the parent Feed class after processing the request.
|
HttpResponse: The HTTP response generated by the parent Feed class after processing the request.
|
||||||
"""
|
"""
|
||||||
self._limit = _query_limit(request)
|
if request.GET.get("limit"):
|
||||||
self._hide_paid = _query_bool(request, "hide_paid")
|
try:
|
||||||
|
self._limit = int(request.GET.get("limit", 200))
|
||||||
|
except ValueError, TypeError:
|
||||||
|
self._limit = None
|
||||||
return super().__call__(request, *args, **kwargs)
|
return super().__call__(request, *args, **kwargs)
|
||||||
|
|
||||||
def items(self) -> list[DropCampaign]:
|
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
|
limit: int = self._limit if self._limit is not None else 200
|
||||||
queryset: QuerySet[DropCampaign] = _active_drop_campaigns(
|
queryset: QuerySet[DropCampaign] = _active_drop_campaigns(
|
||||||
DropCampaign.objects.order_by("-start_at"),
|
DropCampaign.objects.order_by("-start_at"),
|
||||||
)
|
)
|
||||||
if self._hide_paid:
|
|
||||||
queryset = _campaigns_with_free_drops(queryset)
|
|
||||||
return list(_with_campaign_related(queryset)[:limit])
|
return list(_with_campaign_related(queryset)[:limit])
|
||||||
|
|
||||||
def item_title(self, item: DropCampaign) -> SafeText:
|
def item_title(self, item: DropCampaign) -> SafeText:
|
||||||
|
|
@ -1061,8 +1015,8 @@ class DropCampaignFeed(TTVDropsBaseFeed):
|
||||||
parts.extend(generate_item_image(item))
|
parts.extend(generate_item_image(item))
|
||||||
parts.extend(generate_description_html(item=item))
|
parts.extend(generate_description_html(item=item))
|
||||||
parts.extend(generate_date_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_drops_summary_html(item=item))
|
||||||
parts.extend(generate_channels_html(item, hide_paid=self._hide_paid))
|
parts.extend(generate_channels_html(item))
|
||||||
parts.extend(generate_details_link_html(item))
|
parts.extend(generate_details_link_html(item))
|
||||||
|
|
||||||
return SafeText("".join(str(p) for p in parts))
|
return SafeText("".join(str(p) for p in parts))
|
||||||
|
|
@ -1156,7 +1110,6 @@ class GameCampaignFeed(TTVDropsBaseFeed):
|
||||||
|
|
||||||
item_guid_is_permalink = True
|
item_guid_is_permalink = True
|
||||||
_limit: int | None = None
|
_limit: int | None = None
|
||||||
_hide_paid: bool = False
|
|
||||||
|
|
||||||
def __call__(
|
def __call__(
|
||||||
self,
|
self,
|
||||||
|
|
@ -1164,18 +1117,21 @@ class GameCampaignFeed(TTVDropsBaseFeed):
|
||||||
*args: str | int,
|
*args: str | int,
|
||||||
**kwargs: str | int,
|
**kwargs: str | int,
|
||||||
) -> HttpResponse:
|
) -> HttpResponse:
|
||||||
"""Override to capture supported feed query parameters from request.
|
"""Override to capture limit parameter from request.
|
||||||
|
|
||||||
Args:
|
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.
|
*args: Additional positional arguments.
|
||||||
**kwargs: Additional keyword arguments.
|
**kwargs: Additional keyword arguments.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
HttpResponse: The HTTP response generated by the parent Feed class after processing the request.
|
HttpResponse: The HTTP response generated by the parent Feed class after processing the request.
|
||||||
"""
|
"""
|
||||||
self._limit = _query_limit(request)
|
if request.GET.get("limit"):
|
||||||
self._hide_paid = _query_bool(request, "hide_paid")
|
try:
|
||||||
|
self._limit = int(request.GET.get("limit", 200))
|
||||||
|
except ValueError, TypeError:
|
||||||
|
self._limit = None
|
||||||
return super().__call__(request, *args, **kwargs)
|
return super().__call__(request, *args, **kwargs)
|
||||||
|
|
||||||
def get_object(self, request: HttpRequest, twitch_id: str) -> Game: # noqa: ARG002
|
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}"
|
return f"Latest drop campaigns for {obj.display_name}"
|
||||||
|
|
||||||
def items(self, obj: Game) -> list[DropCampaign]:
|
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
|
limit: int = self._limit if self._limit is not None else 200
|
||||||
queryset: QuerySet[DropCampaign] = _active_drop_campaigns(
|
queryset: QuerySet[DropCampaign] = _active_drop_campaigns(
|
||||||
DropCampaign.objects.filter(
|
DropCampaign.objects.filter(
|
||||||
game=obj,
|
game=obj,
|
||||||
).order_by("-start_at"),
|
).order_by("-start_at"),
|
||||||
)
|
)
|
||||||
if self._hide_paid:
|
|
||||||
queryset = _campaigns_with_free_drops(queryset)
|
|
||||||
return list(_with_campaign_related(queryset)[:limit])
|
return list(_with_campaign_related(queryset)[:limit])
|
||||||
|
|
||||||
def item_title(self, item: DropCampaign) -> SafeText:
|
def item_title(self, item: DropCampaign) -> SafeText:
|
||||||
|
|
@ -1226,8 +1180,8 @@ class GameCampaignFeed(TTVDropsBaseFeed):
|
||||||
parts.extend(generate_item_image_tag(item))
|
parts.extend(generate_item_image_tag(item))
|
||||||
parts.extend(generate_details_link(item))
|
parts.extend(generate_details_link(item))
|
||||||
parts.extend(generate_date_html(item))
|
parts.extend(generate_date_html(item))
|
||||||
parts.extend(generate_drops_summary_html(item, hide_paid=self._hide_paid))
|
parts.extend(generate_drops_summary_html(item))
|
||||||
parts.extend(generate_channels_html(item, hide_paid=self._hide_paid))
|
parts.extend(generate_channels_html(item))
|
||||||
|
|
||||||
return SafeText("".join(str(p) for p in parts))
|
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_item_image(item))
|
||||||
parts.extend(generate_description_html(item=item))
|
parts.extend(generate_description_html(item=item))
|
||||||
parts.extend(generate_discord_date_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_drops_summary_html(item=item))
|
||||||
parts.extend(generate_channels_html(item, hide_paid=self._hide_paid))
|
parts.extend(generate_channels_html(item))
|
||||||
parts.extend(generate_details_link_html(item))
|
parts.extend(generate_details_link_html(item))
|
||||||
|
|
||||||
return SafeText("".join(str(p) for p in parts))
|
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_item_image_tag(item))
|
||||||
parts.extend(generate_details_link(item))
|
parts.extend(generate_details_link(item))
|
||||||
parts.extend(generate_discord_date_html(item))
|
parts.extend(generate_discord_date_html(item))
|
||||||
parts.extend(generate_drops_summary_html(item, hide_paid=self._hide_paid))
|
parts.extend(generate_drops_summary_html(item))
|
||||||
parts.extend(generate_channels_html(item, hide_paid=self._hide_paid))
|
parts.extend(generate_channels_html(item))
|
||||||
|
|
||||||
return SafeText("".join(str(p) for p in parts))
|
return SafeText("".join(str(p) for p in parts))
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -91,10 +91,6 @@ class Organization(auto_prefetch.Model):
|
||||||
"name",
|
"name",
|
||||||
"display_name",
|
"display_name",
|
||||||
"slug",
|
"slug",
|
||||||
"box_art",
|
|
||||||
"box_art_file",
|
|
||||||
"box_art_width",
|
|
||||||
"box_art_height",
|
|
||||||
).order_by("display_name"),
|
).order_by("display_name"),
|
||||||
to_attr="games_for_detail",
|
to_attr="games_for_detail",
|
||||||
),
|
),
|
||||||
|
|
@ -344,8 +340,6 @@ class Game(auto_prefetch.Model):
|
||||||
"box_art_file",
|
"box_art_file",
|
||||||
"box_art_width",
|
"box_art_width",
|
||||||
"box_art_height",
|
"box_art_height",
|
||||||
"added_at",
|
|
||||||
"updated_at",
|
|
||||||
)
|
)
|
||||||
.prefetch_related(
|
.prefetch_related(
|
||||||
Prefetch(
|
Prefetch(
|
||||||
|
|
@ -809,8 +803,6 @@ class DropCampaign(auto_prefetch.Model):
|
||||||
"start_at",
|
"start_at",
|
||||||
"end_at",
|
"end_at",
|
||||||
"game",
|
"game",
|
||||||
"game__twitch_id",
|
|
||||||
"game__name",
|
|
||||||
"game__display_name",
|
"game__display_name",
|
||||||
)
|
)
|
||||||
.prefetch_related(
|
.prefetch_related(
|
||||||
|
|
@ -862,19 +854,12 @@ class DropCampaign(auto_prefetch.Model):
|
||||||
"image_height",
|
"image_height",
|
||||||
"start_at",
|
"start_at",
|
||||||
"end_at",
|
"end_at",
|
||||||
"allow_is_enabled",
|
|
||||||
"operation_names",
|
|
||||||
"added_at",
|
"added_at",
|
||||||
"updated_at",
|
"updated_at",
|
||||||
"is_fully_imported",
|
|
||||||
"game__twitch_id",
|
"game__twitch_id",
|
||||||
"game__name",
|
"game__name",
|
||||||
"game__display_name",
|
"game__display_name",
|
||||||
"game__slug",
|
"game__slug",
|
||||||
"game__box_art",
|
|
||||||
"game__box_art_file",
|
|
||||||
"game__box_art_width",
|
|
||||||
"game__box_art_height",
|
|
||||||
)
|
)
|
||||||
.prefetch_related(
|
.prefetch_related(
|
||||||
Prefetch(
|
Prefetch(
|
||||||
|
|
|
||||||
|
|
@ -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]</a>' in content
|
|
||||||
|
|
@ -104,90 +104,6 @@ class RSSFeedTestCase(TestCase):
|
||||||
game=self.game,
|
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:
|
def test_organization_feed(self) -> None:
|
||||||
"""Test organization feed returns 200."""
|
"""Test organization feed returns 200."""
|
||||||
url: str = reverse("core:organization_feed")
|
url: str = reverse("core:organization_feed")
|
||||||
|
|
@ -578,43 +494,6 @@ class RSSFeedTestCase(TestCase):
|
||||||
assert "Past Campaign" not in content
|
assert "Past Campaign" not in content
|
||||||
assert "Upcoming 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:
|
def test_campaign_feed_enclosure_helpers(self) -> None:
|
||||||
"""Helper methods for campaigns should respect new fields."""
|
"""Helper methods for campaigns should respect new fields."""
|
||||||
feed = DropCampaignFeed()
|
feed = DropCampaignFeed()
|
||||||
|
|
|
||||||
|
|
@ -2504,227 +2504,6 @@ class TestChannelListView:
|
||||||
assert "g-rss" in response.content.decode("utf-8")
|
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
|
@pytest.mark.django_db
|
||||||
class TestSEOHelperFunctions:
|
class TestSEOHelperFunctions:
|
||||||
"""Tests for SEO helper functions."""
|
"""Tests for SEO helper functions."""
|
||||||
|
|
|
||||||
|
|
@ -3,7 +3,6 @@ from typing import TYPE_CHECKING
|
||||||
from django.urls import path
|
from django.urls import path
|
||||||
|
|
||||||
from twitch import views
|
from twitch import views
|
||||||
from twitch.api import api as twitch_api_v1
|
|
||||||
|
|
||||||
if TYPE_CHECKING:
|
if TYPE_CHECKING:
|
||||||
from django.urls.resolvers import URLPattern
|
from django.urls.resolvers import URLPattern
|
||||||
|
|
@ -13,8 +12,6 @@ app_name = "twitch"
|
||||||
|
|
||||||
|
|
||||||
urlpatterns: list[URLPattern | URLResolver] = [
|
urlpatterns: list[URLPattern | URLResolver] = [
|
||||||
# /twitch/api/v1/
|
|
||||||
path("api/v1/", twitch_api_v1.urls),
|
|
||||||
# /twitch/
|
# /twitch/
|
||||||
path("", views.dashboard, name="dashboard"),
|
path("", views.dashboard, name="dashboard"),
|
||||||
# /twitch/badges/
|
# /twitch/badges/
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue