diff --git a/core/data.py b/core/data.py new file mode 100644 index 0000000..4aafb47 --- /dev/null +++ b/core/data.py @@ -0,0 +1,54 @@ +import datetime +from dataclasses import dataclass + + +@dataclass +class WebhookData: + """The webhook data.""" + + name: str | None = None + url: str | None = None + avatar: str | None = None + status: str | None = None + response: str | None = None + + +@dataclass +class DropContext: + """The drop.""" + + drops_id: str | None = None + image_url: str | None = None + name: str | None = None + limit: int | None = None + required_minutes_watched: int | None = None + required_subs: int | None = None + + +@dataclass +class CampaignContext: + """Drops are grouped into campaigns.""" + + drop_id: str | None = None + name: str | None = None + image_url: str | None = None + status: str | None = None + account_link_url: str | None = None + description: str | None = None + details_url: str | None = None + ios_available: bool | None = None + start_at: datetime.datetime | None = None + end_at: datetime.datetime | None = None + drops: list[DropContext] | None = None + + +@dataclass +class GameContext: + """Campaigns are under a game.""" + + game_id: str | None = None + campaigns: list[CampaignContext] | None = None + image_url: str | None = None + display_name: str | None = None + twitch_url: str | None = None + slug: str | None = None diff --git a/core/urls.py b/core/urls.py index a6f1be0..5e7eb69 100644 --- a/core/urls.py +++ b/core/urls.py @@ -2,16 +2,18 @@ from __future__ import annotations from django.urls import URLPattern, URLResolver, path -from . import views +from .views.games import GameView +from .views.index import index +from .views.webhooks import WebhooksView app_name: str = "core" urlpatterns: list[URLPattern | URLResolver] = [ - path(route="", view=views.index, name="index"), + path(route="", view=index, name="index"), path( route="games/", - view=views.GameView.as_view(), + view=GameView.as_view(), name="games", ), - path("webhooks/", views.WebhooksView.as_view(), name="webhooks"), + path("webhooks/", WebhooksView.as_view(), name="webhooks"), ] diff --git a/core/views/__init__.py b/core/views/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/core/views/games.py b/core/views/games.py new file mode 100644 index 0000000..a4d9a00 --- /dev/null +++ b/core/views/games.py @@ -0,0 +1,18 @@ +from __future__ import annotations + +import logging + +from django.views.generic import ListView + +from twitch_app.models import ( + Game, +) + +logger: logging.Logger = logging.getLogger(__name__) + + +class GameView(ListView): + model = Game + template_name: str = "games.html" + context_object_name: str = "games" + paginate_by = 100 diff --git a/core/views.py b/core/views/index.py similarity index 57% rename from core/views.py rename to core/views/index.py index 32733f6..8bc9b57 100644 --- a/core/views.py +++ b/core/views/index.py @@ -2,16 +2,14 @@ from __future__ import annotations import datetime import logging -from dataclasses import dataclass -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING import hishel from django.conf import settings -from django.contrib import messages -from django.http.response import HttpResponse +from django.http import HttpRequest, HttpResponse from django.template.response import TemplateResponse -from django.views.generic import FormView, ListView +from core.data import CampaignContext, DropContext, GameContext, WebhookData from twitch_app.models import ( DropBenefit, DropCampaign, @@ -19,8 +17,6 @@ from twitch_app.models import ( TimeBasedDrop, ) -from .forms import DiscordSettingForm - if TYPE_CHECKING: from pathlib import Path @@ -29,6 +25,12 @@ if TYPE_CHECKING: HttpResponse, ) from httpx import Response +if TYPE_CHECKING: + from django.http import ( + HttpRequest, + HttpResponse, + ) + from httpx import Response logger: logging.Logger = logging.getLogger(__name__) @@ -42,80 +44,6 @@ controller = hishel.Controller( ) -@dataclass -class DropContext: - """The drop.""" - - drops_id: str | None = None - image_url: str | None = None - name: str | None = None - limit: int | None = None - required_minutes_watched: int | None = None - required_subs: int | None = None - - -@dataclass -class CampaignContext: - """Drops are grouped into campaigns.""" - - drop_id: str | None = None - name: str | None = None - image_url: str | None = None - status: str | None = None - account_link_url: str | None = None - description: str | None = None - details_url: str | None = None - ios_available: bool | None = None - start_at: datetime.datetime | None = None - end_at: datetime.datetime | None = None - drops: list[DropContext] | None = None - - -@dataclass -class GameContext: - """Campaigns are under a game.""" - - game_id: str | None = None - campaigns: list[CampaignContext] | None = None - image_url: str | None = None - display_name: str | None = None - twitch_url: str | None = None - slug: str | None = None - - -def get_avatar(webhook_response: Response) -> str: - """Get the avatar URL from the webhook response.""" - avatar: str = "https://cdn.discordapp.com/embed/avatars/0.png" - if webhook_response.is_success and webhook_response.json().get("id") and webhook_response.json().get("avatar"): - avatar = f'https://cdn.discordapp.com/avatars/{webhook_response.json().get("id")}/{webhook_response.json().get("avatar")}.png' - return avatar - - -def get_webhook_data(webhook: str) -> WebhookData: - """Get the webhook data.""" - with hishel.CacheClient(storage=storage, controller=controller) as client: - webhook_response: Response = client.get(url=webhook, extensions={"cache_metadata": True}) - - return WebhookData( - name=webhook_response.json().get("name") if webhook_response.is_success else "Unknown", - url=webhook, - avatar=get_avatar(webhook_response), - status="Success" if webhook_response.is_success else "Failed", - response=webhook_response.text, - ) - - -def get_webhooks(request: HttpRequest) -> list[str]: - """Get the webhooks from the cookie.""" - cookie: str = request.COOKIES.get("webhooks", "") - return list(filter(None, cookie.split(","))) - - -def fetch_games() -> list[Game]: - """Fetch all games with necessary fields.""" - return list(Game.objects.all().only("id", "image_url", "display_name", "slug")) - - def fetch_campaigns(game: Game) -> list[CampaignContext]: """Fetch active campaigns for a given game.""" campaigns: list[CampaignContext] = [] @@ -184,10 +112,28 @@ def fetch_drops(campaign: DropCampaign) -> list[DropContext]: return drops +def sort_games_by_campaign_start(list_of_games: list[GameContext]) -> list[GameContext]: + """Sort games by the start date of the first campaign and reverse the list so the latest games are first.""" + if list_of_games and list_of_games[0].campaigns: + list_of_games.sort( + key=lambda x: x.campaigns[0].start_at + if x.campaigns and x.campaigns[0].start_at is not None + else datetime.datetime.min, + ) + list_of_games.reverse() + return list_of_games + + +def get_webhooks(request: HttpRequest) -> list[str]: + """Get the webhooks from the cookie.""" + cookie: str = request.COOKIES.get("webhooks", "") + return list(filter(None, cookie.split(","))) + + def prepare_game_contexts() -> list[GameContext]: """Prepare game contexts with their respective campaigns and drops.""" list_of_games: list[GameContext] = [] - for game in fetch_games(): + for game in list(Game.objects.all().only("id", "image_url", "display_name", "slug")): campaigns: list[CampaignContext] = fetch_campaigns(game) if not campaigns: continue @@ -203,16 +149,26 @@ def prepare_game_contexts() -> list[GameContext]: return list_of_games -def sort_games_by_campaign_start(list_of_games: list[GameContext]) -> list[GameContext]: - """Sort games by the start date of the first campaign and reverse the list so the latest games are first.""" - if list_of_games and list_of_games[0].campaigns: - list_of_games.sort( - key=lambda x: x.campaigns[0].start_at - if x.campaigns and x.campaigns[0].start_at is not None - else datetime.datetime.min, - ) - list_of_games.reverse() - return list_of_games +def get_avatar(webhook_response: Response) -> str: + """Get the avatar URL from the webhook response.""" + avatar: str = "https://cdn.discordapp.com/embed/avatars/0.png" + if webhook_response.is_success and webhook_response.json().get("id") and webhook_response.json().get("avatar"): + avatar = f'https://cdn.discordapp.com/avatars/{webhook_response.json().get("id")}/{webhook_response.json().get("avatar")}.png' + return avatar + + +def get_webhook_data(webhook: str) -> WebhookData: + """Get the webhook data.""" + with hishel.CacheClient(storage=storage, controller=controller) as client: + webhook_response: Response = client.get(url=webhook, extensions={"cache_metadata": True}) + + return WebhookData( + name=webhook_response.json().get("name") if webhook_response.is_success else "Unknown", + url=webhook, + avatar=get_avatar(webhook_response), + status="Success" if webhook_response.is_success else "Failed", + response=webhook_response.text, + ) def index(request: HttpRequest) -> HttpResponse: @@ -226,73 +182,3 @@ def index(request: HttpRequest) -> HttpResponse: template="index.html", context={"games": sorted_list_of_games, "webhooks": webhooks}, ) - - -class GameView(ListView): - model = Game - template_name: str = "games.html" - context_object_name: str = "games" - paginate_by = 100 - - -@dataclass -class WebhookData: - """The webhook data.""" - - name: str | None = None - url: str | None = None - avatar: str | None = None - status: str | None = None - response: str | None = None - - -class WebhooksView(FormView): - model = Game - template_name = "webhooks.html" - form_class = DiscordSettingForm - context_object_name: str = "webhooks" - paginate_by = 100 - - def get_context_data(self: WebhooksView, **kwargs: dict[str, WebhooksView] | DiscordSettingForm) -> dict[str, Any]: - """Get the context data for the view.""" - context: dict[str, DiscordSettingForm | list[WebhookData]] = super().get_context_data(**kwargs) - webhooks: list[str] = get_webhooks(self.request) - - context.update({ - "webhooks": [get_webhook_data(webhook) for webhook in webhooks], - "form": DiscordSettingForm(), - }) - return context - - def form_valid(self: WebhooksView, form: DiscordSettingForm) -> HttpResponse: - """Handle valid form submission.""" - webhook = str(form.cleaned_data["webhook_url"]) - - with hishel.CacheClient(storage=storage, controller=controller) as client: - webhook_response: Response = client.get(url=webhook, extensions={"cache_metadata": True}) - if not webhook_response.is_success: - messages.error(self.request, "Failed to get webhook information. Is the URL correct?") - return self.render_to_response(self.get_context_data(form=form)) - - webhook_name: str | None = str(webhook_response.json().get("name")) if webhook_response.is_success else None - - cookie: str = self.request.COOKIES.get("webhooks", "") - webhooks: list[str] = cookie.split(",") - webhooks = list(filter(None, webhooks)) - if webhook in webhooks: - if webhook_name: - messages.error(self.request, f"Webhook {webhook_name} already exists.") - else: - messages.error(self.request, "Webhook already exists.") - return self.render_to_response(self.get_context_data(form=form)) - - webhooks.append(webhook) - response: HttpResponse = self.render_to_response(self.get_context_data(form=form)) - response.set_cookie(key="webhooks", value=",".join(webhooks), max_age=315360000) # 10 years - - messages.success(self.request, "Webhook successfully added.") - return response - - def form_invalid(self: WebhooksView, form: DiscordSettingForm) -> HttpResponse: - messages.error(self.request, "Failed to add webhook.") - return self.render_to_response(self.get_context_data(form=form)) diff --git a/core/views/webhooks.py b/core/views/webhooks.py new file mode 100644 index 0000000..075d907 --- /dev/null +++ b/core/views/webhooks.py @@ -0,0 +1,120 @@ +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING, Any + +import hishel +from django.conf import settings +from django.contrib import messages +from django.http.response import HttpResponse +from django.views.generic import FormView + +from core.data import WebhookData +from core.forms import DiscordSettingForm +from twitch_app.models import ( + Game, +) + +if TYPE_CHECKING: + from pathlib import Path + + from django.http import HttpRequest + + +cache_dir: Path = settings.DATA_DIR / "cache" +cache_dir.mkdir(exist_ok=True, parents=True) +storage = hishel.FileStorage(base_path=cache_dir) +controller = hishel.Controller( + cacheable_status_codes=[200, 203, 204, 206, 300, 301, 308, 404, 405, 410, 414, 501], + allow_stale=True, + always_revalidate=True, +) + + +if TYPE_CHECKING: + from django.http import ( + HttpResponse, + ) + from httpx import Response + +logger: logging.Logger = logging.getLogger(__name__) + + +def get_webhooks(request: HttpRequest) -> list[str]: + """Get the webhooks from the cookie.""" + cookie: str = request.COOKIES.get("webhooks", "") + return list(filter(None, cookie.split(","))) + + +def get_avatar(webhook_response: Response) -> str: + """Get the avatar URL from the webhook response.""" + avatar: str = "https://cdn.discordapp.com/embed/avatars/0.png" + if webhook_response.is_success and webhook_response.json().get("id") and webhook_response.json().get("avatar"): + avatar = f'https://cdn.discordapp.com/avatars/{webhook_response.json().get("id")}/{webhook_response.json().get("avatar")}.png' + return avatar + + +def get_webhook_data(webhook: str) -> WebhookData: + """Get the webhook data.""" + with hishel.CacheClient(storage=storage, controller=controller) as client: + webhook_response: Response = client.get(url=webhook, extensions={"cache_metadata": True}) + + return WebhookData( + name=webhook_response.json().get("name") if webhook_response.is_success else "Unknown", + url=webhook, + avatar=get_avatar(webhook_response), + status="Success" if webhook_response.is_success else "Failed", + response=webhook_response.text, + ) + + +class WebhooksView(FormView): + model = Game + template_name = "webhooks.html" + form_class = DiscordSettingForm + context_object_name: str = "webhooks" + paginate_by = 100 + + def get_context_data(self: WebhooksView, **kwargs: dict[str, WebhooksView] | DiscordSettingForm) -> dict[str, Any]: + """Get the context data for the view.""" + context: dict[str, DiscordSettingForm | list[WebhookData]] = super().get_context_data(**kwargs) + webhooks: list[str] = get_webhooks(self.request) + + context.update({ + "webhooks": [get_webhook_data(webhook) for webhook in webhooks], + "form": DiscordSettingForm(), + }) + return context + + def form_valid(self: WebhooksView, form: DiscordSettingForm) -> HttpResponse: + """Handle valid form submission.""" + webhook = str(form.cleaned_data["webhook_url"]) + + with hishel.CacheClient(storage=storage, controller=controller) as client: + webhook_response: Response = client.get(url=webhook, extensions={"cache_metadata": True}) + if not webhook_response.is_success: + messages.error(self.request, "Failed to get webhook information. Is the URL correct?") + return self.render_to_response(self.get_context_data(form=form)) + + webhook_name: str | None = str(webhook_response.json().get("name")) if webhook_response.is_success else None + + cookie: str = self.request.COOKIES.get("webhooks", "") + webhooks: list[str] = cookie.split(",") + webhooks = list(filter(None, webhooks)) + if webhook in webhooks: + if webhook_name: + messages.error(self.request, f"Webhook {webhook_name} already exists.") + else: + messages.error(self.request, "Webhook already exists.") + return self.render_to_response(self.get_context_data(form=form)) + + webhooks.append(webhook) + response: HttpResponse = self.render_to_response(self.get_context_data(form=form)) + response.set_cookie(key="webhooks", value=",".join(webhooks), max_age=315360000) # 10 years + + messages.success(self.request, "Webhook successfully added.") + return response + + def form_invalid(self: WebhooksView, form: DiscordSettingForm) -> HttpResponse: + messages.error(self.request, "Failed to add webhook.") + return self.render_to_response(self.get_context_data(form=form))