146 lines
5.2 KiB
Python
146 lines
5.2 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
from django.db.models import Prefetch
|
|
from django.http import HttpRequest, HttpResponse
|
|
from django.shortcuts import render
|
|
from django.template.response import TemplateResponse
|
|
from django.utils import timezone
|
|
from django.views.decorators.http import require_http_methods
|
|
|
|
from core.models import DropCampaign, Game, TimeBasedDrop
|
|
|
|
if TYPE_CHECKING:
|
|
from django.db.models.query import QuerySet
|
|
from django.http import HttpRequest
|
|
|
|
logger: logging.Logger = logging.getLogger(__name__)
|
|
|
|
|
|
@require_http_methods(request_method_list=["GET", "HEAD"])
|
|
def get_home(request: HttpRequest) -> HttpResponse:
|
|
"""Render the index page with drops grouped hierarchically by game and campaign.
|
|
|
|
This view fetches all currently active drops (where current time is between start_at and end_at),
|
|
and organizes them by game and campaign for display. Drops within each campaign are sorted by
|
|
required minutes watched.
|
|
|
|
Args:
|
|
request (HttpRequest): The request object.
|
|
|
|
Returns:
|
|
HttpResponse: The rendered index template with grouped drops.
|
|
"""
|
|
now: timezone.datetime = timezone.now()
|
|
|
|
try:
|
|
# Dictionary structure: {Game: {Campaign: [TimeBasedDrop, ...]}}
|
|
grouped_drops: dict[Game, dict[DropCampaign, list[TimeBasedDrop]]] = {}
|
|
|
|
# Fetch all currently active drops with optimized queries
|
|
current_drops_qs = (
|
|
TimeBasedDrop.objects.filter(start_at__lte=now, end_at__gte=now)
|
|
.select_related("campaign", "campaign__game", "campaign__owner")
|
|
.prefetch_related("benefits")
|
|
.order_by("campaign__game__display_name", "campaign__name", "required_minutes_watched")
|
|
)
|
|
|
|
# Drops without associated games or campaigns
|
|
orphaned_drops: list[TimeBasedDrop] = []
|
|
|
|
for drop in current_drops_qs:
|
|
# Check if drop has both campaign and game
|
|
if drop.campaign and drop.campaign.game:
|
|
game: Game = drop.campaign.game
|
|
campaign: DropCampaign = drop.campaign
|
|
|
|
# Initialize the game entry if it doesn't exist
|
|
if game not in grouped_drops:
|
|
grouped_drops[game] = {}
|
|
|
|
# Initialize the campaign entry if it doesn't exist
|
|
if campaign not in grouped_drops[game]:
|
|
grouped_drops[game][campaign] = []
|
|
|
|
# Add the drop to the appropriate campaign
|
|
grouped_drops[game][campaign].append(drop)
|
|
else:
|
|
# Store drops without proper association separately
|
|
orphaned_drops.append(drop)
|
|
logger.warning("Drop %s does not have an associated game or campaign.", drop.name or drop.drop_id)
|
|
|
|
# Make sure drops within each campaign are sorted by required_minutes_watched
|
|
for campaigns in grouped_drops.values():
|
|
for drops in campaigns.values():
|
|
drops.sort(key=lambda drop: drop.required_minutes_watched or 0)
|
|
|
|
# Also sort orphaned drops if any
|
|
if orphaned_drops:
|
|
orphaned_drops.sort(key=lambda drop: drop.required_minutes_watched or 0)
|
|
|
|
context: dict[str, Any] = {
|
|
"grouped_drops": grouped_drops,
|
|
"orphaned_drops": orphaned_drops,
|
|
"current_time": now,
|
|
}
|
|
|
|
return render(request, "index.html", context)
|
|
|
|
except Exception:
|
|
logger.exception("Error in get_home view")
|
|
return HttpResponse(
|
|
status=500,
|
|
content="An error occurred while processing your request. Please try again later.",
|
|
)
|
|
|
|
|
|
@require_http_methods(request_method_list=["GET", "HEAD"])
|
|
def get_game(request: HttpRequest, twitch_id: int) -> HttpResponse:
|
|
"""Render the game view page.
|
|
|
|
Args:
|
|
request (HttpRequest): The request object.
|
|
twitch_id (int): The Twitch ID of the game.
|
|
|
|
Returns:
|
|
HttpResponse: The response object.
|
|
"""
|
|
try:
|
|
time_based_drops_prefetch = Prefetch(
|
|
lookup="drops",
|
|
queryset=TimeBasedDrop.objects.prefetch_related("benefits"),
|
|
)
|
|
drop_campaigns_prefetch = Prefetch(
|
|
lookup="drop_campaigns",
|
|
queryset=DropCampaign.objects.prefetch_related(time_based_drops_prefetch),
|
|
)
|
|
game: Game = (
|
|
Game.objects.select_related("org").prefetch_related(drop_campaigns_prefetch).get(twitch_id=twitch_id)
|
|
)
|
|
|
|
except Game.DoesNotExist:
|
|
return HttpResponse(status=404, content="Game not found.")
|
|
except Game.MultipleObjectsReturned:
|
|
return HttpResponse(status=500, content="Multiple games found with the same Twitch ID.")
|
|
|
|
context: dict[str, Any] = {"game": game}
|
|
return TemplateResponse(request=request, template="game.html", context=context)
|
|
|
|
|
|
@require_http_methods(request_method_list=["GET", "HEAD"])
|
|
def get_games(request: HttpRequest) -> HttpResponse:
|
|
"""Render the game view page.
|
|
|
|
Args:
|
|
request (HttpRequest): The request object.
|
|
|
|
Returns:
|
|
HttpResponse: The response object.
|
|
"""
|
|
games: QuerySet[Game] = Game.objects.all()
|
|
|
|
context: dict[str, QuerySet[Game] | str] = {"games": games}
|
|
return TemplateResponse(request=request, template="games.html", context=context)
|