822 lines
24 KiB
Python
822 lines
24 KiB
Python
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)
|