from __future__ import annotations import logging import re from typing import TYPE_CHECKING from typing import Literal from django.contrib.humanize.templatetags.humanize import naturaltime from django.contrib.syndication.views import Feed from django.db.models.query import QuerySet from django.urls import reverse from django.utils import timezone from django.utils.html import format_html from django.utils.html import format_html_join from django.utils.safestring import SafeString from django.utils.safestring import SafeText from twitch.models import Channel 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 if TYPE_CHECKING: import datetime from django.db.models import Model from django.db.models import QuerySet from django.http import HttpRequest logger: logging.Logger = logging.getLogger("ttvdrops") def insert_date_info(item: Model, parts: list[SafeText]) -> None: """Insert start and end date information into parts list. Args: item (Model): The campaign item containing start_at and end_at. parts (list[SafeText]): The list of HTML parts to append to. """ end_at: datetime.datetime | None = getattr(item, "end_at", None) start_at: datetime.datetime | None = getattr(item, "start_at", None) if start_at or end_at: start_part: SafeString = ( format_html("Starts: {} ({})", start_at.strftime("%Y-%m-%d %H:%M %Z"), naturaltime(start_at)) if start_at else SafeText("") ) end_part: SafeString = ( format_html("Ends: {} ({})", end_at.strftime("%Y-%m-%d %H:%M %Z"), naturaltime(end_at)) if end_at else SafeText("") ) # Start date and end date separated by a line break if both present if start_part and end_part: parts.append(format_html("
{}
{}
{}
", start_part)) elif end_part: parts.append(format_html("{}
", end_part)) def _build_drops_data(drops_qs: QuerySet[TimeBasedDrop]) -> list[dict]: """Build a simplified data structure for rendering drops in a template. Returns: list[dict]: A list of dictionaries each containing `name`, `benefits`, `requirements`, and `period` for a drop, suitable for template rendering. """ drops_data: list[dict] = [] for drop in drops_qs: requirements: str = "" required_minutes: int | None = getattr(drop, "required_minutes_watched", None) required_subs: int = getattr(drop, "required_subs", 0) or 0 if required_minutes: requirements = f"{required_minutes} minutes watched" if required_subs > 0: sub_word: Literal["subs", "sub"] = "subs" if required_subs > 1 else "sub" if requirements: requirements += f" and {required_subs} {sub_word} required" else: requirements = f"{required_subs} {sub_word} required" period: str = "" drop_start: datetime.datetime | None = getattr(drop, "start_at", None) drop_end: datetime.datetime | None = getattr(drop, "end_at", None) if drop_start is not None: period += drop_start.strftime("%Y-%m-%d %H:%M %Z") if drop_end is not None: if period: period += " - " + drop_end.strftime("%Y-%m-%d %H:%M %Z") else: period = drop_end.strftime("%Y-%m-%d %H:%M %Z") drops_data.append({ "name": getattr(drop, "name", str(drop)), "benefits": list(drop.benefits.all()), "requirements": requirements, "period": period, }) return drops_data def _build_channels_html(channels: QuerySet[Channel], game: Game | None) -> SafeText: """Render up to max_links channel links asNew Twitch organization added to TTVDrops:
"), SafeText( f"", # noqa: E501 ), )) return SafeText("".join(str(part) for part in description_parts)) def item_link(self, item: Model) -> str: """Return the link to the organization detail.""" if not isinstance(item, Organization): logger.error("item_link called with non-Organization item: %s", type(item)) return reverse("twitch:dashboard") return reverse("twitch:organization_detail", args=[item.twitch_id]) def item_pubdate(self, item: Model) -> 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: Model) -> 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_guid(self, item: Model) -> str: """Return a unique identifier for each organization.""" twitch_id: str = getattr(item, "twitch_id", "unknown") return twitch_id + "@ttvdrops.com" def item_author_name(self, item: Model) -> str: """Return the author name for the organization.""" if not isinstance(item, Organization): logger.error("item_author_name called with non-Organization item: %s", type(item)) return "Twitch" 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." def items(self) -> list[Game]: """Return the latest 200 games.""" return list(Game.objects.order_by("-added_at")[:200]) def item_title(self, item: Model) -> SafeText: """Return the game name as the item title (SafeText for RSS).""" if not isinstance(item, Game): logger.error("item_title called with non-Game item: %s", type(item)) return SafeText("New Twitch game added to TTVDrops") return SafeText(item.get_game_name) def item_description(self, item: Model) -> SafeText: """Return a description of the game.""" # Return all the information we have about the game if not isinstance(item, Game): logger.error("item_description called with non-Game item: %s", type(item)) return SafeText("No description available.") 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"{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: Model) -> str: """Return the link to the game detail.""" if not isinstance(item, Game): logger.error("item_link called with non-Game item: %s", type(item)) return reverse("twitch:dashboard") return reverse("twitch:game_detail", args=[item.twitch_id]) def item_pubdate(self, item: Model) -> 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: Model) -> 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: Model) -> 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: Model) -> 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: Model) -> 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: Model) -> 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: Model) -> 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." def items(self) -> list[DropCampaign]: """Return the latest 200 drop campaigns ordered by most recent start date.""" return list( DropCampaign.objects.select_related("game").order_by("-start_at")[:200], ) def item_title(self, item: Model) -> SafeText: """Return the campaign name as the item title (SafeText for RSS).""" game: Game | None = getattr(item, "game", None) game_name: str = getattr(game, "display_name", str(game)) if game else "" clean_name: str = getattr(item, "clean_name", str(item)) return SafeText(f"{game_name}: {clean_name}") def item_description(self, item: Model) -> 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.select_related().prefetch_related("benefits").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('{}
", 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: Model) -> str: """Return the link to the campaign detail.""" if not isinstance(item, DropCampaign): logger.error("item_link called with non-DropCampaign item: %s", type(item)) return reverse("twitch:dashboard") return reverse("twitch:campaign_detail", args=[item.twitch_id]) def item_pubdate(self, item: Model) -> datetime.datetime: """Returns the publication date to the feed item. Fallback to updated_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: 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.""" categories: list[str] = ["twitch", "drops"] item_game: Game | None = getattr(item, "game", None) if item_game: categories.append(item_game.get_game_name) item_game_owner: Organization | None = getattr(item_game, "owner", None) if item_game_owner: categories.extend((str(item_game_owner.name), str(item_game_owner.twitch_id))) return tuple(categories) def item_guid(self, item: DropCampaign) -> str: """Return a unique identifier for each campaign.""" return item.twitch_id + "@ttvdrops.com" def item_author_name(self, item: DropCampaign) -> str: """Return the author name for the campaign, typically the game name.""" item_game: Game | None = getattr(item, "game", None) if item_game and item_game.display_name: return item_game.display_name return "Twitch" def item_enclosure_url(self, item: DropCampaign) -> str: """Returns the URL of the campaign image for enclosure.""" return item.image_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/{}
", 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: Model) -> 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. """ start_at: datetime.datetime | None = getattr(item, "start_at", None) if start_at: return start_at added_at: datetime.datetime | None = getattr(item, "added_at", None) if added_at: return 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.""" categories: list[str] = ["twitch", "drops"] item_game: Game | None = getattr(item, "game", None) if item_game: categories.append(item_game.get_game_name) item_game_owner: Organization | None = getattr(item_game, "owner", None) if item_game_owner: categories.extend((str(item_game_owner.name), str(item_game_owner.twitch_id))) return tuple(categories) def item_guid(self, item: DropCampaign) -> str: """Return a unique identifier for each campaign.""" return item.twitch_id + "@ttvdrops.com" def item_author_name(self, item: DropCampaign) -> str: """Return the author name for the campaign, typically the game name.""" item_game: Game | None = getattr(item, "game", None) if item_game and item_game.display_name: return item_game.display_name return "Twitch" def item_enclosure_url(self, item: DropCampaign) -> str: """Returns the URL of the campaign image for enclosure.""" return item.image_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/{}
", 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." def items(self) -> list[RewardCampaign]: """Return the latest 200 reward campaigns.""" return list( RewardCampaign.objects.select_related("game").order_by("-added_at")[:200], ) def item_title(self, item: Model) -> SafeText: """Return the reward campaign name as the item title.""" brand: str = getattr(item, "brand", "") name: str = getattr(item, "name", str(item)) if brand: return SafeText(f"{brand}: {name}") return SafeText(name) def item_description(self, item: Model) -> SafeText: """Return a description of the reward campaign.""" parts: list[SafeText] = [] summary: str | None = getattr(item, "summary", None) if summary: parts.append(format_html("{}
", summary)) # Insert start and end date info (uses starts_at/ends_at instead of start_at/end_at) ends_at: datetime.datetime | None = getattr(item, "ends_at", None) starts_at: datetime.datetime | None = getattr(item, "starts_at", None) if starts_at or ends_at: start_part: SafeString = ( format_html("Starts: {} ({})", starts_at.strftime("%Y-%m-%d %H:%M %Z"), naturaltime(starts_at)) if starts_at else SafeText("") ) end_part: SafeString = ( format_html("Ends: {} ({})", ends_at.strftime("%Y-%m-%d %H:%M %Z"), naturaltime(ends_at)) if ends_at else SafeText("") ) if start_part and end_part: parts.append(format_html("{}
{}
{}
", start_part)) elif end_part: parts.append(format_html("{}
", end_part)) is_sitewide: bool = getattr(item, "is_sitewide", False) if is_sitewide: parts.append(SafeText("This is a sitewide reward campaign
")) else: game: Game | None = getattr(item, "game", None) if game: parts.append(format_html("Game: {}
", game.display_name or game.name)) about_url: str | None = getattr(item, "about_url", None) if about_url: parts.append(format_html('', about_url)) external_url: str | None = getattr(item, "external_url", None) if external_url: parts.append(format_html('', external_url)) return SafeText("".join(str(p) for p in parts)) def item_link(self, item: Model) -> str: """Return the link to the reward campaign (external URL or dashboard).""" external_url: str | None = getattr(item, "external_url", None) if external_url: return external_url return reverse("twitch:dashboard") def item_pubdate(self, item: Model) -> 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. """ starts_at: datetime.datetime | None = getattr(item, "starts_at", None) if starts_at: return starts_at added_at: datetime.datetime | None = getattr(item, "added_at", None) if added_at: return added_at return timezone.now() 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.""" categories: list[str] = ["twitch", "rewards", "quests"] brand: str | None = getattr(item, "brand", None) if brand: categories.append(brand) item_game: Game | None = getattr(item, "game", None) if item_game: categories.append(item_game.get_game_name) return tuple(categories) def item_guid(self, item: RewardCampaign) -> str: """Return a unique identifier for each reward campaign.""" return item.twitch_id + "@ttvdrops.com" def item_author_name(self, item: RewardCampaign) -> str: """Return the author name for the reward campaign.""" brand: str | None = getattr(item, "brand", None) if brand: return brand item_game: Game | None = getattr(item, "game", None) if item_game and item_game.display_name: return item_game.display_name return "Twitch"