TheLovinator/twitch-drop-notifier
1
Fork 0
Code Issues Actions Packages Releases Activity

Separate views into their own file

Browse Source
This commit is contained in:
Joakim Hellsén
2024-07-09 15:11:27 +02:00
parent e0dbfb1c46
commit ec75ce1ccb
6 changed files with 246 additions and 166 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
core/data.py Normal file
View File

@ -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

10
core/urls.py
View File

@ -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"),
]

0
core/views/__init__.py Normal file
View File

18
core/views/games.py Normal file
View File

@ -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

210
core/views.py → core/views/index.py
View File

@ -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))

120
core/views/webhooks.py Normal file
View File

@ -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))