with up to max_links channel links, count of more, or fallback link.
""" # noqa: E501
max_links = 5
channels_all: list[Channel] = list(channels.all())
total: int = len(channels_all)
if channels_all:
items: list[SafeString] = [
format_html(
"- "
'{}'
"
",
ch.name,
ch.display_name,
ch.display_name,
)
for ch in channels_all[:max_links]
]
if total > max_links:
items.append(format_html("- ... and {} more
", total - max_links))
return format_html(
"",
format_html_join("", "{}", [(item,) for item in items]),
)
if not game:
logger.warning("No game associated with drop campaign for channel fallback link")
return format_html("{}", "- Drop has no game and no channels connected to the drop.
")
if not game.twitch_directory_url:
logger.warning("Game %s has no Twitch directory URL for channel fallback link", game)
if getattr(game, "details_url", "") == "https://help.twitch.tv/s/article/twitch-chat-badges-guide ":
# TODO(TheLovinator): Improve detection of global emotes # noqa: TD003
return format_html("{}", "")
return format_html("{}", "- Failed to get Twitch category URL :(
")
# If no channel is associated, the drop is category-wide; link to the game's Twitch directory
display_name: str = getattr(game, "display_name", "this game")
return format_html(
"",
game.twitch_directory_url,
display_name,
display_name,
)
def _get_channel_name_from_drops(drops: QuerySet[TimeBasedDrop]) -> str | None:
for d in drops:
campaign: DropCampaign | None = getattr(d, "campaign", None)
if campaign:
allow_channels: QuerySet[Channel] | None = getattr(campaign, "allow_channels", None)
if allow_channels:
channels: QuerySet[Channel, Channel] = allow_channels.all()
if channels:
return channels[0].name
return None
def get_channel_from_benefit(benefit: Model) -> str | None:
"""Get the Twitch channel name associated with a drop benefit.
Args:
benefit (Model): The drop benefit model instance.
Returns:
str | None: The Twitch channel name if found, else None.
"""
drop_obj: QuerySet[TimeBasedDrop] | None = getattr(benefit, "drops", None)
if drop_obj and hasattr(drop_obj, "all"):
try:
return _get_channel_name_from_drops(drop_obj.all())
except AttributeError:
logger.exception("Exception occurred while resolving channel name for benefit")
return None
def _resolve_channel_name(drop: dict) -> str | None:
"""Try to resolve the Twitch channel name for a drop dict's benefits or fallback keys.
Args:
drop (dict): The drop data dictionary.
Returns:
str | None: The Twitch channel name if found, else None.
"""
benefits: list[Model] = drop.get("benefits", [])
benefit0: Model | None = benefits[0] if benefits else None
if benefit0 and hasattr(benefit0, "drops"):
channel_name: str | None = get_channel_from_benefit(benefit0)
if channel_name:
return channel_name
return None
def _construct_drops_summary(drops_data: list[dict]) -> SafeText:
"""Construct a safe HTML summary of drops and their benefits.
If the requirements indicate a subscription is required, link the benefit names to the Twitch channel.
Args:
drops_data (list[dict]): List of drop data dicts.
Returns:
SafeText: A single safe HTML line summarizing all drops, or empty SafeText if none.
"""
if not drops_data:
return SafeText("")
badge_titles: set[str] = set()
for drop in drops_data:
for b in drop.get("benefits", []):
if getattr(b, "distribution_type", "") == "BADGE" and getattr(b, "name", ""):
badge_titles.add(b.name)
badge_descriptions_by_title: dict[str, str] = {}
if badge_titles:
badge_descriptions_by_title = dict(
ChatBadge.objects.filter(title__in=badge_titles).values_list("title", "description"),
)
def sort_key(drop: dict) -> tuple[bool, int]:
req: str = drop.get("requirements", "")
m: re.Match[str] | None = re.search(r"(\d+) minutes watched", req)
minutes: int | None = int(m.group(1)) if m else None
is_sub: bool = "sub required" in req or "subs required" in req
return (is_sub, minutes if minutes is not None else 99999)
sorted_drops: list[dict] = sorted(drops_data, key=sort_key)
items: list[SafeText] = []
for drop in sorted_drops:
requirements: str = drop.get("requirements", "")
benefits: list[DropBenefit] = drop.get("benefits", [])
channel_name: str | None = _resolve_channel_name(drop)
is_sub_required: bool = "sub required" in requirements or "subs required" in requirements
benefit_names: list[tuple[str]] = []
for b in benefits:
benefit_name: str = getattr(b, "name", str(b))
badge_desc: str | None = badge_descriptions_by_title.get(benefit_name)
if is_sub_required and channel_name:
linked_name: SafeString = format_html(
'{}',
channel_name,
benefit_name,
)
if badge_desc:
benefit_names.append((format_html("{} ({})", linked_name, badge_desc),))
else:
benefit_names.append((linked_name,))
elif badge_desc:
benefit_names.append((format_html("{} ({})", benefit_name, badge_desc),))
else:
benefit_names.append((benefit_name,))
benefits_str: SafeString = format_html_join(", ", "{}", benefit_names) if benefit_names else SafeText("")
if requirements:
items.append(format_html("- {}: {}
", requirements, benefits_str))
else:
items.append(format_html("- {}
", benefits_str))
return format_html("", format_html_join("", "{}", [(item,) for item in items]))
# MARK: /rss/organizations/
class OrganizationRSSFeed(Feed):
"""RSS feed for latest organizations."""
# Spec: https://cyber.harvard.edu/rss/rss.html
feed_type = feedgenerator.Rss201rev2Feed
title: str = "TTVDrops Organizations"
link: str = "/organizations/"
description: str = "Latest organizations on TTVDrops"
feed_copyright: str = "Information wants to be free."
_limit: int | None = None
def __call__(self, request: HttpRequest, *args: object, **kwargs: object) -> HttpResponse:
"""Override to capture limit parameter from request.
Args:
request (HttpRequest): The incoming HTTP request, potentially containing a 'limit' query parameter.
*args: Additional positional arguments.
**kwargs: Additional keyword arguments.
Returns:
HttpResponse: The HTTP response generated by the parent Feed class after processing the request.
"""
if request.GET.get("limit"):
try:
self._limit = int(request.GET.get("limit", 200))
except ValueError, TypeError:
self._limit = None
return super().__call__(request, *args, **kwargs)
def items(self) -> QuerySet[Organization]:
"""Return the latest organizations (default 200, or limited by ?limit query param)."""
limit: int = self._limit if self._limit is not None else 200
return Organization.objects.order_by("-added_at")[:limit]
def item_title(self, item: Organization) -> SafeText:
"""Return the organization name as the item title."""
return SafeText(getattr(item, "name", str(item)))
def item_description(self, item: Organization) -> SafeText:
"""Return a description of the organization."""
return SafeText(item.feed_description)
def item_link(self, item: Organization) -> str:
"""Return the link to the organization detail."""
return reverse("twitch:organization_detail", args=[item.twitch_id])
def item_pubdate(self, item: Organization) -> datetime.datetime:
"""Returns the publication date to the feed item.
Fallback to added_at or now if missing.
"""
added_at: datetime.datetime | None = getattr(item, "added_at", None)
if added_at:
return added_at
return timezone.now()
def item_updateddate(self, item: Organization) -> datetime.datetime:
"""Returns the organization's last update time."""
updated_at: datetime.datetime | None = getattr(item, "updated_at", None)
if updated_at:
return updated_at
return timezone.now()
def item_author_name(self, item: Organization) -> str:
"""Return the author name for the organization."""
return getattr(item, "name", "Twitch")
# MARK: /rss/games/
class GameFeed(Feed):
"""RSS feed for latest games."""
title: str = "Games - TTVDrops"
link: str = "/games/"
description: str = "Latest games on TTVDrops"
feed_copyright: str = "Information wants to be free."
_limit: int | None = None
def __call__(self, request: HttpRequest, *args: object, **kwargs: object) -> HttpResponse:
"""Override to capture limit parameter from request.
Args:
request (HttpRequest): The incoming HTTP request, potentially containing a 'limit' query parameter.
*args: Additional positional arguments.
**kwargs: Additional keyword arguments.
Returns:
HttpResponse: The HTTP response generated by the parent Feed class after processing the request.
"""
if request.GET.get("limit"):
try:
self._limit = int(request.GET.get("limit", 200))
except ValueError, TypeError:
self._limit = None
return super().__call__(request, *args, **kwargs)
def items(self) -> list[Game]:
"""Return the latest games (default 200, or limited by ?limit query param)."""
limit: int = self._limit if self._limit is not None else 200
return list(Game.objects.order_by("-added_at")[:limit])
def item_title(self, item: Game) -> SafeText:
"""Return the game name as the item title (SafeText for RSS)."""
return SafeText(item.get_game_name)
def item_description(self, item: Game) -> SafeText:
"""Return a description of the game."""
twitch_id: str = getattr(item, "twitch_id", "")
slug: str = getattr(item, "slug", "")
name: str = getattr(item, "name", "")
display_name: str = getattr(item, "display_name", "")
box_art: str | None = getattr(item, "box_art", None)
owner: Organization | None = getattr(item, "owner", None)
description_parts: list[SafeText] = []
game_name: str = display_name or name or slug or twitch_id
game_owner: str = owner.name if owner else "Unknown Owner"
if box_art:
description_parts.append(
SafeText(f"
"),
)
if slug:
description_parts.append(
SafeText(
f"{game_name} by {game_owner}
", # noqa: E501
),
)
else:
description_parts.append(SafeText(f"{game_name} by {game_owner}
"))
if twitch_id:
description_parts.append(SafeText(f"Twitch ID: {twitch_id}"))
return SafeText("".join(str(part) for part in description_parts))
def item_link(self, item: Game) -> str:
"""Return the link to the game detail."""
return reverse("twitch:game_detail", args=[item.twitch_id])
def item_pubdate(self, item: Game) -> datetime.datetime:
"""Returns the publication date to the feed item.
Fallback to added_at or now if missing.
"""
added_at: datetime.datetime | None = getattr(item, "added_at", None)
if added_at:
return added_at
return timezone.now()
def item_updateddate(self, item: Game) -> datetime.datetime:
"""Returns the game's last update time."""
updated_at: datetime.datetime | None = getattr(item, "updated_at", None)
if updated_at:
return updated_at
return timezone.now()
def item_guid(self, item: Game) -> str:
"""Return a unique identifier for each game."""
twitch_id: str = getattr(item, "twitch_id", "unknown")
return twitch_id + "@ttvdrops.com"
def item_author_name(self, item: Game) -> str:
"""Return the author name for the game, typically the owner organization name."""
owner: Organization | None = getattr(item, "owner", None)
if owner and owner.name:
return owner.name
return "Twitch"
def item_enclosure_url(self, item: Game) -> str:
"""Returns the URL of the game's box art for enclosure."""
box_art: str | None = getattr(item, "box_art", None)
if box_art:
return box_art
return ""
def item_enclosure_length(self, item: Game) -> int: # noqa: ARG002
"""Returns the length of the enclosure."""
# TODO(TheLovinator): Track image size for proper length # noqa: TD003
return 0
def item_enclosure_mime_type(self, item: Game) -> str: # noqa: ARG002
"""Returns the MIME type of the enclosure."""
# TODO(TheLovinator): Determine actual MIME type if needed # noqa: TD003
return "image/jpeg"
# MARK: /rss/campaigns/
class DropCampaignFeed(Feed):
"""RSS feed for latest drop campaigns."""
title: str = "Twitch Drop Campaigns"
link: str = "/campaigns/"
description: str = "Latest Twitch drop campaigns on TTVDrops"
feed_url: str = "/rss/campaigns/"
feed_copyright: str = "Information wants to be free."
_limit: int | None = None
def __call__(self, request: HttpRequest, *args: object, **kwargs: object) -> HttpResponse:
"""Override to capture limit parameter from request.
Args:
request (HttpRequest): The incoming HTTP request, potentially containing a 'limit' query parameter.
*args: Additional positional arguments.
**kwargs: Additional keyword arguments.
Returns:
HttpResponse: The HTTP response generated by the parent Feed class after processing the request.
"""
if request.GET.get("limit"):
try:
self._limit = int(request.GET.get("limit", 200))
except ValueError, TypeError:
self._limit = None
return super().__call__(request, *args, **kwargs)
def items(self) -> list[DropCampaign]:
"""Return the latest drop campaigns ordered by most recent start date (default 200, or limited by ?limit query param).""" # noqa: E501
limit: int = self._limit if self._limit is not None else 200
queryset: QuerySet[DropCampaign] = DropCampaign.objects.order_by("-start_at")
return list(_with_campaign_related(queryset)[:limit])
def item_title(self, item: DropCampaign) -> SafeText:
"""Return the campaign name as the item title (SafeText for RSS)."""
return SafeText(item.get_feed_title())
def item_description(self, item: DropCampaign) -> SafeText:
"""Return a description of the campaign."""
drops_data: list[dict] = []
drops: QuerySet[TimeBasedDrop] | None = getattr(item, "time_based_drops", None)
if drops:
drops_data = _build_drops_data(drops.all())
parts: list[SafeText] = []
image_url: str | None = getattr(item, "image_url", None)
if image_url:
item_name: str = getattr(item, "name", str(object=item))
parts.append(
format_html('
', image_url, item_name),
)
desc_text: str | None = getattr(item, "description", None)
if desc_text:
parts.append(format_html("{}
", desc_text))
# Insert start and end date info
insert_date_info(item, parts)
if drops_data:
parts.append(format_html("{}
", _construct_drops_summary(drops_data)))
# Only show channels if drop is not subscription only
if not getattr(item, "is_subscription_only", False):
channels: QuerySet[Channel] | None = getattr(item, "allow_channels", None)
if channels is not None:
game: Game | None = getattr(item, "game", None)
parts.append(_build_channels_html(channels, game=game))
details_url: str | None = getattr(item, "details_url", None)
if details_url:
parts.append(format_html('About', details_url))
return SafeText("".join(str(p) for p in parts))
def item_link(self, item: DropCampaign) -> str:
"""Return the link to the campaign detail."""
return item.get_feed_link()
def item_pubdate(self, item: DropCampaign) -> datetime.datetime:
"""Returns the publication date to the feed item.
Fallback to updated_at or now if missing.
"""
return item.get_feed_pubdate()
def item_updateddate(self, item: DropCampaign) -> datetime.datetime:
"""Returns the campaign's last update time."""
return item.updated_at
def item_categories(self, item: DropCampaign) -> tuple[str, ...]:
"""Returns the associated game's name as a category."""
return item.get_feed_categories()
def item_guid(self, item: DropCampaign) -> str:
"""Return a unique identifier for each campaign."""
return item.get_feed_guid()
def item_author_name(self, item: DropCampaign) -> str:
"""Return the author name for the campaign, typically the game name."""
return item.get_feed_author_name()
def item_enclosure_url(self, item: DropCampaign) -> str:
"""Returns the URL of the campaign image for enclosure."""
return item.get_feed_enclosure_url()
def item_enclosure_length(self, item: DropCampaign) -> int: # noqa: ARG002
"""Returns the length of the enclosure."""
# TODO(TheLovinator): Track image size for proper length # noqa: TD003
return 0
def item_enclosure_mime_type(self, item: DropCampaign) -> str: # noqa: ARG002
"""Returns the MIME type of the enclosure."""
# TODO(TheLovinator): Determine actual MIME type if needed # noqa: TD003
return "image/jpeg"
# MARK: /rss/games//campaigns/
class GameCampaignFeed(Feed):
"""RSS feed for the latest drop campaigns of a specific game."""
feed_copyright: str = "Information wants to be free."
_limit: int | None = None
def __call__(self, request: HttpRequest, *args: object, **kwargs: object) -> HttpResponse:
"""Override to capture limit parameter from request.
Args:
request (HttpRequest): The incoming HTTP request, potentially containing a 'limit' query parameter
*args: Additional positional arguments.
**kwargs: Additional keyword arguments.
Returns:
HttpResponse: The HTTP response generated by the parent Feed class after processing the request.
"""
if request.GET.get("limit"):
try:
self._limit = int(request.GET.get("limit", 200))
except ValueError, TypeError:
self._limit = None
return super().__call__(request, *args, **kwargs)
def get_object(self, request: HttpRequest, twitch_id: str) -> Game: # noqa: ARG002
"""Retrieve the Game instance for the given Twitch ID.
Returns:
Game: The corresponding Game object.
"""
return Game.objects.get(twitch_id=twitch_id)
def item_link(self, item: DropCampaign) -> str:
"""Return the link to the campaign detail."""
return reverse("twitch:campaign_detail", args=[item.twitch_id])
def title(self, obj: Game) -> str:
"""Return the feed title for the game campaigns."""
return f"TTVDrops: {obj.display_name} Campaigns"
def link(self, obj: Game) -> str:
"""Return the absolute URL to the game detail page."""
return reverse("twitch:game_detail", args=[obj.twitch_id])
def description(self, obj: Game) -> str:
"""Return a description for the feed."""
return f"Latest drop campaigns for {obj.display_name}"
def feed_url(self, obj: Game) -> str:
"""Return the URL to the RSS feed itself."""
return reverse("twitch:game_campaign_feed", args=[obj.twitch_id])
def items(self, obj: Game) -> list[DropCampaign]:
"""Return the latest drop campaigns for this game, ordered by most recent start date (default 200, or limited by ?limit query param).""" # noqa: E501
limit: int = self._limit if self._limit is not None else 200
queryset: QuerySet[DropCampaign] = DropCampaign.objects.filter(game=obj).order_by("-start_at")
return list(_with_campaign_related(queryset)[:limit])
def item_title(self, item: DropCampaign) -> SafeText:
"""Return the campaign name as the item title (SafeText for RSS)."""
return SafeText(item.get_feed_title())
def item_description(self, item: DropCampaign) -> SafeText:
"""Return a description of the campaign."""
drops_data: list[dict] = []
drops: QuerySet[TimeBasedDrop] | None = getattr(item, "time_based_drops", None)
if drops:
drops_data = _build_drops_data(drops.all())
parts: list[SafeText] = []
image_url: str | None = getattr(item, "image_url", None)
if image_url:
item_name: str = getattr(item, "name", str(object=item))
parts.append(
format_html('', image_url, item_name),
)
desc_text: str | None = getattr(item, "description", None)
if desc_text:
parts.append(format_html("{}
", desc_text))
# Insert start and end date info
insert_date_info(item, parts)
if drops_data:
parts.append(format_html("{}
", _construct_drops_summary(drops_data)))
# Only show channels if drop is not subscription only
if not getattr(item, "is_subscription_only", False):
channels: QuerySet[Channel] | None = getattr(item, "allow_channels", None)
if channels is not None:
game: Game | None = getattr(item, "game", None)
parts.append(_build_channels_html(channels, game=game))
details_url: str | None = getattr(item, "details_url", None)
if details_url:
parts.append(format_html('About', details_url))
account_link_url: str | None = getattr(item, "account_link_url", None)
if account_link_url:
parts.append(format_html(' | Link Account', account_link_url))
return SafeText("".join(str(p) for p in parts))
def item_pubdate(self, item: DropCampaign) -> datetime.datetime:
"""Returns the publication date to the feed item.
Uses start_at (when the drop starts). Fallback to added_at or now if missing.
"""
if isinstance(item, DropCampaign):
if item.start_at:
return item.start_at
if item.added_at:
return item.added_at
return timezone.now()
def item_updateddate(self, item: DropCampaign) -> datetime.datetime:
"""Returns the campaign's last update time."""
return item.updated_at
def item_categories(self, item: DropCampaign) -> tuple[str, ...]:
"""Returns the associated game's name as a category."""
return item.get_feed_categories()
def item_guid(self, item: DropCampaign) -> str:
"""Return a unique identifier for each campaign."""
return item.get_feed_guid()
def item_author_name(self, item: DropCampaign) -> str:
"""Return the author name for the campaign, typically the game name."""
return item.get_feed_author_name()
def item_enclosure_url(self, item: DropCampaign) -> str:
"""Returns the URL of the campaign image for enclosure."""
return item.get_feed_enclosure_url()
def item_enclosure_length(self, item: DropCampaign) -> int: # noqa: ARG002
"""Returns the length of the enclosure."""
# TODO(TheLovinator): Track image size for proper length # noqa: TD003
return 0
def item_enclosure_mime_type(self, item: DropCampaign) -> str: # noqa: ARG002
"""Returns the MIME type of the enclosure."""
# TODO(TheLovinator): Determine actual MIME type if needed # noqa: TD003
return "image/jpeg"
# MARK: /rss/organizations//campaigns/
class OrganizationCampaignFeed(Feed):
"""RSS feed for campaigns of a specific organization."""
_limit: int | None = None
def __call__(self, request: HttpRequest, *args: object, **kwargs: object) -> HttpResponse:
"""Override to capture limit parameter from request.
Args:
request (HttpRequest): The incoming HTTP request, potentially containing a 'limit' query parameter
*args: Additional positional arguments.
**kwargs: Additional keyword arguments.
Returns:
HttpResponse: The HTTP response generated by the parent Feed class after processing the request.
"""
if request.GET.get("limit"):
try:
self._limit = int(request.GET.get("limit", 200))
except ValueError, TypeError:
self._limit = None
return super().__call__(request, *args, **kwargs)
def get_object(self, request: HttpRequest, twitch_id: str) -> Organization: # noqa: ARG002
"""Retrieve the Organization instance for the given Twitch ID.
Returns:
Organization: The corresponding Organization object.
"""
return Organization.objects.get(twitch_id=twitch_id)
def item_link(self, item: DropCampaign) -> str:
"""Return the link to the campaign detail."""
return reverse("twitch:campaign_detail", args=[item.twitch_id])
def title(self, obj: Organization) -> str:
"""Return the feed title for the organization's campaigns."""
return f"TTVDrops: {obj.name} Campaigns"
def link(self, obj: Organization) -> str:
"""Return the absolute URL to the organization detail page."""
return reverse("twitch:organization_detail", args=[obj.twitch_id])
def description(self, obj: Organization) -> str:
"""Return a description for the feed."""
return f"Latest drop campaigns for organization {obj.name}"
def items(self, obj: Organization) -> list[DropCampaign]:
"""Return the latest drop campaigns for this organization, ordered by most recent start date (default 200, or limited by ?limit query param).""" # noqa: E501
limit: int = self._limit if self._limit is not None else 200
queryset: QuerySet[DropCampaign] = DropCampaign.objects.filter(game__owners=obj).order_by("-start_at")
return list(_with_campaign_related(queryset)[:limit])
def item_author_name(self, item: DropCampaign) -> str:
"""Return the author name for the campaign, typically the game name."""
return item.get_feed_author_name()
def item_guid(self, item: DropCampaign) -> str:
"""Return a unique identifier for each campaign."""
return item.get_feed_guid()
def item_enclosure_url(self, item: DropCampaign) -> str:
"""Returns the URL of the campaign image for enclosure."""
return item.get_feed_enclosure_url()
def item_enclosure_length(self, item: DropCampaign) -> int: # noqa: ARG002
"""Returns the length of the enclosure."""
# TODO(TheLovinator): Track image size for proper length # noqa: TD003
return 0
def item_enclosure_mime_type(self, item: DropCampaign) -> str: # noqa: ARG002
"""Returns the MIME type of the enclosure."""
# TODO(TheLovinator): Determine actual MIME type if needed # noqa: TD003
return "image/jpeg"
def item_categories(self, item: DropCampaign) -> tuple[str, ...]:
"""Returns the associated game's name as a category."""
return item.get_feed_categories()
def item_updateddate(self, item: DropCampaign) -> datetime.datetime:
"""Returns the campaign's last update time."""
return item.updated_at
def item_pubdate(self, item: DropCampaign) -> datetime.datetime:
"""Returns the publication date to the feed item.
Uses start_at (when the drop starts). Fallback to added_at or now if missing.
"""
if item.start_at:
return item.start_at
if item.added_at:
return item.added_at
return timezone.now()
def item_description(self, item: DropCampaign) -> SafeText:
"""Return a description of the campaign."""
drops_data: list[dict] = []
drops: QuerySet[TimeBasedDrop] | None = getattr(item, "time_based_drops", None)
if drops:
drops_data = _build_drops_data(drops.all())
parts: list[SafeText] = []
image_url: str | None = getattr(item, "image_url", None)
if image_url:
item_name: str = getattr(item, "name", str(object=item))
parts.append(
format_html('', image_url, item_name),
)
desc_text: str | None = getattr(item, "description", None)
if desc_text:
parts.append(format_html("{}
", desc_text))
# Insert start and end date info
insert_date_info(item, parts)
if drops_data:
parts.append(format_html("{}
", _construct_drops_summary(drops_data)))
# Only show channels if drop is not subscription only
if not getattr(item, "is_subscription_only", False):
channels: QuerySet[Channel] | None = getattr(item, "allow_channels", None)
if channels is not None:
game: Game | None = getattr(item, "game", None)
parts.append(_build_channels_html(channels, game=game))
details_url: str | None = getattr(item, "details_url", None)
if details_url:
parts.append(format_html('About', details_url))
return SafeText("".join(str(p) for p in parts))
# MARK: /rss/reward-campaigns/
class RewardCampaignFeed(Feed):
"""RSS feed for latest reward campaigns (Quest rewards)."""
title: str = "Twitch Reward Campaigns (Quest Rewards)"
link: str = "/campaigns/"
description: str = "Latest Twitch reward campaigns (Quest rewards) on TTVDrops"
feed_url: str = "/rss/reward-campaigns/"
feed_copyright: str = "Information wants to be free."
_limit: int | None = None
def __call__(self, request: HttpRequest, *args: object, **kwargs: object) -> HttpResponse:
"""Override to capture limit parameter from request.
Args:
request (HttpRequest): The incoming HTTP request, potentially containing a 'limit' query parameter.
*args: Additional positional arguments.
**kwargs: Additional keyword arguments.
Returns:
HttpResponse: The HTTP response generated by the parent Feed class after processing the request.
"""
if request.GET.get("limit"):
try:
self._limit = int(request.GET.get("limit", 200))
except ValueError, TypeError:
self._limit = None
return super().__call__(request, *args, **kwargs)
def items(self) -> list[RewardCampaign]:
"""Return the latest reward campaigns (default 200, or limited by ?limit query param)."""
limit: int = self._limit if self._limit is not None else 200
return list(
RewardCampaign.objects.select_related("game").order_by("-added_at")[:limit],
)
def item_title(self, item: RewardCampaign) -> SafeText:
"""Return the reward campaign name as the item title."""
return SafeText(item.get_feed_title())
def item_description(self, item: RewardCampaign) -> SafeText:
"""Return a description of the reward campaign."""
return SafeText(item.get_feed_description())
def item_link(self, item: RewardCampaign) -> str:
"""Return the link to the reward campaign (external URL or dashboard)."""
return item.get_feed_link()
def item_pubdate(self, item: RewardCampaign) -> datetime.datetime:
"""Returns the publication date to the feed item.
Uses starts_at (when the reward starts). Fallback to added_at or now if missing.
"""
return item.get_feed_pubdate()
def item_updateddate(self, item: RewardCampaign) -> datetime.datetime:
"""Returns the reward campaign's last update time."""
return item.updated_at
def item_categories(self, item: RewardCampaign) -> tuple[str, ...]:
"""Returns the associated game's name and brand as categories."""
return item.get_feed_categories()
def item_guid(self, item: RewardCampaign) -> str:
"""Return a unique identifier for each reward campaign."""
return item.get_feed_guid()
def item_author_name(self, item: RewardCampaign) -> str:
"""Return the author name for the reward campaign."""
return item.get_feed_author_name()