import logging
from typing import TYPE_CHECKING
import auto_prefetch
from django.contrib.humanize.templatetags.humanize import naturaltime
from django.contrib.postgres.indexes import GinIndex
from django.db import models
from django.urls import reverse
from django.utils import timezone
from django.utils.html import format_html
from django.utils.safestring import SafeText
from twitch.utils import normalize_twitch_box_art_url
if TYPE_CHECKING:
import datetime
logger: logging.Logger = logging.getLogger("ttvdrops")
# MARK: Organization
class Organization(auto_prefetch.Model):
"""Represents an organization on Twitch that can own drop campaigns."""
twitch_id = models.TextField(
unique=True,
verbose_name="Organization ID",
editable=False,
help_text="The unique Twitch identifier for the organization.",
)
name = models.TextField(
unique=True,
verbose_name="Name",
help_text="Display name of the organization.",
)
added_at = models.DateTimeField(
auto_now_add=True,
verbose_name="Added At",
editable=False,
help_text="Timestamp when this organization record was created.",
)
updated_at = models.DateTimeField(
auto_now=True,
verbose_name="Updated At",
editable=False,
help_text="Timestamp when this organization record was last updated.",
)
class Meta(auto_prefetch.Model.Meta):
ordering = ["name"]
indexes = [
models.Index(fields=["name"]),
models.Index(fields=["twitch_id"]),
models.Index(fields=["added_at"]),
models.Index(fields=["updated_at"]),
]
def __str__(self) -> str:
"""Return a string representation of the organization."""
return self.name or self.twitch_id
def feed_description(self: Organization) -> str:
"""Return a description of the organization for RSS feeds."""
name: str = self.name or "Unknown Organization"
url: str = reverse("twitch:organization_detail", args=[self.twitch_id])
return format_html(
'
',
url,
name,
)
# MARK: Game
class Game(auto_prefetch.Model):
"""Represents a game on Twitch."""
twitch_id = models.TextField(verbose_name="Twitch game ID", unique=True)
slug = models.TextField(
max_length=200,
blank=True,
default="",
verbose_name="Slug",
help_text="Short unique identifier for the game.",
)
name = models.TextField(blank=True, default="", verbose_name="Name")
display_name = models.TextField(blank=True, default="", verbose_name="Display name")
box_art = models.URLField( # noqa: DJ001
max_length=500,
blank=True,
null=True,
default="",
verbose_name="Box art URL",
)
box_art_file = models.ImageField(
upload_to="games/box_art/",
blank=True,
null=True,
width_field="box_art_width",
height_field="box_art_height",
help_text="Locally cached box art image served from this site.",
)
box_art_width = models.PositiveIntegerField(
null=True,
blank=True,
editable=False,
help_text="Width of cached box art image in pixels.",
)
box_art_height = models.PositiveIntegerField(
null=True,
blank=True,
editable=False,
help_text="Height of cached box art image in pixels.",
)
owners = models.ManyToManyField(
Organization,
related_name="games",
blank=True,
verbose_name="Organizations",
help_text="Organizations that own this game.",
)
added_at = models.DateTimeField(
auto_now_add=True,
help_text="Timestamp when this game record was created.",
)
updated_at = models.DateTimeField(
auto_now=True,
help_text="Timestamp when this game record was last updated.",
)
class Meta(auto_prefetch.Model.Meta):
ordering = ["display_name"]
indexes = [
models.Index(fields=["display_name"]),
models.Index(fields=["name"]),
models.Index(fields=["slug"]),
models.Index(fields=["twitch_id"]),
models.Index(fields=["added_at"]),
models.Index(fields=["updated_at"]),
# For games_grid_view grouping by owners + display_name
# ManyToManyField does not support direct indexing, so skip these
]
def __str__(self) -> str:
"""Return a string representation of the game."""
if (self.display_name and self.name) and (self.display_name != self.name):
logger.warning(
"Game display name '%s' does not match name '%s'.",
self.display_name,
self.name,
)
return f"{self.display_name} ({self.name})"
return self.display_name or self.name or self.slug or self.twitch_id
def get_absolute_url(self) -> str:
"""Return canonical URL to the game details page."""
return reverse("game_detail", kwargs={"twitch_id": self.twitch_id})
@property
def organizations(self) -> models.QuerySet[Organization]:
"""Return orgs that own games with campaigns for this game."""
return Organization.objects.filter(games__drop_campaigns__game=self).distinct()
@property
def get_game_name(self) -> str:
"""Return the best available name for the game."""
if self.display_name:
return self.display_name
if self.name:
return self.name
if self.slug:
return self.slug
return self.twitch_id
@property
def twitch_directory_url(self) -> str:
"""Return Twitch directory URL with drops filter when slug exists."""
# TODO(TheLovinator): If no slug, get from Twitch API or IGDB? # noqa: TD003
if self.slug:
return f"https://www.twitch.tv/directory/category/{self.slug}?filter=drops"
return ""
@property
def box_art_best_url(self) -> str:
"""Return the best available URL for the game's box art (local first)."""
try:
if self.box_art_file and getattr(self.box_art_file, "url", None):
return self.box_art_file.url
except (AttributeError, OSError, ValueError) as exc:
logger.debug("Failed to resolve Game.box_art_file url: %s", exc)
return normalize_twitch_box_art_url(self.box_art or "")
# MARK: TwitchGame
class TwitchGameData(auto_prefetch.Model):
"""Represents game metadata returned from the Twitch API.
This mirrors the public Twitch API fields for a game and is tied to the
local `Game` model where possible.
Fields:
id: Twitch game id (primary key)
game: Optional FK to the local Game object
name: Display name of the game
box_art_url: URL template for box art with {width}x{height} placeholder
igdb_id: Optional IGDB id for the game
"""
twitch_id = models.TextField(
verbose_name="Twitch Game ID",
unique=True,
help_text="The Twitch ID for this game.",
)
game = auto_prefetch.ForeignKey(
Game,
on_delete=models.SET_NULL,
related_name="twitch_game_data",
null=True,
blank=True,
verbose_name="Game",
help_text=("Optional link to the local Game record for this Twitch game."),
)
name = models.TextField(blank=True, default="", verbose_name="Name")
box_art_url = models.URLField(
max_length=500,
blank=True,
default="",
verbose_name="Box art URL",
help_text=(
"URL template with {width}x{height} placeholders for the box art image."
),
)
igdb_id = models.TextField(blank=True, default="", verbose_name="IGDB ID")
added_at = models.DateTimeField(
auto_now_add=True,
help_text="Record creation time.",
)
updated_at = models.DateTimeField(
auto_now=True,
help_text="Record last update time.",
)
class Meta(auto_prefetch.Model.Meta):
ordering = ["name"]
indexes = [
models.Index(fields=["name"]),
models.Index(fields=["twitch_id"]),
models.Index(fields=["game"]),
models.Index(fields=["igdb_id"]),
models.Index(fields=["added_at"]),
models.Index(fields=["updated_at"]),
]
def __str__(self) -> str:
return self.name or self.twitch_id
# MARK: Channel
class Channel(auto_prefetch.Model):
"""Represents a Twitch channel that can participate in drop campaigns."""
twitch_id = models.TextField(
verbose_name="Channel ID",
help_text="The unique Twitch identifier for the channel.",
unique=True,
)
name = models.TextField(
verbose_name="Username",
help_text="The lowercase username of the channel.",
)
display_name = models.TextField(
verbose_name="Display Name",
help_text="The display name of the channel (with proper capitalization).",
)
added_at = models.DateTimeField(
auto_now_add=True,
help_text="Timestamp when this channel record was created.",
)
updated_at = models.DateTimeField(
auto_now=True,
help_text="Timestamp when this channel record was last updated.",
)
class Meta(auto_prefetch.Model.Meta):
ordering = ["display_name"]
indexes = [
models.Index(fields=["display_name"]),
models.Index(fields=["name"]),
models.Index(fields=["twitch_id"]),
models.Index(fields=["added_at"]),
models.Index(fields=["updated_at"]),
]
def __str__(self) -> str:
"""Return a string representation of the channel."""
return self.display_name or self.name or self.twitch_id
# MARK: DropCampaign
class DropCampaign(auto_prefetch.Model):
"""Represents a Twitch drop campaign."""
twitch_id = models.TextField(
unique=True,
editable=False,
help_text="The Twitch ID for this campaign.",
)
name = models.TextField(help_text="Name of the drop campaign.")
description = models.TextField(
blank=True,
help_text="Detailed description of the campaign.",
)
details_url = models.URLField(
max_length=500,
blank=True,
default="",
help_text="URL with campaign details.",
)
account_link_url = models.URLField(
max_length=500,
blank=True,
default="",
help_text="URL to link a Twitch account for the campaign.",
)
image_url = models.URLField(
max_length=500,
blank=True,
default="",
help_text="URL to an image representing the campaign.",
)
image_file = models.ImageField(
upload_to="campaigns/images/",
blank=True,
null=True,
width_field="image_width",
height_field="image_height",
help_text="Locally cached campaign image served from this site.",
)
image_width = models.PositiveIntegerField(
null=True,
blank=True,
editable=False,
help_text="Width of cached image in pixels.",
)
image_height = models.PositiveIntegerField(
null=True,
blank=True,
editable=False,
help_text="Height of cached image in pixels.",
)
start_at = models.DateTimeField(
null=True,
blank=True,
help_text="Datetime when the campaign starts.",
)
end_at = models.DateTimeField(
null=True,
blank=True,
help_text="Datetime when the campaign ends.",
)
allow_is_enabled = models.BooleanField(
default=True,
help_text="Whether the campaign allows participation.",
)
allow_channels = models.ManyToManyField(
Channel,
blank=True,
related_name="allowed_campaigns",
help_text="Channels that are allowed to participate in this campaign.",
)
game = auto_prefetch.ForeignKey(
Game,
on_delete=models.CASCADE,
related_name="drop_campaigns",
verbose_name="Game",
help_text="Game associated with this campaign.",
)
operation_names = models.JSONField(
default=list,
blank=True,
help_text="List of GraphQL operation names used to fetch this campaign data (e.g., ['ViewerDropsDashboard', 'Inventory']).",
)
added_at = models.DateTimeField(
auto_now_add=True,
help_text="Timestamp when this campaign record was created.",
)
updated_at = models.DateTimeField(
auto_now=True,
help_text="Timestamp when this campaign record was last updated.",
)
class Meta(auto_prefetch.Model.Meta):
ordering = ["-start_at"]
indexes = [
models.Index(fields=["-start_at"]),
models.Index(fields=["end_at"]),
models.Index(fields=["game"]),
models.Index(fields=["twitch_id"]),
models.Index(fields=["name"]),
models.Index(fields=["description"]),
models.Index(fields=["allow_is_enabled"]),
GinIndex(fields=["operation_names"], name="twitch_drop_operati_gin_idx"),
models.Index(fields=["added_at"]),
models.Index(fields=["updated_at"]),
# Composite indexes for common queries
models.Index(fields=["game", "-start_at"]),
models.Index(fields=["start_at", "end_at"]),
# For dashboard and game_detail active campaign filtering
models.Index(fields=["start_at", "end_at", "game"]),
models.Index(fields=["end_at", "-start_at"]),
]
def __str__(self) -> str:
return self.name
@property
def is_active(self) -> bool:
"""Check if the campaign is currently active."""
now: datetime.datetime = timezone.now()
if self.start_at is None or self.end_at is None:
return False
return self.start_at <= now <= self.end_at
@property
def clean_name(self) -> str:
"""Return the campaign name without the game name prefix.
Examples:
"Ravendawn - July 2" -> "July 2"
"Party Animals Twitch Drop" -> "Twitch Drop"
"Skull & Bones - Closed Beta" -> "Closed Beta" (& is replaced
with "and")
"""
if not self.game or not self.game.display_name:
return self.name
game_variations = [self.game.display_name]
if "&" in self.game.display_name:
game_variations.append(self.game.display_name.replace("&", "and"))
if "and" in self.game.display_name:
game_variations.append(self.game.display_name.replace("and", "&"))
for game_name in game_variations:
# Check for different separators after the game name
for separator in [" - ", " | ", " "]:
prefix_to_check = game_name + separator
name: str = self.name
if name.startswith(prefix_to_check):
return name.removeprefix(prefix_to_check).strip()
return self.name
@property
def image_best_url(self) -> str:
"""Return the best URL for the campaign image.
Priority:
1. Local cached image file
2. Campaign image URL
3. First benefit image URL (if campaign has no image)
"""
try:
if self.image_file and getattr(self.image_file, "url", None):
return self.image_file.url
except (AttributeError, OSError, ValueError) as exc:
logger.debug("Failed to resolve DropCampaign.image_file url: %s", exc)
if self.image_url:
return self.image_url
# If no campaign image, use the first benefit image
for drop in self.time_based_drops.all(): # pyright: ignore[reportAttributeAccessIssue]
for benefit in drop.benefits.all(): # pyright: ignore[reportAttributeAccessIssue]
benefit_image_url: str = benefit.image_best_url
if benefit_image_url:
return benefit_image_url
return ""
@property
def duration_iso(self) -> str:
"""Return the campaign duration in ISO 8601 format (e.g., 'P3DT4H30M').
This is used for the