feat: Add Twitch Drops Tracker application with campaign management

- Implemented models for DropCampaign, Game, Organization, DropBenefit, TimeBasedDrop, and DropBenefitEdge.
- Created views for listing and detailing drop campaigns.
- Added templates for dashboard, campaign list, and campaign detail.
- Developed management command to import drop campaigns from JSON files.
- Configured admin interface for managing campaigns and related models.
- Updated URL routing for the application.
- Enhanced README with installation instructions and project structure.
This commit is contained in:
Joakim Hellsén 2025-07-09 22:46:23 +02:00
commit 5c482c1729
15 changed files with 1145 additions and 10 deletions

View file

@ -1 +1,81 @@
# Register your models here.
from __future__ import annotations
from django.contrib import admin
from twitch.models import DropBenefit, DropBenefitEdge, DropCampaign, Game, Organization, TimeBasedDrop
@admin.register(Game)
class GameAdmin(admin.ModelAdmin):
"""Admin configuration for Game model."""
list_display = ("id", "display_name", "slug")
search_fields = ("id", "display_name", "slug")
@admin.register(Organization)
class OrganizationAdmin(admin.ModelAdmin):
"""Admin configuration for Organization model."""
list_display = ("id", "name")
search_fields = ("id", "name")
class TimeBasedDropInline(admin.TabularInline):
"""Inline admin for TimeBasedDrop model."""
model = TimeBasedDrop
extra = 0
@admin.register(DropCampaign)
class DropCampaignAdmin(admin.ModelAdmin):
"""Admin configuration for DropCampaign model."""
list_display = ("id", "name", "game", "owner", "status", "start_at", "end_at", "is_active")
list_filter = ("status", "game", "owner")
search_fields = ("id", "name", "description")
inlines = [TimeBasedDropInline]
readonly_fields = ("created_at", "updated_at")
class DropBenefitEdgeInline(admin.TabularInline):
"""Inline admin for DropBenefitEdge model."""
model = DropBenefitEdge
extra = 0
@admin.register(TimeBasedDrop)
class TimeBasedDropAdmin(admin.ModelAdmin):
"""Admin configuration for TimeBasedDrop model."""
list_display = (
"id",
"name",
"campaign",
"required_minutes_watched",
"required_subs",
"start_at",
"end_at",
)
list_filter = ("campaign__game", "campaign")
search_fields = ("id", "name")
inlines = [DropBenefitEdgeInline]
@admin.register(DropBenefit)
class DropBenefitAdmin(admin.ModelAdmin):
"""Admin configuration for DropBenefit model."""
list_display = (
"id",
"name",
"game",
"owner_organization",
"distribution_type",
"entitlement_limit",
"created_at",
)
list_filter = ("game", "owner_organization", "distribution_type")
search_fields = ("id", "name")

View file

View file

View file

@ -0,0 +1,151 @@
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
from django.core.management.base import BaseCommand, CommandError, CommandParser
from twitch.models import DropBenefit, DropBenefitEdge, DropCampaign, Game, Organization, TimeBasedDrop
class Command(BaseCommand):
"""Import Twitch drop campaign data from a JSON file."""
help = "Import Twitch drop campaign data from a JSON file"
def add_arguments(self, parser: CommandParser) -> None:
"""Add command arguments.
Args:
parser: The command argument parser.
"""
parser.add_argument(
"json_file",
type=str,
help="Path to the JSON file containing the drop campaign data",
)
def handle(self, **options: str) -> None:
"""Execute the command.
Args:
**options: Arbitrary keyword arguments.
Raises:
CommandError: If the file doesn't exist, isn't a JSON file,
or has an invalid JSON structure.
"""
json_file_path: str = options["json_file"]
file_path = Path(json_file_path)
# Validate file exists and is a JSON file
if not file_path.exists():
msg = f"File {json_file_path} does not exist"
raise CommandError(msg)
if not json_file_path.endswith(".json"):
msg = f"File {json_file_path} is not a JSON file"
raise CommandError(msg)
# Load JSON data
try:
with file_path.open(encoding="utf-8") as f:
data = json.load(f)
except json.JSONDecodeError as e:
msg = f"Error decoding JSON: {e}"
raise CommandError(msg) from e
# Check if the JSON has the expected structure
if "data" not in data or "user" not in data["data"] or "dropCampaign" not in data["data"]["user"]:
msg = "Invalid JSON structure: Missing data.user.dropCampaign"
raise CommandError(msg)
# Extract drop campaign data
drop_campaign_data = data["data"]["user"]["dropCampaign"]
# Process the data
self._import_drop_campaign(drop_campaign_data)
self.stdout.write(self.style.SUCCESS(f"Successfully imported drop campaign: {drop_campaign_data['name']}"))
def _import_drop_campaign(self, campaign_data: dict[str, Any]) -> None:
"""Import drop campaign data into the database.
Args:
campaign_data: The drop campaign data to import.
"""
# First, create or update the game
game_data = campaign_data["game"]
game, _ = Game.objects.update_or_create(
id=game_data["id"],
defaults={
"slug": game_data.get("slug", ""),
"display_name": game_data["displayName"],
},
)
# Create or update the organization
org_data = campaign_data["owner"]
organization, _ = Organization.objects.update_or_create(
id=org_data["id"],
defaults={"name": org_data["name"]},
)
# Create or update the drop campaign
drop_campaign, _ = DropCampaign.objects.update_or_create(
id=campaign_data["id"],
defaults={
"name": campaign_data["name"],
"description": campaign_data["description"],
"details_url": campaign_data.get("detailsURL", ""),
"account_link_url": campaign_data.get("accountLinkURL", ""),
"image_url": campaign_data.get("imageURL", ""),
"start_at": campaign_data["startAt"],
"end_at": campaign_data["endAt"],
"status": campaign_data["status"],
"is_account_connected": campaign_data["self"]["isAccountConnected"],
"game": game,
"owner": organization,
},
)
# Process time-based drops
for drop_data in campaign_data.get("timeBasedDrops", []):
time_based_drop, _ = TimeBasedDrop.objects.update_or_create(
id=drop_data["id"],
defaults={
"name": drop_data["name"],
"required_minutes_watched": drop_data["requiredMinutesWatched"],
"required_subs": drop_data.get("requiredSubs", 0),
"start_at": drop_data["startAt"],
"end_at": drop_data["endAt"],
"campaign": drop_campaign,
},
)
# Process benefits
for benefit_edge in drop_data.get("benefitEdges", []):
benefit_data = benefit_edge["benefit"]
benefit, _ = DropBenefit.objects.update_or_create(
id=benefit_data["id"],
defaults={
"name": benefit_data["name"],
"image_asset_url": benefit_data.get("imageAssetURL", ""),
"created_at": benefit_data["createdAt"],
"entitlement_limit": benefit_data.get("entitlementLimit", 1),
"is_ios_available": benefit_data.get("isIosAvailable", False),
"distribution_type": benefit_data["distributionType"],
"game": game,
"owner_organization": organization,
},
)
# Create the relationship between drop and benefit
DropBenefitEdge.objects.update_or_create(
drop=time_based_drop,
benefit=benefit,
defaults={
"entitlement_limit": benefit_edge.get("entitlementLimit", 1),
},
)

View file

@ -0,0 +1,101 @@
# Generated by Django 5.2.4 on 2025-07-09 20:44
import django.db.models.deletion
from django.db import migrations, models
class Migration(migrations.Migration):
initial = True
dependencies = [
]
operations = [
migrations.CreateModel(
name='DropBenefit',
fields=[
('id', models.TextField(primary_key=True, serialize=False)),
('name', models.TextField()),
('image_asset_url', models.URLField(blank=True, default='', max_length=500)),
('created_at', models.DateTimeField()),
('entitlement_limit', models.PositiveIntegerField(default=1)),
('is_ios_available', models.BooleanField(default=False)),
('distribution_type', models.TextField(choices=[('DIRECT_ENTITLEMENT', 'Direct Entitlement'), ('CODE', 'Code')])),
],
),
migrations.CreateModel(
name='Game',
fields=[
('id', models.TextField(primary_key=True, serialize=False)),
('slug', models.TextField(blank=True, default='')),
('display_name', models.TextField()),
],
),
migrations.CreateModel(
name='Organization',
fields=[
('id', models.TextField(primary_key=True, serialize=False)),
('name', models.TextField()),
],
),
migrations.CreateModel(
name='DropBenefitEdge',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('entitlement_limit', models.PositiveIntegerField(default=1)),
('benefit', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='twitch.dropbenefit')),
],
),
migrations.AddField(
model_name='dropbenefit',
name='game',
field=models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='drop_benefits', to='twitch.game'),
),
migrations.CreateModel(
name='DropCampaign',
fields=[
('id', models.TextField(primary_key=True, serialize=False)),
('name', models.TextField()),
('description', models.TextField(blank=True)),
('details_url', models.URLField(blank=True, default='', max_length=500)),
('account_link_url', models.URLField(blank=True, default='', max_length=500)),
('image_url', models.URLField(blank=True, default='', max_length=500)),
('start_at', models.DateTimeField()),
('end_at', models.DateTimeField()),
('status', models.TextField(choices=[('ACTIVE', 'Active'), ('UPCOMING', 'Upcoming'), ('EXPIRED', 'Expired')])),
('is_account_connected', models.BooleanField(default=False)),
('created_at', models.DateTimeField(auto_now_add=True)),
('updated_at', models.DateTimeField(auto_now=True)),
('game', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='drop_campaigns', to='twitch.game')),
('owner', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='drop_campaigns', to='twitch.organization')),
],
),
migrations.AddField(
model_name='dropbenefit',
name='owner_organization',
field=models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='drop_benefits', to='twitch.organization'),
),
migrations.CreateModel(
name='TimeBasedDrop',
fields=[
('id', models.TextField(primary_key=True, serialize=False)),
('name', models.TextField()),
('required_minutes_watched', models.PositiveIntegerField()),
('required_subs', models.PositiveIntegerField(default=0)),
('start_at', models.DateTimeField()),
('end_at', models.DateTimeField()),
('benefits', models.ManyToManyField(related_name='drops', through='twitch.DropBenefitEdge', to='twitch.dropbenefit')),
('campaign', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='time_based_drops', to='twitch.dropcampaign')),
],
),
migrations.AddField(
model_name='dropbenefitedge',
name='drop',
field=models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='twitch.timebaseddrop'),
),
migrations.AlterUniqueTogether(
name='dropbenefitedge',
unique_together={('drop', 'benefit')},
),
]

View file

@ -1 +1,127 @@
# Create your models here.
from __future__ import annotations
from typing import ClassVar
from django.db import models
from django.utils import timezone
class Game(models.Model):
"""Represents a game on Twitch."""
id = models.TextField(primary_key=True)
slug = models.TextField(blank=True, default="")
display_name = models.TextField()
def __str__(self) -> str:
"""Return a string representation of the game."""
return self.display_name
class Organization(models.Model):
"""Represents an organization on Twitch that can own drop campaigns."""
id = models.TextField(primary_key=True)
name = models.TextField()
def __str__(self) -> str:
"""Return a string representation of the organization."""
return self.name
class DropCampaign(models.Model):
"""Represents a Twitch drop campaign."""
STATUS_CHOICES: ClassVar[list[tuple[str, str]]] = [
("ACTIVE", "Active"),
("UPCOMING", "Upcoming"),
("EXPIRED", "Expired"),
]
id = models.TextField(primary_key=True)
name = models.TextField()
description = models.TextField(blank=True)
details_url = models.URLField(max_length=500, blank=True, default="")
account_link_url = models.URLField(max_length=500, blank=True, default="")
image_url = models.URLField(max_length=500, blank=True, default="")
start_at = models.DateTimeField()
end_at = models.DateTimeField()
status = models.TextField(choices=STATUS_CHOICES)
is_account_connected = models.BooleanField(default=False)
# Foreign keys
game = models.ForeignKey(Game, on_delete=models.CASCADE, related_name="drop_campaigns")
owner = models.ForeignKey(Organization, on_delete=models.CASCADE, related_name="drop_campaigns")
# Tracking fields
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
def __str__(self) -> str:
"""Return a string representation of the drop campaign."""
return self.name
@property
def is_active(self) -> bool:
"""Check if the campaign is currently active."""
now = timezone.now()
return self.start_at <= now <= self.end_at and self.status == "ACTIVE"
class DropBenefit(models.Model):
"""Represents a benefit that can be earned from a drop."""
DISTRIBUTION_TYPES: ClassVar[list[tuple[str, str]]] = [
("DIRECT_ENTITLEMENT", "Direct Entitlement"),
("CODE", "Code"),
]
id = models.TextField(primary_key=True)
name = models.TextField()
image_asset_url = models.URLField(max_length=500, blank=True, default="")
created_at = models.DateTimeField()
entitlement_limit = models.PositiveIntegerField(default=1)
is_ios_available = models.BooleanField(default=False)
distribution_type = models.TextField(choices=DISTRIBUTION_TYPES)
# Foreign keys
game = models.ForeignKey(Game, on_delete=models.CASCADE, related_name="drop_benefits")
owner_organization = models.ForeignKey(Organization, on_delete=models.CASCADE, related_name="drop_benefits")
def __str__(self) -> str:
"""Return a string representation of the drop benefit."""
return self.name
class TimeBasedDrop(models.Model):
"""Represents a time-based drop in a drop campaign."""
id = models.TextField(primary_key=True)
name = models.TextField()
required_minutes_watched = models.PositiveIntegerField()
required_subs = models.PositiveIntegerField(default=0)
start_at = models.DateTimeField()
end_at = models.DateTimeField()
# Foreign keys
campaign = models.ForeignKey(DropCampaign, on_delete=models.CASCADE, related_name="time_based_drops")
benefits = models.ManyToManyField(DropBenefit, through="DropBenefitEdge", related_name="drops")
def __str__(self) -> str:
"""Return a string representation of the time-based drop."""
return self.name
class DropBenefitEdge(models.Model):
"""Represents the relationship between a TimeBasedDrop and a DropBenefit."""
drop = models.ForeignKey(TimeBasedDrop, on_delete=models.CASCADE)
benefit = models.ForeignKey(DropBenefit, on_delete=models.CASCADE)
entitlement_limit = models.PositiveIntegerField(default=1)
class Meta:
unique_together = ("drop", "benefit")
def __str__(self) -> str:
"""Return a string representation of the drop benefit edge."""
return f"{self.drop.name} - {self.benefit.name}"

13
twitch/urls.py Normal file
View file

@ -0,0 +1,13 @@
from __future__ import annotations
from django.urls import path
from twitch import views
app_name = "twitch"
urlpatterns = [
path("", views.dashboard, name="dashboard"),
path("campaigns/", views.DropCampaignListView.as_view(), name="campaign_list"),
path("campaigns/<str:pk>/", views.DropCampaignDetailView.as_view(), name="campaign_detail"),
]

View file

@ -1 +1,122 @@
# Create your views here.
from __future__ import annotations
from typing import TYPE_CHECKING, Any
from django.shortcuts import render
from django.utils import timezone
from django.views.generic import DetailView, ListView
from twitch.models import DropCampaign, Game, TimeBasedDrop
if TYPE_CHECKING:
from django.db.models import QuerySet
from django.http import HttpRequest, HttpResponse
class DropCampaignListView(ListView):
"""List view for drop campaigns."""
model = DropCampaign
template_name = "twitch/campaign_list.html"
context_object_name = "campaigns"
def get_queryset(self) -> QuerySet[DropCampaign]:
"""Get queryset of drop campaigns.
Returns:
QuerySet: Filtered drop campaigns.
"""
queryset = super().get_queryset()
status_filter = self.request.GET.get("status")
game_filter = self.request.GET.get("game")
# Filter by status if provided
if status_filter:
queryset = queryset.filter(status=status_filter)
# Filter by game if provided
if game_filter:
queryset = queryset.filter(game__id=game_filter)
return queryset.select_related("game", "owner").order_by("-start_at")
def get_context_data(self, **kwargs: Any) -> dict[str, Any]:
"""Add additional context data.
Args:
**kwargs: Additional arguments.
Returns:
dict: Context data.
"""
context = super().get_context_data(**kwargs)
# Add games for filtering
context["games"] = Game.objects.all()
# Add status options for filtering
context["status_options"] = [status[0] for status in DropCampaign.STATUS_CHOICES]
# Add selected filters
context["selected_status"] = self.request.GET.get("status", "")
context["selected_game"] = self.request.GET.get("game", "")
# Current time for active campaign highlighting
context["now"] = timezone.now()
return context
class DropCampaignDetailView(DetailView):
"""Detail view for a drop campaign."""
model = DropCampaign
template_name = "twitch/campaign_detail.html"
context_object_name = "campaign"
def get_context_data(self, **kwargs: Any) -> dict[str, Any]:
"""Add additional context data.
Args:
**kwargs: Additional arguments.
Returns:
dict: Context data.
"""
context = super().get_context_data(**kwargs)
# Add drops for this campaign with benefits preloaded
context["drops"] = (
TimeBasedDrop.objects.filter(campaign=self.get_object())
.select_related("campaign")
.prefetch_related("benefits")
.order_by("required_minutes_watched")
)
# Current time for active campaign highlighting
context["now"] = timezone.now()
return context
def dashboard(request: HttpRequest) -> HttpResponse:
"""Dashboard view showing active campaigns and progress.
Args:
request: The HTTP request.
Returns:
HttpResponse: The rendered dashboard template.
"""
# Get active campaigns
now = timezone.now()
active_campaigns = DropCampaign.objects.filter(start_at__lte=now, end_at__gte=now, status="ACTIVE").select_related("game", "owner")
return render(
request,
"twitch/dashboard.html",
{
"active_campaigns": active_campaigns,
"now": now,
},
)