Move Twitch stuff to /twitch/

This commit is contained in:
Joakim Hellsén 2026-03-16 15:27:33 +01:00
commit 6f6116c3c7
Signed by: Joakim Hellsén
SSH key fingerprint: SHA256:/9h/CsExpFp+PRhsfA0xznFx2CGfTT5R/kpuFfUgEQk
31 changed files with 1150 additions and 984 deletions

View file

@ -759,7 +759,7 @@ class OrganizationRSSFeed(TTVDropsBaseFeed):
def feed_url(self) -> str:
"""Return the absolute URL for this feed."""
return reverse("twitch:organization_feed")
return reverse("core:organization_feed")
# MARK: /rss/games/
@ -829,7 +829,7 @@ class GameFeed(TTVDropsBaseFeed):
# Get the full URL for TTVDrops game detail page
game_url: str = reverse("twitch:game_detail", args=[twitch_id])
rss_feed_url: str = reverse("twitch:game_campaign_feed", args=[twitch_id])
rss_feed_url: str = reverse("core:game_campaign_feed", args=[twitch_id])
twitch_directory_url: str = getattr(item, "twitch_directory_url", "")
description_parts.append(
@ -911,7 +911,7 @@ class GameFeed(TTVDropsBaseFeed):
def feed_url(self) -> str:
"""Return the URL to the RSS feed itself."""
return reverse("twitch:game_feed")
return reverse("core:game_feed")
# MARK: /rss/campaigns/
@ -1054,7 +1054,7 @@ class DropCampaignFeed(TTVDropsBaseFeed):
def feed_url(self) -> str:
"""Return the URL to the RSS feed itself."""
return reverse("twitch:campaign_feed")
return reverse("core:campaign_feed")
# MARK: /rss/games/<twitch_id>/campaigns/
@ -1230,7 +1230,7 @@ class GameCampaignFeed(TTVDropsBaseFeed):
def feed_url(self, obj: Game) -> str:
"""Return the URL to the RSS feed itself."""
return reverse("twitch:game_campaign_feed", args=[obj.twitch_id])
return reverse("core:game_campaign_feed", args=[obj.twitch_id])
# MARK: /rss/reward-campaigns/
@ -1422,7 +1422,7 @@ class RewardCampaignFeed(TTVDropsBaseFeed):
def feed_url(self) -> str:
"""Return the URL to the RSS feed itself."""
return reverse("twitch:reward_campaign_feed")
return reverse("core:reward_campaign_feed")
# Atom feed variants: reuse existing logic but switch the feed generator to Atom
@ -1433,7 +1433,7 @@ class OrganizationAtomFeed(TTVDropsAtomBaseFeed, OrganizationRSSFeed):
def feed_url(self) -> str:
"""Return the URL to the Atom feed itself."""
return reverse("twitch:organization_feed_atom")
return reverse("core:organization_feed_atom")
class GameAtomFeed(TTVDropsAtomBaseFeed, GameFeed):
@ -1443,7 +1443,7 @@ class GameAtomFeed(TTVDropsAtomBaseFeed, GameFeed):
def feed_url(self) -> str:
"""Return the URL to the Atom feed itself."""
return reverse("twitch:game_feed_atom")
return reverse("core:game_feed_atom")
class DropCampaignAtomFeed(TTVDropsAtomBaseFeed, DropCampaignFeed):
@ -1453,7 +1453,7 @@ class DropCampaignAtomFeed(TTVDropsAtomBaseFeed, DropCampaignFeed):
def feed_url(self) -> str:
"""Return the URL to the Atom feed itself."""
return reverse("twitch:campaign_feed_atom")
return reverse("core:campaign_feed_atom")
class GameCampaignAtomFeed(TTVDropsAtomBaseFeed, GameCampaignFeed):
@ -1461,7 +1461,7 @@ class GameCampaignAtomFeed(TTVDropsAtomBaseFeed, GameCampaignFeed):
def feed_url(self, obj: Game) -> str:
"""Return the URL to the Atom feed itself."""
return reverse("twitch:game_campaign_feed_atom", args=[obj.twitch_id])
return reverse("core:game_campaign_feed_atom", args=[obj.twitch_id])
class RewardCampaignAtomFeed(TTVDropsAtomBaseFeed, RewardCampaignFeed):
@ -1471,7 +1471,7 @@ class RewardCampaignAtomFeed(TTVDropsAtomBaseFeed, RewardCampaignFeed):
def feed_url(self) -> str:
"""Return the URL to the Atom feed itself."""
return reverse("twitch:reward_campaign_feed_atom")
return reverse("core:reward_campaign_feed_atom")
# Discord feed variants: Atom feeds with Discord relative timestamps
@ -1482,7 +1482,7 @@ class OrganizationDiscordFeed(TTVDropsAtomBaseFeed, OrganizationRSSFeed):
def feed_url(self) -> str:
"""Return the URL to the Discord feed itself."""
return reverse("twitch:organization_feed_discord")
return reverse("core:organization_feed_discord")
class GameDiscordFeed(TTVDropsAtomBaseFeed, GameFeed):
@ -1492,7 +1492,7 @@ class GameDiscordFeed(TTVDropsAtomBaseFeed, GameFeed):
def feed_url(self) -> str:
"""Return the URL to the Discord feed itself."""
return reverse("twitch:game_feed_discord")
return reverse("core:game_feed_discord")
class DropCampaignDiscordFeed(TTVDropsAtomBaseFeed, DropCampaignFeed):
@ -1515,7 +1515,7 @@ class DropCampaignDiscordFeed(TTVDropsAtomBaseFeed, DropCampaignFeed):
def feed_url(self) -> str:
"""Return the URL to the Discord feed itself."""
return reverse("twitch:campaign_feed_discord")
return reverse("core:campaign_feed_discord")
class GameCampaignDiscordFeed(TTVDropsAtomBaseFeed, GameCampaignFeed):
@ -1535,7 +1535,7 @@ class GameCampaignDiscordFeed(TTVDropsAtomBaseFeed, GameCampaignFeed):
def feed_url(self, obj: Game) -> str:
"""Return the URL to the Discord feed itself."""
return reverse("twitch:game_campaign_feed_discord", args=[obj.twitch_id])
return reverse("core:game_campaign_feed_discord", args=[obj.twitch_id])
class RewardCampaignDiscordFeed(TTVDropsAtomBaseFeed, RewardCampaignFeed):
@ -1602,4 +1602,4 @@ class RewardCampaignDiscordFeed(TTVDropsAtomBaseFeed, RewardCampaignFeed):
def feed_url(self) -> str:
"""Return the URL to the Discord feed itself."""
return reverse("twitch:reward_campaign_feed_discord")
return reverse("core:reward_campaign_feed_discord")

View file

@ -288,7 +288,7 @@ class TestDatasetBackupViews:
monkeypatch.setattr(settings, "DATA_DIR", datasets_dir.parent)
response: _MonkeyPatchedWSGIResponse = client.get(
reverse("twitch:dataset_backups"),
reverse("core:dataset_backups"),
)
assert response.status_code == 200
@ -305,7 +305,7 @@ class TestDatasetBackupViews:
monkeypatch.setattr(settings, "DATA_DIR", datasets_dir.parent)
response: _MonkeyPatchedWSGIResponse = client.get(
reverse("twitch:dataset_backups"),
reverse("core:dataset_backups"),
)
assert response.status_code == 200
@ -339,7 +339,7 @@ class TestDatasetBackupViews:
os.utime(newer_backup, (newer_time, newer_time))
response: _MonkeyPatchedWSGIResponse = client.get(
reverse("twitch:dataset_backups"),
reverse("core:dataset_backups"),
)
content = response.content.decode()
@ -361,7 +361,7 @@ class TestDatasetBackupViews:
response: _MonkeyPatchedWSGIResponse = client.get(
reverse(
"twitch:dataset_backup_download",
"core:dataset_backup_download",
args=["ttvdrops-20260210-120000.sql.zst"],
),
)
@ -382,7 +382,7 @@ class TestDatasetBackupViews:
# Attempt path traversal
response = client.get(
reverse("twitch:dataset_backup_download", args=["../../../etc/passwd"]),
reverse("core:dataset_backup_download", args=["../../../etc/passwd"]),
)
assert response.status_code == 404
@ -400,7 +400,7 @@ class TestDatasetBackupViews:
invalid_file.write_text("not a backup")
response = client.get(
reverse("twitch:dataset_backup_download", args=["malicious.exe"]),
reverse("core:dataset_backup_download", args=["malicious.exe"]),
)
assert response.status_code == 404
@ -414,7 +414,7 @@ class TestDatasetBackupViews:
monkeypatch.setattr(settings, "DATA_DIR", datasets_dir.parent)
response = client.get(
reverse("twitch:dataset_backup_download", args=["nonexistent.sql.zst"]),
reverse("core:dataset_backup_download", args=["nonexistent.sql.zst"]),
)
assert response.status_code == 404
@ -429,7 +429,7 @@ class TestDatasetBackupViews:
monkeypatch.setattr(settings, "DATA_DIR", datasets_dir.parent)
response: _MonkeyPatchedWSGIResponse = client.get(
reverse("twitch:dataset_backups"),
reverse("core:dataset_backups"),
)
assert response.status_code == 200
@ -452,7 +452,7 @@ class TestDatasetBackupViews:
(datasets_dir / "old_backup.gz").write_bytes(b"should be ignored")
response: _MonkeyPatchedWSGIResponse = client.get(
reverse("twitch:dataset_backups"),
reverse("core:dataset_backups"),
)
content = response.content.decode()
@ -481,7 +481,7 @@ class TestDatasetBackupViews:
handle.write("-- Test\n")
response: _MonkeyPatchedWSGIResponse = client.get(
reverse("twitch:dataset_backup_download", args=["2026/02/backup.sql.zst"]),
reverse("core:dataset_backup_download", args=["2026/02/backup.sql.zst"]),
)
assert response.status_code == 200

View file

@ -155,7 +155,7 @@ class TestBadgeSearch:
ChatBadgeSet.objects.create(set_id="vip")
ChatBadgeSet.objects.create(set_id="subscriber")
response = client.get(reverse("twitch:search"), {"q": "vip"})
response = client.get(reverse("core:search"), {"q": "vip"})
assert response.status_code == 200
content = response.content.decode()
@ -175,7 +175,7 @@ class TestBadgeSearch:
description="Test description",
)
response = client.get(reverse("twitch:search"), {"q": "Moderator"})
response = client.get(reverse("core:search"), {"q": "Moderator"})
assert response.status_code == 200
content = response.content.decode()
@ -195,7 +195,7 @@ class TestBadgeSearch:
description="Unique description text",
)
response = client.get(reverse("twitch:search"), {"q": "Unique description"})
response = client.get(reverse("core:search"), {"q": "Unique description"})
assert response.status_code == 200
content = response.content.decode()

View file

@ -44,7 +44,7 @@ class ExportViewsTestCase(TestCase):
def test_export_campaigns_csv(self) -> None:
"""Test CSV export of campaigns."""
response = self.client.get("/export/campaigns/csv/")
response = self.client.get("/twitch/export/campaigns/csv/")
assert response.status_code == 200
assert response["Content-Type"] == "text/csv"
assert b"Twitch ID" in response.content
@ -53,7 +53,7 @@ class ExportViewsTestCase(TestCase):
def test_export_campaigns_json(self) -> None:
"""Test JSON export of campaigns."""
response = self.client.get("/export/campaigns/json/")
response = self.client.get("/twitch/export/campaigns/json/")
assert response.status_code == 200
assert response["Content-Type"] == "application/json"
@ -66,7 +66,7 @@ class ExportViewsTestCase(TestCase):
def test_export_games_csv(self) -> None:
"""Test CSV export of games."""
response = self.client.get("/export/games/csv/")
response = self.client.get("/twitch/export/games/csv/")
assert response.status_code == 200
assert response["Content-Type"] == "text/csv"
assert b"Twitch ID" in response.content
@ -75,7 +75,7 @@ class ExportViewsTestCase(TestCase):
def test_export_games_json(self) -> None:
"""Test JSON export of games."""
response = self.client.get("/export/games/json/")
response = self.client.get("/twitch/export/games/json/")
assert response.status_code == 200
assert response["Content-Type"] == "application/json"
@ -87,7 +87,7 @@ class ExportViewsTestCase(TestCase):
def test_export_organizations_csv(self) -> None:
"""Test CSV export of organizations."""
response = self.client.get("/export/organizations/csv/")
response = self.client.get("/twitch/export/organizations/csv/")
assert response.status_code == 200
assert response["Content-Type"] == "text/csv"
assert b"Twitch ID" in response.content
@ -96,7 +96,7 @@ class ExportViewsTestCase(TestCase):
def test_export_organizations_json(self) -> None:
"""Test JSON export of organizations."""
response = self.client.get("/export/organizations/json/")
response = self.client.get("/twitch/export/organizations/json/")
assert response.status_code == 200
assert response["Content-Type"] == "application/json"
@ -108,13 +108,13 @@ class ExportViewsTestCase(TestCase):
def test_export_campaigns_csv_with_filters(self) -> None:
"""Test CSV export of campaigns with status filter."""
response = self.client.get("/export/campaigns/csv/?status=active")
response = self.client.get("/twitch/export/campaigns/csv/?status=active")
assert response.status_code == 200
assert b"campaign123" in response.content
def test_export_campaigns_json_with_filters(self) -> None:
"""Test JSON export of campaigns with status filter."""
response = self.client.get("/export/campaigns/json/?status=active")
response = self.client.get("/twitch/export/campaigns/json/?status=active")
assert response.status_code == 200
data = json.loads(response.content)

View file

@ -106,7 +106,7 @@ class RSSFeedTestCase(TestCase):
def test_organization_feed(self) -> None:
"""Test organization feed returns 200."""
url: str = reverse("twitch:organization_feed")
url: str = reverse("core:organization_feed")
response: _MonkeyPatchedWSGIResponse = self.client.get(url)
assert response.status_code == 200
assert response["Content-Type"] == "application/xml; charset=utf-8"
@ -114,7 +114,7 @@ class RSSFeedTestCase(TestCase):
def test_game_feed(self) -> None:
"""Test game feed returns 200."""
url: str = reverse("twitch:game_feed")
url: str = reverse("core:game_feed")
response: _MonkeyPatchedWSGIResponse = self.client.get(url)
assert response.status_code == 200
assert response["Content-Type"] == "application/xml; charset=utf-8"
@ -123,7 +123,7 @@ class RSSFeedTestCase(TestCase):
assert "Owned by Test Organization." in content
expected_rss_link: str = reverse(
"twitch:game_campaign_feed",
"core:game_campaign_feed",
args=[self.game.twitch_id],
)
assert expected_rss_link in content
@ -137,7 +137,7 @@ class RSSFeedTestCase(TestCase):
def test_organization_atom_feed(self) -> None:
"""Test organization Atom feed returns 200 and Atom XML."""
url: str = reverse("twitch:organization_feed_atom")
url: str = reverse("core:organization_feed_atom")
response: _MonkeyPatchedWSGIResponse = self.client.get(url)
msg: str = f"Expected 200 OK, got {response.status_code} with content: {response.content.decode('utf-8')}"
@ -151,14 +151,14 @@ class RSSFeedTestCase(TestCase):
def test_game_atom_feed(self) -> None:
"""Test game Atom feed returns 200 and contains expected content."""
url: str = reverse("twitch:game_feed_atom")
url: str = reverse("core:game_feed_atom")
response: _MonkeyPatchedWSGIResponse = self.client.get(url)
assert response.status_code == 200
assert response["Content-Type"] == "application/xml; charset=utf-8"
content: str = response.content.decode("utf-8")
assert "Owned by Test Organization." in content
expected_atom_link: str = reverse(
"twitch:game_campaign_feed",
"core:game_campaign_feed",
args=[self.game.twitch_id],
)
assert expected_atom_link in content
@ -167,7 +167,7 @@ class RSSFeedTestCase(TestCase):
def test_campaign_atom_feed_uses_url_ids_and_correct_self_link(self) -> None:
"""Atom campaign feed should use URL ids and a matching self link."""
url: str = reverse("twitch:campaign_feed_atom")
url: str = reverse("core:campaign_feed_atom")
response: _MonkeyPatchedWSGIResponse = self.client.get(url)
assert response.status_code == 200
@ -180,33 +180,35 @@ class RSSFeedTestCase(TestCase):
assert 'href="http://testserver/atom/campaigns/"' in content, msg
msg: str = f"Expected entry ID to be the campaign URL, got: {content}"
assert "<id>http://testserver/campaigns/test-campaign-123/</id>" in content, msg
assert (
"<id>http://testserver/twitch/campaigns/test-campaign-123/</id>" in content
), msg
def test_all_atom_feeds_use_url_ids_and_correct_self_links(self) -> None:
"""All Atom feeds should use absolute URL entry IDs and matching self links."""
atom_feed_cases: list[tuple[str, dict[str, str], str]] = [
(
"twitch:campaign_feed_atom",
"core:campaign_feed_atom",
{},
f"http://testserver{reverse('twitch:campaign_detail', args=[self.campaign.twitch_id])}",
),
(
"twitch:game_feed_atom",
"core:game_feed_atom",
{},
f"http://testserver{reverse('twitch:game_detail', args=[self.game.twitch_id])}",
),
(
"twitch:game_campaign_feed_atom",
"core:game_campaign_feed_atom",
{"twitch_id": self.game.twitch_id},
f"http://testserver{reverse('twitch:campaign_detail', args=[self.campaign.twitch_id])}",
),
(
"twitch:organization_feed_atom",
"core:organization_feed_atom",
{},
f"http://testserver{reverse('twitch:organization_detail', args=[self.org.twitch_id])}",
),
(
"twitch:reward_campaign_feed_atom",
"core:reward_campaign_feed_atom",
{},
f"http://testserver{reverse('twitch:reward_campaign_detail', args=[self.reward_campaign.twitch_id])}",
),
@ -246,7 +248,7 @@ class RSSFeedTestCase(TestCase):
)
drop.benefits.add(benefit)
url: str = reverse("twitch:campaign_feed_atom")
url: str = reverse("core:campaign_feed_atom")
response: _MonkeyPatchedWSGIResponse = self.client.get(url)
assert response.status_code == 200
@ -257,11 +259,11 @@ class RSSFeedTestCase(TestCase):
def test_atom_feeds_include_stylesheet_processing_instruction(self) -> None:
"""Atom feeds should include an xml-stylesheet processing instruction."""
feed_urls: list[str] = [
reverse("twitch:campaign_feed_atom"),
reverse("twitch:game_feed_atom"),
reverse("twitch:game_campaign_feed_atom", args=[self.game.twitch_id]),
reverse("twitch:organization_feed_atom"),
reverse("twitch:reward_campaign_feed_atom"),
reverse("core:campaign_feed_atom"),
reverse("core:game_feed_atom"),
reverse("core:game_campaign_feed_atom", args=[self.game.twitch_id]),
reverse("core:organization_feed_atom"),
reverse("core:reward_campaign_feed_atom"),
]
for url in feed_urls:
@ -297,12 +299,12 @@ class RSSFeedTestCase(TestCase):
self.campaign.save()
feed_urls: list[str] = [
reverse("twitch:game_feed"),
reverse("twitch:campaign_feed"),
reverse("twitch:game_campaign_feed", args=[self.game.twitch_id]),
reverse("twitch:game_feed_atom"),
reverse("twitch:campaign_feed_atom"),
reverse("twitch:game_campaign_feed_atom", args=[self.game.twitch_id]),
reverse("core:game_feed"),
reverse("core:campaign_feed"),
reverse("core:game_campaign_feed", args=[self.game.twitch_id]),
reverse("core:game_feed_atom"),
reverse("core:campaign_feed_atom"),
reverse("core:game_campaign_feed_atom", args=[self.game.twitch_id]),
]
for url in feed_urls:
@ -333,14 +335,14 @@ class RSSFeedTestCase(TestCase):
self.reward_campaign.save()
feed_urls: list[str] = [
reverse("twitch:game_feed"),
reverse("twitch:campaign_feed"),
reverse("twitch:game_campaign_feed", args=[self.game.twitch_id]),
reverse("twitch:reward_campaign_feed"),
reverse("twitch:game_feed_atom"),
reverse("twitch:campaign_feed_atom"),
reverse("twitch:game_campaign_feed_atom", args=[self.game.twitch_id]),
reverse("twitch:reward_campaign_feed_atom"),
reverse("core:game_feed"),
reverse("core:campaign_feed"),
reverse("core:game_campaign_feed", args=[self.game.twitch_id]),
reverse("core:reward_campaign_feed"),
reverse("core:game_feed_atom"),
reverse("core:campaign_feed_atom"),
reverse("core:game_campaign_feed_atom", args=[self.game.twitch_id]),
reverse("core:reward_campaign_feed_atom"),
]
for url in feed_urls:
@ -378,7 +380,7 @@ class RSSFeedTestCase(TestCase):
def test_campaign_feed(self) -> None:
"""Test campaign feed returns 200."""
url: str = reverse("twitch:campaign_feed")
url: str = reverse("core:campaign_feed")
response: _MonkeyPatchedWSGIResponse = self.client.get(url)
assert response.status_code == 200
assert response["Content-Type"] == "application/xml; charset=utf-8"
@ -392,11 +394,11 @@ class RSSFeedTestCase(TestCase):
def test_rss_feeds_include_stylesheet_processing_instruction(self) -> None:
"""RSS feeds should include an xml-stylesheet processing instruction."""
feed_urls: list[str] = [
reverse("twitch:campaign_feed"),
reverse("twitch:game_feed"),
reverse("twitch:game_campaign_feed", args=[self.game.twitch_id]),
reverse("twitch:organization_feed"),
reverse("twitch:reward_campaign_feed"),
reverse("core:campaign_feed"),
reverse("core:game_feed"),
reverse("core:game_campaign_feed", args=[self.game.twitch_id]),
reverse("core:organization_feed"),
reverse("core:reward_campaign_feed"),
]
for url in feed_urls:
@ -443,11 +445,11 @@ class RSSFeedTestCase(TestCase):
def test_rss_feeds_include_shared_metadata_fields(self) -> None:
"""RSS output should contain base feed metadata fields."""
feed_urls: list[str] = [
reverse("twitch:campaign_feed"),
reverse("twitch:game_feed"),
reverse("twitch:game_campaign_feed", args=[self.game.twitch_id]),
reverse("twitch:organization_feed"),
reverse("twitch:reward_campaign_feed"),
reverse("core:campaign_feed"),
reverse("core:game_feed"),
reverse("core:game_campaign_feed", args=[self.game.twitch_id]),
reverse("core:organization_feed"),
reverse("core:reward_campaign_feed"),
]
for url in feed_urls:
@ -480,7 +482,7 @@ class RSSFeedTestCase(TestCase):
operation_names=["DropCampaignDetails"],
)
url: str = reverse("twitch:campaign_feed")
url: str = reverse("core:campaign_feed")
response: _MonkeyPatchedWSGIResponse = self.client.get(url)
assert response.status_code == 200
content: str = response.content.decode("utf-8")
@ -539,7 +541,7 @@ class RSSFeedTestCase(TestCase):
description="This badge was earned by subscribing.",
)
url: str = reverse("twitch:campaign_feed")
url: str = reverse("core:campaign_feed")
response: _MonkeyPatchedWSGIResponse = self.client.get(url)
assert response.status_code == 200
content: str = response.content.decode("utf-8")
@ -547,7 +549,7 @@ class RSSFeedTestCase(TestCase):
def test_game_campaign_feed(self) -> None:
"""Test game-specific campaign feed returns 200."""
url: str = reverse("twitch:game_campaign_feed", args=[self.game.twitch_id])
url: str = reverse("core:game_campaign_feed", args=[self.game.twitch_id])
response: _MonkeyPatchedWSGIResponse = self.client.get(url)
assert response.status_code == 200
assert response["Content-Type"] == "application/xml; charset=utf-8"
@ -576,7 +578,7 @@ class RSSFeedTestCase(TestCase):
)
# Get feed for first game
url: str = reverse("twitch:game_campaign_feed", args=[self.game.twitch_id])
url: str = reverse("core:game_campaign_feed", args=[self.game.twitch_id])
response: _MonkeyPatchedWSGIResponse = self.client.get(url)
content: str = response.content.decode("utf-8")
@ -609,7 +611,7 @@ class RSSFeedTestCase(TestCase):
operation_names=["DropCampaignDetails"],
)
url: str = reverse("twitch:game_campaign_feed", args=[self.game.twitch_id])
url: str = reverse("core:game_campaign_feed", args=[self.game.twitch_id])
response: _MonkeyPatchedWSGIResponse = self.client.get(url)
assert response.status_code == 200
content: str = response.content.decode("utf-8")
@ -664,7 +666,7 @@ class RSSFeedTestCase(TestCase):
game=self.game,
)
url: str = reverse("twitch:reward_campaign_feed")
url: str = reverse("core:reward_campaign_feed")
response: _MonkeyPatchedWSGIResponse = self.client.get(url)
assert response.status_code == 200
content: str = response.content.decode("utf-8")
@ -855,7 +857,7 @@ def test_campaign_feed_queries_bounded(
for i in range(3):
_build_campaign(game, i)
url: str = reverse("twitch:campaign_feed")
url: str = reverse("core:campaign_feed")
# TODO(TheLovinator): 14 queries is still quite high for a feed - we should be able to optimize this further, but this is a good starting point to prevent regressions for now. # noqa: TD003
with django_assert_num_queries(14, exact=False):
response: _MonkeyPatchedWSGIResponse = client.get(url)
@ -911,7 +913,7 @@ def test_campaign_feed_queries_do_not_scale_with_items(
)
drop.benefits.add(benefit)
url: str = reverse("twitch:campaign_feed")
url: str = reverse("core:campaign_feed")
# N+1 safeguard: query count should not scale linearly with campaign count.
with django_assert_num_queries(40, exact=False):
@ -941,7 +943,7 @@ def test_game_campaign_feed_queries_bounded(
for i in range(3):
_build_campaign(game, i)
url: str = reverse("twitch:game_campaign_feed", args=[game.twitch_id])
url: str = reverse("core:game_campaign_feed", args=[game.twitch_id])
with django_assert_num_queries(6, exact=False):
response: _MonkeyPatchedWSGIResponse = client.get(url)
@ -970,7 +972,7 @@ def test_game_campaign_feed_queries_do_not_scale_with_items(
for i in range(50):
_build_campaign(game, i)
url: str = reverse("twitch:game_campaign_feed", args=[game.twitch_id])
url: str = reverse("core:game_campaign_feed", args=[game.twitch_id])
with django_assert_num_queries(6, exact=False):
response: _MonkeyPatchedWSGIResponse = client.get(url)
@ -987,7 +989,7 @@ def test_organization_feed_queries_bounded(
for i in range(5):
Organization.objects.create(twitch_id=f"org-feed-{i}", name=f"Org Feed {i}")
url: str = reverse("twitch:organization_feed")
url: str = reverse("core:organization_feed")
with django_assert_num_queries(1, exact=True):
response: _MonkeyPatchedWSGIResponse = client.get(url)
@ -1014,7 +1016,7 @@ def test_game_feed_queries_bounded(
)
game.owners.add(org)
url: str = reverse("twitch:game_feed")
url: str = reverse("core:game_feed")
# One query for games + one prefetch query for owners.
with django_assert_num_queries(2, exact=True):
response: _MonkeyPatchedWSGIResponse = client.get(url)
@ -1043,7 +1045,7 @@ def test_reward_campaign_feed_queries_bounded(
for i in range(3):
_build_reward_campaign(game, i)
url: str = reverse("twitch:reward_campaign_feed")
url: str = reverse("core:reward_campaign_feed")
with django_assert_num_queries(1, exact=True):
response: _MonkeyPatchedWSGIResponse = client.get(url)
@ -1076,7 +1078,7 @@ def test_docs_rss_queries_bounded(
_build_campaign(game, i)
_build_reward_campaign(game, i)
url: str = reverse("twitch:docs_rss")
url: str = reverse("core:docs_rss")
# TODO(TheLovinator): 31 queries is still quite high for a feed - we should be able to optimize this further, but this is a good starting point to prevent regressions for now. # noqa: TD003
with django_assert_num_queries(31, exact=False):
@ -1093,8 +1095,8 @@ URL_NAMES: list[tuple[str, dict[str, str]]] = [
("twitch:campaign_detail", {"twitch_id": "test-campaign-123"}),
("twitch:channel_list", {}),
("twitch:channel_detail", {"twitch_id": "test-channel-123"}),
("twitch:debug", {}),
("twitch:docs_rss", {}),
("core:debug", {}),
("core:docs_rss", {}),
("twitch:emote_gallery", {}),
("twitch:games_grid", {}),
("twitch:games_list", {}),
@ -1103,12 +1105,12 @@ URL_NAMES: list[tuple[str, dict[str, str]]] = [
("twitch:organization_detail", {"twitch_id": "test-org-123"}),
("twitch:reward_campaign_list", {}),
("twitch:reward_campaign_detail", {"twitch_id": "test-reward-123"}),
("twitch:search", {}),
("twitch:campaign_feed", {}),
("twitch:game_feed", {}),
("twitch:game_campaign_feed", {"twitch_id": "test-game-123"}),
("twitch:organization_feed", {}),
("twitch:reward_campaign_feed", {}),
("core:search", {}),
("core:campaign_feed", {}),
("core:game_feed", {}),
("core:game_campaign_feed", {"twitch_id": "test-game-123"}),
("core:organization_feed", {}),
("core:reward_campaign_feed", {}),
]
@ -1251,7 +1253,7 @@ class DiscordFeedTestCase(TestCase):
def test_organization_discord_feed(self) -> None:
"""Test organization Discord feed returns 200."""
url: str = reverse("twitch:organization_feed_discord")
url: str = reverse("core:organization_feed_discord")
response: _MonkeyPatchedWSGIResponse = self.client.get(url)
assert response.status_code == 200
assert response["Content-Type"] == "application/xml; charset=utf-8"
@ -1262,7 +1264,7 @@ class DiscordFeedTestCase(TestCase):
def test_game_discord_feed(self) -> None:
"""Test game Discord feed returns 200."""
url: str = reverse("twitch:game_feed_discord")
url: str = reverse("core:game_feed_discord")
response: _MonkeyPatchedWSGIResponse = self.client.get(url)
assert response.status_code == 200
assert response["Content-Type"] == "application/xml; charset=utf-8"
@ -1272,7 +1274,7 @@ class DiscordFeedTestCase(TestCase):
def test_campaign_discord_feed(self) -> None:
"""Test campaign Discord feed returns 200 with Discord timestamps."""
url: str = reverse("twitch:campaign_feed_discord")
url: str = reverse("core:campaign_feed_discord")
response: _MonkeyPatchedWSGIResponse = self.client.get(url)
assert response.status_code == 200
assert response["Content-Type"] == "application/xml; charset=utf-8"
@ -1286,7 +1288,7 @@ class DiscordFeedTestCase(TestCase):
def test_game_campaign_discord_feed(self) -> None:
"""Test game-specific campaign Discord feed returns 200."""
url: str = reverse(
"twitch:game_campaign_feed_discord",
"core:game_campaign_feed_discord",
args=[self.game.twitch_id],
)
response: _MonkeyPatchedWSGIResponse = self.client.get(url)
@ -1298,7 +1300,7 @@ class DiscordFeedTestCase(TestCase):
def test_reward_campaign_discord_feed(self) -> None:
"""Test reward campaign Discord feed returns 200."""
url: str = reverse("twitch:reward_campaign_feed_discord")
url: str = reverse("core:reward_campaign_feed_discord")
response: _MonkeyPatchedWSGIResponse = self.client.get(url)
assert response.status_code == 200
assert response["Content-Type"] == "application/xml; charset=utf-8"
@ -1313,27 +1315,27 @@ class DiscordFeedTestCase(TestCase):
"""All Discord feeds should use absolute URL entry IDs and matching self links."""
discord_feed_cases: list[tuple[str, dict[str, str], str]] = [
(
"twitch:campaign_feed_discord",
"core:campaign_feed_discord",
{},
f"http://testserver{reverse('twitch:campaign_detail', args=[self.campaign.twitch_id])}",
),
(
"twitch:game_feed_discord",
"core:game_feed_discord",
{},
f"http://testserver{reverse('twitch:game_detail', args=[self.game.twitch_id])}",
),
(
"twitch:game_campaign_feed_discord",
"core:game_campaign_feed_discord",
{"twitch_id": self.game.twitch_id},
f"http://testserver{reverse('twitch:campaign_detail', args=[self.campaign.twitch_id])}",
),
(
"twitch:organization_feed_discord",
"core:organization_feed_discord",
{},
f"http://testserver{reverse('twitch:organization_detail', args=[self.org.twitch_id])}",
),
(
"twitch:reward_campaign_feed_discord",
"core:reward_campaign_feed_discord",
{},
f"http://testserver{reverse('twitch:reward_campaign_detail', args=[self.reward_campaign.twitch_id])}",
),
@ -1359,11 +1361,11 @@ class DiscordFeedTestCase(TestCase):
def test_discord_feeds_include_stylesheet_processing_instruction(self) -> None:
"""Discord feeds should include an xml-stylesheet processing instruction."""
feed_urls: list[str] = [
reverse("twitch:campaign_feed_discord"),
reverse("twitch:game_feed_discord"),
reverse("twitch:game_campaign_feed_discord", args=[self.game.twitch_id]),
reverse("twitch:organization_feed_discord"),
reverse("twitch:reward_campaign_feed_discord"),
reverse("core:campaign_feed_discord"),
reverse("core:game_feed_discord"),
reverse("core:game_campaign_feed_discord", args=[self.game.twitch_id]),
reverse("core:organization_feed_discord"),
reverse("core:reward_campaign_feed_discord"),
]
for url in feed_urls:
@ -1378,7 +1380,7 @@ class DiscordFeedTestCase(TestCase):
def test_discord_campaign_feed_contains_discord_timestamps(self) -> None:
"""Discord campaign feed should contain Discord relative timestamps."""
url: str = reverse("twitch:campaign_feed_discord")
url: str = reverse("core:campaign_feed_discord")
response: _MonkeyPatchedWSGIResponse = self.client.get(url)
assert response.status_code == 200
content: str = response.content.decode("utf-8")
@ -1392,7 +1394,7 @@ class DiscordFeedTestCase(TestCase):
def test_discord_reward_campaign_feed_contains_discord_timestamps(self) -> None:
"""Discord reward campaign feed should contain Discord relative timestamps."""
url: str = reverse("twitch:reward_campaign_feed_discord")
url: str = reverse("core:reward_campaign_feed_discord")
response: _MonkeyPatchedWSGIResponse = self.client.get(url)
assert response.status_code == 200
content: str = response.content.decode("utf-8")

View file

@ -327,7 +327,7 @@ class TestChannelListView:
def test_channel_list_loads(self, client: Client) -> None:
"""Test that channel list view loads successfully."""
response: _MonkeyPatchedWSGIResponse = client.get("/channels/")
response: _MonkeyPatchedWSGIResponse = client.get("/twitch/channels/")
assert response.status_code == 200
def test_campaign_count_annotation(
@ -342,7 +342,7 @@ class TestChannelListView:
channel: Channel = channel_with_campaigns["channel"] # type: ignore[assignment]
campaigns: list[DropCampaign] = channel_with_campaigns["campaigns"] # type: ignore[assignment]
response: _MonkeyPatchedWSGIResponse = client.get("/channels/")
response: _MonkeyPatchedWSGIResponse = client.get("/twitch/channels/")
context: ContextList | dict[str, Any] = response.context # type: ignore[assignment]
if isinstance(context, list):
context = context[-1]
@ -375,7 +375,7 @@ class TestChannelListView:
display_name="NoCampaigns",
)
response: _MonkeyPatchedWSGIResponse = client.get("/channels/")
response: _MonkeyPatchedWSGIResponse = client.get("/twitch/channels/")
context: ContextList | dict[str, Any] = response.context # type: ignore[assignment]
if isinstance(context, list):
context = context[-1]
@ -420,7 +420,7 @@ class TestChannelListView:
)
campaign.allow_channels.add(channel2)
response: _MonkeyPatchedWSGIResponse = client.get("/channels/")
response: _MonkeyPatchedWSGIResponse = client.get("/twitch/channels/")
context: ContextList | dict[str, Any] = response.context # type: ignore[assignment]
if isinstance(context, list):
context = context[-1]
@ -462,7 +462,7 @@ class TestChannelListView:
)
response: _MonkeyPatchedWSGIResponse = client.get(
f"/channels/?search={channel.name}",
f"/twitch/channels/?search={channel.name}",
)
context: ContextList | dict[str, Any] = response.context # type: ignore[assignment]
if isinstance(context, list):
@ -527,7 +527,7 @@ class TestChannelListView:
@pytest.mark.django_db
def test_debug_view(self, client: Client) -> None:
"""Test debug view returns 200 and has games_without_owner in context."""
response: _MonkeyPatchedWSGIResponse = client.get(reverse("twitch:debug"))
response: _MonkeyPatchedWSGIResponse = client.get(reverse("core:debug"))
assert response.status_code == 200
assert "games_without_owner" in response.context
@ -1014,7 +1014,7 @@ class TestChannelListView:
@pytest.mark.django_db
def test_docs_rss_view(self, client: Client) -> None:
"""Test docs RSS view returns 200 and has feeds in context."""
response: _MonkeyPatchedWSGIResponse = client.get(reverse("twitch:docs_rss"))
response: _MonkeyPatchedWSGIResponse = client.get(reverse("core:docs_rss"))
assert response.status_code == 200
assert "feeds" in response.context
assert "filtered_feeds" in response.context
@ -1268,7 +1268,7 @@ class TestSEOMetaTags:
def test_noindex_pages_have_robots_directive(self, client: Client) -> None:
"""Test that pages with noindex have proper robots directive."""
response: _MonkeyPatchedWSGIResponse = client.get(
reverse("twitch:dataset_backups"),
reverse("core:dataset_backups"),
)
assert response.status_code == 200
assert "robots_directive" in response.context
@ -1405,7 +1405,7 @@ class TestSitemapView:
channel: Channel = sample_entities["channel"]
response: _MonkeyPatchedWSGIResponse = client.get("/sitemap.xml")
content: str = response.content.decode()
assert f"/channels/{channel.twitch_id}/" in content
assert f"/twitch/channels/{channel.twitch_id}/" in content
def test_sitemap_contains_badge_detail_pages(
self,

View file

@ -3,21 +3,6 @@ from typing import TYPE_CHECKING
from django.urls import path
from twitch import views
from twitch.feeds import DropCampaignAtomFeed
from twitch.feeds import DropCampaignDiscordFeed
from twitch.feeds import DropCampaignFeed
from twitch.feeds import GameAtomFeed
from twitch.feeds import GameCampaignAtomFeed
from twitch.feeds import GameCampaignDiscordFeed
from twitch.feeds import GameCampaignFeed
from twitch.feeds import GameDiscordFeed
from twitch.feeds import GameFeed
from twitch.feeds import OrganizationAtomFeed
from twitch.feeds import OrganizationDiscordFeed
from twitch.feeds import OrganizationRSSFeed
from twitch.feeds import RewardCampaignAtomFeed
from twitch.feeds import RewardCampaignDiscordFeed
from twitch.feeds import RewardCampaignFeed
if TYPE_CHECKING:
from django.urls.resolvers import URLPattern
@ -27,129 +12,82 @@ app_name = "twitch"
urlpatterns: list[URLPattern | URLResolver] = [
# /twitch/
path("", views.dashboard, name="dashboard"),
# /twitch/badges/
path("badges/", views.badge_list_view, name="badge_list"),
# /twitch/badges/<set_id>/
path("badges/<str:set_id>/", views.badge_set_detail_view, name="badge_set_detail"),
# /twitch/campaigns/
path("campaigns/", views.drop_campaign_list_view, name="campaign_list"),
# /twitch/campaigns/<twitch_id>/
path(
"campaigns/<str:twitch_id>/",
views.drop_campaign_detail_view,
name="campaign_detail",
),
# /twitch/channels/
path("channels/", views.ChannelListView.as_view(), name="channel_list"),
# /twitch/channels/<twitch_id>/
path(
"channels/<str:twitch_id>/",
views.ChannelDetailView.as_view(),
name="channel_detail",
),
path("debug/", views.debug_view, name="debug"),
path("datasets/", views.dataset_backups_view, name="dataset_backups"),
path(
"datasets/download/<path:relative_path>/",
views.dataset_backup_download_view,
name="dataset_backup_download",
),
path("docs/rss/", views.docs_rss_view, name="docs_rss"),
# /twitch/emotes/
path("emotes/", views.emote_gallery_view, name="emote_gallery"),
# /twitch/games/
path("games/", views.GamesGridView.as_view(), name="games_grid"),
# /twitch/games/list/
path("games/list/", views.GamesListView.as_view(), name="games_list"),
# /twitch/games/<twitch_id>/
path("games/<str:twitch_id>/", views.GameDetailView.as_view(), name="game_detail"),
# /twitch/organizations/
path("organizations/", views.org_list_view, name="org_list"),
# /twitch/organizations/<twitch_id>/
path(
"organizations/<str:twitch_id>/",
views.organization_detail_view,
name="organization_detail",
),
# /twitch/reward-campaigns/
path(
"reward-campaigns/",
views.reward_campaign_list_view,
name="reward_campaign_list",
),
# /twitch/reward-campaigns/<twitch_id>/
path(
"reward-campaigns/<str:twitch_id>/",
views.reward_campaign_detail_view,
name="reward_campaign_detail",
),
path("search/", views.search_view, name="search"),
# /twitch/export/campaigns/csv/
path(
"export/campaigns/csv/",
views.export_campaigns_csv,
name="export_campaigns_csv",
),
# /twitch/export/campaigns/json/
path(
"export/campaigns/json/",
views.export_campaigns_json,
name="export_campaigns_json",
),
# /twitch/export/games/csv/
path("export/games/csv/", views.export_games_csv, name="export_games_csv"),
# /twitch/export/games/json/
path("export/games/json/", views.export_games_json, name="export_games_json"),
# /twitch/export/organizations/csv/
path(
"export/organizations/csv/",
views.export_organizations_csv,
name="export_organizations_csv",
),
# /twitch/export/organizations/json/
path(
"export/organizations/json/",
views.export_organizations_json,
name="export_organizations_json",
),
# RSS feeds
# /rss/campaigns/ - all active campaigns
path("rss/campaigns/", DropCampaignFeed(), name="campaign_feed"),
# /rss/games/ - newly added games
path("rss/games/", GameFeed(), name="game_feed"),
# /rss/games/<twitch_id>/campaigns/ - active campaigns for a specific game
path(
"rss/games/<str:twitch_id>/campaigns/",
GameCampaignFeed(),
name="game_campaign_feed",
),
# /rss/organizations/ - newly added organizations
path(
"rss/organizations/",
OrganizationRSSFeed(),
name="organization_feed",
),
# /rss/reward-campaigns/ - all active reward campaigns
path(
"rss/reward-campaigns/",
RewardCampaignFeed(),
name="reward_campaign_feed",
),
# Atom feeds (added alongside RSS to preserve backward compatibility)
path("atom/campaigns/", DropCampaignAtomFeed(), name="campaign_feed_atom"),
path("atom/games/", GameAtomFeed(), name="game_feed_atom"),
path(
"atom/games/<str:twitch_id>/campaigns/",
GameCampaignAtomFeed(),
name="game_campaign_feed_atom",
),
path(
"atom/organizations/",
OrganizationAtomFeed(),
name="organization_feed_atom",
),
path(
"atom/reward-campaigns/",
RewardCampaignAtomFeed(),
name="reward_campaign_feed_atom",
),
# Discord feeds (Atom feeds with Discord relative timestamps)
path("discord/campaigns/", DropCampaignDiscordFeed(), name="campaign_feed_discord"),
path("discord/games/", GameDiscordFeed(), name="game_feed_discord"),
path(
"discord/games/<str:twitch_id>/campaigns/",
GameCampaignDiscordFeed(),
name="game_campaign_feed_discord",
),
path(
"discord/organizations/",
OrganizationDiscordFeed(),
name="organization_feed_discord",
),
path(
"discord/reward-campaigns/",
RewardCampaignDiscordFeed(),
name="reward_campaign_feed_discord",
),
]

View file

@ -2,36 +2,26 @@ import csv
import datetime
import json
import logging
import operator
from collections import OrderedDict
from collections import defaultdict
from copy import copy
from typing import TYPE_CHECKING
from typing import Any
from typing import Literal
from django.conf import settings
from django.core.paginator import EmptyPage
from django.core.paginator import Page
from django.core.paginator import PageNotAnInteger
from django.core.paginator import Paginator
from django.core.serializers import serialize
from django.db import connection
from django.db.models import Case
from django.db.models import Count
from django.db.models import Exists
from django.db.models import F
from django.db.models import OuterRef
from django.db.models import Prefetch
from django.db.models import Q
from django.db.models import When
from django.db.models.functions import Trim
from django.db.models.query import QuerySet
from django.http import FileResponse
from django.http import Http404
from django.http import HttpResponse
from django.shortcuts import render
from django.template.defaultfilters import filesizeformat
from django.urls import reverse
from django.utils import timezone
from django.views.generic import DetailView
@ -40,21 +30,6 @@ from pygments import highlight
from pygments.formatters import HtmlFormatter
from pygments.lexers.data import JsonLexer
from twitch.feeds import DropCampaignAtomFeed
from twitch.feeds import DropCampaignDiscordFeed
from twitch.feeds import DropCampaignFeed
from twitch.feeds import GameAtomFeed
from twitch.feeds import GameCampaignAtomFeed
from twitch.feeds import GameCampaignDiscordFeed
from twitch.feeds import GameCampaignFeed
from twitch.feeds import GameDiscordFeed
from twitch.feeds import GameFeed
from twitch.feeds import OrganizationAtomFeed
from twitch.feeds import OrganizationDiscordFeed
from twitch.feeds import OrganizationRSSFeed
from twitch.feeds import RewardCampaignAtomFeed
from twitch.feeds import RewardCampaignDiscordFeed
from twitch.feeds import RewardCampaignFeed
from twitch.models import Channel
from twitch.models import ChatBadge
from twitch.models import ChatBadgeSet
@ -66,11 +41,6 @@ from twitch.models import RewardCampaign
from twitch.models import TimeBasedDrop
if TYPE_CHECKING:
from collections.abc import Callable
from os import stat_result
from pathlib import Path
from debug_toolbar.utils import QueryDict
from django.db.models import QuerySet
from django.http import HttpRequest
@ -274,105 +244,6 @@ def emote_gallery_view(request: HttpRequest) -> HttpResponse:
return render(request, "twitch/emote_gallery.html", context)
# MARK: /search/
def search_view(request: HttpRequest) -> HttpResponse:
"""Search view for all models.
Args:
request: The HTTP request.
Returns:
HttpResponse: The rendered search results.
"""
query: str = request.GET.get("q", "")
results: dict[str, QuerySet] = {}
if query:
if len(query) < MIN_QUERY_LENGTH_FOR_FTS:
results["organizations"] = Organization.objects.filter(
name__istartswith=query,
)
results["games"] = Game.objects.filter(
Q(name__istartswith=query) | Q(display_name__istartswith=query),
)
results["campaigns"] = DropCampaign.objects.filter(
Q(name__istartswith=query) | Q(description__icontains=query),
).select_related("game")
results["drops"] = TimeBasedDrop.objects.filter(
name__istartswith=query,
).select_related("campaign")
results["benefits"] = DropBenefit.objects.filter(
name__istartswith=query,
).prefetch_related("drops__campaign")
results["reward_campaigns"] = RewardCampaign.objects.filter(
Q(name__istartswith=query)
| Q(brand__istartswith=query)
| Q(summary__icontains=query),
).select_related("game")
results["badge_sets"] = ChatBadgeSet.objects.filter(
set_id__istartswith=query,
)
results["badges"] = ChatBadge.objects.filter(
Q(title__istartswith=query) | Q(description__icontains=query),
).select_related("badge_set")
else:
results["organizations"] = Organization.objects.filter(
name__icontains=query,
)
results["games"] = Game.objects.filter(
Q(name__icontains=query) | Q(display_name__icontains=query),
)
results["campaigns"] = DropCampaign.objects.filter(
Q(name__icontains=query) | Q(description__icontains=query),
).select_related("game")
results["drops"] = TimeBasedDrop.objects.filter(
name__icontains=query,
).select_related("campaign")
results["benefits"] = DropBenefit.objects.filter(
name__icontains=query,
).prefetch_related("drops__campaign")
results["reward_campaigns"] = RewardCampaign.objects.filter(
Q(name__icontains=query)
| Q(brand__icontains=query)
| Q(summary__icontains=query),
).select_related("game")
results["badge_sets"] = ChatBadgeSet.objects.filter(set_id__icontains=query)
results["badges"] = ChatBadge.objects.filter(
Q(title__icontains=query) | Q(description__icontains=query),
).select_related("badge_set")
total_results_count: int = sum(len(qs) for qs in results.values())
# TODO(TheLovinator): Make the description more informative by including counts of each result type, e.g. "Found 5 games, 3 campaigns, and 10 drops for 'rust'." # noqa: TD003
if query:
page_title: str = f"Search Results for '{query}'"[:60]
page_description: str = f"Found {total_results_count} results for '{query}'."
else:
page_title = "Search"
page_description = "Search for drops, games, channels, and organizations."
seo_context: dict[str, Any] = _build_seo_context(
page_title=page_title,
page_description=page_description,
)
return render(
request,
"twitch/search_results.html",
{"query": query, "results": results, **seo_context},
)
# MARK: /organizations/
def org_list_view(request: HttpRequest) -> HttpResponse:
"""Function-based view for organization list.
@ -624,111 +495,6 @@ def format_and_color_json(data: dict[str, Any] | list[dict] | str) -> str:
return highlight(formatted_code, JsonLexer(), HtmlFormatter())
# MARK: /datasets/
def dataset_backups_view(request: HttpRequest) -> HttpResponse:
"""View to list database backup datasets on disk.
Args:
request: The HTTP request.
Returns:
HttpResponse: The rendered dataset backups page.
"""
# TODO(TheLovinator): Instead of only using sql we should also support other formats like parquet, csv, or json. # noqa: TD003
# TODO(TheLovinator): Upload to s3 instead. # noqa: TD003
# TODO(TheLovinator): https://developers.google.com/search/docs/appearance/structured-data/dataset#json-ld
datasets_root: Path = settings.DATA_DIR / "datasets"
search_dirs: list[Path] = [datasets_root]
seen_paths: set[str] = set()
datasets: list[dict[str, Any]] = []
for folder in search_dirs:
if not folder.exists() or not folder.is_dir():
continue
# Only include .zst files
for path in folder.glob("*.zst"):
if not path.is_file():
continue
key = str(path.resolve())
if key in seen_paths:
continue
seen_paths.add(key)
stat: stat_result = path.stat()
updated_at: datetime.datetime = datetime.datetime.fromtimestamp(
stat.st_mtime,
tz=timezone.get_current_timezone(),
)
try:
display_path = str(path.relative_to(datasets_root))
download_path: str | None = display_path
except ValueError:
display_path: str = path.name
download_path: str | None = None
datasets.append({
"name": path.name,
"display_path": display_path,
"download_path": download_path,
"size": filesizeformat(stat.st_size),
"updated_at": updated_at,
})
datasets.sort(key=operator.itemgetter("updated_at"), reverse=True)
seo_context: dict[str, Any] = _build_seo_context(
page_title="Twitch Dataset",
page_description="Database backups and datasets available for download.",
)
context: dict[str, Any] = {
"datasets": datasets,
"data_dir": str(datasets_root),
"dataset_count": len(datasets),
**seo_context,
}
return render(request, "twitch/dataset_backups.html", context)
def dataset_backup_download_view(
request: HttpRequest, # noqa: ARG001
relative_path: str,
) -> FileResponse:
"""Download a dataset backup from the data directory.
Args:
request: The HTTP request.
relative_path: The path relative to the data directory.
Returns:
FileResponse: The file response for the requested dataset.
Raises:
Http404: When the file is not found or is outside the data directory.
"""
# TODO(TheLovinator): Use s3 instead of local disk. # noqa: TD003
datasets_root: Path = settings.DATA_DIR / "datasets"
requested_path: Path = (datasets_root / relative_path).resolve()
data_root: Path = datasets_root.resolve()
try:
requested_path.relative_to(data_root)
except ValueError as exc:
msg = "File not found"
raise Http404(msg) from exc
if not requested_path.exists() or not requested_path.is_file():
msg = "File not found"
raise Http404(msg)
if not requested_path.name.endswith(".zst"):
msg = "File not found"
raise Http404(msg)
return FileResponse(
requested_path.open("rb"),
as_attachment=True,
filename=requested_path.name,
)
def _enhance_drops_with_context(
drops: QuerySet[TimeBasedDrop],
now: datetime.datetime,
@ -1626,148 +1392,6 @@ def reward_campaign_detail_view(request: HttpRequest, twitch_id: str) -> HttpRes
return render(request, "twitch/reward_campaign_detail.html", context)
# MARK: /debug/
def debug_view(request: HttpRequest) -> HttpResponse:
"""Debug view showing potentially broken or inconsistent data.
Returns:
HttpResponse: Rendered debug template or redirect if unauthorized.
"""
now: datetime.datetime = timezone.now()
# Games with no assigned owner organization
games_without_owner: QuerySet[Game] = Game.objects.filter(
owners__isnull=True,
).order_by("display_name")
# Campaigns with no images at all (no direct URL and no benefit image fallbacks)
broken_image_campaigns: QuerySet[DropCampaign] = (
DropCampaign.objects
.filter(
Q(image_url__isnull=True)
| Q(image_url__exact="")
| ~Q(image_url__startswith="http"),
)
.exclude(
Exists(
TimeBasedDrop.objects.filter(campaign=OuterRef("pk")).filter(
benefits__image_asset_url__startswith="http",
),
),
)
.select_related("game")
)
# Benefits with missing images
broken_benefit_images: QuerySet[DropBenefit] = DropBenefit.objects.annotate(
trimmed_url=Trim("image_asset_url"),
).filter(
Q(image_asset_url__isnull=True)
| Q(trimmed_url__exact="")
| ~Q(image_asset_url__startswith="http"),
)
# Time-based drops without any benefits
drops_without_benefits: QuerySet[TimeBasedDrop] = TimeBasedDrop.objects.filter(
benefits__isnull=True,
).select_related("campaign__game")
# Campaigns with invalid dates (start after end or missing either)
invalid_date_campaigns: QuerySet[DropCampaign] = DropCampaign.objects.filter(
Q(start_at__gt=F("end_at")) | Q(start_at__isnull=True) | Q(end_at__isnull=True),
).select_related("game")
# Duplicate campaign names per game.
# We retrieve the game's name for user-friendly display.
duplicate_name_campaigns: QuerySet[DropCampaign, dict[str, Any]] = (
DropCampaign.objects
.values("game__display_name", "name", "game__twitch_id")
.annotate(name_count=Count("twitch_id"))
.filter(name_count__gt=1)
.order_by("game__display_name", "name")
)
# Active campaigns with no images at all
active_missing_image: QuerySet[DropCampaign] = (
DropCampaign.objects
.filter(start_at__lte=now, end_at__gte=now)
.filter(
Q(image_url__isnull=True)
| Q(image_url__exact="")
| ~Q(image_url__startswith="http"),
)
.exclude(
Exists(
TimeBasedDrop.objects.filter(campaign=OuterRef("pk")).filter(
benefits__image_asset_url__startswith="http",
),
),
)
.select_related("game")
)
# Distinct GraphQL operation names used to fetch campaigns with counts
# Since operation_names is now a JSON list field, we need to flatten and count
operation_names_counter: dict[str, int] = {}
for campaign in DropCampaign.objects.only("operation_names"):
for op_name in campaign.operation_names:
if op_name and op_name.strip():
operation_names_counter[op_name.strip()] = (
operation_names_counter.get(op_name.strip(), 0) + 1
)
operation_names_with_counts: list[dict[str, Any]] = [
{"trimmed_op": op_name, "count": count}
for op_name, count in sorted(operation_names_counter.items())
]
# Campaigns missing DropCampaignDetails operation name
# Need to handle SQLite separately since it doesn't support JSONField lookups
# Sqlite is used when testing
if connection.vendor == "sqlite":
all_campaigns: QuerySet[DropCampaign] = DropCampaign.objects.select_related(
"game",
).order_by("game__display_name", "name")
campaigns_missing_dropcampaigndetails: list[DropCampaign] = [
c
for c in all_campaigns
if c.operation_names is None
or "DropCampaignDetails" not in c.operation_names
]
else:
campaigns_missing_dropcampaigndetails: list[DropCampaign] = list(
DropCampaign.objects
.filter(
Q(operation_names__isnull=True)
| ~Q(operation_names__contains=["DropCampaignDetails"]),
)
.select_related("game")
.order_by("game__display_name", "name"),
)
context: dict[str, Any] = {
"now": now,
"games_without_owner": games_without_owner,
"broken_image_campaigns": broken_image_campaigns,
"broken_benefit_images": broken_benefit_images,
"drops_without_benefits": drops_without_benefits,
"invalid_date_campaigns": invalid_date_campaigns,
"duplicate_name_campaigns": duplicate_name_campaigns,
"active_missing_image": active_missing_image,
"operation_names_with_counts": operation_names_with_counts,
"campaigns_missing_dropcampaigndetails": campaigns_missing_dropcampaigndetails,
}
seo_context: dict[str, Any] = _build_seo_context(
page_title="Debug",
page_description="Debug view showing potentially broken or inconsistent data.",
robots_directive="noindex, nofollow",
)
context.update(seo_context)
return render(request, "twitch/debug.html", context)
# MARK: /games/list/
class GamesListView(GamesGridView):
"""List view for games in simple list format."""
@ -1775,184 +1399,6 @@ class GamesListView(GamesGridView):
template_name: str = "twitch/games_list.html"
# MARK: /docs/rss/
def docs_rss_view(request: HttpRequest) -> HttpResponse:
"""View for /docs/rss that lists all available RSS feeds.
Args:
request: The HTTP request object.
Returns:
Rendered HTML response with list of RSS feeds.
"""
def absolute(path: str) -> str:
try:
return request.build_absolute_uri(path)
except Exception:
logger.exception("Failed to build absolute URL for %s", path)
return path
def _pretty_example(xml_str: str, max_items: int = 1) -> str:
try:
trimmed: str = xml_str.strip()
first_item: int = trimmed.find("<item")
if first_item != -1 and max_items == 1:
second_item: int = trimmed.find("<item", first_item + 5)
if second_item != -1:
end_channel: int = trimmed.find("</channel>", second_item)
if end_channel != -1:
trimmed = trimmed[:second_item] + trimmed[end_channel:]
formatted: str = trimmed.replace("><", ">\n<")
return "\n".join(line for line in formatted.splitlines() if line.strip())
except Exception:
logger.exception("Failed to pretty-print RSS example")
return xml_str
def render_feed(feed_view: Callable[..., HttpResponse], *args: object) -> str:
try:
limited_request: HttpRequest = copy(request)
# Add limit=1 to GET parameters
get_data: QueryDict = request.GET.copy()
get_data["limit"] = "1"
limited_request.GET = get_data # pyright: ignore[reportAttributeAccessIssue]
response: HttpResponse = feed_view(limited_request, *args)
return _pretty_example(response.content.decode("utf-8"))
except Exception:
logger.exception(
"Failed to render %s for RSS docs",
feed_view.__class__.__name__,
)
return ""
show_atom: bool = bool(request.GET.get("show_atom"))
feeds: list[dict[str, str]] = [
{
"title": "All Organizations",
"description": "Latest organizations added to TTVDrops",
"url": absolute(reverse("twitch:organization_feed")),
"atom_url": absolute(reverse("twitch:organization_feed_atom")),
"discord_url": absolute(reverse("twitch:organization_feed_discord")),
"example_xml": render_feed(OrganizationRSSFeed()),
"example_xml_atom": render_feed(OrganizationAtomFeed())
if show_atom
else "",
"example_xml_discord": render_feed(OrganizationDiscordFeed())
if show_atom
else "",
},
{
"title": "All Games",
"description": "Latest games added to TTVDrops",
"url": absolute(reverse("twitch:game_feed")),
"atom_url": absolute(reverse("twitch:game_feed_atom")),
"discord_url": absolute(reverse("twitch:game_feed_discord")),
"example_xml": render_feed(GameFeed()),
"example_xml_atom": render_feed(GameAtomFeed()) if show_atom else "",
"example_xml_discord": render_feed(GameDiscordFeed()) if show_atom else "",
},
{
"title": "All Drop Campaigns",
"description": "Latest drop campaigns across all games",
"url": absolute(reverse("twitch:campaign_feed")),
"atom_url": absolute(reverse("twitch:campaign_feed_atom")),
"discord_url": absolute(reverse("twitch:campaign_feed_discord")),
"example_xml": render_feed(DropCampaignFeed()),
"example_xml_atom": render_feed(DropCampaignAtomFeed())
if show_atom
else "",
"example_xml_discord": render_feed(DropCampaignDiscordFeed())
if show_atom
else "",
},
{
"title": "All Reward Campaigns",
"description": "Latest reward campaigns (Quest rewards) on Twitch",
"url": absolute(reverse("twitch:reward_campaign_feed")),
"atom_url": absolute(reverse("twitch:reward_campaign_feed_atom")),
"discord_url": absolute(reverse("twitch:reward_campaign_feed_discord")),
"example_xml": render_feed(RewardCampaignFeed()),
"example_xml_atom": render_feed(RewardCampaignAtomFeed())
if show_atom
else "",
"example_xml_discord": render_feed(RewardCampaignDiscordFeed())
if show_atom
else "",
},
]
sample_game: Game | None = Game.objects.order_by("-added_at").first()
sample_org: Organization | None = Organization.objects.order_by("-added_at").first()
if sample_org is None and sample_game is not None:
sample_org = sample_game.owners.order_by("-pk").first()
filtered_feeds: list[dict[str, str | bool]] = [
{
"title": "Campaigns for a Single Game",
"description": "Latest drop campaigns for one game.",
"url": (
absolute(
reverse("twitch:game_campaign_feed", args=[sample_game.twitch_id]),
)
if sample_game
else absolute("/rss/games/<game_id>/campaigns/")
),
"atom_url": (
absolute(
reverse(
"twitch:game_campaign_feed_atom",
args=[sample_game.twitch_id],
),
)
if sample_game
else absolute("/atom/games/<game_id>/campaigns/")
),
"discord_url": (
absolute(
reverse(
"twitch:game_campaign_feed_discord",
args=[sample_game.twitch_id],
),
)
if sample_game
else absolute("/discord/games/<game_id>/campaigns/")
),
"has_sample": bool(sample_game),
"example_xml": render_feed(GameCampaignFeed(), sample_game.twitch_id)
if sample_game
else "",
"example_xml_atom": (
render_feed(GameCampaignAtomFeed(), sample_game.twitch_id)
if sample_game and show_atom
else ""
),
"example_xml_discord": (
render_feed(GameCampaignDiscordFeed(), sample_game.twitch_id)
if sample_game and show_atom
else ""
),
},
]
seo_context: dict[str, Any] = _build_seo_context(
page_title="Twitch RSS Feeds",
page_description="RSS feeds for Twitch drops.",
)
return render(
request,
"twitch/docs_rss.html",
{
"feeds": feeds,
"filtered_feeds": filtered_feeds,
"sample_game": sample_game,
"sample_org": sample_org,
**seo_context,
},
)
# MARK: /channels/
class ChannelListView(ListView):
"""List view for channels."""
@ -2647,143 +2093,3 @@ def export_organizations_json(request: HttpRequest) -> HttpResponse: # noqa: AR
response["Content-Disposition"] = "attachment; filename=organizations.json"
return response
# MARK: /sitemap.xml
def sitemap_view(request: HttpRequest) -> HttpResponse: # noqa: PLR0915
"""Generate a dynamic XML sitemap for search engines.
Args:
request: The HTTP request.
Returns:
HttpResponse: XML sitemap.
"""
base_url: str = f"{request.scheme}://{request.get_host()}"
# Start building sitemap XML
sitemap_urls: list[dict[str, str | dict[str, str]]] = []
# Static pages
sitemap_urls.extend([
{"url": f"{base_url}/", "priority": "1.0", "changefreq": "daily"},
{"url": f"{base_url}/campaigns/", "priority": "0.9", "changefreq": "daily"},
{
"url": f"{base_url}/reward-campaigns/",
"priority": "0.9",
"changefreq": "daily",
},
{"url": f"{base_url}/games/", "priority": "0.9", "changefreq": "weekly"},
{
"url": f"{base_url}/organizations/",
"priority": "0.8",
"changefreq": "weekly",
},
{"url": f"{base_url}/channels/", "priority": "0.8", "changefreq": "weekly"},
{"url": f"{base_url}/badges/", "priority": "0.7", "changefreq": "monthly"},
{"url": f"{base_url}/emotes/", "priority": "0.7", "changefreq": "monthly"},
{"url": f"{base_url}/search/", "priority": "0.6", "changefreq": "monthly"},
])
# Dynamic detail pages - Games
games: QuerySet[Game] = Game.objects.all()
for game in games:
entry: dict[str, str | dict[str, str]] = {
"url": f"{base_url}{reverse('twitch:game_detail', args=[game.twitch_id])}",
"priority": "0.8",
"changefreq": "weekly",
}
if game.updated_at:
entry["lastmod"] = game.updated_at.isoformat()
sitemap_urls.append(entry)
# Dynamic detail pages - Campaigns
campaigns: QuerySet[DropCampaign] = DropCampaign.objects.all()
for campaign in campaigns:
resource_url: str = reverse("twitch:campaign_detail", args=[campaign.twitch_id])
full_url: str = f"{base_url}{resource_url}"
entry: dict[str, str | dict[str, str]] = {
"url": full_url,
"priority": "0.7",
"changefreq": "weekly",
}
if campaign.updated_at:
entry["lastmod"] = campaign.updated_at.isoformat()
sitemap_urls.append(entry)
# Dynamic detail pages - Organizations
orgs: QuerySet[Organization] = Organization.objects.all()
for org in orgs:
resource_url = reverse("twitch:organization_detail", args=[org.twitch_id])
full_url: str = f"{base_url}{resource_url}"
entry: dict[str, str | dict[str, str]] = {
"url": full_url,
"priority": "0.7",
"changefreq": "weekly",
}
if org.updated_at:
entry["lastmod"] = org.updated_at.isoformat()
sitemap_urls.append(entry)
# Dynamic detail pages - Channels
channels: QuerySet[Channel] = Channel.objects.all()
for channel in channels:
resource_url = reverse("twitch:channel_detail", args=[channel.twitch_id])
full_url: str = f"{base_url}{resource_url}"
entry: dict[str, str | dict[str, str]] = {
"url": full_url,
"priority": "0.6",
"changefreq": "weekly",
}
if channel.updated_at:
entry["lastmod"] = channel.updated_at.isoformat()
sitemap_urls.append(entry)
# Dynamic detail pages - Badges
badge_sets: QuerySet[ChatBadgeSet] = ChatBadgeSet.objects.all()
for badge_set in badge_sets:
resource_url = reverse("twitch:badge_set_detail", args=[badge_set.set_id])
full_url: str = f"{base_url}{resource_url}"
sitemap_urls.append({
"url": full_url,
"priority": "0.5",
"changefreq": "monthly",
})
# Dynamic detail pages - Reward Campaigns
reward_campaigns: QuerySet[RewardCampaign] = RewardCampaign.objects.all()
for reward_campaign in reward_campaigns:
resource_url = reverse(
"twitch:reward_campaign_detail",
args=[
reward_campaign.twitch_id,
],
)
full_url: str = f"{base_url}{resource_url}"
entry: dict[str, str | dict[str, str]] = {
"url": full_url,
"priority": "0.6",
"changefreq": "weekly",
}
if reward_campaign.updated_at:
entry["lastmod"] = reward_campaign.updated_at.isoformat()
sitemap_urls.append(entry)
# Build XML
xml_content = '<?xml version="1.0" encoding="UTF-8"?>\n'
xml_content += '<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">\n'
for url_entry in sitemap_urls:
xml_content += " <url>\n"
xml_content += f" <loc>{url_entry['url']}</loc>\n"
if url_entry.get("lastmod"):
xml_content += f" <lastmod>{url_entry['lastmod']}</lastmod>\n"
xml_content += (
f" <changefreq>{url_entry.get('changefreq', 'monthly')}</changefreq>\n"
)
xml_content += f" <priority>{url_entry.get('priority', '0.5')}</priority>\n"
xml_content += " </url>\n"
xml_content += "</urlset>"
return HttpResponse(xml_content, content_type="application/xml")