Refactor HTML
This commit is contained in:
parent
a12b34a665
commit
05eb0d92e3
27 changed files with 776 additions and 393 deletions
|
|
@ -48,7 +48,15 @@ def _with_campaign_related(queryset: QuerySet[DropCampaign]) -> QuerySet[DropCam
|
|||
queryset=TimeBasedDrop.objects.prefetch_related("benefits"),
|
||||
)
|
||||
|
||||
return queryset.select_related("game").prefetch_related("game__owners", "allow_channels", drops_prefetch)
|
||||
return queryset.select_related("game").prefetch_related(
|
||||
"game__owners",
|
||||
Prefetch(
|
||||
"allow_channels",
|
||||
queryset=Channel.objects.order_by("display_name"),
|
||||
to_attr="channels_ordered",
|
||||
),
|
||||
drops_prefetch,
|
||||
)
|
||||
|
||||
|
||||
def insert_date_info(item: Model, parts: list[SafeText]) -> None:
|
||||
|
|
@ -122,30 +130,27 @@ def _build_drops_data(drops_qs: QuerySet[TimeBasedDrop]) -> list[dict]:
|
|||
return drops_data
|
||||
|
||||
|
||||
def _build_channels_html(channels: QuerySet[Channel], game: Game | None) -> SafeText:
|
||||
def _build_channels_html(channels: list[Channel] | QuerySet[Channel], game: Game | None) -> SafeText:
|
||||
"""Render up to max_links channel links as <li>, then a count of additional channels, or fallback to game category link.
|
||||
|
||||
If only one channel and drop_requirements is '1 subscriptions required',
|
||||
merge the Twitch link with the '1 subs' row.
|
||||
|
||||
Args:
|
||||
channels (QuerySet[Channel]): The queryset of channels.
|
||||
channels (list[Channel] | QuerySet[Channel]): The channels (already ordered).
|
||||
game (Game | None): The game object for fallback link.
|
||||
|
||||
Returns:
|
||||
SafeText: HTML <ul> with up to max_links channel links, count of more, or fallback link.
|
||||
""" # noqa: E501
|
||||
max_links = 5
|
||||
channels_all: list[Channel] = list(channels.all())
|
||||
channels_all: list[Channel] = list(channels) if isinstance(channels, list) else list(channels.all())
|
||||
total: int = len(channels_all)
|
||||
|
||||
if channels_all:
|
||||
items: list[SafeString] = [
|
||||
format_html(
|
||||
"<li>"
|
||||
'<a href="https://twitch.tv/{}" target="_blank" rel="noopener noreferrer"'
|
||||
' title="Watch {} on Twitch">{}</a>'
|
||||
"</li>",
|
||||
'<li><a href="https://twitch.tv/{}" title="Watch {} on Twitch">{}</a></li>',
|
||||
ch.name,
|
||||
ch.display_name,
|
||||
ch.display_name,
|
||||
|
|
@ -175,10 +180,7 @@ def _build_channels_html(channels: QuerySet[Channel], game: Game | None) -> Safe
|
|||
# If no channel is associated, the drop is category-wide; link to the game's Twitch directory
|
||||
display_name: str = getattr(game, "display_name", "this game")
|
||||
return format_html(
|
||||
"<ul><li>"
|
||||
'<a href="{}" target="_blank" rel="noopener noreferrer"'
|
||||
' title="Browse {} category">Category-wide for {}</a>'
|
||||
"</li></ul>",
|
||||
'<ul><li><a href="{}" title="Browse {} category">Category-wide for {}</a></li></ul>',
|
||||
game.twitch_directory_url,
|
||||
display_name,
|
||||
display_name,
|
||||
|
|
@ -189,11 +191,9 @@ def _get_channel_name_from_drops(drops: QuerySet[TimeBasedDrop]) -> str | None:
|
|||
for d in drops:
|
||||
campaign: DropCampaign | None = getattr(d, "campaign", None)
|
||||
if campaign:
|
||||
allow_channels: QuerySet[Channel] | None = getattr(campaign, "allow_channels", None)
|
||||
if allow_channels:
|
||||
channels: QuerySet[Channel, Channel] = allow_channels.all()
|
||||
if channels:
|
||||
return channels[0].name
|
||||
channels: list[Channel] | None = getattr(campaign, "channels_ordered", None)
|
||||
if channels:
|
||||
return channels[0].name
|
||||
return None
|
||||
|
||||
|
||||
|
|
@ -279,7 +279,7 @@ def _construct_drops_summary(drops_data: list[dict]) -> SafeText:
|
|||
badge_desc: str | None = badge_descriptions_by_title.get(benefit_name)
|
||||
if is_sub_required and channel_name:
|
||||
linked_name: SafeString = format_html(
|
||||
'<a href="https://twitch.tv/{}" target="_blank">{}</a>',
|
||||
'<a href="https://twitch.tv/{}" >{}</a>',
|
||||
channel_name,
|
||||
benefit_name,
|
||||
)
|
||||
|
|
@ -427,7 +427,7 @@ class GameFeed(Feed):
|
|||
if slug:
|
||||
description_parts.append(
|
||||
SafeText(
|
||||
f"<p><a href='https://www.twitch.tv/directory/game/{slug}' target='_blank' rel='noopener noreferrer'>{game_name} by {game_owner}</a></p>", # noqa: E501
|
||||
f"<p><a href='https://www.twitch.tv/directory/game/{slug}'>{game_name} by {game_owner}</a></p>",
|
||||
),
|
||||
)
|
||||
else:
|
||||
|
|
@ -559,7 +559,7 @@ class DropCampaignFeed(Feed):
|
|||
|
||||
# Only show channels if drop is not subscription only
|
||||
if not getattr(item, "is_subscription_only", False):
|
||||
channels: QuerySet[Channel] | None = getattr(item, "allow_channels", None)
|
||||
channels: list[Channel] | None = getattr(item, "channels_ordered", None)
|
||||
if channels is not None:
|
||||
game: Game | None = getattr(item, "game", None)
|
||||
parts.append(_build_channels_html(channels, game=game))
|
||||
|
|
@ -704,7 +704,7 @@ class GameCampaignFeed(Feed):
|
|||
|
||||
# Only show channels if drop is not subscription only
|
||||
if not getattr(item, "is_subscription_only", False):
|
||||
channels: QuerySet[Channel] | None = getattr(item, "allow_channels", None)
|
||||
channels: list[Channel] | None = getattr(item, "channels_ordered", None)
|
||||
if channels is not None:
|
||||
game: Game | None = getattr(item, "game", None)
|
||||
parts.append(_build_channels_html(channels, game=game))
|
||||
|
|
@ -888,7 +888,7 @@ class OrganizationCampaignFeed(Feed):
|
|||
|
||||
# Only show channels if drop is not subscription only
|
||||
if not getattr(item, "is_subscription_only", False):
|
||||
channels: QuerySet[Channel] | None = getattr(item, "allow_channels", None)
|
||||
channels: list[Channel] | None = getattr(item, "channels_ordered", None)
|
||||
if channels is not None:
|
||||
game: Game | None = getattr(item, "game", None)
|
||||
parts.append(_build_channels_html(channels, game=game))
|
||||
|
|
|
|||
|
|
@ -66,8 +66,7 @@ class Organization(auto_prefetch.Model):
|
|||
url: str = reverse("twitch:organization_detail", args=[self.twitch_id])
|
||||
|
||||
return format_html(
|
||||
"<p>New Twitch organization added to TTVDrops:</p>\n"
|
||||
'<p><a href="{}" target="_blank" rel="noopener noreferrer">{}</a></p>',
|
||||
'<p>New Twitch organization added to TTVDrops:</p>\n<p><a href="{}">{}</a></p>',
|
||||
url,
|
||||
name,
|
||||
)
|
||||
|
|
@ -456,6 +455,38 @@ class DropCampaign(auto_prefetch.Model):
|
|||
)
|
||||
return self.image_url or ""
|
||||
|
||||
@property
|
||||
def duration_iso(self) -> str:
|
||||
"""Return the campaign duration in ISO 8601 format (e.g., 'P3DT4H30M').
|
||||
|
||||
This is used for the <time> element's datetime attribute to provide machine-readable duration.
|
||||
If start_at or end_at is missing, returns an empty string.
|
||||
"""
|
||||
if not self.start_at or not self.end_at:
|
||||
return ""
|
||||
|
||||
total_seconds: int = int((self.end_at - self.start_at).total_seconds())
|
||||
if total_seconds < 0:
|
||||
total_seconds = abs(total_seconds)
|
||||
|
||||
days, remainder = divmod(total_seconds, 86400)
|
||||
hours, remainder = divmod(remainder, 3600)
|
||||
minutes, seconds = divmod(remainder, 60)
|
||||
|
||||
time_parts: list[str] = []
|
||||
if hours:
|
||||
time_parts.append(f"{hours}H")
|
||||
if minutes:
|
||||
time_parts.append(f"{minutes}M")
|
||||
if seconds or not time_parts:
|
||||
time_parts.append(f"{seconds}S")
|
||||
|
||||
if days and time_parts:
|
||||
return f"P{days}DT{''.join(time_parts)}"
|
||||
if days:
|
||||
return f"P{days}D"
|
||||
return f"PT{''.join(time_parts)}"
|
||||
|
||||
@property
|
||||
def is_subscription_only(self) -> bool:
|
||||
"""Determine if the campaign is subscription only based on its benefits."""
|
||||
|
|
|
|||
|
|
@ -126,27 +126,20 @@ class TestBackupCommand:
|
|||
assert output_dir.exists()
|
||||
assert len(list(output_dir.glob("test-*.sql.zst"))) == 1
|
||||
|
||||
def test_backup_uses_default_directory(self) -> None:
|
||||
def test_backup_uses_default_directory(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Test that backup uses DATA_DIR/datasets by default."""
|
||||
# Create test data so tables exist
|
||||
Organization.objects.create(twitch_id="test004", name="Test Org")
|
||||
|
||||
datasets_dir = settings.DATA_DIR / "datasets"
|
||||
monkeypatch.setattr(settings, "DATA_DIR", tmp_path)
|
||||
datasets_dir = tmp_path / "datasets"
|
||||
datasets_dir.mkdir(exist_ok=True, parents=True)
|
||||
|
||||
# Clean up any existing test backups
|
||||
for old_backup in datasets_dir.glob("ttvdrops-*.sql.zst"):
|
||||
old_backup.unlink()
|
||||
|
||||
call_command("backup_db")
|
||||
|
||||
backup_files = list(datasets_dir.glob("ttvdrops-*.sql.zst"))
|
||||
assert len(backup_files) >= 1
|
||||
|
||||
# Clean up
|
||||
for backup in backup_files:
|
||||
backup.unlink()
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
class TestBackupHelperFunctions:
|
||||
|
|
|
|||
|
|
@ -118,7 +118,6 @@ class TestBadgeSetDetailView:
|
|||
content = response.content.decode()
|
||||
|
||||
assert "vip" in content
|
||||
assert "Total Versions:" in content
|
||||
assert "1" in content
|
||||
|
||||
def test_badge_set_detail_json_data(self, client: Client) -> None:
|
||||
|
|
|
|||
125
twitch/tests/test_exports.py
Normal file
125
twitch/tests/test_exports.py
Normal file
|
|
@ -0,0 +1,125 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from datetime import timedelta
|
||||
|
||||
from django.test import Client
|
||||
from django.test import TestCase
|
||||
from django.utils import timezone
|
||||
|
||||
from twitch.models import DropCampaign
|
||||
from twitch.models import Game
|
||||
from twitch.models import Organization
|
||||
|
||||
|
||||
class ExportViewsTestCase(TestCase):
|
||||
"""Test export views for CSV and JSON formats."""
|
||||
|
||||
def setUp(self) -> None:
|
||||
"""Set up test data."""
|
||||
self.client = Client()
|
||||
|
||||
# Create test organization
|
||||
self.org = Organization.objects.create(
|
||||
twitch_id="org123",
|
||||
name="Test Organization",
|
||||
)
|
||||
|
||||
# Create test game
|
||||
self.game = Game.objects.create(
|
||||
twitch_id="game123",
|
||||
name="Test Game",
|
||||
display_name="Test Game Display",
|
||||
)
|
||||
self.game.owners.add(self.org)
|
||||
|
||||
# Create test campaign
|
||||
now = timezone.now()
|
||||
self.campaign = DropCampaign.objects.create(
|
||||
twitch_id="campaign123",
|
||||
name="Test Campaign",
|
||||
description="A test campaign description",
|
||||
game=self.game,
|
||||
start_at=now - timedelta(days=1),
|
||||
end_at=now + timedelta(days=1),
|
||||
)
|
||||
|
||||
def test_export_campaigns_csv(self) -> None:
|
||||
"""Test CSV export of campaigns."""
|
||||
response = self.client.get("/export/campaigns/csv/")
|
||||
assert response.status_code == 200
|
||||
assert response["Content-Type"] == "text/csv"
|
||||
assert b"Twitch ID" in response.content
|
||||
assert b"campaign123" in response.content
|
||||
assert b"Test Campaign" in response.content
|
||||
|
||||
def test_export_campaigns_json(self) -> None:
|
||||
"""Test JSON export of campaigns."""
|
||||
response = self.client.get("/export/campaigns/json/")
|
||||
assert response.status_code == 200
|
||||
assert response["Content-Type"] == "application/json"
|
||||
|
||||
data = json.loads(response.content)
|
||||
assert isinstance(data, list)
|
||||
assert len(data) == 1
|
||||
assert data[0]["twitch_id"] == "campaign123"
|
||||
assert data[0]["name"] == "Test Campaign"
|
||||
assert data[0]["status"] == "Active"
|
||||
|
||||
def test_export_games_csv(self) -> None:
|
||||
"""Test CSV export of games."""
|
||||
response = self.client.get("/export/games/csv/")
|
||||
assert response.status_code == 200
|
||||
assert response["Content-Type"] == "text/csv"
|
||||
assert b"Twitch ID" in response.content
|
||||
assert b"game123" in response.content
|
||||
assert b"Test Game Display" in response.content
|
||||
|
||||
def test_export_games_json(self) -> None:
|
||||
"""Test JSON export of games."""
|
||||
response = self.client.get("/export/games/json/")
|
||||
assert response.status_code == 200
|
||||
assert response["Content-Type"] == "application/json"
|
||||
|
||||
data = json.loads(response.content)
|
||||
assert isinstance(data, list)
|
||||
assert len(data) == 1
|
||||
assert data[0]["twitch_id"] == "game123"
|
||||
assert data[0]["display_name"] == "Test Game Display"
|
||||
|
||||
def test_export_organizations_csv(self) -> None:
|
||||
"""Test CSV export of organizations."""
|
||||
response = self.client.get("/export/organizations/csv/")
|
||||
assert response.status_code == 200
|
||||
assert response["Content-Type"] == "text/csv"
|
||||
assert b"Twitch ID" in response.content
|
||||
assert b"org123" in response.content
|
||||
assert b"Test Organization" in response.content
|
||||
|
||||
def test_export_organizations_json(self) -> None:
|
||||
"""Test JSON export of organizations."""
|
||||
response = self.client.get("/export/organizations/json/")
|
||||
assert response.status_code == 200
|
||||
assert response["Content-Type"] == "application/json"
|
||||
|
||||
data = json.loads(response.content)
|
||||
assert isinstance(data, list)
|
||||
assert len(data) == 1
|
||||
assert data[0]["twitch_id"] == "org123"
|
||||
assert data[0]["name"] == "Test Organization"
|
||||
|
||||
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")
|
||||
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")
|
||||
assert response.status_code == 200
|
||||
|
||||
data = json.loads(response.content)
|
||||
assert isinstance(data, list)
|
||||
assert len(data) == 1
|
||||
assert data[0]["status"] == "Active"
|
||||
|
|
@ -442,8 +442,8 @@ URL_NAMES: list[tuple[str, dict[str, str]]] = [
|
|||
("twitch:debug", {}),
|
||||
("twitch:docs_rss", {}),
|
||||
("twitch:emote_gallery", {}),
|
||||
("twitch:game_list", {}),
|
||||
("twitch:game_list_simple", {}),
|
||||
("twitch:games_grid", {}),
|
||||
("twitch:games_list", {}),
|
||||
("twitch:game_detail", {"twitch_id": "test-game-123"}),
|
||||
("twitch:org_list", {}),
|
||||
("twitch:organization_detail", {"twitch_id": "test-org-123"}),
|
||||
|
|
|
|||
|
|
@ -522,14 +522,14 @@ class TestChannelListView:
|
|||
@pytest.mark.django_db
|
||||
def test_games_grid_view(self, client: Client) -> None:
|
||||
"""Test games grid view returns 200 and has games in context."""
|
||||
response: _MonkeyPatchedWSGIResponse = client.get(reverse("twitch:game_list"))
|
||||
response: _MonkeyPatchedWSGIResponse = client.get(reverse("twitch:games_grid"))
|
||||
assert response.status_code == 200
|
||||
assert "games" in response.context
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_games_list_view(self, client: Client) -> None:
|
||||
"""Test games list view returns 200 and has games in context."""
|
||||
response: _MonkeyPatchedWSGIResponse = client.get(reverse("twitch:game_list_simple"))
|
||||
response: _MonkeyPatchedWSGIResponse = client.get(reverse("twitch:games_list"))
|
||||
assert response.status_code == 200
|
||||
assert "games" in response.context
|
||||
|
||||
|
|
|
|||
|
|
@ -35,14 +35,20 @@ urlpatterns: list[URLPattern] = [
|
|||
),
|
||||
path("docs/rss/", views.docs_rss_view, name="docs_rss"),
|
||||
path("emotes/", views.emote_gallery_view, name="emote_gallery"),
|
||||
path("games/", views.GamesGridView.as_view(), name="game_list"),
|
||||
path("games/list/", views.GamesListView.as_view(), name="game_list_simple"),
|
||||
path("games/", views.GamesGridView.as_view(), name="games_grid"),
|
||||
path("games/list/", views.GamesListView.as_view(), name="games_list"),
|
||||
path("games/<str:twitch_id>/", views.GameDetailView.as_view(), name="game_detail"),
|
||||
path("organizations/", views.org_list_view, name="org_list"),
|
||||
path("organizations/<str:twitch_id>/", views.organization_detail_view, name="organization_detail"),
|
||||
path("reward-campaigns/", views.reward_campaign_list_view, name="reward_campaign_list"),
|
||||
path("reward-campaigns/<str:twitch_id>/", views.reward_campaign_detail_view, name="reward_campaign_detail"),
|
||||
path("search/", views.search_view, name="search"),
|
||||
path("export/campaigns/csv/", views.export_campaigns_csv, name="export_campaigns_csv"),
|
||||
path("export/campaigns/json/", views.export_campaigns_json, name="export_campaigns_json"),
|
||||
path("export/games/csv/", views.export_games_csv, name="export_games_csv"),
|
||||
path("export/games/json/", views.export_games_json, name="export_games_json"),
|
||||
path("export/organizations/csv/", views.export_organizations_csv, name="export_organizations_csv"),
|
||||
path("export/organizations/json/", views.export_organizations_json, name="export_organizations_json"),
|
||||
path("rss/campaigns/", DropCampaignFeed(), name="campaign_feed"),
|
||||
path("rss/games/", GameFeed(), name="game_feed"),
|
||||
path("rss/games/<str:twitch_id>/campaigns/", GameCampaignFeed(), name="game_campaign_feed"),
|
||||
|
|
|
|||
312
twitch/views.py
312
twitch/views.py
|
|
@ -1,5 +1,6 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import csv
|
||||
import datetime
|
||||
import json
|
||||
import logging
|
||||
|
|
@ -473,7 +474,14 @@ def drop_campaign_detail_view(request: HttpRequest, twitch_id: str) -> HttpRespo
|
|||
Http404: If the campaign is not found.
|
||||
"""
|
||||
try:
|
||||
campaign: DropCampaign = DropCampaign.objects.prefetch_related("game__owners").get(
|
||||
campaign: DropCampaign = DropCampaign.objects.prefetch_related(
|
||||
"game__owners",
|
||||
Prefetch(
|
||||
"allow_channels",
|
||||
queryset=Channel.objects.order_by("display_name"),
|
||||
to_attr="channels_ordered",
|
||||
),
|
||||
).get(
|
||||
twitch_id=twitch_id,
|
||||
)
|
||||
except DropCampaign.DoesNotExist as exc:
|
||||
|
|
@ -591,7 +599,7 @@ def drop_campaign_detail_view(request: HttpRequest, twitch_id: str) -> HttpRespo
|
|||
"drops": enhanced_drops,
|
||||
"campaign_data": format_and_color_json(campaign_data[0]),
|
||||
"owners": list(campaign.game.owners.all()),
|
||||
"allowed_channels": campaign.allow_channels.all().order_by("display_name"),
|
||||
"allowed_channels": getattr(campaign, "channels_ordered", []),
|
||||
}
|
||||
|
||||
return render(request, "twitch/campaign_detail.html", context)
|
||||
|
|
@ -809,7 +817,6 @@ class GameDetailView(DetailView):
|
|||
"start_at",
|
||||
"end_at",
|
||||
"allow_is_enabled",
|
||||
"allow_channels",
|
||||
"game",
|
||||
"operation_names",
|
||||
"added_at",
|
||||
|
|
@ -821,12 +828,15 @@ class GameDetailView(DetailView):
|
|||
)
|
||||
game_data[0]["fields"]["campaigns"] = campaigns_data
|
||||
|
||||
owners: list[Organization] = list(game.owners.all())
|
||||
|
||||
context.update(
|
||||
{
|
||||
"active_campaigns": active_campaigns,
|
||||
"upcoming_campaigns": upcoming_campaigns,
|
||||
"expired_campaigns": expired_campaigns,
|
||||
"owners": list(game.owners.all()),
|
||||
"owner": owners[0] if owners else None,
|
||||
"owners": owners,
|
||||
"drop_awarded_badges": drop_awarded_badges,
|
||||
"now": now,
|
||||
"game_data": format_and_color_json(game_data[0]),
|
||||
|
|
@ -853,7 +863,11 @@ def dashboard(request: HttpRequest) -> HttpResponse:
|
|||
.select_related("game")
|
||||
.prefetch_related("game__owners")
|
||||
.prefetch_related(
|
||||
"allow_channels",
|
||||
Prefetch(
|
||||
"allow_channels",
|
||||
queryset=Channel.objects.order_by("display_name"),
|
||||
to_attr="channels_ordered",
|
||||
),
|
||||
)
|
||||
.order_by("-start_at")
|
||||
)
|
||||
|
|
@ -874,7 +888,10 @@ def dashboard(request: HttpRequest) -> HttpResponse:
|
|||
"campaigns": [],
|
||||
}
|
||||
|
||||
campaigns_by_game[game_id]["campaigns"].append(campaign)
|
||||
campaigns_by_game[game_id]["campaigns"].append({
|
||||
"campaign": campaign,
|
||||
"allowed_channels": getattr(campaign, "channels_ordered", []),
|
||||
})
|
||||
|
||||
# Get active reward campaigns (Quest rewards)
|
||||
active_reward_campaigns: QuerySet[RewardCampaign] = (
|
||||
|
|
@ -1519,3 +1536,286 @@ def badge_set_detail_view(request: HttpRequest, set_id: str) -> HttpResponse:
|
|||
}
|
||||
|
||||
return render(request, "twitch/badge_set_detail.html", context)
|
||||
|
||||
|
||||
# MARK: Export Views
|
||||
def export_campaigns_csv(request: HttpRequest) -> HttpResponse:
|
||||
"""Export drop campaigns to CSV format.
|
||||
|
||||
Args:
|
||||
request: The HTTP request.
|
||||
|
||||
Returns:
|
||||
HttpResponse: CSV file response.
|
||||
"""
|
||||
# Get filters from query parameters
|
||||
game_filter: str | None = request.GET.get("game")
|
||||
status_filter: str | None = request.GET.get("status")
|
||||
|
||||
queryset: QuerySet[DropCampaign] = DropCampaign.objects.all()
|
||||
|
||||
if game_filter:
|
||||
queryset = queryset.filter(game__twitch_id=game_filter)
|
||||
|
||||
queryset = queryset.prefetch_related("game__owners").order_by("-start_at")
|
||||
|
||||
now: datetime.datetime = timezone.now()
|
||||
if status_filter == "active":
|
||||
queryset = queryset.filter(start_at__lte=now, end_at__gte=now)
|
||||
elif status_filter == "upcoming":
|
||||
queryset = queryset.filter(start_at__gt=now)
|
||||
elif status_filter == "expired":
|
||||
queryset = queryset.filter(end_at__lt=now)
|
||||
|
||||
# Create CSV response
|
||||
response = HttpResponse(content_type="text/csv")
|
||||
response["Content-Disposition"] = "attachment; filename=campaigns.csv"
|
||||
|
||||
writer = csv.writer(response)
|
||||
writer.writerow([
|
||||
"Twitch ID",
|
||||
"Name",
|
||||
"Description",
|
||||
"Game",
|
||||
"Status",
|
||||
"Start Date",
|
||||
"End Date",
|
||||
"Details URL",
|
||||
"Created At",
|
||||
"Updated At",
|
||||
])
|
||||
|
||||
for campaign in queryset:
|
||||
# Determine campaign status
|
||||
if campaign.start_at and campaign.end_at:
|
||||
if campaign.start_at <= now <= campaign.end_at:
|
||||
status = "Active"
|
||||
elif campaign.start_at > now:
|
||||
status = "Upcoming"
|
||||
else:
|
||||
status = "Expired"
|
||||
else:
|
||||
status = "Unknown"
|
||||
|
||||
writer.writerow([
|
||||
campaign.twitch_id,
|
||||
campaign.name,
|
||||
campaign.description[:100] if campaign.description else "", # Truncate for CSV
|
||||
campaign.game.name if campaign.game else "",
|
||||
status,
|
||||
campaign.start_at.isoformat() if campaign.start_at else "",
|
||||
campaign.end_at.isoformat() if campaign.end_at else "",
|
||||
campaign.details_url,
|
||||
campaign.added_at.isoformat() if campaign.added_at else "",
|
||||
campaign.updated_at.isoformat() if campaign.updated_at else "",
|
||||
])
|
||||
|
||||
return response
|
||||
|
||||
|
||||
def export_campaigns_json(request: HttpRequest) -> HttpResponse:
|
||||
"""Export drop campaigns to JSON format.
|
||||
|
||||
Args:
|
||||
request: The HTTP request.
|
||||
|
||||
Returns:
|
||||
HttpResponse: JSON file response.
|
||||
"""
|
||||
# Get filters from query parameters
|
||||
game_filter: str | None = request.GET.get("game")
|
||||
status_filter: str | None = request.GET.get("status")
|
||||
|
||||
queryset: QuerySet[DropCampaign] = DropCampaign.objects.all()
|
||||
|
||||
if game_filter:
|
||||
queryset = queryset.filter(game__twitch_id=game_filter)
|
||||
|
||||
queryset = queryset.prefetch_related("game__owners").order_by("-start_at")
|
||||
|
||||
now = timezone.now()
|
||||
if status_filter == "active":
|
||||
queryset = queryset.filter(start_at__lte=now, end_at__gte=now)
|
||||
elif status_filter == "upcoming":
|
||||
queryset = queryset.filter(start_at__gt=now)
|
||||
elif status_filter == "expired":
|
||||
queryset = queryset.filter(end_at__lt=now)
|
||||
|
||||
# Build data list
|
||||
campaigns_data: list[dict[str, Any]] = []
|
||||
for campaign in queryset:
|
||||
# Determine campaign status
|
||||
if campaign.start_at and campaign.end_at:
|
||||
if campaign.start_at <= now <= campaign.end_at:
|
||||
status = "Active"
|
||||
elif campaign.start_at > now:
|
||||
status = "Upcoming"
|
||||
else:
|
||||
status = "Expired"
|
||||
else:
|
||||
status = "Unknown"
|
||||
|
||||
campaigns_data.append({
|
||||
"twitch_id": campaign.twitch_id,
|
||||
"name": campaign.name,
|
||||
"description": campaign.description,
|
||||
"game": campaign.game.name if campaign.game else None,
|
||||
"game_twitch_id": campaign.game.twitch_id if campaign.game else None,
|
||||
"status": status,
|
||||
"start_at": campaign.start_at.isoformat() if campaign.start_at else None,
|
||||
"end_at": campaign.end_at.isoformat() if campaign.end_at else None,
|
||||
"details_url": campaign.details_url,
|
||||
"account_link_url": campaign.account_link_url,
|
||||
"added_at": campaign.added_at.isoformat() if campaign.added_at else None,
|
||||
"updated_at": campaign.updated_at.isoformat() if campaign.updated_at else None,
|
||||
})
|
||||
|
||||
# Create JSON response
|
||||
response = HttpResponse(
|
||||
json.dumps(campaigns_data, indent=2),
|
||||
content_type="application/json",
|
||||
)
|
||||
response["Content-Disposition"] = "attachment; filename=campaigns.json"
|
||||
|
||||
return response
|
||||
|
||||
|
||||
def export_games_csv(request: HttpRequest) -> HttpResponse: # noqa: ARG001 # noqa: ARG001
|
||||
"""Export games to CSV format.
|
||||
|
||||
Args:
|
||||
request: The HTTP request.
|
||||
|
||||
Returns:
|
||||
HttpResponse: CSV file response.
|
||||
"""
|
||||
queryset: QuerySet[Game] = Game.objects.all().order_by("display_name")
|
||||
|
||||
# Create CSV response
|
||||
response = HttpResponse(content_type="text/csv")
|
||||
response["Content-Disposition"] = "attachment; filename=games.csv"
|
||||
|
||||
writer = csv.writer(response)
|
||||
writer.writerow([
|
||||
"Twitch ID",
|
||||
"Name",
|
||||
"Display Name",
|
||||
"Slug",
|
||||
"Box Art URL",
|
||||
"Added At",
|
||||
"Updated At",
|
||||
])
|
||||
|
||||
for game in queryset:
|
||||
writer.writerow([
|
||||
game.twitch_id,
|
||||
game.name,
|
||||
game.display_name,
|
||||
game.slug,
|
||||
game.box_art,
|
||||
game.added_at.isoformat() if game.added_at else "",
|
||||
game.updated_at.isoformat() if game.updated_at else "",
|
||||
])
|
||||
|
||||
return response
|
||||
|
||||
|
||||
def export_games_json(request: HttpRequest) -> HttpResponse: # noqa: ARG001 # noqa: ARG001
|
||||
"""Export games to JSON format.
|
||||
|
||||
Args:
|
||||
request: The HTTP request.
|
||||
|
||||
Returns:
|
||||
HttpResponse: JSON file response.
|
||||
"""
|
||||
queryset: QuerySet[Game] = Game.objects.all().order_by("display_name")
|
||||
|
||||
# Build data list
|
||||
games_data: list[dict[str, Any]] = [
|
||||
{
|
||||
"twitch_id": game.twitch_id,
|
||||
"name": game.name,
|
||||
"display_name": game.display_name,
|
||||
"slug": game.slug,
|
||||
"box_art_url": game.box_art,
|
||||
"added_at": game.added_at.isoformat() if game.added_at else None,
|
||||
"updated_at": game.updated_at.isoformat() if game.updated_at else None,
|
||||
}
|
||||
for game in queryset
|
||||
]
|
||||
|
||||
# Create JSON response
|
||||
response = HttpResponse(
|
||||
json.dumps(games_data, indent=2),
|
||||
content_type="application/json",
|
||||
)
|
||||
response["Content-Disposition"] = "attachment; filename=games.json"
|
||||
|
||||
return response
|
||||
|
||||
|
||||
def export_organizations_csv(request: HttpRequest) -> HttpResponse: # noqa: ARG001
|
||||
"""Export organizations to CSV format.
|
||||
|
||||
Args:
|
||||
request: The HTTP request.
|
||||
|
||||
Returns:
|
||||
HttpResponse: CSV file response.
|
||||
"""
|
||||
queryset: QuerySet[Organization] = Organization.objects.all().order_by("name")
|
||||
|
||||
# Create CSV response
|
||||
response = HttpResponse(content_type="text/csv")
|
||||
response["Content-Disposition"] = "attachment; filename=organizations.csv"
|
||||
|
||||
writer = csv.writer(response)
|
||||
writer.writerow([
|
||||
"Twitch ID",
|
||||
"Name",
|
||||
"Added At",
|
||||
"Updated At",
|
||||
])
|
||||
|
||||
for org in queryset:
|
||||
writer.writerow([
|
||||
org.twitch_id,
|
||||
org.name,
|
||||
org.added_at.isoformat() if org.added_at else "",
|
||||
org.updated_at.isoformat() if org.updated_at else "",
|
||||
])
|
||||
|
||||
return response
|
||||
|
||||
|
||||
def export_organizations_json(request: HttpRequest) -> HttpResponse: # noqa: ARG001
|
||||
"""Export organizations to JSON format.
|
||||
|
||||
Args:
|
||||
request: The HTTP request.
|
||||
|
||||
Returns:
|
||||
HttpResponse: JSON file response.
|
||||
"""
|
||||
queryset: QuerySet[Organization] = Organization.objects.all().order_by("name")
|
||||
|
||||
# Build data list
|
||||
orgs_data: list[dict[str, Any]] = [
|
||||
{
|
||||
"twitch_id": org.twitch_id,
|
||||
"name": org.name,
|
||||
"added_at": org.added_at.isoformat() if org.added_at else None,
|
||||
"updated_at": org.updated_at.isoformat() if org.updated_at else None,
|
||||
}
|
||||
for org in queryset
|
||||
]
|
||||
|
||||
# Create JSON response
|
||||
response = HttpResponse(
|
||||
json.dumps(orgs_data, indent=2),
|
||||
content_type="application/json",
|
||||
)
|
||||
response["Content-Disposition"] = "attachment; filename=organizations.json"
|
||||
|
||||
return response
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue