diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 69361ac..f53995d 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -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 diff --git a/.vscode/settings.json b/.vscode/settings.json index 9260d8b..ee505c3 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -71,6 +71,7 @@ "sitewide", "speculationrules", "staticfiles", + "Sunkwi", "testchannel", "testpass", "thelovinator", diff --git a/config/tests/test_site_endpoint_smoke.py b/config/tests/test_site_endpoint_smoke.py index 90537c0..8814395 100644 --- a/config/tests/test_site_endpoint_smoke.py +++ b/config/tests/test_site_endpoint_smoke.py @@ -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 diff --git a/conftest.py b/conftest.py index 883c39d..e3b4683 100644 --- a/conftest.py +++ b/conftest.py @@ -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. diff --git a/core/views.py b/core/views.py index 36d0748..b8e1af7 100644 --- a/core/views.py +++ b/core/views.py @@ -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( diff --git a/kick/tests/test_kick.py b/kick/tests/test_kick.py index d404ccd..3f2925e 100644 --- a/kick/tests/test_kick.py +++ b/kick/tests/test_kick.py @@ -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) diff --git a/templates/core/debug.html b/templates/core/debug.html index 6c0c547..32c41e3 100644 --- a/templates/core/debug.html +++ b/templates/core/debug.html @@ -170,4 +170,73 @@

None ✅

{% endif %} +
+

Data Source Diff — Drop Campaigns

+

Comparing campaigns by data_source field in the DB.

+

Only in SunkwiBOT/twitch-drops-api ({{ sunkibot_only_drops_count }})

+ {% if sunkibot_only_drops %} + + {% if sunkibot_only_drops_count > 200 %} +

… and {{ sunkibot_only_drops_count|add:"-200" }} more (showing first 200)

+ {% endif %} + {% else %} +

None ✅

+ {% endif %} +

Only in TwitchDropsMiner ({{ miner_only_drops_count }})

+ {% if miner_only_drops %} + + {% if miner_only_drops_count > 200 %} +

… and {{ miner_only_drops_count|add:"-200" }} more (showing first 200)

+ {% endif %} + {% else %} +

None ✅

+ {% endif %} +
+
+

Data Source Diff — Reward Campaigns

+

Only in SunkwiBOT/twitch-drops-api ({{ sunkibot_only_rewards_count }})

+ {% if sunkibot_only_rewards %} + + {% if sunkibot_only_rewards_count > 200 %} +

… and {{ sunkibot_only_rewards_count|add:"-200" }} more (showing first 200)

+ {% endif %} + {% else %} +

None ✅

+ {% endif %} +

Only in TwitchDropsMiner ({{ miner_only_rewards_count }})

+ {% if miner_only_rewards %} + + {% if miner_only_rewards_count > 200 %} +

… and {{ miner_only_rewards_count|add:"-200" }} more (showing first 200)

+ {% endif %} + {% else %} +

None ✅

+ {% endif %} +
{% endblock content %} diff --git a/templates/twitch/campaign_detail.html b/templates/twitch/campaign_detail.html index 08dd892..85ec252 100644 --- a/templates/twitch/campaign_detail.html +++ b/templates/twitch/campaign_detail.html @@ -47,48 +47,65 @@ {% endif %}

{{ campaign.description|linebreaksbr }}

-
- Published - ({{ campaign.added_at|timesince }} ago) -
-
- Last updated - ({{ campaign.updated_at|timesince }} ago) -
- -
- {% if campaign.end_at < now %} - Ended +
+
Published
+
+ ({{ campaign.added_at|timesince }} ago) +
+
Last updated
+
+ ({{ campaign.updated_at|timesince }} ago) +
+ +
+ {% if campaign.end_at < now %} + Ended + {% else %} + Ends in + {% endif %} +
+
({{ campaign.end_at|timesince }} ago) - {% else %} - Ends in - (in {{ campaign.end_at|timeuntil }}) - {% endif %} -
- -
- {% if campaign.start_at > now %} - Starts in + title="{{ campaign.end_at|date:'DATETIME_FORMAT' }}">{{ campaign.end_at|date:"M d, Y H:i" }} + {% if campaign.end_at < now %} + ({{ campaign.end_at|timesince }} ago) + {% else %} + (in {{ campaign.end_at|timeuntil }}) + {% endif %} + + +
+ {% if campaign.start_at > now %} + Starts in + {% else %} + Started + {% endif %} +
+
(in {{ campaign.start_at|timeuntil }}) - {% else %} - Started - ({{ campaign.start_at|timesince }} ago) - {% endif %} -
- -
- Duration is - -
+ title="{{ campaign.start_at|date:'DATETIME_FORMAT' }}">{{ campaign.start_at|date:"M d, Y H:i" }} + {% if campaign.start_at > now %} + (in {{ campaign.start_at|timeuntil }}) + {% else %} + ({{ campaign.start_at|timesince }} ago) + {% endif %} + + +
Duration
+
+ +
+ +
Data source
+
+ {{ campaign.data_source }} +
+
@@ -103,9 +120,9 @@ title="Atom feed for {{ campaign.game.display_name }} campaigns">[atom] [discord] + [explain] [api] - [explain] {% endif %}
diff --git a/templates/twitch/reward_campaign_detail.html b/templates/twitch/reward_campaign_detail.html index bdca15a..535da41 100644 --- a/templates/twitch/reward_campaign_detail.html +++ b/templates/twitch/reward_campaign_detail.html @@ -50,58 +50,71 @@ {% if reward_campaign.summary %}

{{ reward_campaign.summary|linebreaksbr }}

{% endif %} - -
- Status: - {% if is_active %} - Active - {% elif reward_campaign.starts_at > now %} - Upcoming - {% else %} - Expired - {% endif %} -
-
Added at {{ reward_campaign.added_at|date:"M d, Y H:i" }} ({{ reward_campaign.added_at|timesince }} ago)
-
- Last updated at {{ reward_campaign.updated_at|date:"M d, Y H:i" }} ({{ reward_campaign.updated_at|timesince }} ago) -
- -
- {% if reward_campaign.starts_at > now %} - Starts - {% else %} - Started at - {% endif %} -
- -
- {% if reward_campaign.ends_at > now %} - Ends - {% else %} - Ended at - {% endif %} -
-
- {% if reward_campaign.instructions %}{{ reward_campaign.instructions|linebreaksbr }}{% endif %} -
+ + +
Data source
+
+ {{ reward_campaign.data_source }} +
+ +
{% if reward_campaign.instructions %}{{ reward_campaign.instructions|linebreaksbr }}{% endif %}
Twitch > Reward Campaigns
-

This is an archive of old Twitch reward campaigns because we currently do not scrape them.

- Feel free to submit a pull request on GitHub - with a working implementation :-). + Data is scraped from SunkwiBOT/twitch-drops-api every 15 minutes.

@@ -36,84 +36,81 @@ title="Atom feed for all reward campaigns">[atom] [discord] + [explain] [api] - [explain]
{% if reward_campaigns %} - {% comment %} -
Active Reward Campaigns
- - - {% for campaign in reward_campaigns %} - {% if campaign.starts_at <= now and campaign.ends_at >= now %} - - + + + {% endif %} + {% endfor %} + +
- - {% if campaign.brand %} - {{ campaign.brand }}: {{ campaign.name }} - {% else %} - {{ campaign.name }} + {% with active_campaigns=reward_campaigns|dictsortreversed:"ends_at" %} +
Active Reward Campaigns
+ + + {% for campaign in active_campaigns %} + {% if campaign.starts_at <= now and campaign.ends_at >= now %} + + - - - - {% endif %} - {% endfor %} - -
+ + {% if campaign.brand %} + {{ campaign.brand }}: {{ campaign.name }} + {% else %} + {{ campaign.name }} + {% endif %} + + {% if campaign.summary %} +
+ {{ campaign.summary }} {% endif %} - - {% if campaign.summary %} -
- {{ campaign.summary }} - {% endif %} - 		- {% if campaign.game %} - {{ campaign.game.display_name }} - {% elif campaign.is_sitewide %} - Site-wide - {% endif %} - - Ends in {{ campaign.ends_at|timeuntil }} -
- {% endcomment %} - {% comment %} -
Upcoming Reward Campaigns
- - - {% for campaign in reward_campaigns %} - - {% if campaign.starts_at > now %} - - + - - - - {% endif %} - {% endfor %} - -
- - {% if campaign.brand %} - {{ campaign.brand }}: {{ campaign.name }} - {% else %} - {{ campaign.name }} + + {% if campaign.game %} + {{ campaign.game.display_name }} + {% elif campaign.is_sitewide %} + Site-wide {% endif %} - - {% if campaign.summary %} -
- {{ campaign.summary }} - {% endif %} - 		- {% if campaign.game %} - {{ campaign.game.display_name }} - {% elif campaign.is_sitewide %} - Site-wide - {% endif %} - - Starts in {{ campaign.starts_at|timeuntil }} -
- {% endcomment %} + 		+ Ends in {{ campaign.ends_at|timeuntil }} +
+
Upcoming Reward Campaigns
+ + + {% for campaign in active_campaigns %} + {% if campaign.starts_at > now %} + + + + + + {% endif %} + {% endfor %} + +
+ + {% if campaign.brand %} + {{ campaign.brand }}: {{ campaign.name }} + {% else %} + {{ campaign.name }} + {% endif %} + + {% if campaign.summary %} +
+ {{ campaign.summary }} + {% endif %} + 		+ {% if campaign.game %} + {{ campaign.game.display_name }} + {% elif campaign.is_sitewide %} + Site-wide + {% endif %} + + Starts in {{ campaign.starts_at|timeuntil }} +
+ {% endwith %}
Past Reward Campaigns
diff --git a/tools/extract_historical_drops.py b/tools/extract_historical_drops.py new file mode 100755 index 0000000..65db357 --- /dev/null +++ b/tools/extract_historical_drops.py @@ -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() diff --git a/twitch/api.py b/twitch/api.py index a657e31..30c6460 100644 --- a/twitch/api.py +++ b/twitch/api.py @@ -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], ) diff --git a/twitch/apps.py b/twitch/apps.py index 9b8d77f..d15fdbd 100644 --- a/twitch/apps.py +++ b/twitch/apps.py @@ -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. diff --git a/twitch/feeds.py b/twitch/feeds.py index c1c07d4..81c539e 100644 --- a/twitch/feeds.py +++ b/twitch/feeds.py @@ -1,6 +1,5 @@ import logging import re -from collections.abc import Callable from typing import TYPE_CHECKING from typing import Literal diff --git a/twitch/management/commands/backfill_image_dimensions.py b/twitch/management/commands/backfill_image_dimensions.py index aa72903..9d876b0 100644 --- a/twitch/management/commands/backfill_image_dimensions.py +++ b/twitch/management/commands/backfill_image_dimensions.py @@ -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...") diff --git a/twitch/management/commands/better_import_drops.py b/twitch/management/commands/better_import_drops.py index e729930..81acda0 100644 --- a/twitch/management/commands/better_import_drops.py +++ b/twitch/management/commands/better_import_drops.py @@ -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 diff --git a/twitch/management/commands/download_box_art.py b/twitch/management/commands/download_box_art.py index ca27157..3d1b513 100644 --- a/twitch/management/commands/download_box_art.py +++ b/twitch/management/commands/download_box_art.py @@ -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) diff --git a/twitch/management/commands/download_campaign_images.py b/twitch/management/commands/download_campaign_images.py index 66a7d5c..68b82f2 100644 --- a/twitch/management/commands/download_campaign_images.py +++ b/twitch/management/commands/download_campaign_images.py @@ -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( diff --git a/twitch/management/commands/import_twitch_drops_api.py b/twitch/management/commands/import_twitch_drops_api.py new file mode 100644 index 0000000..f8119ec --- /dev/null +++ b/twitch/management/commands/import_twitch_drops_api.py @@ -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 `_ +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 diff --git a/twitch/migrations/0024_dropcampaign_data_source_rewardcampaign_data_source.py b/twitch/migrations/0024_dropcampaign_data_source_rewardcampaign_data_source.py new file mode 100644 index 0000000..8cac6b5 --- /dev/null +++ b/twitch/migrations/0024_dropcampaign_data_source_rewardcampaign_data_source.py @@ -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, + ), + ), + ] diff --git a/twitch/models.py b/twitch/models.py index 0e08d39..6aeb0f1 100644 --- a/twitch/models.py +++ b/twitch/models.py @@ -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", diff --git a/twitch/schemas.py b/twitch/schemas.py index b4d0739..d8d8901 100644 --- a/twitch/schemas.py +++ b/twitch/schemas.py @@ -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, } diff --git a/twitch/tasks.py b/twitch/tasks.py index 32e5100..9f57d87 100644 --- a/twitch/tasks.py +++ b/twitch/tasks.py @@ -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) # --------------------------------------------------------------------------- diff --git a/twitch/tests/test_api.py b/twitch/tests/test_api.py index 40edecc..a753a34 100644 --- a/twitch/tests/test_api.py +++ b/twitch/tests/test_api.py @@ -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]' 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 diff --git a/twitch/tests/test_import_twitch_drops_api.py b/twitch/tests/test_import_twitch_drops_api.py new file mode 100644 index 0000000..12e4cca --- /dev/null +++ b/twitch/tests/test_import_twitch_drops_api.py @@ -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() diff --git a/twitch/tests/test_migrations.py b/twitch/tests/test_migrations.py index 9972ff8..a43dcc0 100644 --- a/twitch/tests/test_migrations.py +++ b/twitch/tests/test_migrations.py @@ -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) diff --git a/twitch/tests/test_views.py b/twitch/tests/test_views.py index 397077a..b0f34f3 100644 --- a/twitch/tests/test_views.py +++ b/twitch/tests/test_views.py @@ -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.""" diff --git a/twitch/tests/test_watch_imports.py b/twitch/tests/test_watch_imports.py index 37d0005..f848815 100644 --- a/twitch/tests/test_watch_imports.py +++ b/twitch/tests/test_watch_imports.py @@ -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