TheLovinator/ttvdrops
1
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Scrape SunkwiBOT/twitch-drops-api for data
All checks were successful
Deploy to Server / deploy (push) Successful in 29s
Details

Browse source
This commit is contained in:
Joakim Hellsén 2026-06-14 17:35:24 +02:00
commit 3535d7d2dd
Signed by: Joakim Hellsén
SSH key fingerprint: SHA256:/9h/CsExpFp+PRhsfA0xznFx2CGfTT5R/kpuFfUgEQk
28 changed files with 4272 additions and 576 deletions
Download patch file Download diff file Expand all files Collapse all files

7
.pre-commit-config.yaml
View file

@ -27,7 +27,7 @@ repos:
args: [--target-version, "6.0"]
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.15.12
rev: v0.15.17
hooks:
- id: ruff-check
args: ["--fix", "--exit-non-zero-on-fix"]
@ -38,8 +38,3 @@ repos:
hooks:
- id: pyupgrade
args: ["--py311-plus"]
- repo: https://github.com/rhysd/actionlint
rev: v1.7.12
hooks:
- id: actionlint

1
.vscode/settings.json vendored
View file

@ -71,6 +71,7 @@
"sitewide",
"speculationrules",
"staticfiles",
"Sunkwi",
"testchannel",
"testpass",
"thelovinator",

1
config/tests/test_site_endpoint_smoke.py
View file

@ -1,7 +1,6 @@
from __future__ import annotations
from datetime import timedelta
from pathlib import Path
from typing import TYPE_CHECKING
from django.conf import settings

2
conftest.py
View file

@ -7,7 +7,7 @@ if TYPE_CHECKING:
from collections.abc import Generator
@pytest.fixture(autouse=True)
@pytest.fixture(autouse=True) # noqa: RUF076 — intentional: project-wide N+1 detection with @pytest.mark.no_zeal escape hatch
def use_zeal(request: pytest.FixtureRequest) -> Generator[None]:
"""Enable Zeal N+1 detection context for each pytest test.

65
core/views.py
View file

@ -533,7 +533,7 @@ def docs_rss_view(request: HttpRequest) -> HttpResponse:
# MARK: /debug/
def debug_view(request: HttpRequest) -> HttpResponse:
def debug_view(request: HttpRequest) -> HttpResponse: # noqa: PLR0914
"""Debug view showing potentially broken or inconsistent data.
Returns:
@ -651,6 +651,60 @@ def debug_view(request: HttpRequest) -> HttpResponse:
.order_by("game__display_name", "name"),
)
# ── Data source diff ────────────────────────────────────────────────
# Compare campaigns imported from different data sources
sunkibot_drop_ids: set[str] = set(
DropCampaign.objects.filter(
data_source="SunkwiBOT/twitch-drops-api",
).values_list("twitch_id", flat=True),
)
miner_drop_ids: set[str] = set(
DropCampaign.objects.filter(data_source="TwitchDropsMiner").values_list(
"twitch_id",
flat=True,
),
)
# Drops only in SunkwiBOT (not in TwitchDropsMiner)
sunkibot_only_drop_ids: set[str] = sunkibot_drop_ids - miner_drop_ids
# Drops only in TwitchDropsMiner (not in SunkwiBOT)
miner_only_drop_ids: set[str] = miner_drop_ids - sunkibot_drop_ids
sunkibot_only_drops: QuerySet[DropCampaign] = (
DropCampaign.objects
.filter(twitch_id__in=sunkibot_only_drop_ids)
.select_related("game")
.order_by("name")[:200]
)
miner_only_drops: QuerySet[DropCampaign] = (
DropCampaign.objects
.filter(twitch_id__in=miner_only_drop_ids)
.select_related("game")
.order_by("name")[:200]
)
# Same for reward campaigns
sunkibot_reward_ids: set[str] = set(
RewardCampaign.objects.filter(
data_source="SunkwiBOT/twitch-drops-api",
).values_list("twitch_id", flat=True),
)
miner_reward_ids: set[str] = set(
RewardCampaign.objects.filter(data_source="TwitchDropsMiner").values_list(
"twitch_id",
flat=True,
),
)
sunkibot_only_reward_ids: set[str] = sunkibot_reward_ids - miner_reward_ids
miner_only_reward_ids: set[str] = miner_reward_ids - sunkibot_reward_ids
sunkibot_only_rewards: QuerySet[RewardCampaign] = RewardCampaign.objects.filter(
twitch_id__in=sunkibot_only_reward_ids,
).order_by("name")[:200]
miner_only_rewards: QuerySet[RewardCampaign] = RewardCampaign.objects.filter(
twitch_id__in=miner_only_reward_ids,
).order_by("name")[:200]
context: dict[str, Any] = {
"now": now,
"games_without_owner": games_without_owner,
@ -662,6 +716,15 @@ def debug_view(request: HttpRequest) -> HttpResponse:
"active_missing_image": active_missing_image,
"operation_names_with_counts": operation_names_with_counts,
"campaigns_missing_dropcampaigndetails": campaigns_missing_dropcampaigndetails,
# Data source diff
"sunkibot_only_drops_count": len(sunkibot_only_drop_ids),
"miner_only_drops_count": len(miner_only_drop_ids),
"sunkibot_only_drops": sunkibot_only_drops,
"miner_only_drops": miner_only_drops,
"sunkibot_only_rewards_count": len(sunkibot_only_reward_ids),
"miner_only_rewards_count": len(miner_only_reward_ids),
"sunkibot_only_rewards": sunkibot_only_rewards,
"miner_only_rewards": miner_only_rewards,
}
seo_context: dict[str, Any] = _build_seo_context(

1
kick/tests/test_kick.py
View file

@ -35,7 +35,6 @@ if TYPE_CHECKING:
from django.test.client import _MonkeyPatchedWSGIResponse
from pytest_django.asserts import QuerySet
from kick.schemas import KickDropCampaignSchema
from kick.schemas import KickRewardSchema
# Minimal valid campaign fixture (single campaign, active status)

69
templates/core/debug.html
View file

@ -170,4 +170,73 @@
<p>None ✅</p>
{% endif %}
</section>
<section>
<h2>Data Source Diff — Drop Campaigns</h2>
<p>Comparing campaigns by <code>data_source</code> field in the DB.</p>
<h3>Only in SunkwiBOT/twitch-drops-api ({{ sunkibot_only_drops_count }})</h3>
{% if sunkibot_only_drops %}
<ul>
{% for c in sunkibot_only_drops %}
<li>
<a href="{% url 'twitch:campaign_detail' c.twitch_id %}">{{ c.name }}</a>
(Game: <a href="{% url 'twitch:game_detail' c.game.twitch_id %}">{{ c.game.display_name }}</a>)
</li>
{% endfor %}
</ul>
{% if sunkibot_only_drops_count > 200 %}
<p><em>… and {{ sunkibot_only_drops_count|add:"-200" }} more (showing first 200)</em></p>
{% endif %}
{% else %}
<p>None ✅</p>
{% endif %}
<h3>Only in TwitchDropsMiner ({{ miner_only_drops_count }})</h3>
{% if miner_only_drops %}
<ul>
{% for c in miner_only_drops %}
<li>
<a href="{% url 'twitch:campaign_detail' c.twitch_id %}">{{ c.name }}</a>
(Game: <a href="{% url 'twitch:game_detail' c.game.twitch_id %}">{{ c.game.display_name }}</a>)
</li>
{% endfor %}
</ul>
{% if miner_only_drops_count > 200 %}
<p><em>… and {{ miner_only_drops_count|add:"-200" }} more (showing first 200)</em></p>
{% endif %}
{% else %}
<p>None ✅</p>
{% endif %}
</section>
<section>
<h2>Data Source Diff — Reward Campaigns</h2>
<h3>Only in SunkwiBOT/twitch-drops-api ({{ sunkibot_only_rewards_count }})</h3>
{% if sunkibot_only_rewards %}
<ul>
{% for c in sunkibot_only_rewards %}
<li>
<a href="{% url 'twitch:reward_campaign_detail' c.twitch_id %}">{{ c.name }}</a>
</li>
{% endfor %}
</ul>
{% if sunkibot_only_rewards_count > 200 %}
<p><em>… and {{ sunkibot_only_rewards_count|add:"-200" }} more (showing first 200)</em></p>
{% endif %}
{% else %}
<p>None ✅</p>
{% endif %}
<h3>Only in TwitchDropsMiner ({{ miner_only_rewards_count }})</h3>
{% if miner_only_rewards %}
<ul>
{% for c in miner_only_rewards %}
<li>
<a href="{% url 'twitch:reward_campaign_detail' c.twitch_id %}">{{ c.name }}</a>
</li>
{% endfor %}
</ul>
{% if miner_only_rewards_count > 200 %}
<p><em>… and {{ miner_only_rewards_count|add:"-200" }} more (showing first 200)</em></p>
{% endif %}
{% else %}
<p>None ✅</p>
{% endif %}
</section>
{% endblock content %}

99
templates/twitch/campaign_detail.html
View file

@ -47,48 +47,65 @@
{% endif %}
<!-- Campaign description -->
<p>{{ campaign.description|linebreaksbr }}</p>
<div>
Published
<time datetime="{{ campaign.added_at|date:'c' }}"
title="{{ campaign.added_at|date:'DATETIME_FORMAT' }}">{{ campaign.added_at|date:"M d, Y H:i" }}</time> ({{ campaign.added_at|timesince }} ago)
</div>
<div>
Last updated
<time datetime="{{ campaign.updated_at|date:'c' }}"
title="{{ campaign.updated_at|date:'DATETIME_FORMAT' }}">{{ campaign.updated_at|date:"M d, Y H:i" }}</time> ({{ campaign.updated_at|timesince }} ago)
</div>
<!-- Campaign end times -->
<div>
{% if campaign.end_at < now %}
Ended
<dl>
<dt>Published</dt>
<dd>
<time datetime="{{ campaign.added_at|date:'c' }}"
title="{{ campaign.added_at|date:'DATETIME_FORMAT' }}">{{ campaign.added_at|date:"M d, Y H:i" }}</time> ({{ campaign.added_at|timesince }} ago)
</dd>
<dt>Last updated</dt>
<dd>
<time datetime="{{ campaign.updated_at|date:'c' }}"
title="{{ campaign.updated_at|date:'DATETIME_FORMAT' }}">{{ campaign.updated_at|date:"M d, Y H:i" }}</time> ({{ campaign.updated_at|timesince }} ago)
</dd>
<!-- Campaign end times -->
<dt>
{% if campaign.end_at < now %}
Ended
{% else %}
Ends in
{% endif %}
</dt>
<dd>
<time datetime="{{ campaign.end_at|date:'c' }}"
title="{{ campaign.end_at|date:'DATETIME_FORMAT' }}">{{ campaign.end_at|date:"M d, Y H:i" }}</time> ({{ campaign.end_at|timesince }} ago)
{% else %}
Ends in
<time datetime="{{ campaign.end_at|date:'c' }}"
title="{{ campaign.end_at|date:'DATETIME_FORMAT' }}">{{ campaign.end_at|date:"M d, Y H:i" }}</time> (in {{ campaign.end_at|timeuntil }})
{% endif %}
</div>
<!-- Campaign start times -->
<div>
{% if campaign.start_at > now %}
Starts in
title="{{ campaign.end_at|date:'DATETIME_FORMAT' }}">{{ campaign.end_at|date:"M d, Y H:i" }}</time>
{% if campaign.end_at < now %}
({{ campaign.end_at|timesince }} ago)
{% else %}
(in {{ campaign.end_at|timeuntil }})
{% endif %}
</dd>
<!-- Campaign start times -->
<dt>
{% if campaign.start_at > now %}
Starts in
{% else %}
Started
{% endif %}
</dt>
<dd>
<time datetime="{{ campaign.start_at|date:'c' }}"
title="{{ campaign.start_at|date:'DATETIME_FORMAT' }}">{{ campaign.start_at|date:"M d, Y H:i" }}</time> (in {{ campaign.start_at|timeuntil }})
{% else %}
Started
<time datetime="{{ campaign.start_at|date:'c' }}"
title="{{ campaign.start_at|date:'DATETIME_FORMAT' }}">{{ campaign.start_at|date:"M d, Y H:i" }}</time> ({{ campaign.start_at|timesince }} ago)
{% endif %}
</div>
<!-- Campaign duration -->
<div>
Duration is
<time datetime="{{ campaign.duration_iso }}"
title="{{ campaign.start_at|date:'DATETIME_FORMAT' }} to {{ campaign.end_at|date:'DATETIME_FORMAT' }}">
{{ campaign.end_at|timeuntil:campaign.start_at }}
</time>
</div>
title="{{ campaign.start_at|date:'DATETIME_FORMAT' }}">{{ campaign.start_at|date:"M d, Y H:i" }}</time>
{% if campaign.start_at > now %}
(in {{ campaign.start_at|timeuntil }})
{% else %}
({{ campaign.start_at|timesince }} ago)
{% endif %}
</dd>
<!-- Campaign duration -->
<dt>Duration</dt>
<dd>
<time datetime="{{ campaign.duration_iso }}"
title="{{ campaign.start_at|date:'DATETIME_FORMAT' }} to {{ campaign.end_at|date:'DATETIME_FORMAT' }}">
{{ campaign.end_at|timeuntil:campaign.start_at }}
</time>
</dd>
<!-- Data source -->
<dt>Data source</dt>
<dd>
{{ campaign.data_source }}
</dd>
</dl>
<!-- Buttons -->
<div>
<!-- Campaign Detail URL -->
@ -103,9 +120,9 @@
title="Atom feed for {{ campaign.game.display_name }} campaigns">[atom]</a>
<a href="{% url 'core:game_campaign_feed_discord' campaign.game.twitch_id %}"
title="Discord feed for {{ campaign.game.display_name }} campaigns">[discord]</a>
<a href="{% url 'core:docs_rss' %}" title="RSS feed documentation">[explain]</a>
<a href="{% url 'twitch:twitch-api-v1:get_campaign' campaign.twitch_id %}"
title="Twitch campaign API">[api]</a>
<a href="{% url 'core:docs_rss' %}" title="RSS feed documentation">[explain]</a>
{% endif %}
</div>
</div>

97
templates/twitch/reward_campaign_detail.html
View file

@ -50,58 +50,71 @@
</div>
<!-- Campaign Summary -->
{% if reward_campaign.summary %}<p id="campaign-summary">{{ reward_campaign.summary|linebreaksbr }}</p>{% endif %}
<!-- Campaign Status -->
<div>
<strong>Status:</strong>
{% if is_active %}
Active
{% elif reward_campaign.starts_at > now %}
Upcoming
{% else %}
Expired
{% endif %}
</div>
<div>Added at {{ reward_campaign.added_at|date:"M d, Y H:i" }} ({{ reward_campaign.added_at|timesince }} ago)</div>
<div>
Last updated at {{ reward_campaign.updated_at|date:"M d, Y H:i" }} ({{ reward_campaign.updated_at|timesince }} ago)
</div>
<!-- Reward Start Time -->
<div>
{% if reward_campaign.starts_at > now %}
Starts
{% else %}
Started at
{% endif %}
<time datetime="{{ reward_campaign.starts_at|date:'c' }}"
title="{{ reward_campaign.starts_at|date:'DATETIME_FORMAT' }}">
{{ reward_campaign.starts_at|date:"M d, Y H:i" }}
<dl>
<!-- Campaign Status -->
<dt>Status</dt>
<dd>
{% if is_active %}
Active
{% elif reward_campaign.starts_at > now %}
Upcoming
{% else %}
Expired
{% endif %}
</dd>
<dt>Added</dt>
<dd>
{{ reward_campaign.added_at|date:"M d, Y H:i" }} ({{ reward_campaign.added_at|timesince }} ago)
</dd>
<dt>Last updated</dt>
<dd>
{{ reward_campaign.updated_at|date:"M d, Y H:i" }} ({{ reward_campaign.updated_at|timesince }} ago)
</dd>
<!-- Reward Start Time -->
<dt>
{% if reward_campaign.starts_at > now %}
Starts
{% else %}
Started at
{% endif %}
</dt>
<dd>
<time datetime="{{ reward_campaign.starts_at|date:'c' }}"
title="{{ reward_campaign.starts_at|date:'DATETIME_FORMAT' }}">
{{ reward_campaign.starts_at|date:"M d, Y H:i" }}
</time>
{% if reward_campaign.starts_at > now %}
(in {{ reward_campaign.starts_at|timeuntil }})
{% else %}
({{ reward_campaign.starts_at|timesince }} ago)
{% endif %}
</time>
</div>
<!-- Reward End Time -->
<div>
{% if reward_campaign.ends_at > now %}
Ends
{% else %}
Ended at
{% endif %}
<time datetime="{{ reward_campaign.ends_at|date:'c' }}"
title="{{ reward_campaign.ends_at|date:'DATETIME_FORMAT' }}">
{{ reward_campaign.ends_at|date:"M d, Y H:i" }}
</dd>
<!-- Reward End Time -->
<dt>
{% if reward_campaign.ends_at > now %}
Ends
{% else %}
Ended at
{% endif %}
</dt>
<dd>
<time datetime="{{ reward_campaign.ends_at|date:'c' }}"
title="{{ reward_campaign.ends_at|date:'DATETIME_FORMAT' }}">
{{ reward_campaign.ends_at|date:"M d, Y H:i" }}
</time>
{% if reward_campaign.ends_at > now %}
(in {{ reward_campaign.ends_at|timeuntil }})
{% else %}
({{ reward_campaign.ends_at|timesince }} ago)
{% endif %}
</time>
</div>
<div>
{% if reward_campaign.instructions %}{{ reward_campaign.instructions|linebreaksbr }}{% endif %}
</div>
</dd>
<!-- Data source -->
<dt>Data source</dt>
<dd>
{{ reward_campaign.data_source }}
</dd>
</dl>
<div>{% if reward_campaign.instructions %}{{ reward_campaign.instructions|linebreaksbr }}{% endif %}</div>
<!-- RSS Feeds -->
<div style="margin-bottom: 1rem;">
<a href="{% url 'core:reward_campaign_feed' %}"

147
templates/twitch/reward_campaign_list.html
View file

@ -23,10 +23,10 @@
<div>
<a href="{% url 'twitch:dashboard' %}">Twitch</a> > Reward Campaigns
</div>
<p>This is an archive of old Twitch reward campaigns because we currently do not scrape them.</p>
<p>
Feel free to submit a pull request on <a href="https://github.com/TheLovinator1/ttvdrops">GitHub</a>
with a working implementation :-).
Data is scraped from <a href="https://github.com/SunkwiBOT/twitch-drops-api"
target="_blank"
rel="noopener">SunkwiBOT/twitch-drops-api</a> every 15 minutes.
</p>
<!-- RSS Feeds -->
<div>
@ -36,84 +36,81 @@
title="Atom feed for all reward campaigns">[atom]</a>
<a href="{% url 'core:reward_campaign_feed_discord' %}"
title="Discord feed for all reward campaigns">[discord]</a>
<a href="{% url 'core:docs_rss' %}" title="RSS feed documentation">[explain]</a>
<a href="{% url 'twitch:twitch-api-v1:list_reward_campaigns' %}"
title="Twitch reward campaigns API">[api]</a>
<a href="{% url 'core:docs_rss' %}" title="RSS feed documentation">[explain]</a>
</div>
{% if reward_campaigns %}
{% comment %}
<h5>Active Reward Campaigns</h5>
<table>
<tbody>
{% for campaign in reward_campaigns %}
{% if campaign.starts_at <= now and campaign.ends_at >= now %}
<tr>
<td>
<a href="{% url 'twitch:reward_campaign_detail' campaign.twitch_id %}">
{% if campaign.brand %}
{{ campaign.brand }}: {{ campaign.name }}
{% else %}
{{ campaign.name }}
{% with active_campaigns=reward_campaigns|dictsortreversed:"ends_at" %}
<h5>Active Reward Campaigns</h5>
<table>
<tbody>
{% for campaign in active_campaigns %}
{% if campaign.starts_at <= now and campaign.ends_at >= now %}
<tr>
<td>
<a href="{% url 'twitch:reward_campaign_detail' campaign.twitch_id %}">
{% if campaign.brand %}
{{ campaign.brand }}: {{ campaign.name }}
{% else %}
{{ campaign.name }}
{% endif %}
</a>
{% if campaign.summary %}
<br />
<small>{{ campaign.summary }}</small>
{% endif %}
</a>
{% if campaign.summary %}
<br />
<small>{{ campaign.summary }}</small>
{% endif %}
</td>
<td>
{% if campaign.game %}
<a href="{% url 'twitch:game_detail' campaign.game.twitch_id %}">{{ campaign.game.display_name }}</a>
{% elif campaign.is_sitewide %}
Site-wide
{% endif %}
</td>
<td>
<span title="{{ campaign.ends_at|date:'M d, Y H:i' }}">Ends in {{ campaign.ends_at|timeuntil }}</span>
</td>
</tr>
{% endif %}
{% endfor %}
</tbody>
</table>
{% endcomment %}
{% comment %}
<h5>Upcoming Reward Campaigns</h5>
<table>
<tbody>
{% for campaign in reward_campaigns %}
{% if campaign.starts_at > now %}
<tr>
<td>
<a href="{% url 'twitch:reward_campaign_detail' campaign.twitch_id %}">
{% if campaign.brand %}
{{ campaign.brand }}: {{ campaign.name }}
{% else %}
{{ campaign.name }}
</td>
<td>
{% if campaign.game %}
<a href="{% url 'twitch:game_detail' campaign.game.twitch_id %}">{{ campaign.game.display_name }}</a>
{% elif campaign.is_sitewide %}
Site-wide
{% endif %}
</a>
{% if campaign.summary %}
<br />
<small>{{ campaign.summary }}</small>
{% endif %}
</td>
<td>
{% if campaign.game %}
<a href="{% url 'twitch:game_detail' campaign.game.twitch_id %}">{{ campaign.game.display_name }}</a>
{% elif campaign.is_sitewide %}
Site-wide
{% endif %}
</td>
<td>
<span title="Starts on {{ campaign.starts_at|date:'M d, Y H:i' }}">Starts in {{ campaign.starts_at|timeuntil }}</span>
</td>
</tr>
{% endif %}
{% endfor %}
</tbody>
</table>
{% endcomment %}
</td>
<td>
<span title="{{ campaign.ends_at|date:'M d, Y H:i' }}">Ends in {{ campaign.ends_at|timeuntil }}</span>
</td>
</tr>
{% endif %}
{% endfor %}
</tbody>
</table>
<h5>Upcoming Reward Campaigns</h5>
<table>
<tbody>
{% for campaign in active_campaigns %}
{% if campaign.starts_at > now %}
<tr>
<td>
<a href="{% url 'twitch:reward_campaign_detail' campaign.twitch_id %}">
{% if campaign.brand %}
{{ campaign.brand }}: {{ campaign.name }}
{% else %}
{{ campaign.name }}
{% endif %}
</a>
{% if campaign.summary %}
<br />
<small>{{ campaign.summary }}</small>
{% endif %}
</td>
<td>
{% if campaign.game %}
<a href="{% url 'twitch:game_detail' campaign.game.twitch_id %}">{{ campaign.game.display_name }}</a>
{% elif campaign.is_sitewide %}
Site-wide
{% endif %}
</td>
<td>
<span title="Starts on {{ campaign.starts_at|date:'M d, Y H:i' }}">Starts in {{ campaign.starts_at|timeuntil }}</span>
</td>
</tr>
{% endif %}
{% endfor %}
</tbody>
</table>
{% endwith %}
<h5>Past Reward Campaigns</h5>
<table>
<tbody>

364
tools/extract_historical_drops.py Executable file
View file

@ -0,0 +1,364 @@
#!/usr/bin/env python3
"""Extract all unique historical versions of drops.json/rewards.json from the SunkwiBOT/twitch-drops-api repo and merge them into a single JSON file.
For each campaign (identified by its ``id``), channels from ALL commits are
accumulated and deduplicated. The output file can be imported in a single
ORM pass via ``--from-file``, which is much faster than iterating 14k+
git commits through Django.
Usage:
python tools/extract_historical_drops.py [--rewards] [--git-dir DIR] [--output FILE]
"""
from __future__ import annotations
import argparse
import hashlib
import json
import shutil
import subprocess # noqa: S404
import tempfile
from pathlib import Path
from typing import Any
REPO_URL = "https://github.com/SunkwiBOT/twitch-drops-api"
# Fields that should be taken from the *first* occurrence of a campaign
# (game metadata, names, descriptions, timestamps, etc.)
CAMPAIGN_FIRST_FIELDS = {
"gameId",
"gameDisplayName",
"gameBoxArtURL",
"startAt",
"endAt",
}
# Fields within individual rewards that should be first-occurrence only
REWARD_FIRST_FIELDS: set[str] = set()
# Fields that should be *merged* across all occurrences (channels, time-based drops)
REWARD_MERGE_FIELDS = {"channels"} # mapped from allow.channels
def run_git(git_dir: str, *args: str, check: bool = True) -> str:
"""Run a git command and return stdout.
Args:
git_dir: Path to the git repository.
*args: Git command arguments.
check: If True, raise on non-zero exit.
Returns:
The stdout output of the git command.
Raises:
subprocess.CalledProcessError: If the command fails and check is True.
FileNotFoundError: If the git executable is not found.
"""
try:
git: str | None = shutil.which("git")
if not git:
msg = "Git executable not found in PATH."
raise FileNotFoundError(msg)
result: subprocess.CompletedProcess[str] = subprocess.run( # noqa: S603
[git, "--git-dir", git_dir, *args],
capture_output=True,
text=True,
check=check,
)
except subprocess.CalledProcessError:
if check:
raise
return ""
else:
return result.stdout
def extract(
git_dir: str,
file_path: str,
*,
skip_latest: bool = True,
) -> list[dict[str, Any]]:
"""Extract and merge all historical versions of *file_path* from the repo.
Returns:
A list of top-level entries (drop groups or reward campaigns)
with channels accumulated across all commits.
"""
log_output = run_git(
git_dir,
"log",
"--all",
"--format=%H %ct",
"--follow",
"--reverse",
"--",
file_path,
)
lines = [ln.strip() for ln in log_output.strip().split("\n") if ln.strip()]
if skip_latest and lines:
lines = lines[:-1]
# Per-campaign-id accumulators
# Structure: {twitch_id: {"_first_data": {...}, "_channels": {channel_id: channel_data}}}
campaigns: dict[str, dict[str, Any]] = {}
# For drops.json, track groups separately
groups: dict[str, dict[str, Any]] = {}
total_processed = 0
seen_hashes: set[str] = set()
for line in lines:
sha = line.split(None, 1)[0]
try:
content = run_git(git_dir, "show", f"{sha}:{file_path}", check=True)
except subprocess.CalledProcessError:
continue
content_hash = hashlib.sha256(content.encode()).hexdigest()
if content_hash in seen_hashes:
continue
seen_hashes.add(content_hash)
try:
data = json.loads(content)
except json.JSONDecodeError:
continue
if not isinstance(data, list):
continue
if file_path == "drops.json":
_merge_drops_version(data, groups, campaigns)
else:
_merge_rewards_version(data, campaigns)
total_processed += 1
# Build final output from accumulated data
result: list[dict[str, Any]] = []
if file_path == "drops.json":
# Sort groups by their startAt for stable output
sorted_groups = sorted(groups.values(), key=lambda g: g.get("startAt", ""))
for group in sorted_groups:
group_rewards = [
_build_reward_output(campaigns[cid])
for cid in group.get("_reward_ids", [])
if cid in campaigns
]
if group_rewards:
entry = {k: v for k, v in group.items() if not k.startswith("_")}
entry["rewards"] = group_rewards
result.append(entry)
else:
sorted_campaigns = sorted(
campaigns.values(),
key=lambda c: c.get("_first_data", {}).get("startAt", ""),
)
result.extend(_build_reward_output(camp) for camp in sorted_campaigns)
return result
def _merge_drops_version(
data: list[dict[str, Any]],
groups: dict[str, dict[str, Any]],
campaigns: dict[str, dict[str, Any]],
) -> None:
"""Merge a single commit's drops.json data into the accumulators."""
for item in data:
game_id = item.get("gameId", "")
if game_id not in groups:
groups[game_id] = {k: v for k, v in item.items() if k != "rewards"}
groups[game_id]["_reward_ids"] = []
for reward in item.get("rewards", []):
rid = reward.get("id", "")
if not rid:
continue
if rid not in campaigns:
# First occurrence: store all fields
campaigns[rid] = {
"_first_data": dict(reward),
"_channels": {},
}
groups[game_id]["_reward_ids"].append(rid)
else:
# Subsequent occurrence: merge metadata + channels
_merge_metadata(campaigns[rid], reward)
_merge_channels(campaigns[rid], reward)
# Always merge channels from every commit
_merge_channels(campaigns[rid], reward)
def _merge_rewards_version(
data: list[dict[str, Any]],
campaigns: dict[str, dict[str, Any]],
) -> None:
"""Merge a single commit's rewards.json data into the accumulators."""
for reward in data:
rid = reward.get("id", "")
if not rid:
continue
if rid not in campaigns:
campaigns[rid] = {
"_first_data": dict(reward),
"_channels": {},
}
else:
# Subsequent occurrence: merge metadata
_merge_metadata(campaigns[rid], reward)
_merge_channels(campaigns[rid], reward)
def _merge_metadata(
campaign: dict[str, Any],
reward_data: dict[str, Any],
) -> None:
"""Fill empty/missing fields in ``_first_data`` with values from later commits.
For each top-level field in *reward_data*, if the stored value in
``_first_data`` is empty/None and the new value is not, update it.
This ensures campaigns get the best available description, dates,
image URL, etc. across all historical versions.
"""
first = campaign["_first_data"]
# Top-level string/primitive fields to merge (first non-empty wins)
for key in (
"name",
"description",
"imageURL",
"detailsURL",
"accountLinkURL",
"startAt",
"endAt",
"status",
):
old_val = first.get(key)
new_val = reward_data.get(key)
if not old_val and new_val:
first[key] = new_val
# Nested fields: allow.isEnabled
old_allow = first.get("allow")
new_allow = reward_data.get("allow")
if (
isinstance(old_allow, dict)
and isinstance(new_allow, dict)
and not old_allow.get("isEnabled")
and new_allow.get("isEnabled")
):
old_allow["isEnabled"] = new_allow["isEnabled"]
def _merge_channels(
campaign: dict[str, Any],
reward_data: dict[str, Any],
) -> None:
"""Merge channels from *reward_data* into the accumulated *campaign*."""
allow = reward_data.get("allow")
if not allow or not isinstance(allow, dict):
return
channels = allow.get("channels")
if not channels or not isinstance(channels, list):
return
for ch in channels:
ch_id = ch.get("id", "")
if ch_id and ch_id not in campaign["_channels"]:
campaign["_channels"][ch_id] = dict(ch)
def _build_reward_output(campaign: dict[str, Any]) -> dict[str, Any]:
"""Build the final reward dict from accumulated data.
Returns:
The merged reward dict with all accumulated channels.
"""
first = campaign["_first_data"]
# Start with all first-occurrence data
result = dict(first)
# Build the allow field with merged channels
merged_channels = list(campaign["_channels"].values())
if merged_channels or "allow" in first:
allow = (
dict(first.get("allow", {})) if isinstance(first.get("allow"), dict) else {}
)
allow["channels"] = merged_channels
allow["isEnabled"] = (
first.get("allow", {}).get("isEnabled", True)
if isinstance(first.get("allow"), dict)
else True
)
result["allow"] = allow
return result
def main() -> None:
"""Entry point: clone repo, extract historical data, write output JSON."""
parser = argparse.ArgumentParser(
description="Extract historical drops data from SunkwiBOT/twitch-drops-api",
)
parser.add_argument(
"--rewards",
action="store_true",
help="Extract rewards.json instead of drops.json",
)
parser.add_argument(
"--git-dir",
type=str,
default="",
help="Path to an existing bare clone (avoids re-cloning)",
)
parser.add_argument(
"--output",
type=str,
default="",
help="Output file path (default: drops_merged.json or rewards_merged.json)",
)
args = parser.parse_args()
file_path = "rewards.json" if args.rewards else "drops.json"
default_output = "rewards_merged.json" if args.rewards else "drops_merged.json"
output_path = args.output or default_output
git_dir = args.git_dir
cleanup = False
tmp: Path | None = None
if not git_dir:
tmp = Path(tempfile.mkdtemp())
run_git(
str(tmp),
"clone",
"--filter=blob:none",
"--bare",
REPO_URL,
str(tmp / "repo.git"),
check=True,
)
git_dir = str(tmp / "repo.git")
cleanup = True
try:
result = extract(git_dir, file_path)
with Path(output_path).open("w", encoding="utf-8") as f:
json.dump(result, f, indent=2)
finally:
if cleanup and tmp is not None:
shutil.rmtree(str(tmp), ignore_errors=True)
if __name__ == "__main__":
main()

20
twitch/api.py
View file

@ -409,7 +409,8 @@ def _serialize_game_summary(
display_name=game.display_name,
box_art_url=_game_box_art_url(game),
organizations=[
_serialize_organization_summary(org) for org in game.owners.all()
_serialize_organization_summary(org)
for org in getattr(game, "owners_for_detail", game.owners.all())
],
campaign_count=_game_campaign_count(game),
active_campaign_count=_game_active_campaign_count(game, now),
@ -424,7 +425,8 @@ def _serialize_game(game: Game, now: datetime.datetime) -> V1GameSchema:
display_name=game.display_name,
box_art_url=_game_box_art_url(game),
organizations=[
_serialize_organization_summary(org) for org in game.owners.all()
_serialize_organization_summary(org)
for org in getattr(game, "owners_for_detail", game.owners.all())
],
campaign_count=_game_campaign_count(game),
active_campaign_count=_game_active_campaign_count(game, now),
@ -520,7 +522,11 @@ def _serialize_campaign_detail(
operation_names=campaign.operation_names,
allowed_channels=[
_serialize_channel_summary(channel)
for channel in campaign.allow_channels.all()
for channel in getattr(
campaign,
"channels_ordered",
campaign.allow_channels.all(),
)
],
drops=[_serialize_drop(drop) for drop in campaign.time_based_drops.all()], # pyright: ignore[reportAttributeAccessIssue]
)
@ -643,7 +649,8 @@ def get_game(request: HttpRequest, twitch_id: str) -> V1GameDetailSchema:
display_name=game.display_name,
box_art_url=_game_box_art_url(game),
organizations=[
_serialize_organization_summary(org) for org in game.owners.all()
_serialize_organization_summary(org)
for org in getattr(game, "owners_for_detail", game.owners.all())
],
campaign_count=_game_campaign_count(game),
active_campaign_count=_game_active_campaign_count(game, now),
@ -700,7 +707,10 @@ def get_organization(
name=org.name,
added_at=org.added_at,
updated_at=org.updated_at,
games=[_serialize_game_summary(game, now) for game in org.games.all()], # pyright: ignore[reportAttributeAccessIssue]
games=[
_serialize_game_summary(game, now)
for game in getattr(org, "games_for_detail", org.games.all()) # pyright: ignore[reportAttributeAccessIssue]
],
campaigns=[_serialize_campaign_detail(campaign, now) for campaign in campaigns],
)

6
twitch/apps.py
View file

@ -21,7 +21,7 @@ class TwitchConfig(AppConfig):
# Patch FieldFile.open to swallow FileNotFoundError and provide
# an empty in-memory file-like object so image dimension
# calculations don't crash when the on-disk file was removed.
try:
if hasattr(FieldFile, "open"):
orig_open: Callable[..., FieldFile] = FieldFile.open
def _safe_open(self: FieldFile, mode: str = "rb") -> FieldFile:
@ -34,8 +34,8 @@ class TwitchConfig(AppConfig):
return self
FieldFile.open = _safe_open
except (AttributeError, TypeError) as exc:
logger.debug("Failed to patch FieldFile.open: %s", exc)
else:
logger.debug("FieldFile has no 'open' attribute; skipping patch")
# Register post_save signal handlers that dispatch image download tasks
# when new Twitch records are created.

1
twitch/feeds.py
View file

@ -1,6 +1,5 @@
import logging
import re
from collections.abc import Callable
from typing import TYPE_CHECKING
from typing import Literal

63
twitch/management/commands/backfill_image_dimensions.py
View file

@ -26,23 +26,25 @@ class Command(BaseCommand):
or not game.box_art_mime_type
or not game.box_art_width
)
if game.box_art_file and needs_update:
try:
if not game.box_art_width:
game.box_art_file.open()
if not (game.box_art_file and needs_update):
continue
# populate size and mime if available
with contextlib.suppress(Exception):
game.box_art_size_bytes = game.box_art_file.size
mime, _ = mimetypes.guess_type(game.box_art_file.name or "")
if mime:
game.box_art_mime_type = mime
try:
if not game.box_art_width:
game.box_art_file.open()
with contextlib.suppress(Exception):
game.box_art_size_bytes = game.box_art_file.size
except (OSError, ValueError, AttributeError) as exc:
self.stdout.write(self.style.ERROR(f" Failed {game}: {exc}"))
continue
game.save()
total_updated += 1
self.stdout.write(self.style.SUCCESS(f" Updated {game}"))
except (OSError, ValueError, AttributeError) as exc:
self.stdout.write(self.style.ERROR(f" Failed {game}: {exc}"))
mime, _ = mimetypes.guess_type(game.box_art_file.name or "")
if mime:
game.box_art_mime_type = mime
game.save()
total_updated += 1
self.stdout.write(self.style.SUCCESS(f" Updated {game}"))
# Update DropCampaign images
self.stdout.write("Processing DropCampaign image_file...")
@ -52,22 +54,25 @@ class Command(BaseCommand):
or not campaign.image_mime_type
or not campaign.image_width
)
if campaign.image_file and needs_update:
try:
if not campaign.image_width:
campaign.image_file.open()
if not (campaign.image_file and needs_update):
continue
with contextlib.suppress(Exception):
campaign.image_size_bytes = campaign.image_file.size
mime, _ = mimetypes.guess_type(campaign.image_file.name or "")
if mime:
campaign.image_mime_type = mime
try:
if not campaign.image_width:
campaign.image_file.open()
with contextlib.suppress(Exception):
campaign.image_size_bytes = campaign.image_file.size
except (OSError, ValueError, AttributeError) as exc:
self.stdout.write(self.style.ERROR(f" Failed {campaign}: {exc}"))
continue
campaign.save()
total_updated += 1
self.stdout.write(self.style.SUCCESS(f" Updated {campaign}"))
except (OSError, ValueError, AttributeError) as exc:
self.stdout.write(self.style.ERROR(f" Failed {campaign}: {exc}"))
mime, _ = mimetypes.guess_type(campaign.image_file.name or "")
if mime:
campaign.image_mime_type = mime
campaign.save()
total_updated += 1
self.stdout.write(self.style.SUCCESS(f" Updated {campaign}"))
# Update DropBenefit images
self.stdout.write("Processing DropBenefit image_file...")

590
twitch/management/commands/better_import_drops.py
View file

@ -1,7 +1,6 @@
import json
import os
import sys
from collections.abc import Mapping
from datetime import UTC
from datetime import datetime
from pathlib import Path
@ -340,7 +339,41 @@ def extract_operation_name_from_parsed(
return None
def repair_partially_broken_json(raw_text: str) -> str: # noqa: PLR0915
def _validate_repair_result(
fixed: str,
parsed_data: dict | list | str,
) -> str | None:
"""Validate that a repaired JSON result contains valid GraphQL responses.
Args:
fixed: The repaired JSON string.
parsed_data: The parsed JSON data.
Returns:
The string to return if valid, or None to continue trying other strategies.
"""
# If it's a list, validate all items are GraphQL responses
if isinstance(parsed_data, list):
# Filter to only keep GraphQL responses
filtered = [
item
for item in parsed_data
if isinstance(item, dict) and ("data" in item or "extensions" in item)
]
if filtered:
# If we filtered anything out, return the filtered version
if len(filtered) < len(parsed_data):
return json.dumps(filtered)
# Otherwise return as-is
return fixed
# Single dict - check if it's a GraphQL response
elif isinstance(parsed_data, dict):
if "data" in parsed_data or "extensions" in parsed_data:
return fixed
return None
def repair_partially_broken_json(raw_text: str) -> str:
"""Attempt to repair partially broken JSON with multiple fallback strategies.
Handles "half-bad" JSON by:
@ -357,27 +390,10 @@ def repair_partially_broken_json(raw_text: str) -> str: # noqa: PLR0915
# Strategy 1: Direct repair attempt
try:
fixed: str = json_repair.repair_json(raw_text, logging=False)
# Validate it produces valid JSON
parsed_data = json.loads(fixed)
# If it's a list, validate all items are GraphQL responses
if isinstance(parsed_data, list):
# Filter to only keep GraphQL responses
filtered = [
item
for item in parsed_data
if isinstance(item, dict) and ("data" in item or "extensions" in item)
]
if filtered:
# If we filtered anything out, return the filtered version
if len(filtered) < len(parsed_data):
return json.dumps(filtered)
# Otherwise return as-is
return fixed
# Single dict - check if it's a GraphQL response
elif isinstance(parsed_data, dict):
if "data" in parsed_data or "extensions" in parsed_data:
return fixed
result: str | None = _validate_repair_result(fixed, parsed_data)
if result is not None:
return result
except ValueError, TypeError, json.JSONDecodeError:
pass
@ -1142,30 +1158,92 @@ class Command(BaseCommand):
f"{Fore.GREEN}✓{Style.RESET_ALL} {action} reward campaign: {display_name}",
)
def handle(self, *args, **options) -> None: # noqa: ARG002
"""Main entry point for the command.
def _dispatch_path_processing(
self,
input_path: Path,
options: dict,
) -> None:
"""Dispatch processing based on whether the path is a file or directory.
Args:
input_path: Resolved path to process.
options: Command options.
Raises:
CommandError: If the provided path does not exist.
CommandError: If the path does not exist.
"""
if input_path.is_file():
self.process_file(file_path=input_path, options=options)
elif input_path.is_dir():
self.process_json_files(input_path=input_path, options=options)
else:
msg: str = f"Path does not exist: {input_path}"
raise CommandError(msg)
def handle(self, *args, **options) -> None: # noqa: ARG002
"""Main entry point for the command."""
colorama_init(autoreset=True)
input_path: Path = Path(options["path"]).resolve()
try:
if input_path.is_file():
self.process_file(file_path=input_path, options=options)
elif input_path.is_dir():
self.process_json_files(input_path=input_path, options=options)
else:
msg: str = f"Path does not exist: {input_path}"
raise CommandError(msg)
self._dispatch_path_processing(input_path, options)
except KeyboardInterrupt:
tqdm.write(self.style.WARNING("\n\nInterrupted by user!"))
tqdm.write(self.style.WARNING("Shutting down gracefully..."))
sys.exit(130) # 128 + 2 (Keyboard Interrupt)
def _process_single_file_result(
self,
file_path: Path,
options: dict,
progress_bar: tqdm,
) -> tuple[int, int, int]:
"""Process a single file result and update progress output.
Args:
file_path: Path to the JSON file.
options: Command options.
progress_bar: tqdm progress bar instance.
Returns:
Tuple of (success_increment, failed_increment, error_increment).
"""
try:
result: dict[str, bool | str] = self.process_file_worker(
file_path=file_path,
options=options,
)
except (OSError, ValueError, KeyError) as e:
progress_bar.write(
f"{Fore.RED}✗{Style.RESET_ALL} {file_path.name} (error: {e})",
)
return 0, 0, 1
if result["success"]:
if options.get("verbose"):
progress_bar.write(
f"{Fore.GREEN}✓{Style.RESET_ALL} {file_path.name}",
)
return 1, 0, 0
reason: bool | str | None = (
result.get("reason") if isinstance(result, dict) else None
)
if reason:
progress_bar.write(
f"{Fore.RED}✗{Style.RESET_ALL} "
f"{file_path.name} → {result['broken_dir']}/"
f"{file_path.name} ({reason})",
)
else:
progress_bar.write(
f"{Fore.RED}✗{Style.RESET_ALL} "
f"{file_path.name} → {result['broken_dir']}/"
f"{file_path.name}",
)
return 0, 1, 0
def process_json_files(self, input_path: Path, options: dict) -> None:
"""Process multiple JSON files in a directory.
@ -1191,39 +1269,14 @@ class Command(BaseCommand):
dynamic_ncols=True,
) as progress_bar:
for file_path in json_files:
try:
result: dict[str, bool | str] = self.process_file_worker(
file_path=file_path,
options=options,
)
if result["success"]:
success_count += 1
if options.get("verbose"):
progress_bar.write(
f"{Fore.GREEN}✓{Style.RESET_ALL} {file_path.name}",
)
else:
failed_count += 1
reason: bool | str | None = (
result.get("reason") if isinstance(result, dict) else None
)
if reason:
progress_bar.write(
f"{Fore.RED}✗{Style.RESET_ALL} "
f"{file_path.name} → {result['broken_dir']}/"
f"{file_path.name} ({reason})",
)
else:
progress_bar.write(
f"{Fore.RED}✗{Style.RESET_ALL} "
f"{file_path.name} → {result['broken_dir']}/"
f"{file_path.name}",
)
except (OSError, ValueError, KeyError) as e:
error_count += 1
progress_bar.write(
f"{Fore.RED}✗{Style.RESET_ALL} {file_path.name} (error: {e})",
)
s_inc, f_inc, e_inc = self._process_single_file_result(
file_path=file_path,
options=options,
progress_bar=progress_bar,
)
success_count += s_inc
failed_count += f_inc
error_count += e_inc
# Update postfix with statistics
progress_bar.set_postfix_str(
@ -1385,6 +1438,111 @@ class Command(BaseCommand):
return []
return []
def _run_worker_inner(
self,
file_path: Path,
options: dict,
) -> dict[str, bool | str]:
"""Inner processing logic for the file worker.
Args:
file_path: Path to the JSON file to process.
options: Command options.
Returns:
Dict with success status and optional broken_dir path.
"""
raw_text: str = file_path.read_text(encoding="utf-8", errors="ignore")
# Repair potentially broken JSON with multiple fallback strategies
fixed_json_str: str = repair_partially_broken_json(raw_text)
parsed_json: (
JSONReturnType | tuple[JSONReturnType, list[dict[str, str]]] | str
) = json.loads(fixed_json_str)
operation_name: str | None = extract_operation_name_from_parsed(parsed_json)
# Check for error-only responses first
error_description: str | None = detect_error_only_response(parsed_json)
if error_description:
if not options.get("skip_broken_moves"):
broken_dir: Path | None = move_file_to_broken_subdir(
file_path,
error_description,
operation_name=operation_name,
)
return {
"success": False,
"broken_dir": str(broken_dir),
"reason": error_description,
}
return {
"success": False,
"broken_dir": "(skipped)",
"reason": error_description,
}
matched: str | None = detect_non_campaign_keyword(raw_text)
if matched:
if not options.get("skip_broken_moves"):
broken_dir: Path | None = move_file_to_broken_subdir(
file_path,
matched,
operation_name=operation_name,
)
return {
"success": False,
"broken_dir": str(broken_dir),
"reason": f"matched '{matched}'",
}
return {
"success": False,
"broken_dir": "(skipped)",
"reason": f"matched '{matched}'",
}
if "dropCampaign" not in raw_text:
if not options.get("skip_broken_moves"):
broken_dir: Path | None = move_file_to_broken_subdir(
file_path,
"no_dropCampaign",
operation_name=operation_name,
)
return {
"success": False,
"broken_dir": str(broken_dir),
"reason": "no dropCampaign present",
}
return {
"success": False,
"broken_dir": "(skipped)",
"reason": "no dropCampaign present",
}
# Normalize and filter to dict responses only
responses: list[dict[str, Any]] = self._normalize_responses(parsed_json)
processed, broken_dir = self.process_responses(
responses=responses,
file_path=file_path,
options=options,
)
if not processed:
# File was already moved to broken during validation
return {
"success": False,
"broken_dir": str(broken_dir) if broken_dir else "(unknown)",
"reason": "validation failed",
}
campaign_structure: str | None = self._detect_campaign_structure(
responses[0] if responses else {},
)
move_completed_file(
file_path=file_path,
operation_name=operation_name,
campaign_structure=campaign_structure,
)
return {"success": True}
def process_file_worker(
self,
file_path: Path,
@ -1404,96 +1562,7 @@ class Command(BaseCommand):
json.JSONDecodeError: If the JSON file cannot be parsed
"""
try:
raw_text: str = file_path.read_text(encoding="utf-8", errors="ignore")
# Repair potentially broken JSON with multiple fallback strategies
fixed_json_str: str = repair_partially_broken_json(raw_text)
parsed_json: (
JSONReturnType | tuple[JSONReturnType, list[dict[str, str]]] | str
) = json.loads(fixed_json_str)
operation_name: str | None = extract_operation_name_from_parsed(parsed_json)
# Check for error-only responses first
error_description: str | None = detect_error_only_response(parsed_json)
if error_description:
if not options.get("skip_broken_moves"):
broken_dir: Path | None = move_file_to_broken_subdir(
file_path,
error_description,
operation_name=operation_name,
)
return {
"success": False,
"broken_dir": str(broken_dir),
"reason": error_description,
}
return {
"success": False,
"broken_dir": "(skipped)",
"reason": error_description,
}
matched: str | None = detect_non_campaign_keyword(raw_text)
if matched:
if not options.get("skip_broken_moves"):
broken_dir: Path | None = move_file_to_broken_subdir(
file_path,
matched,
operation_name=operation_name,
)
return {
"success": False,
"broken_dir": str(broken_dir),
"reason": f"matched '{matched}'",
}
return {
"success": False,
"broken_dir": "(skipped)",
"reason": f"matched '{matched}'",
}
if "dropCampaign" not in raw_text:
if not options.get("skip_broken_moves"):
broken_dir: Path | None = move_file_to_broken_subdir(
file_path,
"no_dropCampaign",
operation_name=operation_name,
)
return {
"success": False,
"broken_dir": str(broken_dir),
"reason": "no dropCampaign present",
}
return {
"success": False,
"broken_dir": "(skipped)",
"reason": "no dropCampaign present",
}
# Normalize and filter to dict responses only
responses: list[dict[str, Any]] = self._normalize_responses(parsed_json)
processed, broken_dir = self.process_responses(
responses=responses,
file_path=file_path,
options=options,
)
if not processed:
# File was already moved to broken during validation
return {
"success": False,
"broken_dir": str(broken_dir) if broken_dir else "(unknown)",
"reason": "validation failed",
}
campaign_structure: str | None = self._detect_campaign_structure(
responses[0] if responses else {},
)
move_completed_file(
file_path=file_path,
operation_name=operation_name,
campaign_structure=campaign_structure,
)
return self._run_worker_inner(file_path, options)
except ValidationError, json.JSONDecodeError:
if options["crash_on_error"]:
raise
@ -1511,8 +1580,116 @@ class Command(BaseCommand):
)
return {"success": False, "broken_dir": str(broken_dir)}
return {"success": False, "broken_dir": "(skipped)"}
else:
return {"success": True}
def _run_file_processing_inner(
self,
file_path: Path,
options: dict,
progress_bar: tqdm,
) -> None:
"""Inner file processing logic with progress reporting.
Args:
file_path: Path to the JSON file.
options: Command options.
progress_bar: tqdm progress bar instance.
"""
raw_text: str = file_path.read_text(encoding="utf-8", errors="ignore")
# Repair potentially broken JSON with multiple fallback strategies
fixed_json_str: str = repair_partially_broken_json(raw_text)
parsed_json: (
JSONReturnType | tuple[JSONReturnType, list[dict[str, str]]] | str
) = json.loads(fixed_json_str)
operation_name: str | None = extract_operation_name_from_parsed(
parsed_json,
)
# Check for error-only responses first
error_description: str | None = detect_error_only_response(parsed_json)
if error_description:
if not options.get("skip_broken_moves"):
broken_dir: Path | None = move_file_to_broken_subdir(
file_path,
error_description,
operation_name=operation_name,
)
progress_bar.write(
f"{Fore.RED}✗{Style.RESET_ALL} {file_path.name} → "
f"{broken_dir}/{file_path.name} "
f"({error_description})",
)
else:
progress_bar.write(
f"{Fore.RED}✗{Style.RESET_ALL} {file_path.name} ({error_description}, move skipped)",
)
return
matched: str | None = detect_non_campaign_keyword(raw_text)
if matched:
if not options.get("skip_broken_moves"):
broken_dir: Path | None = move_file_to_broken_subdir(
file_path,
matched,
operation_name=operation_name,
)
progress_bar.write(
f"{Fore.RED}✗{Style.RESET_ALL} {file_path.name} → "
f"{broken_dir}/{file_path.name} "
f"(matched '{matched}')",
)
else:
progress_bar.write(
f"{Fore.RED}✗{Style.RESET_ALL} {file_path.name} (matched '{matched}', move skipped)",
)
return
if "dropCampaign" not in raw_text:
if not options.get("skip_broken_moves"):
broken_dir = move_file_to_broken_subdir(
file_path,
"no_dropCampaign",
operation_name=operation_name,
)
progress_bar.write(
f"{Fore.RED}✗{Style.RESET_ALL} {file_path.name} → "
f"{broken_dir}/{file_path.name} "
f"(no dropCampaign present)",
)
else:
progress_bar.write(
f"{Fore.RED}✗{Style.RESET_ALL} {file_path.name} (no dropCampaign present, move skipped)",
)
return
# Normalize and filter to dict responses only
responses: list[dict[str, Any]] = self._normalize_responses(parsed_json)
processed, broken_dir = self.process_responses(
responses=responses,
file_path=file_path,
options=options,
)
if not processed:
# File already moved during validation; nothing more to do here.
progress_bar.write(
f"{Fore.RED}✗{Style.RESET_ALL} {file_path.name} → "
f"{broken_dir}/{file_path.name} (validation failed)",
)
return
campaign_structure: str | None = self._detect_campaign_structure(
responses[0] if responses else {},
)
move_completed_file(
file_path=file_path,
operation_name=operation_name,
campaign_structure=campaign_structure,
)
progress_bar.update(1)
progress_bar.write(f"{Fore.GREEN}✓{Style.RESET_ALL} {file_path.name}")
def process_file(self, file_path: Path, options: dict) -> None:
"""Reads a JSON file and processes the campaign data.
@ -1533,102 +1710,7 @@ class Command(BaseCommand):
dynamic_ncols=True,
) as progress_bar:
try:
raw_text: str = file_path.read_text(encoding="utf-8", errors="ignore")
# Repair potentially broken JSON with multiple fallback strategies
fixed_json_str: str = repair_partially_broken_json(raw_text)
parsed_json: (
JSONReturnType | tuple[JSONReturnType, list[dict[str, str]]] | str
) = json.loads(fixed_json_str)
operation_name: str | None = extract_operation_name_from_parsed(
parsed_json,
)
# Check for error-only responses first
error_description: str | None = detect_error_only_response(parsed_json)
if error_description:
if not options.get("skip_broken_moves"):
broken_dir: Path | None = move_file_to_broken_subdir(
file_path,
error_description,
operation_name=operation_name,
)
progress_bar.write(
f"{Fore.RED}✗{Style.RESET_ALL} {file_path.name} → "
f"{broken_dir}/{file_path.name} "
f"({error_description})",
)
else:
progress_bar.write(
f"{Fore.RED}✗{Style.RESET_ALL} {file_path.name} ({error_description}, move skipped)",
)
return
matched: str | None = detect_non_campaign_keyword(raw_text)
if matched:
if not options.get("skip_broken_moves"):
broken_dir: Path | None = move_file_to_broken_subdir(
file_path,
matched,
operation_name=operation_name,
)
progress_bar.write(
f"{Fore.RED}✗{Style.RESET_ALL} {file_path.name} → "
f"{broken_dir}/{file_path.name} "
f"(matched '{matched}')",
)
else:
progress_bar.write(
f"{Fore.RED}✗{Style.RESET_ALL} {file_path.name} (matched '{matched}', move skipped)",
)
return
if "dropCampaign" not in raw_text:
if not options.get("skip_broken_moves"):
broken_dir = move_file_to_broken_subdir(
file_path,
"no_dropCampaign",
operation_name=operation_name,
)
progress_bar.write(
f"{Fore.RED}✗{Style.RESET_ALL} {file_path.name} → "
f"{broken_dir}/{file_path.name} "
f"(no dropCampaign present)",
)
else:
progress_bar.write(
f"{Fore.RED}✗{Style.RESET_ALL} {file_path.name} (no dropCampaign present, move skipped)",
)
return
# Normalize and filter to dict responses only
responses: list[dict[str, Any]] = self._normalize_responses(parsed_json)
processed, broken_dir = self.process_responses(
responses=responses,
file_path=file_path,
options=options,
)
if not processed:
# File already moved during validation; nothing more to do here.
progress_bar.write(
f"{Fore.RED}✗{Style.RESET_ALL} {file_path.name} → "
f"{broken_dir}/{file_path.name} (validation failed)",
)
return
campaign_structure: str | None = self._detect_campaign_structure(
responses[0] if responses else {},
)
move_completed_file(
file_path=file_path,
operation_name=operation_name,
campaign_structure=campaign_structure,
)
progress_bar.update(1)
progress_bar.write(f"{Fore.GREEN}✓{Style.RESET_ALL} {file_path.name}")
self._run_file_processing_inner(file_path, options, progress_bar)
except ValidationError, json.JSONDecodeError:
if options["crash_on_error"]:
raise

77
twitch/management/commands/download_box_art.py
View file

@ -139,45 +139,48 @@ class Command(BaseCommand):
image_path: Absolute path to the downloaded image file
"""
try:
source_path = Path(image_path)
if not source_path.exists() or source_path.suffix.lower() not in {
".jpg",
".jpeg",
".png",
}:
return
base_path = source_path.with_suffix("")
webp_path = base_path.with_suffix(".webp")
avif_path = base_path.with_suffix(".avif")
with Image.open(source_path) as img:
# Convert to RGB if needed
if img.mode in {"RGBA", "LA"} or (
img.mode == "P" and "transparency" in img.info
):
background = Image.new("RGB", img.size, (255, 255, 255))
rgba_img = img.convert("RGBA") if img.mode == "P" else img
background.paste(
rgba_img,
mask=rgba_img.split()[-1]
if rgba_img.mode in {"RGBA", "LA"}
else None,
)
rgb_img = background
elif img.mode != "RGB":
rgb_img = img.convert("RGB")
else:
rgb_img = img
# Save WebP
rgb_img.save(webp_path, "WEBP", quality=85, method=6)
# Save AVIF
rgb_img.save(avif_path, "AVIF", quality=85, speed=4)
self._perform_image_conversion(image_path)
except (OSError, ValueError) as e:
# Don't fail the download if conversion fails
self.stdout.write(
self.style.WARNING(f"Failed to convert {image_path}: {e}"),
)
def _perform_image_conversion(self, image_path: str) -> None:
"""Perform the actual image conversion to WebP and AVIF."""
source_path = Path(image_path)
if not source_path.exists() or source_path.suffix.lower() not in {
".jpg",
".jpeg",
".png",
}:
return
base_path = source_path.with_suffix("")
webp_path = base_path.with_suffix(".webp")
avif_path = base_path.with_suffix(".avif")
with Image.open(source_path) as img:
# Convert to RGB if needed
if img.mode in {"RGBA", "LA"} or (
img.mode == "P" and "transparency" in img.info
):
background = Image.new("RGB", img.size, (255, 255, 255))
rgba_img = img.convert("RGBA") if img.mode == "P" else img
background.paste(
rgba_img,
mask=rgba_img.split()[-1]
if rgba_img.mode in {"RGBA", "LA"}
else None,
)
rgb_img = background
elif img.mode != "RGB":
rgb_img = img.convert("RGB")
else:
rgb_img = img
# Save WebP
rgb_img.save(webp_path, "WEBP", quality=85, method=6)
# Save AVIF
rgb_img.save(avif_path, "AVIF", quality=85, speed=4)

96
twitch/management/commands/download_campaign_images.py
View file

@ -325,56 +325,66 @@ class Command(BaseCommand):
return "failed"
def _convert_image_to_formats(
self,
source_path: Path,
) -> None:
"""Load an image and save it as WebP and AVIF.
Args:
source_path: Path to the source image file.
"""
base_path: Path = source_path.with_suffix("")
webp_path: Path = base_path.with_suffix(".webp")
avif_path: Path = base_path.with_suffix(".avif")
with Image.open(source_path) as img:
# Convert to RGB if needed
if img.mode in {"RGBA", "LA"} or (
img.mode == "P" and "transparency" in img.info
):
background: Image.Image = Image.new(
"RGB",
img.size,
(255, 255, 255),
)
rgba_img: Image.Image | ImageFile = (
img.convert("RGBA") if img.mode == "P" else img
)
background.paste(
rgba_img,
mask=rgba_img.split()[-1]
if rgba_img.mode in {"RGBA", "LA"}
else None,
)
rgb_img = background
elif img.mode != "RGB":
rgb_img: Image.Image = img.convert("RGB")
else:
rgb_img: Image.Image = img
# Save WebP
rgb_img.save(webp_path, "WEBP", quality=85, method=6)
# Save AVIF
rgb_img.save(avif_path, "AVIF", quality=85, speed=4)
def _convert_to_modern_formats(self, image_path: str) -> None:
"""Convert downloaded image to WebP and AVIF formats.
Args:
image_path: Absolute path to the downloaded image file
"""
source_path = Path(image_path)
if not source_path.exists() or source_path.suffix.lower() not in {
".jpg",
".jpeg",
".png",
}:
return
try:
source_path = Path(image_path)
if not source_path.exists() or source_path.suffix.lower() not in {
".jpg",
".jpeg",
".png",
}:
return
base_path: Path = source_path.with_suffix("")
webp_path: Path = base_path.with_suffix(".webp")
avif_path: Path = base_path.with_suffix(".avif")
with Image.open(source_path) as img:
# Convert to RGB if needed
if img.mode in {"RGBA", "LA"} or (
img.mode == "P" and "transparency" in img.info
):
background: Image.Image = Image.new(
"RGB",
img.size,
(255, 255, 255),
)
rgba_img: Image.Image | ImageFile = (
img.convert("RGBA") if img.mode == "P" else img
)
background.paste(
rgba_img,
mask=rgba_img.split()[-1]
if rgba_img.mode in {"RGBA", "LA"}
else None,
)
rgb_img = background
elif img.mode != "RGB":
rgb_img: Image.Image = img.convert("RGB")
else:
rgb_img: Image.Image = img
# Save WebP
rgb_img.save(webp_path, "WEBP", quality=85, method=6)
# Save AVIF
rgb_img.save(avif_path, "AVIF", quality=85, speed=4)
self._convert_image_to_formats(source_path)
except (OSError, ValueError) as e:
# Don't fail the download if conversion fails
self.stdout.write(

1637
twitch/management/commands/import_twitch_drops_api.py Normal file
View file

@ -0,0 +1,1637 @@
"""Import drop and reward campaign data from the SunkwiBOT/twitch-drops-api repo.
This command fetches ``drops.json`` and ``rewards.json`` from the
`SunkwiBOT/twitch-drops-api <https://github.com/SunkwiBOT/twitch-drops-api>`_
GitHub repository and imports the data into the local database.
Usage::
# Current data (latest commit)
uv run python manage.py import_twitch_drops_api
uv run python manage.py import_twitch_drops_api --source "SunkwiBOT/twitch-drops-api"
# Historical data (all commits, oldest-first — captures expired campaigns)
uv run python manage.py import_twitch_drops_api --historical
uv run python manage.py import_twitch_drops_api --historical --max-commits 50
uv run python manage.py import_twitch_drops_api --historical --git-dir /tmp/mirror
The ``--source`` argument controls the value stored in the
``data_source`` field on imported records (defaults to
``"SunkwiBOT/twitch-drops-api"``).
"""
from __future__ import annotations
import contextlib
import hashlib
import json
import os
import pathlib
import subprocess # noqa: S404
import tempfile
from typing import TYPE_CHECKING
from typing import Any
import httpx
from colorama import Fore
from colorama import Style
from colorama import init as colorama_init
from django.core.management.base import BaseCommand
from django.core.management.base import CommandError
from pydantic import ValidationError
from twitch.models import Channel
from twitch.models import DropBenefit
from twitch.models import DropBenefitEdge
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
from twitch.schemas import SunkwiBotDropGroupSchema
from twitch.schemas import SunkwiBotRewardCampaignSchema
from twitch.utils import parse_date
if TYPE_CHECKING:
from django.core.management.base import CommandParser
from django.db.models import Model
from twitch.schemas import SunkwiBotBenefitEdgeSchema
from twitch.schemas import SunkwiBotDropCampaignACLSchema
from twitch.schemas import SunkwiBotRewardSchema
from twitch.schemas import SunkwiBotTimeBasedDropSchema
# Default raw GitHub URLs for the SunkwiBOT/twitch-drops-api repo
DEFAULT_DROPS_URL = (
"https://raw.githubusercontent.com/SunkwiBOT/twitch-drops-api/main/drops.json"
)
DEFAULT_REWARDS_URL = (
"https://raw.githubusercontent.com/SunkwiBOT/twitch-drops-api/main/rewards.json"
)
GIT_CLONE_URL = "https://github.com/SunkwiBOT/twitch-drops-api.git"
HTTP_TIMEOUT = 60 # seconds
class Command(BaseCommand):
"""Import drop and reward campaigns from the SunkwiBOT twitch-drops-api repo."""
help = (
"Import drop and reward campaigns from "
"SunkwiBOT/twitch-drops-api (GitHub raw JSON)."
)
requires_migrations_checks = True
def add_arguments(self, parser: CommandParser) -> None:
"""Populate the command with arguments."""
parser.add_argument(
"--source",
type=str,
default="SunkwiBOT/twitch-drops-api",
help=(
"Value to store in the data_source field. "
"Default: SunkwiBOT/twitch-drops-api"
),
)
parser.add_argument(
"--drops-url",
type=str,
default=DEFAULT_DROPS_URL,
help=f"URL to fetch drops.json from. Default: {DEFAULT_DROPS_URL}",
)
parser.add_argument(
"--rewards-url",
type=str,
default=DEFAULT_REWARDS_URL,
help=f"URL to fetch rewards.json from. Default: {DEFAULT_REWARDS_URL}",
)
parser.add_argument(
"--drops-only",
action="store_true",
help="Only import drops.json (skip rewards.json).",
)
parser.add_argument(
"--rewards-only",
action="store_true",
help="Only import rewards.json (skip drops.json).",
)
parser.add_argument(
"--crash-on-error",
action="store_true",
help="Crash on first validation/import error instead of continuing.",
)
parser.add_argument(
"--verbose",
action="store_true",
help="Print per-record success messages.",
)
parser.add_argument(
"--report",
action="store_true",
help=(
"Compare SunkwiBOT data against the database without importing. "
"Prints a summary of what's new, what exists, and field-level "
"differences for overlapping campaigns."
),
)
parser.add_argument(
"--merge",
action="store_true",
help=(
"For existing campaigns, fill in empty/blank fields with "
"SunkwiBOT data without overwriting populated ones. The "
"original data_source is preserved."
),
)
parser.add_argument(
"--update-existing",
action="store_true",
help=(
"By default, campaigns already in the database are skipped "
"(existing data_source is preserved). Use this flag to "
"update existing campaigns with the incoming data instead."
),
)
parser.add_argument(
"--historical",
action="store_true",
help=(
"Iterate through all git commits (oldest first) to capture "
"expired campaigns that are no longer in the latest JSON. "
"Implies --skip-latest unless --no-skip-latest is given."
),
)
parser.add_argument(
"--no-skip-latest",
action="store_true",
help="When used with --historical, also process the latest commit.",
)
parser.add_argument(
"--max-commits",
type=int,
default=0,
help="Maximum number of *unique-content* commits to process (0 = all).",
)
parser.add_argument(
"--git-dir",
type=str,
default="",
help=(
"Path to a local bare clone of the repo. If not set, a temp "
"clone is created and cleaned up."
),
)
parser.add_argument(
"--from-file",
type=str,
default="",
help=(
"Path to a pre-extracted JSON file (e.g. drops_merged.json "
"from tools/extract_historical_drops.py). Import this file "
"directly instead of fetching from git history or URLs. "
"Much faster than --historical for repeated imports."
),
)
def handle(self, *args: Any, **options: Any) -> None: # noqa: ANN401, ARG002, PLR0914, PLR0915
"""Main entry point for the command.
Raises:
CommandError: If the --from-file path does not exist or is invalid.
"""
colorama_init(autoreset=True)
source: str = options["source"]
verbose: bool = options["verbose"]
crash_on_error: bool = options["crash_on_error"]
report_mode: bool = options.get("report", False)
merge_mode: bool = options.get("merge", False)
skip_existing: bool = not options.get("update_existing")
historical: bool = options.get("historical", False)
# Caches for the duration of the command to avoid redundant DB lookups
self._org_cache: dict[str, Organization] = {}
self._game_cache: dict[str, Game] = {}
self._channel_cache: dict[str, Channel] = {}
self._benefit_cache: dict[str, DropBenefit] = {}
# Merge mode: fill empty fields on existing campaigns instead of skipping
self._merge_mode: bool = merge_mode
if merge_mode:
skip_existing = False
drops_count = 0
rewards_count = 0
errors: list[str] = []
if report_mode:
self._generate_report(
drops_url=options["drops_url"],
rewards_url=options["rewards_url"],
crash_on_error=crash_on_error,
)
return
from_file: str = options.get("from_file", "") or ""
if from_file:
# Import from a pre-extracted merged JSON file (fast path).
# The file is produced by tools/extract_historical_drops.py.
from_file_path = pathlib.Path(from_file)
if not from_file_path.exists():
msg = f"File not found: {from_file}"
raise CommandError(msg)
with from_file_path.open(encoding="utf-8") as f:
raw_data = json.load(f)
if not isinstance(raw_data, list):
msg = f"Expected a JSON array in {from_file}, got {type(raw_data).__name__}"
raise CommandError(msg)
# Detect file type by content structure
is_drops = any("gameId" in item for item in raw_data)
if is_drops:
self.stdout.write(
f"Importing drops from {from_file} ({len(raw_data)} groups) …",
)
drops_count = self._parse_and_import_drops(
raw_data=raw_data,
source=source,
verbose=verbose,
crash_on_error=crash_on_error,
label=from_file_path.name,
skip_existing=skip_existing,
)
else:
self.stdout.write(
f"Importing rewards from {from_file} ({len(raw_data)} campaigns) …",
)
rewards_count = self._parse_and_import_rewards(
raw_data=raw_data,
source=source,
verbose=verbose,
crash_on_error=crash_on_error,
label=from_file_path.name,
skip_existing=skip_existing,
)
elif historical:
# Historical mode: clone repo and iterate commits
hist_drops, hist_rewards, hist_errors = self._process_historical(
source=source,
verbose=verbose,
crash_on_error=crash_on_error,
skip_latest=not options["no_skip_latest"],
max_commits=options["max_commits"],
skip_existing=skip_existing,
git_dir=options["git_dir"],
)
drops_count += hist_drops
rewards_count += hist_rewards
errors.extend(hist_errors)
else:
# Normal mode: fetch latest from URLs
if not options["rewards_only"]:
try:
drops_count += self._import_drops(
url=options["drops_url"],
source=source,
verbose=verbose,
crash_on_error=crash_on_error,
skip_existing=skip_existing,
)
except Exception as exc: # noqa: BLE001
msg = f"Failed to import drops.json: {exc}"
self.stderr.write(self.style.ERROR(msg))
errors.append(msg)
if not options["drops_only"]:
try:
rewards_count += self._import_rewards(
url=options["rewards_url"],
source=source,
verbose=verbose,
crash_on_error=crash_on_error,
skip_existing=skip_existing,
)
except Exception as exc: # noqa: BLE001
msg = f"Failed to import rewards.json: {exc}"
self.stderr.write(self.style.ERROR(msg))
errors.append(msg)
# Summary
self.stdout.write("\n" + "=" * 50)
if historical:
self.stdout.write(
self.style.SUCCESS(
f"✓ Historical drops imported/updated: {drops_count}",
),
)
self.stdout.write(
self.style.SUCCESS(
f"✓ Historical rewards imported/updated: {rewards_count}",
),
)
else:
self.stdout.write(
self.style.SUCCESS(
f"✓ Drop campaigns imported/updated: {drops_count}",
),
)
self.stdout.write(
self.style.SUCCESS(
f"✓ Reward campaigns imported/updated: {rewards_count}",
),
)
if errors:
for err in errors:
self.stdout.write(self.style.ERROR(f"✗ {err}"))
self.stdout.write("=" * 50)
# ------------------------------------------------------------------
# Historical import via git clone
# ------------------------------------------------------------------
def _process_historical( # noqa: PLR0913
self,
source: str,
*,
verbose: bool,
crash_on_error: bool,
skip_latest: bool,
max_commits: int,
git_dir: str,
skip_existing: bool = False,
) -> tuple[int, int, list[str]]:
"""Clone the repo and iterate commits oldest-first to import historical data.
Skips commits whose file content is byte-identical to the previously
processed commit (deduplicates bot spam).
Args:
source: data_source label.
verbose: Print per-record messages.
crash_on_error: Raise on first error.
skip_latest: If True, skip the most recent commit (already imported
by the non-historical path).
max_commits: Max unique-content commits to process (0 = all).
git_dir: Path to existing bare clone, or '' to create a temp one.
skip_existing: If True, skip campaigns already in the database.
Returns:
Tuple of (drops_count, rewards_count, error_messages).
"""
own_clone = not git_dir
if own_clone:
git_dir_obj = tempfile.mkdtemp(prefix="twitch-drops-api-")
git_dir = git_dir_obj
self.stdout.write(f"Cloning {GIT_CLONE_URL} …")
self._git(
git_dir,
"clone",
"--filter=blob:none",
"--bare",
GIT_CLONE_URL,
git_dir,
)
else:
self.stdout.write(f"Using existing clone at {git_dir} …")
drops_count = 0
rewards_count = 0
errors: list[str] = []
try:
drops_count += self._import_historical_file(
git_dir=git_dir,
file_path="drops.json",
source=source,
verbose=verbose,
crash_on_error=crash_on_error,
skip_latest=skip_latest,
max_commits=max_commits,
skip_existing=skip_existing,
)
rewards_count += self._import_historical_file(
git_dir=git_dir,
file_path="rewards.json",
source=source,
verbose=verbose,
crash_on_error=crash_on_error,
skip_latest=skip_latest,
max_commits=max_commits,
skip_existing=skip_existing,
)
except Exception as exc: # noqa: BLE001
errors.append(str(exc))
finally:
if own_clone:
self._rmtree(git_dir)
return drops_count, rewards_count, errors
def _import_historical_file( # noqa: PLR0913
self,
git_dir: str,
file_path: str,
source: str,
*,
verbose: bool,
crash_on_error: bool,
skip_latest: bool,
max_commits: int,
skip_existing: bool,
) -> int:
"""Import all historical versions of a single file from git history.
Args:
git_dir: Path to bare clone.
file_path: Path within repo (e.g. "drops.json").
source: data_source label.
verbose: Print per-record messages.
crash_on_error: Raise on first error.
skip_latest: If True, skip the newest commit.
max_commits: Max unique-content commits (0 = all).
skip_existing: If True, skip campaigns already in the database.
Returns:
Number of campaigns imported/updated.
"""
label = file_path.replace(".json", "")
self.stdout.write(f"\nCollecting commits for {file_path} …")
# Get all commits that touched this file, oldest first
log_output = self._git(
git_dir,
"log",
"--all",
"--format=%H %ct",
"--follow",
"--reverse",
"--",
file_path,
)
lines = [ln.strip() for ln in log_output.strip().split("\n") if ln.strip()]
total = len(lines)
self.stdout.write(f" Found {total} total commits for {file_path}")
if skip_latest and total > 0:
lines = lines[:-1]
self.stdout.write(f" Skipping latest commit, processing {len(lines)}")
if max_commits and len(lines) > max_commits:
lines = lines[:max_commits]
self.stdout.write(f" Limited to {max_commits} commits")
if not lines:
self.stdout.write(f" No commits to process for {file_path}")
return 0
total_campaigns = 0
seen_hashes: set[str] = set()
processed = 0
skipped_identical = 0
for line in lines:
parts = line.split(None, 1)
if not parts:
continue
sha = parts[0]
# Get file content at this commit
try:
content = self._git(
git_dir,
"show",
f"{sha}:{file_path}",
check=True,
)
except subprocess.CalledProcessError:
# File didn't exist at this commit yet
continue
# Skip if content is byte-identical to last processed
content_hash = hashlib.sha256(content.encode()).hexdigest()
if content_hash in seen_hashes:
skipped_identical += 1
continue
seen_hashes.add(content_hash)
try:
data = json.loads(content)
except json.JSONDecodeError:
if verbose:
self.stdout.write(
f"{Fore.YELLOW}→{Style.RESET_ALL} {sha[:7]}: "
f"invalid JSON, skipping",
)
continue
if not isinstance(data, list):
continue
short_sha = sha[:7]
self.stdout.write(
f" [{processed + 1}/{len(lines)}] Importing {label} @ {short_sha} …",
)
if file_path == "drops.json":
count = self._parse_and_import_drops(
raw_data=data,
source=source,
verbose=verbose,
crash_on_error=crash_on_error,
label=f"{label}@{short_sha}",
skip_existing=skip_existing,
)
else:
count = self._parse_and_import_rewards(
raw_data=data,
source=source,
verbose=verbose,
crash_on_error=crash_on_error,
label=f"{label}@{short_sha}",
skip_existing=skip_existing,
)
total_campaigns += count
processed += 1
self.stdout.write(
f" Done: {processed} versions processed, "
f"{skipped_identical} skipped (identical content), "
f"{total_campaigns} total campaigns imported for {file_path}",
)
return total_campaigns
# ------------------------------------------------------------------
# Git / filesystem helpers
# ------------------------------------------------------------------
def _git(self, git_dir: str, *args: str, check: bool = True) -> str:
"""Run a git command in the bare clone directory.
Args:
git_dir: Path to the bare clone.
*args: Git sub-command and arguments.
check: If True, raise on non-zero exit.
Returns:
Stdout of the git command.
"""
cmd = ["git", "--git-dir", git_dir, *args]
result = subprocess.run(cmd, capture_output=True, text=True, check=check) # noqa: S603
return result.stdout
def _rmtree(self, path: str) -> None:
"""Recursively remove a directory tree.
Args:
path: Path to remove.
"""
try:
for root_str, dirs, files in os.walk(path, topdown=False):
root_path = pathlib.Path(root_str)
for name in files:
(root_path / name).unlink()
for name in dirs:
(root_path / name).rmdir()
pathlib.Path(path).rmdir()
except OSError:
pass
# ------------------------------------------------------------------
# drops.json import
# ------------------------------------------------------------------
def _fetch_json(self, url: str, label: str) -> list[dict[str, Any]]:
"""Fetch a JSON file from a URL and parse it into a list of dicts.
Args:
url: The URL to fetch.
label: Human-readable label for error messages.
Returns:
Parsed JSON data as a list of dicts.
Raises:
TypeError: If the response is not a JSON array.
"""
self.stdout.write(f"Fetching {label} from {url} …")
with httpx.Client(timeout=HTTP_TIMEOUT, follow_redirects=True) as client:
response = client.get(url)
response.raise_for_status()
data: Any = response.json()
if not isinstance(data, list):
msg = f"Expected a JSON array from {url}, got {type(data).__name__}"
raise TypeError(msg)
return data
def _import_drops(
self,
url: str,
source: str,
*,
verbose: bool,
crash_on_error: bool,
skip_existing: bool = False,
) -> int:
"""Import drop campaigns from a drops.json URL.
Args:
url: Raw GitHub URL for drops.json.
source: Data source label to store on records.
verbose: Print per-record messages.
crash_on_error: Raise on first error instead of continuing.
skip_existing: If True, skip campaigns already in the database.
Returns:
Number of drop campaigns imported/updated.
"""
raw_data = self._fetch_json(url, "drops.json")
return self._parse_and_import_drops(
raw_data=raw_data,
source=source,
verbose=verbose,
crash_on_error=crash_on_error,
label="drops.json",
skip_existing=skip_existing,
)
@staticmethod
def _strip_typename(data: Any) -> Any: # noqa: ANN401
"""Recursively remove ``__typename`` keys from parsed JSON data.
Old commits in the repo included GraphQL ``__typename`` metadata which
the SunkwiBot schemas reject with ``extra="forbid"``. Stripping them
before validation keeps the schemas strict while still accepting
historical payloads.
Args:
data: Raw parsed JSON (dict, list, or scalar).
Returns:
Data with all ``__typename`` keys removed.
"""
if isinstance(data, dict):
return {
k: Command._strip_typename(v)
for k, v in data.items()
if k != "__typename"
}
if isinstance(data, list):
return [Command._strip_typename(item) for item in data]
return data
def _generate_report( # noqa: PLR0914, PLR0915
self,
drops_url: str,
rewards_url: str,
*,
crash_on_error: bool,
) -> None:
"""Compare SunkwiBOT data against the database without importing.
Fetches drops.json and rewards.json, validates with schemas, then
compares against the database to show:
- How many campaigns are new
- How many already exist
- Field-level differences for overlapping campaigns
- Total counts per source
Args:
drops_url: URL for drops.json.
rewards_url: URL for rewards.json.
crash_on_error: Raise on first error instead of continuing.
Raises:
ValidationError: If any item fails schema validation and crash_on_error is True.
"""
from collections import Counter # noqa: PLC0415
self.stdout.write("=" * 60)
self.stdout.write(self.style.SUCCESS("COMPARISON REPORT"))
self.stdout.write(
"Comparing SunkwiBOT data against local database…",
)
self.stdout.write("=" * 60)
# Track differences for field-level comparison
sample_limit = 5
for label, raw_url, model_cls, id_field in [ # noqa: PLR1702
("drops", drops_url, DropCampaign, "twitch_id"),
("rewards", rewards_url, RewardCampaign, "twitch_id"),
]:
self.stdout.write(f"\n--- {label.upper()} ---")
try:
raw_data = self._fetch_json(raw_url, f"{label}.json")
except Exception as exc: # noqa: BLE001
self.stderr.write(
self.style.ERROR(f" Failed to fetch {label}.json: {exc}"),
)
continue
# Validate
validated_ids: set[str] = set()
validated: list[Any] = []
groups: list[SunkwiBotDropGroupSchema] = []
if label == "drops":
for i, raw_item in enumerate(raw_data):
try:
stripped_item = self._strip_typename(raw_item)
groups.append(
SunkwiBotDropGroupSchema.model_validate(stripped_item),
)
except ValidationError:
if crash_on_error:
raise
self.stdout.write(
f" {Fore.YELLOW}⚠{Style.RESET_ALL} "
f"item [{i}] failed validation, skipping",
)
for group in groups:
validated_ids.update(reward.twitch_id for reward in group.rewards)
else:
for i, raw_item in enumerate(raw_data):
try:
stripped_item = self._strip_typename(raw_item)
validated.append(
SunkwiBotRewardCampaignSchema.model_validate(stripped_item),
)
except ValidationError:
if crash_on_error:
raise
self.stdout.write(
f" {Fore.YELLOW}⚠{Style.RESET_ALL} "
f"item [{i}] failed validation, skipping",
)
validated_ids.update(c.twitch_id for c in validated)
# Compare against DB
existing_ids: set[str] = set(
model_cls.objects.values_list(id_field, flat=True),
)
new_ids = validated_ids - existing_ids
overlap_ids = validated_ids & existing_ids
self.stdout.write(
f" Total in SunkwiBOT data: {len(validated_ids)}",
)
self.stdout.write(
f" Total in local database: {len(existing_ids)}",
)
self.stdout.write(
f" New (SunkwiBOT only): {len(new_ids)}",
)
self.stdout.write(
f" Overlap (both): {len(overlap_ids)}",
)
# Field-level comparison for overlapping campaigns
if overlap_ids and label == "drops":
field_diffs: list[tuple[str, str, str]] = []
fields_to_check = [
"name",
"description",
"image_url",
"start_at",
"end_at",
]
for group in groups:
for reward in group.rewards:
if reward.twitch_id not in overlap_ids:
continue
try:
db_obj = DropCampaign.objects.get(
twitch_id=reward.twitch_id,
)
except DropCampaign.DoesNotExist:
continue
for field in fields_to_check:
ours = getattr(db_obj, field, None)
theirs = getattr(reward, field, None)
# Compare string forms for dates
str_ours = str(ours) if ours is not None else ""
str_theirs = str(theirs) if theirs is not None else ""
if str_ours.strip() != str_theirs.strip():
field_diffs.append(
(field, str_ours, str_theirs),
)
if field_diffs:
diff_counter: Counter = Counter()
for diff_field, _ours_val, _theirs_val in field_diffs:
diff_counter[diff_field] += 1
self.stdout.write(
"\n Field differences (overlapping campaigns):",
)
for field, count in diff_counter.most_common():
ours_empty = sum(
1 for f, o, _ in field_diffs if f == field and not o.strip()
)
theirs_empty = sum(
1
for f, _o, t in field_diffs
if f == field and not t.strip()
)
self.stdout.write(
f" {field}: {count} campaigns differ "
f"(we have empty: {ours_empty}, "
f"they have empty: {theirs_empty})",
)
self.stdout.write(
" Use --merge to fill empty fields from SunkwiBOT data.",
)
else:
self.stdout.write(
" ✓ No field differences in overlapping campaigns.",
)
elif overlap_ids and label == "rewards":
fields_to_check = [
"name",
"brand",
"summary",
"status",
]
reward_diffs = 0
for c_id in overlap_ids:
try:
db_obj = RewardCampaign.objects.get(twitch_id=c_id)
except RewardCampaign.DoesNotExist:
continue
# Re-find the schema entry
for c_entry in validated:
if c_entry.twitch_id == c_id:
for field in fields_to_check:
ours = getattr(db_obj, field, None)
theirs = getattr(c_entry, field, None)
if str(ours or "") != str(theirs or ""):
reward_diffs += 1
break
if reward_diffs:
self.stdout.write(
f" {reward_diffs} field differences found in "
f"overlapping reward campaigns.",
)
else:
self.stdout.write(
" ✓ No field differences in overlapping reward campaigns.",
)
if new_ids:
self.stdout.write(
f"\n {Fore.GREEN}Sample new campaigns:{Style.RESET_ALL}",
)
sample_new = list(new_ids)[:5]
for sid in sample_new:
self.stdout.write(f" - {sid}")
if len(new_ids) > sample_limit:
self.stdout.write(
f" … and {len(new_ids) - sample_limit} more",
)
self.stdout.write("\n" + "=" * 60)
def _parse_and_import_drops( # noqa: PLR0913
self,
raw_data: list[dict[str, Any]],
source: str,
*,
verbose: bool,
crash_on_error: bool,
label: str = "drops.json",
skip_existing: bool = False,
) -> int:
"""Validate and import raw drops.json data (from URL or git history).
Args:
raw_data: Parsed drops.json array.
source: Data source label.
verbose: Print per-record messages.
crash_on_error: Raise on first error.
label: Human label for error messages (e.g. "drops.json@abc1234").
skip_existing: If True, skip campaigns already in the database.
Returns:
Number of drop campaigns imported/updated.
Raises:
ValidationError: If crash_on_error is True and validation fails.
"""
raw_data = self._strip_typename(raw_data)
drop_groups: list[SunkwiBotDropGroupSchema] = []
for i, item in enumerate(raw_data):
try:
group = SunkwiBotDropGroupSchema.model_validate(item)
drop_groups.append(group)
except ValidationError as e:
msg = f"{label} item [{i}] failed validation: {e}"
if crash_on_error:
raise
self.stderr.write(self.style.ERROR(msg))
continue
total_campaigns = 0
for group in drop_groups:
# Get or create the shared Game for this group
# Use the first reward's owner as the game's organization
first_owner = group.rewards[0].owner if group.rewards else None
org_obj = None
if first_owner:
org_obj = self._get_or_create_organization(
twitch_id=first_owner.twitch_id,
name=first_owner.name,
verbose=verbose,
)
game_obj = self._get_or_create_game(
twitch_id=group.game_id,
display_name=group.game_display_name,
box_art_url=group.game_box_art_url,
source=source,
verbose=verbose,
org_obj=org_obj,
)
for reward in group.rewards:
try:
self._process_sunkwibot_reward(
reward=reward,
game_obj=game_obj,
source=source,
verbose=verbose,
skip_existing=skip_existing,
)
total_campaigns += 1
except Exception as exc:
msg = (
f"Failed to import reward {reward.twitch_id} "
f"({reward.name}): {exc}"
)
if crash_on_error:
raise
self.stderr.write(self.style.ERROR(msg))
return total_campaigns
def _process_sunkwibot_reward( # noqa: PLR0915
self,
reward: SunkwiBotRewardSchema,
game_obj: Game,
source: str,
*,
verbose: bool,
skip_existing: bool = False,
) -> None:
"""Process a single reward item from drops.json into a DropCampaign.
Args:
reward: The validated reward schema from drops.json.
game_obj: The Game instance for this campaign.
source: Data source label.
verbose: Print per-record messages.
skip_existing: If True, skip if campaign already exists.
"""
# Owner organization
self._get_or_create_organization(
twitch_id=reward.owner.twitch_id,
name=reward.owner.name,
verbose=verbose,
)
# Parse dates
start_at_dt = parse_date(reward.start_at)
end_at_dt = parse_date(reward.end_at)
defaults: dict[str, Any] = {
"name": reward.name,
"description": reward.description,
"image_url": reward.image_url,
"game": game_obj,
"start_at": start_at_dt,
"end_at": end_at_dt,
"details_url": reward.details_url,
"account_link_url": reward.account_link_url,
"data_source": source,
}
existing: DropCampaign | None = DropCampaign.objects.filter(
twitch_id=reward.twitch_id,
).first()
# Merge mode: fill empty fields, never overwrite populated ones
if existing is not None and getattr(self, "_merge_mode", False):
changed = False
merge_fields: dict[str, str | object | None] = {
"description": reward.description,
"image_url": reward.image_url,
"details_url": reward.details_url,
}
for field, new_value in merge_fields.items():
old_value = getattr(existing, field, None)
if not old_value and new_value:
setattr(existing, field, new_value)
changed = True
if changed:
existing.save(update_fields=list(merge_fields.keys()))
# Track enrichment
if source not in existing.operation_names:
existing.operation_names.append(source)
existing.save(update_fields=["operation_names"])
if verbose:
action = "Merged" if changed else "Skipped (merged)"
self.stdout.write(
f"{Fore.CYAN}→{Style.RESET_ALL} {action} campaign: {reward.name}",
)
# Still process time-based drops and channels for existing
if reward.time_based_drops:
for tbd_schema in reward.time_based_drops:
self._process_time_based_drop(
tbd_schema=tbd_schema,
campaign_obj=existing,
source=source,
verbose=verbose,
)
if reward.allow:
self._process_allow_channels(
allow_schema=reward.allow,
campaign_obj=existing,
verbose=verbose,
)
if not existing.is_fully_imported:
existing.is_fully_imported = True
existing.save(update_fields=["is_fully_imported"])
return
if skip_existing and existing is not None:
if verbose:
self.stdout.write(
f"{Fore.YELLOW}→{Style.RESET_ALL} Skipped campaign: "
f"{reward.name} (already exists)",
)
# Still accumulate channels and time-based drops for skipped
# campaigns so historical channels are never lost.
if reward.time_based_drops:
for tbd_schema in reward.time_based_drops:
self._process_time_based_drop(
tbd_schema=tbd_schema,
campaign_obj=existing,
source=source,
verbose=verbose,
)
if reward.allow:
self._process_allow_channels(
allow_schema=reward.allow,
campaign_obj=existing,
verbose=verbose,
)
if not existing.is_fully_imported:
existing.is_fully_imported = True
existing.save(update_fields=["is_fully_imported"])
return
campaign_obj, created = DropCampaign.objects.get_or_create(
twitch_id=reward.twitch_id,
defaults=defaults,
)
if not created:
self._save_if_changed(campaign_obj, defaults)
if verbose:
action = "Created" if created else "Updated"
self.stdout.write(
f"{Fore.GREEN}✓{Style.RESET_ALL} {action} campaign: "
f"{reward.name} (source: {source})",
)
# Process time-based drops
if reward.time_based_drops:
for tbd_schema in reward.time_based_drops:
self._process_time_based_drop(
tbd_schema=tbd_schema,
campaign_obj=campaign_obj,
source=source,
verbose=verbose,
)
# Process allowed channels
if reward.allow:
self._process_allow_channels(
allow_schema=reward.allow,
campaign_obj=campaign_obj,
verbose=verbose,
)
# Mark as fully imported
if not campaign_obj.is_fully_imported:
campaign_obj.is_fully_imported = True
campaign_obj.save(update_fields=["is_fully_imported"])
def _process_time_based_drop(
self,
tbd_schema: SunkwiBotTimeBasedDropSchema,
campaign_obj: DropCampaign,
source: str, # noqa: ARG002
*,
verbose: bool,
) -> None:
"""Process a single TimeBasedDrop from the SunkwiBOT schema.
Args:
tbd_schema: A SunkwiBotTimeBasedDropSchema instance.
campaign_obj: The parent DropCampaign.
source: Data source label.
verbose: Print per-record messages.
"""
start_at_dt = parse_date(tbd_schema.start_at)
end_at_dt = parse_date(tbd_schema.end_at)
drop_defaults: dict[str, Any] = {
"campaign": campaign_obj,
"name": tbd_schema.name,
"required_subs": tbd_schema.required_subs,
}
if tbd_schema.required_minutes_watched is not None:
drop_defaults["required_minutes_watched"] = (
tbd_schema.required_minutes_watched
)
if start_at_dt is not None:
drop_defaults["start_at"] = start_at_dt
if end_at_dt is not None:
drop_defaults["end_at"] = end_at_dt
drop_obj, created = TimeBasedDrop.objects.get_or_create(
twitch_id=tbd_schema.twitch_id,
defaults=drop_defaults,
)
if not created:
self._save_if_changed(drop_obj, drop_defaults)
if verbose and created:
self.stdout.write(
f"{Fore.GREEN}✓{Style.RESET_ALL} Created TimeBasedDrop: "
f"{tbd_schema.name}",
)
# Process benefit edges
for edge_schema in tbd_schema.benefit_edges:
self._process_benefit_edge(
edge_schema=edge_schema,
drop_obj=drop_obj,
verbose=verbose,
)
def _process_benefit_edge(
self,
edge_schema: SunkwiBotBenefitEdgeSchema,
drop_obj: TimeBasedDrop,
*,
verbose: bool,
) -> None:
"""Process a benefit edge, creating/updating the benefit and edge.
Args:
edge_schema: A SunkwiBotBenefitEdgeSchema instance.
drop_obj: The parent TimeBasedDrop.
verbose: Print per-record messages.
"""
benefit_schema = edge_schema.benefit
# Get or create benefit
cache_key = benefit_schema.twitch_id
cached = self._benefit_cache.get(cache_key)
if cached is not None:
benefit_obj = cached
else:
benefit_obj = DropBenefit.objects.filter(
twitch_id=benefit_schema.twitch_id,
).first()
if benefit_obj is None:
created_at_dt = (
parse_date(benefit_schema.created_at)
if benefit_schema.created_at
else None
)
benefit_obj = DropBenefit.objects.create(
twitch_id=benefit_schema.twitch_id,
name=benefit_schema.name,
image_asset_url=benefit_schema.image_asset_url,
entitlement_limit=benefit_schema.entitlement_limit,
is_ios_available=benefit_schema.is_ios_available,
distribution_type=benefit_schema.distribution_type or "",
created_at=created_at_dt,
)
if verbose:
self.stdout.write(
f"{Fore.GREEN}✓{Style.RESET_ALL} Created DropBenefit: "
f"{benefit_schema.name}",
)
else:
defaults: dict[str, Any] = {
"name": benefit_schema.name,
"image_asset_url": benefit_schema.image_asset_url,
"entitlement_limit": benefit_schema.entitlement_limit,
"is_ios_available": benefit_schema.is_ios_available,
"distribution_type": benefit_schema.distribution_type or "",
}
if benefit_schema.created_at:
created_at_dt = parse_date(benefit_schema.created_at)
if created_at_dt:
defaults["created_at"] = created_at_dt
self._save_if_changed(benefit_obj, defaults)
self._benefit_cache[cache_key] = benefit_obj
# Create or update the edge
edge_defaults = {"entitlement_limit": edge_schema.entitlement_limit}
edge_obj, edge_created = DropBenefitEdge.objects.get_or_create(
drop=drop_obj,
benefit=benefit_obj,
defaults=edge_defaults,
)
if not edge_created:
self._save_if_changed(edge_obj, edge_defaults)
def _process_allow_channels(
self,
allow_schema: SunkwiBotDropCampaignACLSchema,
campaign_obj: DropCampaign,
*,
verbose: bool,
) -> None:
"""Process the allow.channels list and set on the campaign.
Args:
allow_schema: A SunkwiBotDropCampaignACLSchema instance.
campaign_obj: The DropCampaign to update.
verbose: Print per-record messages.
"""
# Update allow_is_enabled
if campaign_obj.allow_is_enabled != allow_schema.is_enabled:
campaign_obj.allow_is_enabled = allow_schema.is_enabled
campaign_obj.save(update_fields=["allow_is_enabled"])
if not allow_schema.channels:
return
channel_objects: list[Channel] = []
for ch in allow_schema.channels:
display_name: str = ch.display_name or ch.name
channel_obj, created = Channel.objects.get_or_create(
twitch_id=ch.twitch_id,
defaults={
"name": ch.name,
"display_name": display_name,
},
)
if not created:
self._save_if_changed(
channel_obj,
{"name": ch.name, "display_name": display_name},
)
if verbose and created:
self.stdout.write(
f"{Fore.GREEN}✓{Style.RESET_ALL} Created channel: {display_name}",
)
channel_objects.append(channel_obj)
campaign_obj.allow_channels.add(*channel_objects)
# ------------------------------------------------------------------
# rewards.json import
# ------------------------------------------------------------------
def _import_rewards(
self,
url: str,
source: str,
*,
verbose: bool,
crash_on_error: bool,
skip_existing: bool = False,
) -> int:
"""Import reward campaigns from a rewards.json URL.
Args:
url: Raw GitHub URL for rewards.json.
source: Data source label to store on records.
verbose: Print per-record messages.
crash_on_error: Raise on first error instead of continuing.
skip_existing: If True, skip campaigns already in the database.
Returns:
Number of reward campaigns imported/updated.
"""
raw_data = self._fetch_json(url, "rewards.json")
return self._parse_and_import_rewards(
raw_data=raw_data,
source=source,
verbose=verbose,
crash_on_error=crash_on_error,
label="rewards.json",
skip_existing=skip_existing,
)
def _parse_and_import_rewards( # noqa: PLR0913
self,
raw_data: list[dict[str, Any]],
source: str,
*,
verbose: bool,
crash_on_error: bool,
label: str = "rewards.json",
skip_existing: bool = False,
) -> int:
"""Validate and import raw rewards.json data (from URL or git history).
Args:
raw_data: Parsed rewards.json array.
source: Data source label.
verbose: Print per-record messages.
crash_on_error: Raise on first error.
label: Human label for error messages.
skip_existing: If True, skip campaigns already in the database.
Returns:
Number of reward campaigns imported/updated.
Raises:
ValidationError: If crash_on_error is True and validation fails.
"""
raw_data = self._strip_typename(raw_data)
validated: list[SunkwiBotRewardCampaignSchema] = []
for i, item in enumerate(raw_data):
try:
campaign = SunkwiBotRewardCampaignSchema.model_validate(item)
validated.append(campaign)
except ValidationError as e:
msg = f"{label} item [{i}] failed validation: {e}"
if crash_on_error:
raise
self.stderr.write(self.style.ERROR(msg))
continue
total = 0
for campaign in validated:
try:
self._process_reward_campaign(
campaign=campaign,
source=source,
verbose=verbose,
skip_existing=skip_existing,
)
total += 1
except Exception as exc:
msg = (
f"Failed to import reward campaign {campaign.twitch_id} "
f"({campaign.name}): {exc}"
)
if crash_on_error:
raise
self.stderr.write(self.style.ERROR(msg))
return total
def _process_reward_campaign(
self,
campaign: SunkwiBotRewardCampaignSchema,
source: str,
*,
verbose: bool,
skip_existing: bool = False,
) -> None:
"""Process a single reward campaign from rewards.json.
Args:
campaign: Validated reward campaign schema.
source: Data source label.
verbose: Print per-record messages.
skip_existing: If True, skip if campaign already exists.
"""
starts_at_dt = parse_date(campaign.starts_at)
ends_at_dt = parse_date(campaign.ends_at)
# Resolve game if present
game_obj: Game | None = None
if campaign.game and isinstance(campaign.game, dict):
game_id = campaign.game.get("id")
if game_id:
with contextlib.suppress(Game.DoesNotExist):
game_obj = Game.objects.get(twitch_id=game_id)
defaults: dict[str, Any] = {
"name": campaign.name,
"brand": campaign.brand,
"starts_at": starts_at_dt,
"ends_at": ends_at_dt,
"status": campaign.status,
"summary": campaign.summary,
"instructions": campaign.instructions,
"external_url": campaign.external_url,
"reward_value_url_param": campaign.reward_value_url_param,
"about_url": campaign.about_url,
"is_sitewide": campaign.is_sitewide,
"game": game_obj,
"image_url": campaign.image.image1x_url if campaign.image else "",
"data_source": source,
}
existing_reward: RewardCampaign | None = RewardCampaign.objects.filter(
twitch_id=campaign.twitch_id,
).first()
# Merge mode: fill empty fields, never overwrite populated ones
if existing_reward is not None and getattr(self, "_merge_mode", False):
changed = False
merge_fields = {
"summary": campaign.summary,
"instructions": campaign.instructions,
"image_url": campaign.image.image1x_url if campaign.image else "",
"about_url": campaign.about_url,
"external_url": campaign.external_url,
}
for field, new_value in merge_fields.items():
old_value = getattr(existing_reward, field, None)
if not old_value and new_value:
setattr(existing_reward, field, new_value)
changed = True
if changed:
existing_reward.save(
update_fields=list(merge_fields.keys()),
)
if verbose:
action_str = "Merged" if changed else "Skipped (merged)"
display_name = (
f"{campaign.brand}: {campaign.name}"
if campaign.brand
else campaign.name
)
self.stdout.write(
f"{Fore.CYAN}→{Style.RESET_ALL} {action_str} reward "
f"campaign: {display_name}",
)
return
if skip_existing and existing_reward is not None:
if verbose:
self.stdout.write(
f"{Fore.YELLOW}→{Style.RESET_ALL} Skipped reward campaign: "
f"{campaign.name} (already exists)",
)
return
reward_obj, created = RewardCampaign.objects.get_or_create(
twitch_id=campaign.twitch_id,
defaults=defaults,
)
updated = self._save_if_changed(reward_obj, defaults) if not created else True
if verbose and (created or updated):
action = "Created" if created else "Updated"
display_name = (
f"{campaign.brand}: {campaign.name}"
if campaign.brand
else campaign.name
)
self.stdout.write(
f"{Fore.GREEN}✓{Style.RESET_ALL} {action} reward campaign: "
f"{display_name} (source: {source})",
)
# ------------------------------------------------------------------
# Shared helpers
# ------------------------------------------------------------------
def _save_if_changed(self, obj: Model, defaults: dict[str, Any]) -> bool: # type: ignore[valid-type]
"""Save the model instance only when data actually changed.
Args:
obj: The model instance to potentially update.
defaults: Field values to apply.
Returns:
True if the object was saved, False if no changes were detected.
"""
changed_fields: list[str] = []
for field, new_value in defaults.items():
if getattr(obj, field, None) != new_value:
setattr(obj, field, new_value)
changed_fields.append(field)
if not changed_fields:
return False
obj.save(update_fields=changed_fields)
return True
def _get_or_create_organization(
self,
twitch_id: str,
name: str,
*,
verbose: bool,
) -> Organization:
"""Get or create an organization.
Args:
twitch_id: Twitch organization ID.
name: Organization name.
verbose: Print per-record messages.
Returns:
Organization instance.
"""
cached = self._org_cache.get(twitch_id)
if cached is not None:
self._save_if_changed(cached, {"name": name})
return cached
org_obj, created = Organization.objects.get_or_create(
twitch_id=twitch_id,
defaults={"name": name},
)
if not created:
self._save_if_changed(org_obj, {"name": name})
if verbose and created:
self.stdout.write(
f"{Fore.GREEN}✓{Style.RESET_ALL} Created organization: {name}",
)
self._org_cache[twitch_id] = org_obj
return org_obj
def _get_or_create_game( # noqa: PLR0913
self,
twitch_id: str,
display_name: str,
box_art_url: str,
source: str, # noqa: ARG002
*,
verbose: bool,
org_obj: Organization | None = None,
) -> Game:
"""Get or create a game.
Args:
twitch_id: Twitch game ID.
display_name: Game display name.
box_art_url: Box art URL.
source: Data source label.
verbose: Print per-record messages.
org_obj: Optional organization to assign as owner of the game.
Returns:
Game instance.
"""
cached = self._game_cache.get(twitch_id)
if cached is not None:
self._save_if_changed(
cached,
{
"display_name": display_name,
"box_art": box_art_url,
},
)
return cached
defaults: dict[str, Any] = {
"display_name": display_name,
"name": display_name,
"box_art": box_art_url,
}
game_obj, created = Game.objects.get_or_create(
twitch_id=twitch_id,
defaults=defaults,
)
if not created:
self._save_if_changed(
game_obj,
{
"display_name": display_name,
"name": display_name,
"box_art": box_art_url,
},
)
# Assign owner organization if provided
if (
org_obj is not None
and not game_obj.owners.filter(
pk=org_obj.pk,
).exists()
):
game_obj.owners.add(org_obj)
if verbose:
self.stdout.write(
f"{Fore.GREEN}✓{Style.RESET_ALL} Assigned owner "
f"{org_obj.name} to game {display_name}",
)
if verbose and created:
self.stdout.write(
f"{Fore.GREEN}✓{Style.RESET_ALL} Created game: {display_name}",
)
self._game_cache[twitch_id] = game_obj
return game_obj

44
twitch/migrations/0024_dropcampaign_data_source_rewardcampaign_data_source.py Normal file
View file

@ -0,0 +1,44 @@
# Generated by Django 6.0.6 on 2026-06-14 13:39
from django.db import migrations
from django.db import models
class Migration(migrations.Migration):
"""Adds a `data_source` field to DropCampaign and RewardCampaign to track the origin of imported campaign data.
This allows us to distinguish between campaigns imported by different tools (e.g. TwitchDropsMiner vs. SunkwiBOT/twitch-drops-api) and can be useful for debugging, analytics, or future migrations that may need to handle data differently based on its source.
"""
dependencies = [
("twitch", "0023_alter_dropcampaign_url_lengths"),
]
operations = [
migrations.AddField(
model_name="dropcampaign",
name="data_source",
field=models.TextField(
choices=[
("TwitchDropsMiner", "TwitchDropsMiner"),
("SunkwiBOT/twitch-drops-api", "SunkwiBOT/twitch-drops-api"),
],
default="TwitchDropsMiner",
help_text="Identifies the origin tool/script that imported this campaign data.",
max_length=50,
),
),
migrations.AddField(
model_name="rewardcampaign",
name="data_source",
field=models.TextField(
choices=[
("TwitchDropsMiner", "TwitchDropsMiner"),
("SunkwiBOT/twitch-drops-api", "SunkwiBOT/twitch-drops-api"),
],
default="TwitchDropsMiner",
help_text="Identifies the origin tool/script that imported this campaign data.",
max_length=50,
),
),
]

23
twitch/models.py
View file

@ -705,6 +705,16 @@ class DropCampaign(auto_prefetch.Model): # noqa: PLR0904
auto_now=True,
)
data_source = models.TextField(
help_text="Identifies the origin tool/script that imported this campaign data.",
max_length=50,
choices=[
("TwitchDropsMiner", "TwitchDropsMiner"),
("SunkwiBOT/twitch-drops-api", "SunkwiBOT/twitch-drops-api"),
],
default="TwitchDropsMiner",
)
is_fully_imported = models.BooleanField(
help_text="True if all images and formats are imported and ready for display.",
default=False,
@ -912,6 +922,9 @@ class DropCampaign(auto_prefetch.Model): # noqa: PLR0904
"twitch_id",
"name",
"distribution_type",
"created_at",
"entitlement_limit",
"is_ios_available",
"image_asset_url",
"image_file",
"image_width",
@ -1806,6 +1819,16 @@ class RewardCampaign(auto_prefetch.Model):
default=False,
)
data_source = models.TextField(
help_text="Identifies the origin tool/script that imported this campaign data.",
max_length=50,
choices=[
("TwitchDropsMiner", "TwitchDropsMiner"),
("SunkwiBOT/twitch-drops-api", "SunkwiBOT/twitch-drops-api"),
],
default="TwitchDropsMiner",
)
game = auto_prefetch.ForeignKey(
help_text="Game associated with this reward campaign (if any).",
related_name="reward_campaigns",

260
twitch/schemas.py
View file

@ -649,9 +649,267 @@ class GlobalChatBadgesResponse(BaseModel):
data: list[ChatBadgeSetSchema]
# MARK: SunkwiBOT /twitch-drops-api Schemas
#
# The SunkwiBOT API serves simplified JSON without GraphQL metadata (no
# __typename, etc.). These schemas are relaxed to match that format.
class SunkwiBotChannelInfoSchema(BaseModel):
"""Schema for a channel inside a drop's allow.channels from the SunkwiBOT API."""
twitch_id: str = Field(alias="id")
display_name: str | None = Field(default=None, alias="displayName")
name: str
model_config = {
"extra": "forbid",
"validate_assignment": True,
"populate_by_name": True,
}
class SunkwiBotDropCampaignACLSchema(BaseModel):
"""Schema for the allow field within a SunkwiBOT reward (drop campaign)."""
channels: list[SunkwiBotChannelInfoSchema] | None = None
is_enabled: bool = Field(alias="isEnabled")
model_config = {
"extra": "forbid",
"validate_assignment": True,
"populate_by_name": True,
}
class SunkwiBotGameRefSchema(BaseModel):
"""Schema for a game object inside a SunkwiBOT reward."""
twitch_id: str = Field(alias="id")
slug: str
display_name: str = Field(alias="displayName")
model_config = {
"extra": "forbid",
"validate_assignment": True,
"populate_by_name": True,
}
class SunkwiBotOrganizationSchema(BaseModel):
"""Schema for an owner organization inside a SunkwiBOT reward."""
twitch_id: str = Field(alias="id")
name: str
model_config = {
"extra": "forbid",
"validate_assignment": True,
"populate_by_name": True,
}
class SunkwiBotBenefitSchema(BaseModel):
"""Schema for a benefit inside a benefitEdge from the SunkwiBOT API."""
twitch_id: str = Field(alias="id")
created_at: str | None = Field(default=None, alias="createdAt")
entitlement_limit: int = Field(alias="entitlementLimit")
game: dict | None = None
image_asset_url: str = Field(alias="imageAssetURL")
is_ios_available: bool = Field(alias="isIosAvailable")
name: str
owner_organization: dict | None = Field(default=None, alias="ownerOrganization")
distribution_type: str | None = Field(default=None, alias="distributionType")
model_config = {
"extra": "forbid",
"validate_assignment": True,
"populate_by_name": True,
}
class SunkwiBotBenefitEdgeSchema(BaseModel):
"""Schema for a benefitEdge within a timeBasedDrop from the SunkwiBOT API."""
benefit: SunkwiBotBenefitSchema
entitlement_limit: int = Field(alias="entitlementLimit")
model_config = {
"extra": "forbid",
"validate_assignment": True,
"populate_by_name": True,
}
class SunkwiBotTimeBasedDropSchema(BaseModel):
"""Schema for a timeBasedDrop within a SunkwiBOT reward (drop campaign)."""
twitch_id: str = Field(alias="id")
required_subs: int = Field(alias="requiredSubs")
benefit_edges: list[SunkwiBotBenefitEdgeSchema] = Field(
default=[],
alias="benefitEdges",
)
end_at: str = Field(alias="endAt")
name: str
precondition_drops: None = Field(default=None, alias="preconditionDrops")
required_minutes_watched: int | None = Field(
default=None,
alias="requiredMinutesWatched",
)
start_at: str = Field(alias="startAt")
model_config = {
"extra": "forbid",
"validate_assignment": True,
"populate_by_name": True,
}
@field_validator("benefit_edges", mode="before")
@classmethod
def handle_null_benefit_edges(cls, v: list | None) -> list:
"""Convert null benefitEdges to empty list.
Args:
v: The raw benefit_edges value (list or None).
Returns:
Empty list if None, otherwise the list itself.
"""
return v or []
class SunkwiBotRewardSchema(BaseModel):
"""Schema for a single reward item inside the drops.json rewards array.
This maps to a DropCampaign in our database.
"""
twitch_id: str = Field(alias="id")
self_field: dict | None = Field(default=None, alias="self")
allow: SunkwiBotDropCampaignACLSchema | None = None
account_link_url: str = Field(alias="accountLinkURL")
description: str
details_url: str = Field(alias="detailsURL")
end_at: str = Field(alias="endAt")
event_based_drops: list | None = Field(default=None, alias="eventBasedDrops")
game: SunkwiBotGameRefSchema
image_url: str = Field(alias="imageURL")
name: str
owner: SunkwiBotOrganizationSchema
start_at: str = Field(alias="startAt")
status: str
time_based_drops: list[SunkwiBotTimeBasedDropSchema] = Field(
default=[],
alias="timeBasedDrops",
)
model_config = {
"extra": "forbid",
"validate_assignment": True,
"populate_by_name": True,
}
class SunkwiBotDropGroupSchema(BaseModel):
"""Schema for a single top-level item in the drops.json array.
Each item groups multiple drop campaigns under a shared game.
"""
end_at: str = Field(alias="endAt")
game_box_art_url: str = Field(alias="gameBoxArtURL")
game_display_name: str = Field(alias="gameDisplayName")
game_id: str = Field(alias="gameId")
rewards: list[SunkwiBotRewardSchema]
start_at: str = Field(alias="startAt")
model_config = {
"extra": "forbid",
"validate_assignment": True,
"populate_by_name": True,
}
class SunkwiBotRewardImageSetSchema(BaseModel):
"""Schema for an image set inside a rewards.json reward."""
image1x_url: str = Field(alias="image1xURL")
model_config = {
"extra": "forbid",
"validate_assignment": True,
"populate_by_name": True,
}
class SunkwiBotIndividualRewardSchema(BaseModel):
"""Schema for a single reward inside a rewards.json reward campaign."""
twitch_id: str = Field(alias="id")
name: str
banner_image: SunkwiBotRewardImageSetSchema | None = Field(
default=None,
alias="bannerImage",
)
thumbnail_image: SunkwiBotRewardImageSetSchema | None = Field(
default=None,
alias="thumbnailImage",
)
earnable_until: str | None = Field(default=None, alias="earnableUntil")
redemption_instructions: str = Field(default="", alias="redemptionInstructions")
redemption_url: str = Field(default="", alias="redemptionURL")
model_config = {
"extra": "forbid",
"validate_assignment": True,
"populate_by_name": True,
}
class SunkwiBotRewardUnlockRequirementsSchema(BaseModel):
"""Schema for the unlockRequirements inside a rewards.json campaign."""
minute_watched_goal: int = Field(alias="minuteWatchedGoal")
subs_goal: int = Field(alias="subsGoal")
model_config = {
"extra": "forbid",
"validate_assignment": True,
"populate_by_name": True,
}
class SunkwiBotRewardCampaignSchema(BaseModel):
"""Schema for a single top-level item in the rewards.json array.
This maps to a RewardCampaign in our database.
"""
about_url: str = Field(alias="aboutURL")
brand: str
ends_at: str = Field(alias="endsAt")
external_url: str = Field(alias="externalURL")
game: dict | None = None
twitch_id: str = Field(alias="id")
image: SunkwiBotRewardImageSetSchema | None = None
instructions: str
is_sitewide: bool = Field(alias="isSitewide")
name: str
rewards: list[SunkwiBotIndividualRewardSchema] = Field(default=[])
reward_value_url_param: str = Field(default="", alias="rewardValueURLParam")
starts_at: str = Field(alias="startsAt")
status: str
summary: str
unlock_requirements: SunkwiBotRewardUnlockRequirementsSchema | None = Field(
default=None,
alias="unlockRequirements",
)
model_config = {
"extra": "forbid",
"validate_assignment": True,
"strict": True,
"populate_by_name": True,
}

69
twitch/tasks.py
View file

@ -10,14 +10,12 @@ import httpx
from celery import shared_task
from django.core.files.base import ContentFile
from django.core.management import call_command
from PIL.Image import Image
if TYPE_CHECKING:
from urllib.parse import ParseResult
from django.db.models.fields.files import ImageFieldFile
from PIL.Image import Image
from PIL.ImageFile import ImageFile
logger: logging.Logger = logging.getLogger("ttvdrops.tasks")
@ -104,37 +102,46 @@ def _convert_to_modern_formats(source: Path) -> None:
return
try:
from PIL import Image # noqa: PLC0415
except ImportError:
return
try:
with Image.open(source) as raw:
if raw.mode in {"RGBA", "LA"} or (
raw.mode == "P" and "transparency" in raw.info
):
background: Image = Image.new("RGB", raw.size, (255, 255, 255))
rgba: Image | ImageFile = (
raw.convert("RGBA") if raw.mode == "P" else raw
)
mask: Image | None = (
rgba.split()[-1] if rgba.mode in {"RGBA", "LA"} else None
)
background.paste(rgba, mask=mask)
img: Image = background
elif raw.mode != "RGB":
img = raw.convert("RGB")
else:
img: Image = raw.copy()
for fmt, ext in (("WEBP", ".webp"), ("AVIF", ".avif")):
out: Path = source.with_suffix(ext)
try:
img.save(out, fmt, quality=80)
except Exception: # noqa: BLE001
logger.debug("Could not convert %s to %s.", source, fmt)
img = _open_and_prepare_image(source)
except Exception: # noqa: BLE001
logger.debug("Format conversion failed for %s.", source)
return
_save_modern_formats(img, source)
def _open_and_prepare_image(source: Path) -> Image:
"""Open an image and convert it to an RGB-mode PIL Image.
Returns:
An RGB-mode PIL Image ready for saving in modern formats.
"""
from PIL import Image # noqa: PLC0415
with Image.open(source) as raw:
if raw.mode in {"RGBA", "LA"} or (
raw.mode == "P" and "transparency" in raw.info
):
background: Image = Image.new("RGB", raw.size, (255, 255, 255))
rgba: Image = raw.convert("RGBA") if raw.mode == "P" else raw
mask: Image | None = (
rgba.split()[-1] if rgba.mode in {"RGBA", "LA"} else None
)
background.paste(rgba, mask=mask)
return background
if raw.mode != "RGB":
return raw.convert("RGB")
return raw.copy()
def _save_modern_formats(img: Image, source: Path) -> None:
"""Save *img* as WebP and AVIF alongside the original *source*."""
for fmt, ext in (("WEBP", ".webp"), ("AVIF", ".avif")):
out: Path = source.with_suffix(ext)
try:
img.save(out, fmt, quality=80)
except Exception: # noqa: BLE001
logger.debug("Could not convert %s to %s.", source, fmt)
# ---------------------------------------------------------------------------

389
twitch/tests/test_api.py
View file

@ -190,6 +190,106 @@ class TwitchApiV1TestCase(TestCase):
assert data["items"][0]["status"] == "active"
assert data["items"][0]["game"]["twitch_id"] == "game123"
def test_v1_campaign_list_filters_by_game(self) -> None:
"""Filter campaigns by game twitch_id."""
response = self.client.get(
f"/twitch/api/v1/campaigns/?game={self.game.twitch_id}",
)
assert response.status_code == 200
data = response.json()
assert data["total"] == 1
assert data["items"][0]["twitch_id"] == "campaign123"
def test_v1_campaign_list_game_filter_with_no_matches(self) -> None:
"""Return empty list when game filter matches no campaigns."""
response = self.client.get("/twitch/api/v1/campaigns/?game=nonexistent")
assert response.status_code == 200
data = response.json()
assert data["total"] == 0
assert data["items"] == []
def test_v1_campaign_list_status_upcoming(self) -> None:
"""Filter campaigns by status=upcoming."""
now = timezone.now()
DropCampaign.objects.create(
twitch_id="upcoming_campaign",
name="Upcoming Campaign",
game=self.game,
start_at=now + timedelta(days=1),
end_at=now + timedelta(days=2),
operation_names=["DropCampaignDetails"],
is_fully_imported=True,
)
response = self.client.get("/twitch/api/v1/campaigns/?status=upcoming")
assert response.status_code == 200
data = response.json()
assert data["total"] == 1
assert data["items"][0]["status"] == "upcoming"
def test_v1_campaign_list_status_expired(self) -> None:
"""Filter campaigns by status=expired."""
now = timezone.now()
DropCampaign.objects.create(
twitch_id="expired_campaign",
name="Expired Campaign",
game=self.game,
start_at=now - timedelta(days=3),
end_at=now - timedelta(days=1),
operation_names=["DropCampaignDetails"],
is_fully_imported=True,
)
response = self.client.get("/twitch/api/v1/campaigns/?status=expired")
assert response.status_code == 200
data = response.json()
assert data["total"] == 1
assert data["items"][0]["status"] == "expired"
def test_v1_campaign_list_default_page_size(self) -> None:
"""Return all campaigns when no page_size is specified."""
response = self.client.get("/twitch/api/v1/campaigns/")
assert response.status_code == 200
data = response.json()
assert data["total"] >= 1
assert data["page"] == 1
def test_v1_campaign_list_page_size_param(self) -> None:
"""Respect custom page_size parameter."""
response = self.client.get("/twitch/api/v1/campaigns/?page_size=1")
assert response.status_code == 200
data = response.json()
assert data["page_size"] == 1
assert len(data["items"]) == 1
def test_v1_campaign_list_summary_fields(self) -> None:
"""Return correct field shape for campaign summary items."""
response = self.client.get("/twitch/api/v1/campaigns/")
assert response.status_code == 200
data = response.json()
item = data["items"][0]
assert item["twitch_id"] == "campaign123"
assert item["name"] == "Test Campaign"
assert item["description"] == "A test campaign"
assert item["status"] == "active"
assert item["image_url"] == "https://example.com/campaign.png"
assert item["details_url"] == "https://example.com/details"
assert item["account_link_url"] == "https://example.com/link"
assert item["allow_is_enabled"] is True
assert item["is_fully_imported"] is True
assert "start_at" in item
assert "end_at" in item
assert "added_at" in item
assert "updated_at" in item
assert item["game"]["twitch_id"] == "game123"
def test_v1_campaign_detail(self) -> None:
"""Return nested campaign detail data from the v1 endpoint."""
response = self.client.get("/twitch/api/v1/campaigns/campaign123/")
@ -228,6 +328,35 @@ class TwitchApiV1TestCase(TestCase):
data = response.json()
assert data["game"]["box_art_url"] == self.game.box_art_file.url
def test_v1_campaign_detail_avoids_n_plus_one_on_benefits(self) -> None:
"""Fetch campaign detail without N+1 queries on DropBenefit fields.
Regression test: _serialize_benefit accesses created_at,
entitlement_limit, and is_ios_available; these must be included
in the Prefetch .only() in DropCampaign.for_detail_view to avoid
one extra query per benefit.
"""
with CaptureQueriesContext(connection) as capture:
response = self.client.get("/twitch/api/v1/campaigns/campaign123/")
assert response.status_code == 200
data = response.json()
benefits = data["drops"][0]["benefits"]
assert len(benefits) == 1
assert benefits[0]["created_at"] is None
assert benefits[0]["entitlement_limit"] == 1
assert benefits[0]["is_ios_available"] is False
# Expect 7 queries (5 core prefetches + 2 campaign COUNTs):
# 1. campaign + game (select_related)
# 2. game owners (Prefetch → owners_for_detail)
# 3. allow_channels (Prefetch → channels_ordered)
# 4. time_based_drops (Prefetch)
# 5. benefits through DropBenefitEdge (Prefetch)
# 6. campaign count for game
# 7. active campaign count for game
assert len(capture) <= 7
def test_v1_all_endpoints_handle_multiple_rows(self) -> None:
"""Exercise all v1 routes with enough rows to catch deferred loads."""
self._create_secondary_api_fixture()
@ -436,3 +565,263 @@ class TwitchApiV1TestCase(TestCase):
)
assert f'href="{campaign_api_url}"' in content
assert 'title="Twitch campaign API">[api]</a>' in content
# ── Games ──────────────────────────────────────────────────────────
def test_v1_game_list_fields(self) -> None:
"""Return correct field shape for game list items."""
response = self.client.get("/twitch/api/v1/games/")
assert response.status_code == 200
data = response.json()
assert data["total"] >= 1
item = data["items"][0]
assert item["twitch_id"] == "game123"
assert item["slug"] == "test-game"
assert item["name"] == "Test Game"
assert item["display_name"] == "Test Game"
assert isinstance(item["box_art_url"], str)
assert isinstance(item["organizations"], list)
assert item["organizations"][0]["twitch_id"] == "org123"
assert item["campaign_count"] >= 1
assert item["active_campaign_count"] >= 1
assert "added_at" in item
assert "updated_at" in item
def test_v1_game_list_pagination(self) -> None:
"""Paginate game list results."""
response = self.client.get("/twitch/api/v1/games/?page_size=1")
assert response.status_code == 200
data = response.json()
assert data["page_size"] == 1
assert len(data["items"]) == 1
def test_v1_game_detail_campaigns_and_owners(self) -> None:
"""Return campaigns and organizations in game detail."""
response = self.client.get("/twitch/api/v1/games/game123/")
assert response.status_code == 200
data = response.json()
assert data["campaigns"][0]["twitch_id"] == "campaign123"
assert data["campaigns"][0]["status"] == "active"
assert data["organizations"][0]["twitch_id"] == "org123"
def test_v1_game_detail_not_found(self) -> None:
"""Return 404 for missing game."""
response = self.client.get("/twitch/api/v1/games/nonexistent/")
assert response.status_code == 404
# ── Organizations ───────────────────────────────────────────────────
def test_v1_organization_list_fields(self) -> None:
"""Return correct field shape for organization list items."""
response = self.client.get("/twitch/api/v1/organizations/")
assert response.status_code == 200
data = response.json()
assert data["total"] >= 1
item = data["items"][0]
assert item["twitch_id"] == "org123"
assert item["name"] == "Test Organization"
assert "added_at" in item
assert "updated_at" in item
def test_v1_organization_detail_not_found(self) -> None:
"""Return 404 for missing organization."""
response = self.client.get("/twitch/api/v1/organizations/nonexistent/")
assert response.status_code == 404
# ── Channels ────────────────────────────────────────────────────────
def test_v1_channel_list_search(self) -> None:
"""Filter channels by search query."""
response = self.client.get("/twitch/api/v1/channels/?search=testchannel")
assert response.status_code == 200
data = response.json()
assert data["total"] == 1
assert data["items"][0]["twitch_id"] == "channel123"
def test_v1_channel_list_search_no_match(self) -> None:
"""Return empty list when search matches no channels."""
response = self.client.get("/twitch/api/v1/channels/?search=zzzzz")
assert response.status_code == 200
data = response.json()
assert data["total"] == 0
assert data["items"] == []
def test_v1_channel_list_fields(self) -> None:
"""Return correct field shape for channel list items."""
response = self.client.get("/twitch/api/v1/channels/")
assert response.status_code == 200
data = response.json()
item = data["items"][0]
assert item["twitch_id"] == "channel123"
assert item["name"] == "testchannel"
assert item["display_name"] == "TestChannel"
assert item["allowed_campaign_count"] == 1
assert "added_at" in item
assert "updated_at" in item
def test_v1_channel_detail_campaigns(self) -> None:
"""Return campaign summaries in channel detail."""
response = self.client.get("/twitch/api/v1/channels/channel123/")
assert response.status_code == 200
data = response.json()
assert len(data["campaigns"]) == 1
assert data["campaigns"][0]["twitch_id"] == "campaign123"
assert data["campaigns"][0]["game"]["twitch_id"] == "game123"
def test_v1_channel_detail_not_found(self) -> None:
"""Return 404 for missing channel."""
response = self.client.get("/twitch/api/v1/channels/nonexistent/")
assert response.status_code == 404
# ── Reward Campaigns ────────────────────────────────────────────────
def test_v1_reward_campaign_list_filters_by_game(self) -> None:
"""Filter reward campaigns by game twitch_id."""
response = self.client.get(
f"/twitch/api/v1/reward-campaigns/?game={self.game.twitch_id}",
)
assert response.status_code == 200
data = response.json()
assert data["total"] == 1
assert data["items"][0]["twitch_id"] == "reward123"
def test_v1_reward_campaign_list_status_active(self) -> None:
"""Filter reward campaigns by status=active."""
response = self.client.get("/twitch/api/v1/reward-campaigns/?status=active")
assert response.status_code == 200
data = response.json()
assert data["total"] == 1
assert data["items"][0]["computed_status"] == "active"
def test_v1_reward_campaign_list_status_expired(self) -> None:
"""Filter reward campaigns by status=expired."""
now = timezone.now()
RewardCampaign.objects.create(
twitch_id="expired_reward",
name="Expired Reward",
brand="Expired Brand",
starts_at=now - timedelta(days=3),
ends_at=now - timedelta(days=1),
status="ACTIVE",
summary="Expired summary",
)
response = self.client.get("/twitch/api/v1/reward-campaigns/?status=expired")
assert response.status_code == 200
data = response.json()
assert data["total"] == 1
assert data["items"][0]["computed_status"] == "expired"
def test_v1_reward_campaign_list_fields(self) -> None:
"""Return correct field shape for reward campaign list items."""
response = self.client.get("/twitch/api/v1/reward-campaigns/")
assert response.status_code == 200
data = response.json()
item = data["items"][0]
assert item["twitch_id"] == "reward123"
assert item["name"] == "Test Reward"
assert item["brand"] == "Test Brand"
assert item["status"] == "ACTIVE"
assert item["computed_status"] == "active"
assert item["summary"] == "Reward summary"
assert item["is_sitewide"] is False
assert item["game"]["twitch_id"] == "game123"
assert "starts_at" in item
assert "ends_at" in item
assert "added_at" in item
assert "updated_at" in item
def test_v1_reward_campaign_detail_fields(self) -> None:
"""Return correct field shape for reward campaign detail."""
response = self.client.get("/twitch/api/v1/reward-campaigns/reward123/")
assert response.status_code == 200
data = response.json()
assert data["twitch_id"] == "reward123"
assert data["name"] == "Test Reward"
assert data["brand"] == "Test Brand"
assert isinstance(data["instructions"], str)
assert isinstance(data["external_url"], str)
assert isinstance(data["about_url"], str)
assert data["game"]["twitch_id"] == "game123"
def test_v1_reward_campaign_detail_no_game(self) -> None:
"""Return reward campaign detail with null game."""
now = timezone.now()
RewardCampaign.objects.create(
twitch_id="no_game_reward",
name="No Game Reward",
brand="Standalone Brand",
starts_at=now - timedelta(days=1),
ends_at=now + timedelta(days=1),
status="ACTIVE",
summary="No game attached",
is_sitewide=True,
)
response = self.client.get("/twitch/api/v1/reward-campaigns/no_game_reward/")
assert response.status_code == 200
data = response.json()
assert data["game"] is None
assert data["is_sitewide"] is True
def test_v1_reward_campaign_detail_not_found(self) -> None:
"""Return 404 for missing reward campaign."""
response = self.client.get("/twitch/api/v1/reward-campaigns/nonexistent/")
assert response.status_code == 404
# ── Badges ──────────────────────────────────────────────────────────
def test_v1_badge_list_fields(self) -> None:
"""Return correct field shape for badge set list items."""
response = self.client.get("/twitch/api/v1/badges/")
assert response.status_code == 200
data = response.json()
assert data["total"] >= 1
item = data["items"][0]
assert item["set_id"] == "test-badge-set"
assert len(item["badges"]) == 1
assert "added_at" in item
assert "updated_at" in item
def test_v1_badge_detail_fields(self) -> None:
"""Return correct field shape for badge set detail."""
response = self.client.get("/twitch/api/v1/badges/test-badge-set/")
assert response.status_code == 200
data = response.json()
assert data["set_id"] == "test-badge-set"
assert len(data["badges"]) == 1
badge = data["badges"][0]
assert badge["badge_id"] == "1"
assert badge["title"] == "Test Badge"
assert badge["description"] == "Test badge description"
assert "image_url_1x" in badge
assert "image_url_2x" in badge
assert "image_url_4x" in badge
assert "click_action" in badge
assert "click_url" in badge
def test_v1_badge_detail_not_found(self) -> None:
"""Return 404 for missing badge set."""
response = self.client.get("/twitch/api/v1/badges/nonexistent/")
assert response.status_code == 404

714
twitch/tests/test_import_twitch_drops_api.py Normal file
View file

@ -0,0 +1,714 @@
"""Tests for the import_twitch_drops_api management command."""
from __future__ import annotations
from typing import Self
from unittest.mock import MagicMock
from unittest.mock import patch
import httpx
import pytest
from django.test import TestCase
from twitch.management.commands.import_twitch_drops_api import Command
from twitch.models import DropBenefit
from twitch.models import DropBenefitEdge
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
# ---------------------------------------------------------------------------
# Sample data matching the SunkwiBOT/twitch-drops-api JSON formats
# ---------------------------------------------------------------------------
SAMPLE_DROPS_JSON: list[dict] = [
{
"endAt": "2026-06-21T10:59:00Z",
"gameBoxArtURL": "https://static-cdn.jtvnw.net/ttv-boxart/515025_IGDB-285x380.jpg",
"gameDisplayName": "Overwatch 2",
"gameId": "515025",
"rewards": [
{
"id": "reward-campaign-1",
"self": None,
"allow": {
"channels": [
{
"id": "123456789",
"displayName": "SomeStreamer",
"name": "somestreamer",
},
],
"isEnabled": True,
},
"accountLinkURL": "https://www.twitch.tv/drops/campaigns",
"description": "Earn Overwatch 2 drops by watching!",
"detailsURL": "https://www.twitch.tv/drops/campaigns/reward-campaign-1",
"endAt": "2026-06-21T10:59:00Z",
"eventBasedDrops": [],
"game": {
"id": "515025",
"slug": "overwatch-2",
"displayName": "Overwatch 2",
},
"imageURL": "https://example.com/campaign.png",
"name": "Overwatch 2 x Twitch Drops",
"owner": {
"id": "org-blizzard",
"name": "Blizzard Entertainment",
},
"startAt": "2026-06-12T15:00:00Z",
"status": "ACTIVE",
"timeBasedDrops": [
{
"id": "time-drop-1",
"requiredSubs": 0,
"benefitEdges": [
{
"benefit": {
"id": "benefit-1",
"createdAt": "2026-06-10T12:00:00Z",
"entitlementLimit": 1,
"game": {
"id": "515025",
"name": "Overwatch 2",
},
"imageAssetURL": "https://example.com/reward.png",
"isIosAvailable": True,
"name": "Ana's Cybermedic Skin",
"ownerOrganization": {
"id": "org-blizzard",
"name": "Blizzard Entertainment",
},
"distributionType": "MANUAL",
},
"entitlementLimit": 1,
},
],
"endAt": "2026-06-21T10:59:00Z",
"name": "Watch 2 hours",
"preconditionDrops": None,
"requiredMinutesWatched": 120,
"startAt": "2026-06-12T15:00:00Z",
},
],
},
],
"startAt": "2026-06-12T15:00:00Z",
},
]
SAMPLE_REWARDS_JSON: list[dict] = [
{
"aboutURL": "https://example.com/redeem",
"brand": "Mojang",
"endsAt": "2026-06-15T05:59:59.999Z",
"externalURL": "https://example.com/redeem",
"game": {
"displayName": "Minecraft",
"id": "27471",
"slug": "minecraft",
},
"id": "reward-campaign-2",
"image": {
"image1xURL": "https://example.com/reward-campaign.png",
},
"instructions": "",
"isSitewide": False,
"name": "Tubbo's WatchTime",
"rewards": [],
"rewardValueURLParam": "",
"startsAt": "2026-05-31T16:00:00Z",
"status": "UNKNOWN",
"summary": "Watch 5 minutes of Minecraft gameplay!",
"unlockRequirements": {
"minuteWatchedGoal": 5,
"subsGoal": 0,
},
},
]
def mock_httpx_get(data: list[dict]) -> MagicMock:
"""Return a mock httpx.Response that returns the given JSON data.
Args:
data: The JSON data to return from response.json().
Returns:
A MagicMock configured as an httpx response.
"""
mock_response = MagicMock()
mock_response.json.return_value = data
mock_response.raise_for_status.return_value = None
return mock_response
class MockResponse:
"""Minimal mock for httpx.Response used as a context-manager return."""
def __init__(self, json_data: list[dict]) -> None:
"""Initialise with JSON data to return from .json().
Args:
json_data: The data to return.
"""
self._json_data = json_data
def json(self) -> list[dict]:
"""Return the mocked JSON data.
Returns:
The JSON data.
"""
return self._json_data
def raise_for_status(self) -> None:
"""No-op for mock."""
class MockHttpxClient:
"""Mock for httpx.Client that returns canned data from its get() method."""
def __init__(self, drops_data: list[dict], rewards_data: list[dict]) -> None:
"""Initialise with data for each endpoint.
Args:
drops_data: Data to return for the drops URL.
rewards_data: Data to return for the rewards URL.
"""
self.drops_data = drops_data
self.rewards_data = rewards_data
def get(self, url: str, **kwargs: object) -> MockResponse:
"""Return canned data based on the URL.
Args:
url: The request URL.
**kwargs: Ignored.
Returns:
A MockResponse with the appropriate data.
Raises:
ValueError: If the URL is not recognised.
"""
if "drops.json" in url:
return MockResponse(self.drops_data)
if "rewards.json" in url:
return MockResponse(self.rewards_data)
msg = f"Unexpected URL: {url}"
raise ValueError(msg)
def __enter__(self) -> Self:
"""Support context manager usage.
Returns:
self.
"""
return self
def __exit__(self, *args: object) -> None:
"""No-op for context manager."""
@pytest.mark.no_zeal
class ImportTwitchDropsApiTests(TestCase):
"""Tests for the import_twitch_drops_api management command."""
drops_url = (
"https://raw.githubusercontent.com/SunkwiBOT/twitch-drops-api/main/drops.json"
)
rewards_url = (
"https://raw.githubusercontent.com/SunkwiBOT/twitch-drops-api/main/rewards.json"
)
def setUp(self) -> None:
"""Set up the command instance for each test."""
self.command = Command()
self.command._org_cache = {}
self.command._game_cache = {}
self.command._channel_cache = {}
self.command._benefit_cache = {}
# ------------------------------------------------------------------
# drops.json import tests
# ------------------------------------------------------------------
@patch("httpx.Client")
def test_import_drops_creates_campaign(self, mock_client: MagicMock) -> None:
"""Verify a basic drops.json import creates a DropCampaign with correct data."""
mock_client.return_value = MockHttpxClient(
drops_data=SAMPLE_DROPS_JSON,
rewards_data=[],
)
self.command.handle(
drops_url=self.drops_url,
rewards_url=self.rewards_url,
drops_only=True,
rewards_only=False,
source="SunkwiBOT/twitch-drops-api",
verbose=False,
crash_on_error=True,
)
# Game
game = Game.objects.get(twitch_id="515025")
assert game.display_name == "Overwatch 2"
# Organization
org = Organization.objects.get(twitch_id="org-blizzard")
assert org.name == "Blizzard Entertainment"
# DropCampaign
campaign = DropCampaign.objects.get(twitch_id="reward-campaign-1")
assert campaign.name == "Overwatch 2 x Twitch Drops"
assert campaign.data_source == "SunkwiBOT/twitch-drops-api"
assert campaign.is_fully_imported is True
# TimeBasedDrop
tbd = TimeBasedDrop.objects.get(twitch_id="time-drop-1")
assert tbd.name == "Watch 2 hours"
assert tbd.required_minutes_watched == 120
# DropBenefit
benefit = DropBenefit.objects.get(twitch_id="benefit-1")
assert benefit.name == "Ana's Cybermedic Skin"
assert benefit.distribution_type == "MANUAL"
# DropBenefitEdge
edge = DropBenefitEdge.objects.get(drop=tbd, benefit=benefit)
assert edge.entitlement_limit == 1
@patch("httpx.Client")
def test_import_drops_idempotent(self, mock_client: MagicMock) -> None:
"""Verify running the import twice doesn't create duplicates."""
mock_client.return_value = MockHttpxClient(
drops_data=SAMPLE_DROPS_JSON,
rewards_data=[],
)
# First run
self.command.handle(
drops_url=self.drops_url,
rewards_url=self.rewards_url,
drops_only=True,
rewards_only=False,
source="SunkwiBOT/twitch-drops-api",
verbose=False,
crash_on_error=True,
)
# Second run
self.command.handle(
drops_url=self.drops_url,
rewards_url=self.rewards_url,
drops_only=True,
rewards_only=False,
source="SunkwiBOT/twitch-drops-api",
verbose=False,
crash_on_error=True,
)
assert DropCampaign.objects.count() == 1
assert Game.objects.count() == 1
assert Organization.objects.count() == 1
assert TimeBasedDrop.objects.count() == 1
assert DropBenefit.objects.count() == 1
assert DropBenefitEdge.objects.count() == 1
@patch("httpx.Client")
def test_import_drops_sets_data_source(self, mock_client: MagicMock) -> None:
"""Verify the --source argument is stored on the DropCampaign."""
mock_client.return_value = MockHttpxClient(
drops_data=SAMPLE_DROPS_JSON,
rewards_data=[],
)
self.command.handle(
drops_url=self.drops_url,
rewards_url=self.rewards_url,
drops_only=True,
rewards_only=False,
source="CustomSource",
verbose=False,
crash_on_error=True,
)
campaign = DropCampaign.objects.get(twitch_id="reward-campaign-1")
assert campaign.data_source == "CustomSource"
# ------------------------------------------------------------------
# rewards.json import tests
# ------------------------------------------------------------------
@patch("httpx.Client")
def test_import_rewards_creates_reward_campaign(
self,
mock_client: MagicMock,
) -> None:
"""Verify a basic rewards.json import creates a RewardCampaign."""
# Create the game first since rewards.json references existing games
Game.objects.create(
twitch_id="27471",
display_name="Minecraft",
name="Minecraft",
)
mock_client.return_value = MockHttpxClient(
drops_data=[],
rewards_data=SAMPLE_REWARDS_JSON,
)
self.command.handle(
drops_url=self.drops_url,
rewards_url=self.rewards_url,
drops_only=False,
rewards_only=True,
source="SunkwiBOT/twitch-drops-api",
verbose=False,
crash_on_error=True,
)
reward = RewardCampaign.objects.get(twitch_id="reward-campaign-2")
assert reward.name == "Tubbo's WatchTime"
assert reward.brand == "Mojang"
assert reward.data_source == "SunkwiBOT/twitch-drops-api"
assert reward.is_sitewide is False
assert reward.game is not None
assert reward.game.twitch_id == "27471"
@patch("httpx.Client")
def test_import_rewards_sets_data_source(self, mock_client: MagicMock) -> None:
"""Verify the --source argument is stored on the RewardCampaign."""
mock_client.return_value = MockHttpxClient(
drops_data=[],
rewards_data=SAMPLE_REWARDS_JSON,
)
self.command.handle(
drops_url=self.drops_url,
rewards_url=self.rewards_url,
drops_only=False,
rewards_only=True,
source="CustomSource",
verbose=False,
crash_on_error=True,
)
reward = RewardCampaign.objects.get(twitch_id="reward-campaign-2")
assert reward.data_source == "CustomSource"
@patch("httpx.Client")
def test_import_rewards_idempotent(self, mock_client: MagicMock) -> None:
"""Verify running the rewards import twice doesn't create duplicates."""
Game.objects.create(
twitch_id="27471",
display_name="Minecraft",
name="Minecraft",
)
mock_client.return_value = MockHttpxClient(
drops_data=[],
rewards_data=SAMPLE_REWARDS_JSON,
)
# First run
self.command.handle(
drops_url=self.drops_url,
rewards_url=self.rewards_url,
drops_only=False,
rewards_only=True,
source="SunkwiBOT/twitch-drops-api",
verbose=False,
crash_on_error=True,
)
# Second run
self.command.handle(
drops_url=self.drops_url,
rewards_url=self.rewards_url,
drops_only=False,
rewards_only=True,
source="SunkwiBOT/twitch-drops-api",
verbose=False,
crash_on_error=True,
)
assert RewardCampaign.objects.count() == 1
# ------------------------------------------------------------------
# Combined import tests
# ------------------------------------------------------------------
@patch("httpx.Client")
def test_import_both(self, mock_client: MagicMock) -> None:
"""Verify importing both drops.json and rewards.json works together."""
Game.objects.create(
twitch_id="27471",
display_name="Minecraft",
name="Minecraft",
)
mock_client.return_value = MockHttpxClient(
drops_data=SAMPLE_DROPS_JSON,
rewards_data=SAMPLE_REWARDS_JSON,
)
self.command.handle(
drops_url=self.drops_url,
rewards_url=self.rewards_url,
drops_only=False,
rewards_only=False,
source="SunkwiBOT/twitch-drops-api",
verbose=False,
crash_on_error=True,
)
assert DropCampaign.objects.count() == 1
assert RewardCampaign.objects.count() == 1
assert Game.objects.count() == 2 # 515025 + 27471
# ------------------------------------------------------------------
# Error handling tests
# ------------------------------------------------------------------
@patch("httpx.Client")
def test_handles_invalid_json(self, mock_client: MagicMock) -> None:
"""Verify the command handles a non-list response gracefully."""
mock_response = MagicMock()
mock_response.json.return_value = {"not": "a list"}
mock_response.raise_for_status.return_value = None
mock_client.return_value.__enter__.return_value.get.return_value = mock_response
# The error is caught internally and logged; no data should be imported
self.command.handle(
drops_url=self.drops_url,
rewards_url=self.rewards_url,
drops_only=True,
rewards_only=False,
source="SunkwiBOT/twitch-drops-api",
verbose=False,
crash_on_error=False,
)
assert DropCampaign.objects.count() == 0
@patch("httpx.Client")
def test_handles_http_error(self, mock_client: MagicMock) -> None:
"""Verify the command handles HTTP errors gracefully."""
mock_response = MagicMock()
mock_response.raise_for_status.side_effect = httpx.HTTPStatusError(
"404 Not Found",
request=MagicMock(),
response=MagicMock(),
)
mock_client.return_value.__enter__.return_value.get.return_value = mock_response
# Should not crash when crash_on_error is False
self.command.handle(
drops_url=self.drops_url,
rewards_url=self.rewards_url,
drops_only=False,
rewards_only=False,
source="SunkwiBOT/twitch-drops-api",
verbose=False,
crash_on_error=False,
)
# No data should have been imported
assert DropCampaign.objects.count() == 0
assert RewardCampaign.objects.count() == 0
# ---------------------------------------------------------------------------
# _strip_typename tests
# ---------------------------------------------------------------------------
class StripTypenameTests(TestCase):
"""Tests for Command._strip_typename."""
def test_removes_typename_from_dict(self) -> None:
"""Verify __typename is removed from a flat dict."""
result = Command._strip_typename({
"id": "123",
"__typename": "DropCampaign",
"name": "Test",
})
assert result == {"id": "123", "name": "Test"}
def test_removes_typename_nested(self) -> None:
"""Verify __typename is removed from nested dicts."""
result = Command._strip_typename({
"game": {
"id": "g1",
"__typename": "Game",
},
"__typename": "DropCampaign",
})
assert result == {"game": {"id": "g1"}}
def test_removes_typename_in_lists(self) -> None:
"""Verify __typename is removed from items inside lists."""
result = Command._strip_typename({
"rewards": [
{"id": "r1", "__typename": "Reward"},
{"id": "r2", "__typename": "Reward"},
],
})
assert result == {"rewards": [{"id": "r1"}, {"id": "r2"}]}
def test_handles_scalars(self) -> None:
"""Verify scalars and None pass through unchanged."""
assert Command._strip_typename("hello") == "hello"
assert Command._strip_typename(42) == 42
assert Command._strip_typename(None) is None
assert Command._strip_typename([1, 2, 3]) == [1, 2, 3]
def test_removes_typename_deeply_nested(self) -> None:
"""Verify deeply nested __typename is removed."""
data = {
"data": {
"user": {
"id": "u1",
"__typename": "User",
"campaigns": [
{
"id": "c1",
"__typename": "DropCampaign",
"game": {
"id": "g1",
"__typename": "Game",
},
},
],
},
},
"__typename": "Response",
}
result = Command._strip_typename(data)
# Should have no __typename anywhere
raw = str(result)
assert "__typename" not in raw
# ---------------------------------------------------------------------------
# Historical import tests
# ---------------------------------------------------------------------------
@pytest.mark.no_zeal
class HistoricalImportTests(TestCase):
"""Tests for the --historical mode of import_twitch_drops_api."""
def _init_caches(self, command: Command) -> None:
"""Initialise instance caches needed by import helpers."""
command._org_cache = {}
command._game_cache = {}
command._channel_cache = {}
command._benefit_cache = {}
def test_parse_and_import_drops_from_raw(self) -> None:
"""Verify _parse_and_import_drops works with pre-loaded data."""
command = Command()
self._init_caches(command)
count = command._parse_and_import_drops(
raw_data=SAMPLE_DROPS_JSON,
source="SunkwiBOT/twitch-drops-api",
verbose=False,
crash_on_error=True,
)
assert count == 1
assert DropCampaign.objects.count() == 1
campaign = DropCampaign.objects.get(twitch_id="reward-campaign-1")
assert campaign.data_source == "SunkwiBOT/twitch-drops-api"
def test_parse_and_import_rewards_from_raw(self) -> None:
"""Verify _parse_and_import_rewards works with pre-loaded data."""
Game.objects.create(twitch_id="27471", display_name="Minecraft")
command = Command()
self._init_caches(command)
count = command._parse_and_import_rewards(
raw_data=SAMPLE_REWARDS_JSON,
source="SunkwiBOT/twitch-drops-api",
verbose=False,
crash_on_error=True,
)
assert count == 1
assert RewardCampaign.objects.count() == 1
reward = RewardCampaign.objects.get(twitch_id="reward-campaign-2")
assert reward.data_source == "SunkwiBOT/twitch-drops-api"
def test_parse_and_import_drops_handles_typename(self) -> None:
"""Verify _parse_and_import_drops strips __typename before validation."""
raw = [
{
"endAt": "2026-06-21T10:59:00Z",
"gameBoxArtURL": "https://example.com/art.png",
"gameDisplayName": "Test Game",
"gameId": "test-game-1",
"rewards": [
{
"id": "hist-campaign-1",
"self": None,
"allow": {"isEnabled": True},
"accountLinkURL": "https://example.com",
"description": "Historical",
"detailsURL": "https://example.com",
"endAt": "2026-06-21T10:59:00Z",
"eventBasedDrops": [],
"game": {
"id": "test-game-1",
"slug": "test",
"displayName": "Test Game",
},
"imageURL": "",
"name": "Historical Campaign",
"owner": {"id": "org-1", "name": "Test Org"},
"startAt": "2026-06-12T15:00:00Z",
"status": "ACTIVE",
"timeBasedDrops": [],
"__typename": "DropCampaign",
},
],
"startAt": "2026-06-12T15:00:00Z",
"__typename": "DropGroup",
},
]
command = Command()
self._init_caches(command)
count = command._parse_and_import_drops(
raw_data=raw,
source="SunkwiBOT/twitch-drops-api",
verbose=False,
crash_on_error=True,
)
assert count == 1
assert DropCampaign.objects.filter(twitch_id="hist-campaign-1").exists()
@patch.object(Command, "_process_historical")
def test_historical_flag_calls_process_historical(
self,
mock_process: MagicMock,
) -> None:
"""Verify --historical calls _process_historical instead of fetching URLs."""
mock_process.return_value = (5, 2, [])
command = Command()
command.handle(
historical=True,
source="SunkwiBOT/twitch-drops-api",
verbose=False,
crash_on_error=False,
drops_url="",
rewards_url="",
drops_only=False,
rewards_only=False,
no_skip_latest=False,
max_commits=0,
git_dir="",
)
mock_process.assert_called_once()

10
twitch/tests/test_migrations.py
View file

@ -5,7 +5,6 @@ from typing import TYPE_CHECKING
import pytest
from django.db import connection
from django.db.migrations.executor import MigrationExecutor
from django.db.migrations.state import StateApps
if TYPE_CHECKING:
from django.db.migrations.state import StateApps
@ -84,3 +83,12 @@ def test_0021_backfills_allowed_campaign_count() -> None: # noqa: PLR0914
"migration_backfill_channel_2": 1,
"migration_backfill_channel_3": 0,
}
# Restore database to the latest migration so subsequent tests don't
# encounter missing columns (SQLite DDL persists across transaction
# rollbacks, so the schema changes made by this test are permanent).
latest: list[tuple[str, str]] = [
("twitch", "0024_dropcampaign_data_source_rewardcampaign_data_source"),
]
executor = MigrationExecutor(connection)
executor.migrate(latest)

8
twitch/tests/test_views.py
View file

@ -8,7 +8,6 @@ from typing import Literal
import pytest
from django.core.files.base import ContentFile
from django.core.handlers.wsgi import WSGIRequest
from django.core.paginator import Paginator
from django.db import connection
from django.db.models import Max
@ -43,17 +42,10 @@ if TYPE_CHECKING:
from django.test import Client
from django.test.client import _MonkeyPatchedWSGIResponse
from django.test.utils import ContextList
from pytest_django.fixtures import SettingsWrapper
from twitch.views import Page
@pytest.fixture(autouse=True)
def apply_base_url_override(settings: SettingsWrapper) -> None:
"""Ensure BASE_URL is globally overridden for all tests."""
settings.BASE_URL = "https://ttvdrops.lovinator.space" # pyright: ignore[reportAttributeAccessIssue]
@pytest.mark.django_db
class TestSearchView:
"""Tests for the search_view function."""

2
twitch/tests/test_watch_imports.py
View file

@ -1,9 +1,7 @@
from pathlib import Path
from typing import TYPE_CHECKING
from unittest.mock import MagicMock
import pytest
from _pytest.capture import CaptureResult
from django.core.management.base import CommandError
from twitch.management.commands.watch_imports import Command