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

Go back to cookies :(

Browse Source
This commit is contained in:
Joakim Hellsén
2024-08-16 06:37:54 +02:00
parent 324f255a8e
commit 3ff3fe157a
14 changed files with 478 additions and 563 deletions
Download Patch File Download Diff File Expand all files Collapse all files

76
core/discord.py
View File

@ -1,42 +1,10 @@
import logging
from typing import TYPE_CHECKING
from discord_webhook import DiscordWebhook
from django.conf import settings
from django.utils import timezone
from core.models import DropCampaign, Game, GameSubscription, Owner, OwnerSubscription
if TYPE_CHECKING:
from requests import Response
logger: logging.Logger = logging.getLogger(__name__)
def send(message: str, webhook_url: str | None = None) -> None:
"""Send a message to Discord.
Args:
message: The message to send.
webhook_url: The webhook URL to send the message to.
"""
logger.info("Discord message: %s", message)
webhook_url = webhook_url or str(settings.DISCORD_WEBHOOK_URL)
if not webhook_url:
logger.error("No webhook URL provided.")
return
webhook = DiscordWebhook(
url=webhook_url,
content=message,
username="TTVDrops",
rate_limit_retry=True,
)
response: Response = webhook.execute()
logger.debug(response)
def convert_time_to_discord_timestamp(time: timezone.datetime | None) -> str:
"""Discord uses <t:UNIX_TIMESTAMP:R> for timestamps.
@ -47,47 +15,3 @@ def convert_time_to_discord_timestamp(time: timezone.datetime | None) -> str:
str: The Discord timestamp string. If time is None, returns "Unknown".
"""
return f"<t:{int(time.timestamp())}:R>" if time else "Unknown"
def generate_game_message(instance: DropCampaign, game: Game, sub: GameSubscription) -> str:
"""Generate a message for a drop campaign.
Args:
instance (DropCampaign): Drop campaign instance.
game (Game): Game instance.
sub (GameSubscription): Game subscription instance.
Returns:
str: The message to send to Discord.
"""
game_name: str = game.name or "Unknown"
description: str = instance.description or "No description provided."
start_at: str = convert_time_to_discord_timestamp(instance.starts_at)
end_at: str = convert_time_to_discord_timestamp(instance.ends_at)
msg: str = f"{game_name}: {instance.name}\n{description}\nStarts: {start_at}\nEnds: {end_at}"
logger.info("Discord message: '%s' to '%s'", msg, sub.webhook.url)
return msg
def generate_owner_message(instance: DropCampaign, owner: Owner, sub: OwnerSubscription) -> str:
"""Generate a message for a drop campaign.
Args:
instance (DropCampaign): Drop campaign instance.
owner (Owner): Owner instance.
sub (OwnerSubscription): Owner subscription instance.
Returns:
str: The message to send to Discord.
"""
owner_name: str = owner.name or "Unknown"
description: str = instance.description or "No description provided."
start_at: str = convert_time_to_discord_timestamp(instance.starts_at)
end_at: str = convert_time_to_discord_timestamp(instance.ends_at)
msg: str = f"{owner_name}: {instance.name}\n{description}\nStarts: {start_at}\nEnds: {end_at}"
logger.info("Discord message: '%s' to '%s'", msg, sub.webhook.url)
return msg

138
core/management/commands/scrape_twitch.py
View File

@ -68,78 +68,99 @@ def save_json(campaign: dict | None, dir_name: str) -> None:
json.dump(campaign, f, indent=4)
async def add_reward_campaign(campaign: dict | None) -> None: # noqa: C901
async def add_reward_campaign(campaign: dict | None) -> None:
"""Add a reward campaign to the database.
Args:
campaign (dict): The reward campaign to add.
"""
# sourcery skip: low-code-quality
if not campaign:
return
logger.info("Adding reward campaign to database")
logger.info("Adding reward campaign to database %s", campaign["id"])
if "data" in campaign and "rewardCampaignsAvailableToUser" in campaign["data"]:
mappings: dict[str, str] = {
"brand": "brand",
"createdAt": "created_at",
"startsAt": "starts_at",
"endsAt": "ends_at",
"status": "status",
"summary": "summary",
"instructions": "instructions",
"rewardValueURLParam": "reward_value_url_param",
"externalURL": "external_url",
"aboutURL": "about_url",
"isSitewide": "is_site_wide",
}
for reward_campaign in campaign["data"]["rewardCampaignsAvailableToUser"]:
defaults = {new_key: reward_campaign[key] for key, new_key in mappings.items() if reward_campaign.get(key)}
our_reward_campaign = await handle_reward_campaign(reward_campaign)
if reward_campaign.get("unlockRequirements", {}).get("subsGoal"):
defaults["sub_goal"] = reward_campaign["unlockRequirements"]["subsGoal"]
if "rewards" in reward_campaign:
for reward in reward_campaign["rewards"]:
await handle_rewards(reward, our_reward_campaign)
if reward_campaign.get("unlockRequirements", {}).get("minuteWatchedGoal"):
defaults["minute_watched_goal"] = reward_campaign["unlockRequirements"]["minuteWatchedGoal"]
if reward_campaign.get("image"):
defaults["image_url"] = reward_campaign["image"]["image1xURL"]
async def handle_rewards(reward: dict, reward_campaign: RewardCampaign | None) -> None:
"""Add or update a reward in the database.
our_reward_campaign, created = await RewardCampaign.objects.aupdate_or_create(
id=reward_campaign["id"],
defaults=defaults,
)
if created:
logger.info("Added reward campaign %s", our_reward_campaign.id)
Args:
reward (dict): The JSON from Twitch.
reward_campaign (RewardCampaign | None): The reward campaign the reward belongs to.
"""
mappings: dict[str, str] = {
"name": "name",
"earnableUntil": "earnable_until",
"redemptionInstructions": "redemption_instructions",
"redemptionURL": "redemption_url",
}
if reward_campaign["game"]:
# TODO(TheLovinator): Add game to reward campaign # noqa: TD003
# TODO(TheLovinator): Send JSON to Discord # noqa: TD003
logger.error("Not implemented: Add game to reward campaign, JSON: %s", reward_campaign["game"])
defaults: dict = {new_key: reward[key] for key, new_key in mappings.items() if reward.get(key)}
if reward_campaign:
defaults["campaign"] = reward_campaign
# Add rewards
mappings: dict[str, str] = {
"name": "name",
"bannerImage": "banner_image_url",
"thumbnailImage": "thumbnail_image_url",
"earnableUntil": "earnable_until",
"redemptionInstructions": "redemption_instructions",
"redemptionURL": "redemption_url",
}
for reward in reward_campaign["rewards"]:
defaults = {new_key: reward[key] for key, new_key in mappings.items() if reward.get(key)}
if our_reward_campaign:
defaults["campaign"] = our_reward_campaign
our_reward, created = await Reward.objects.aupdate_or_create(
id=reward["id"],
defaults=defaults,
)
if reward.get("bannerImage"):
defaults["banner_image_url"] = reward["bannerImage"]["image1xURL"]
if created:
logger.info("Added reward %s", our_reward.id)
else:
logger.info("Updated reward %s", our_reward.id)
if reward.get("thumbnailImage"):
defaults["thumbnail_image_url"] = reward["thumbnailImage"]["image1xURL"]
reward_instance, created = await Reward.objects.aupdate_or_create(id=reward["id"], defaults=defaults)
if created:
logger.info("Added reward %s", reward_instance.id)
async def handle_reward_campaign(reward_campaign: dict) -> RewardCampaign:
"""Add or update a reward campaign in the database.
Args:
reward_campaign (dict): The reward campaign JSON from Twitch.
Returns:
RewardCampaign: The reward campaign that was added or updated.
"""
mappings: dict[str, str] = {
"brand": "brand",
"createdAt": "created_at",
"startsAt": "starts_at",
"endsAt": "ends_at",
"status": "status",
"summary": "summary",
"instructions": "instructions",
"rewardValueURLParam": "reward_value_url_param",
"externalURL": "external_url",
"aboutURL": "about_url",
"isSitewide": "is_site_wide",
}
defaults: dict = {new_key: reward_campaign[key] for key, new_key in mappings.items() if reward_campaign.get(key)}
unlock_requirements: dict = reward_campaign.get("unlockRequirements", {})
if unlock_requirements.get("subsGoal"):
defaults["sub_goal"] = unlock_requirements["subsGoal"]
if unlock_requirements.get("minuteWatchedGoal"):
defaults["minute_watched_goal"] = unlock_requirements["minuteWatchedGoal"]
if reward_campaign.get("image"):
defaults["image_url"] = reward_campaign["image"]["image1xURL"]
reward_campaign_instance, created = await RewardCampaign.objects.aupdate_or_create(
id=reward_campaign["id"],
defaults=defaults,
)
if created:
logger.info("Added reward campaign %s", reward_campaign_instance.id)
if reward_campaign["game"]:
# TODO(TheLovinator): Add game to reward campaign # noqa: TD003
# TODO(TheLovinator): Send JSON to Discord # noqa: TD003
logger.error("Not implemented: Add game to reward campaign, JSON: %s", reward_campaign["game"])
return reward_campaign_instance
async def add_or_update_game(game_json: dict | None) -> Game | None:
@ -347,7 +368,7 @@ async def process_json_data(num: int, campaign: dict | None) -> None:
return
if not isinstance(campaign, dict):
logger.warning("Campaign is not a dictionary")
logger.warning("Campaign is not a dictionary. %s", campaign)
return
# This is a Reward Campaign
@ -356,12 +377,7 @@ async def process_json_data(num: int, campaign: dict | None) -> None:
await add_reward_campaign(campaign)
if "dropCampaign" in campaign.get("data", {}).get("user", {}):
if not campaign["data"]["user"]["dropCampaign"]:
logger.warning("No drop campaign found")
return
save_json(campaign, "drop_campaign")
if campaign.get("data", {}).get("user", {}).get("dropCampaign"):
await add_drop_campaign(campaign["data"]["user"]["dropCampaign"])

95
core/migrations/0001_initial.py
View File

@ -1,10 +1,9 @@
# Generated by Django 5.1 on 2024-08-15 03:42
# Generated by Django 5.1 on 2024-08-16 02:38
import django.contrib.auth.models
import django.contrib.auth.validators
import django.db.models.deletion
import django.utils.timezone
from django.conf import settings
from django.db import migrations, models
from django.db.migrations.operations.base import Operation
@ -98,8 +97,6 @@ class Migration(migrations.Migration):
),
),
("date_joined", models.DateTimeField(default=django.utils.timezone.now, verbose_name="date joined")),
("subscribe_to_news", models.BooleanField(default=False, help_text="Subscribe to news")),
("subscribe_to_new_games", models.BooleanField(default=False, help_text="Subscribe to new games")),
(
"groups",
models.ManyToManyField(
@ -132,22 +129,6 @@ class Migration(migrations.Migration):
("objects", django.contrib.auth.models.UserManager()),
],
),
migrations.CreateModel(
name="DiscordWebhook",
fields=[
("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")),
("url", models.URLField()),
("name", models.CharField(max_length=255)),
(
"user",
models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE,
related_name="discord_webhooks",
to=settings.AUTH_USER_MODEL,
),
),
],
),
migrations.CreateModel(
name="Channel",
fields=[
@ -169,28 +150,6 @@ class Migration(migrations.Migration):
to="core.game",
),
),
migrations.CreateModel(
name="GameSubscription",
fields=[
("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")),
("game", models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to="core.game")),
("user", models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL)),
("webhook", models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to="core.discordwebhook")),
],
options={
"unique_together": {("user", "game")},
},
),
migrations.AddField(
model_name="user",
name="subscribed_games",
field=models.ManyToManyField(
blank=True,
related_name="subscribed_users",
through="core.GameSubscription",
to="core.game",
),
),
migrations.AddField(
model_name="game",
name="org",
@ -201,28 +160,6 @@ class Migration(migrations.Migration):
to="core.owner",
),
),
migrations.CreateModel(
name="OwnerSubscription",
fields=[
("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")),
("owner", models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to="core.owner")),
("user", models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL)),
("webhook", models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to="core.discordwebhook")),
],
options={
"unique_together": {("user", "owner")},
},
),
migrations.AddField(
model_name="user",
name="subscribed_owners",
field=models.ManyToManyField(
blank=True,
related_name="subscribed_users",
through="core.OwnerSubscription",
to="core.owner",
),
),
migrations.CreateModel(
name="RewardCampaign",
fields=[
@ -319,4 +256,34 @@ class Migration(migrations.Migration):
),
],
),
migrations.CreateModel(
name="Webhook",
fields=[
("avatar", models.TextField(null=True)),
("channel_id", models.TextField(null=True)),
("guild_id", models.TextField(null=True)),
("id", models.TextField(primary_key=True, serialize=False)),
("name", models.TextField(null=True)),
("type", models.TextField(null=True)),
("token", models.TextField()),
("url", models.TextField()),
("seen_drops", models.ManyToManyField(related_name="seen_drops", to="core.dropcampaign")),
(
"subscribed_live_games",
models.ManyToManyField(related_name="subscribed_live_games", to="core.game"),
),
(
"subscribed_live_owners",
models.ManyToManyField(related_name="subscribed_live_owners", to="core.owner"),
),
("subscribed_new_games", models.ManyToManyField(related_name="subscribed_new_games", to="core.game")),
(
"subscribed_new_owners",
models.ManyToManyField(related_name="subscribed_new_owners", to="core.owner"),
),
],
options={
"unique_together": {("id", "token")},
},
),
]

79
core/models.py
View File

@ -1,6 +1,7 @@
from __future__ import annotations
import logging
from typing import ClassVar
from django.contrib.auth.models import AbstractUser
from django.db import models
@ -8,6 +9,9 @@ from django.db import models
logger: logging.Logger = logging.getLogger(__name__)
class User(AbstractUser): ...
class Owner(models.Model):
"""The company or person that owns the game.
@ -182,62 +186,41 @@ class Reward(models.Model):
return self.name or "Reward name unknown"
class User(AbstractUser):
"""Extended User model to include subscriptions."""
class Webhook(models.Model):
"""Discord webhook."""
subscribed_games = models.ManyToManyField(
"Game",
through="GameSubscription",
related_name="subscribed_users",
blank=True,
)
subscribed_owners = models.ManyToManyField(
"Owner",
through="OwnerSubscription",
related_name="subscribed_users",
blank=True,
)
subscribe_to_news = models.BooleanField(default=False, help_text="Subscribe to news")
subscribe_to_new_games = models.BooleanField(default=False, help_text="Subscribe to new games")
avatar = models.TextField(null=True)
channel_id = models.TextField(null=True)
guild_id = models.TextField(null=True)
id = models.TextField(primary_key=True)
name = models.TextField(null=True)
type = models.TextField(null=True)
token = models.TextField()
url = models.TextField()
def __str__(self) -> str:
return self.username
# Get notified when the site finds a new game.
subscribed_new_games = models.ManyToManyField(Game, related_name="subscribed_new_games")
# Get notified when a drop goes live.
subscribed_live_games = models.ManyToManyField(Game, related_name="subscribed_live_games")
class DiscordWebhook(models.Model):
"""A Discord webhook for sending notifications."""
# Get notified when the site finds a new drop campaign for a specific organization.
subscribed_new_owners = models.ManyToManyField(Owner, related_name="subscribed_new_owners")
user = models.ForeignKey(User, on_delete=models.CASCADE, related_name="discord_webhooks")
url = models.URLField()
name = models.CharField(max_length=255)
# Get notified when a drop goes live for a specific organization.
subscribed_live_owners = models.ManyToManyField(Owner, related_name="subscribed_live_owners")
def __str__(self) -> str:
return f"{self.name} ({self.user.username})"
class GameSubscription(models.Model):
"""A subscription to a specific game with a chosen webhook."""
user = models.ForeignKey(User, on_delete=models.CASCADE)
game = models.ForeignKey(Game, on_delete=models.CASCADE)
webhook = models.ForeignKey(DiscordWebhook, on_delete=models.CASCADE)
# So we don't spam the same drop campaign over and over.
seen_drops = models.ManyToManyField(DropCampaign, related_name="seen_drops")
class Meta:
unique_together = ("user", "game")
unique_together: ClassVar[list[str]] = ["id", "token"]
def __str__(self) -> str:
return f"{self.user.username} -> {self.game.name} via {self.webhook.name}"
return f"{self.name} - {self.get_webhook_url()}"
class OwnerSubscription(models.Model):
"""A subscription to a specific owner with a chosen webhook."""
user = models.ForeignKey(User, on_delete=models.CASCADE)
owner = models.ForeignKey(Owner, on_delete=models.CASCADE)
webhook = models.ForeignKey(DiscordWebhook, on_delete=models.CASCADE)
class Meta:
unique_together = ("user", "owner")
def __str__(self) -> str:
return f"{self.user.username} -> {self.owner.name} via {self.webhook.name}"
def get_webhook_url(self) -> str:
try:
return f"https://discord.com/api/webhooks/{self.id}/{self.token}"
except AttributeError:
return ""

23
core/settings.py
View File

@ -1,6 +1,5 @@
import os
from pathlib import Path
from typing import Literal
import sentry_sdk
from django.contrib import messages
@ -76,10 +75,6 @@ INSTALLED_APPS: list[str] = [
"django.contrib.messages",
"django.contrib.staticfiles",
"django.contrib.sites",
"allauth",
"allauth.account",
"allauth.socialaccount",
"allauth.socialaccount.providers.twitch",
"simple_history",
"debug_toolbar",
]
@ -95,7 +90,6 @@ MIDDLEWARE: list[str] = [
"django.contrib.auth.middleware.AuthenticationMiddleware",
"django.contrib.messages.middleware.MessageMiddleware",
"simple_history.middleware.HistoryRequestMiddleware",
"allauth.account.middleware.AccountMiddleware",
]
TEMPLATES = [
@ -185,24 +179,7 @@ CACHES = {
},
}
SITE_ID = 1
AUTHENTICATION_BACKENDS: list[str] = [
"django.contrib.auth.backends.ModelBackend",
"allauth.account.auth_backends.AuthenticationBackend",
]
SOCIALACCOUNT_PROVIDERS = {
"twitch": {
"APP": {"client_id": os.environ["TWITCH_CLIENT_ID"], "secret": os.environ["TWITCH_CLIENT_SECRET"], "key": ""},
"SCOPE": [],
"AUTH_PARAMS": {
"force_verify": "true",
},
},
}
SOCIALACCOUNT_STORE_TOKENS = True
SOCIALACCOUNT_ONLY = True
ACCOUNT_EMAIL_VERIFICATION: Literal["mandatory", "optional", "none"] = "none"
ACCOUNT_SESSION_REMEMBER = True
LOGIN_REDIRECT_URL = "/"
ACCOUNT_LOGOUT_REDIRECT_URL = "/"

123
core/signals.py
View File

@ -3,20 +3,40 @@ import os
from typing import TYPE_CHECKING
from discord_webhook import DiscordWebhook
from django.db.models.manager import BaseManager
from django.db.models.signals import post_save
from django.dispatch import receiver
from core.discord import generate_game_message, generate_owner_message
from core.models import DropCampaign, Game, GameSubscription, Owner, OwnerSubscription, User
from core.discord import convert_time_to_discord_timestamp
from core.models import DropCampaign, Game, Owner, User, Webhook
if TYPE_CHECKING:
import requests
from django.db.models.manager import BaseManager
logger: logging.Logger = logging.getLogger(__name__)
def generate_message(game: Game, drop: DropCampaign) -> str:
"""Generate a message for a game.
Args:
game (Game): The game to generate a message for.
drop (DropCampaign): The drop campaign to generate a message for.
Returns:
str: The message.
"""
# TODO(TheLovinator): Add a twitch link to a stream that has drops enabled. # noqa: TD003
game_name: str = game.name or "Unknown game"
description: str = drop.description or "No description available."
start_at: str = convert_time_to_discord_timestamp(drop.starts_at)
end_at: str = convert_time_to_discord_timestamp(drop.ends_at)
msg: str = f"**{game_name}**\n\n{description}\n\nStarts: {start_at}\nEnds: {end_at}"
logger.debug(msg)
return msg
@receiver(signal=post_save, sender=User)
def handle_user_signed_up(sender: User, instance: User, created: bool, **kwargs) -> None: # noqa: ANN003, ARG001, FBT001
"""Send a message to Discord when a user signs up.
@ -48,9 +68,8 @@ def handle_user_signed_up(sender: User, instance: User, created: bool, **kwargs)
logger.debug(response)
@receiver(signal=post_save, sender=DropCampaign)
def notify_users_of_new_drop(sender: DropCampaign, instance: DropCampaign, created: bool, **kwargs) -> None: # noqa: ANN003, ARG001, FBT001
"""Notify users of a new drop campaign.
"""Send message to all webhooks subscribed to new drops.
Args:
sender (DropCampaign): The model we are sending the signal from.
@ -64,37 +83,81 @@ def notify_users_of_new_drop(sender: DropCampaign, instance: DropCampaign, creat
game: Game | None = instance.game
if not game:
logger.error("No game found. %s", instance)
return
if game.owner: # type: ignore # noqa: PGH003
handle_owner_drops(instance, game)
else:
logger.error("No owner found. %s", instance)
if game := instance.game:
handle_game_drops(instance, game)
else:
logger.error("No game found. %s", instance)
def handle_game_drops(instance: DropCampaign, game: Game) -> None:
"""Send message to all webhooks subscribed to new drops for this game.
Args:
instance (DropCampaign): The drop campaign that was created.
game (Game): The game that the drop campaign is for.
"""
webhooks: list[Webhook] = game.subscribed_new_games.all() # type: ignore # noqa: PGH003
for hook in webhooks:
# Don't spam the same drop campaign.
if hook in hook.seen_drops.all():
logger.error("Already seen drop campaign '%s'.", instance.name)
continue
# Set the webhook as seen so we don't spam it.
hook.seen_drops.add(instance)
# Send the webhook.
webhook_url: str = hook.get_webhook_url()
if not webhook_url:
logger.error("No webhook URL provided.")
continue
webhook = DiscordWebhook(
url=webhook_url,
content=generate_message(game, instance),
username=f"{game.name} Twitch drops",
rate_limit_retry=True,
)
response: requests.Response = webhook.execute()
logger.debug(response)
def handle_owner_drops(instance: DropCampaign, game: Game) -> None:
"""Send message to all webhooks subscribed to new drops for this owner/organization.
Args:
instance (DropCampaign): The drop campaign that was created.
game (Game): The game that the drop campaign is for.
"""
owner: Owner = game.owner # type: ignore # noqa: PGH003
webhooks: list[Webhook] = owner.subscribed_new_games.all() # type: ignore # noqa: PGH003
for hook in webhooks:
# Don't spam the same drop campaign.
if hook in hook.seen_drops.all():
logger.error("Already seen drop campaign '%s'.", instance.name)
continue
# Notify users subscribed to the game
game_subs: BaseManager[GameSubscription] = GameSubscription.objects.filter(game=game)
for sub in game_subs:
if not sub.webhook.url:
# Set the webhook as seen so we don't spam it.
hook.seen_drops.add(instance)
# Send the webhook.
webhook_url: str = hook.get_webhook_url()
if not webhook_url:
logger.error("No webhook URL provided.")
return
continue
webhook = DiscordWebhook(
url=sub.webhook.url,
content=generate_game_message(instance=instance, game=game, sub=sub),
username=f"{game.name} drop.",
rate_limit_retry=True,
)
response: requests.Response = webhook.execute()
logger.debug(response)
# Notify users subscribed to the owner
owner_subs: BaseManager[OwnerSubscription] = OwnerSubscription.objects.filter(owner=owner)
for sub in owner_subs:
if not sub.webhook.url:
logger.error("No webhook URL provided.")
return
webhook = DiscordWebhook(
url=sub.webhook.url,
content=generate_owner_message(instance=instance, owner=owner, sub=sub),
username=f"{owner.name} drop.",
url=webhook_url,
content=generate_message(game, instance),
username=f"{game.name} Twitch drops",
rate_limit_retry=True,
)
response: requests.Response = webhook.execute()

23
core/templates/partials/game_card.html
View File

@ -18,9 +18,7 @@
<!-- Insert nice buttons -->
</div>
{% for campaign in game.drop_campaigns.all %}
{% if not forloop.first %}<br>{% endif %}
<div class="mt-3">
<h3 class="h6">{{ campaign.name }}</h3>
{% if campaign.details_url == campaign.account_link_url %}
<a href="{{ campaign.details_url }}" class="text-decoration-none">Details</a>
{% else %}
@ -31,26 +29,7 @@
<p class="mb-2 text-muted">
Ends in: <abbr title="{{ campaign.starts_at|date:'l d F H:i' }} - {{ campaign.ends_at|date:'l d F H:i e' }}">{{ campaign.ends_at|timeuntil }}</abbr>
</p>
{% if campaign.description != campaign.name %}
{% if campaign.description|length > 300 %}
<p>
<a class="btn btn-link p-0 text-muted"
data-bs-toggle="collapse"
href="#collapseDescription{{ campaign.id }}"
role="button"
aria-expanded="false"
aria-controls="collapseDescription{{ campaign.id }}"
aria-label="Show Description">Show Description</a>
</p>
<div class="collapse" id="collapseDescription{{ campaign.id }}">
<div class="card card-body">{{ campaign.description }}</div>
<br>
</div>
{% else %}
<p>{{ campaign.description }}</p>
{% endif %}
{% endif %}
<div class="row row-cols-1 row-cols-md-2 row-cols-lg-3 g-2">
<div class="row row-cols-1 row-cols-md-2 row-cols-lg-3">
{% for drop in campaign.drops.all %}
{% for benefit in drop.benefits.all %}
<div class="col d-flex align-items-center position-relative">

16
core/templates/partials/header.html
View File

@ -1,4 +1,3 @@
{% load socialaccount %}
<header class="d-flex justify-content-between align-items-center py-3 border-bottom">
<h1 class="h2">
<a href='{% url "index" %}' class="text-decoration-none nav-title">Twitch drops</a>
@ -17,18 +16,9 @@
<li class="nav-item d-none d-sm-block">
<a class="nav-link" href="https://github.com/sponsors/TheLovinator1">Donate</a>
</li>
{% if not user.is_authenticated %}
<li>
<a class="nav-link" href="{% provider_login_url 'twitch' %}">Login with Twitch</a>
</li>
{% else %}
<li>
<form action="{% url 'account_logout' %}" method="post">
{% csrf_token %}
<button type="submit" class="btn btn-link nav-link">Logout</button>
</form>
</li>
{% endif %}
<li class="nav-item d-none d-sm-block">
<a class="nav-link" href='{% url "webhooks" %}'>Webhooks</a>
</li>
</ul>
</nav>
</header>

22
core/templates/webhooks.html
View File

@ -24,5 +24,27 @@
</form>
</div>
<h2 class="mt-5">Webhooks</h2>
{% if webhooks %}
<div class="list-group">
{% for webhook in webhooks %}
<div class="list-group-item d-flex justify-content-between align-items-center">
<span>
{% if webhook.avatar %}
<img src="https://cdn.discordapp.com/avatars/{{ webhook.id }}/a_{{ webhook.avatar }}.png"
alt="Avatar of {{ webhook.name }}"
class="rounded-circle"
height="32"
width="32">
{% endif %}
<a href="https://discord.com/api/webhooks/{{ webhook.id }}/{{ webhook.token }}"
target="_blank"
class="text-decoration-none">{{ webhook.name }}</a>
</span>
</div>
{% endfor %}
</div>
{% else %}
<div class="alert alert-info">No webhooks added</div>
{% endif %}
</div>
{% endblock content %}

20
core/urls.py
View File

@ -2,25 +2,17 @@ from __future__ import annotations
from debug_toolbar.toolbar import debug_toolbar_urls
from django.contrib import admin
from django.urls import URLPattern, URLResolver, include, path
from django.urls import URLPattern, URLResolver, path
from core.views import game_view, index, reward_campaign_view
from core.views import WebhooksView, game_view, index, reward_campaign_view
app_name: str = "core"
urlpatterns: list[URLPattern | URLResolver] = [
path("admin/", admin.site.urls),
path("accounts/", include("allauth.urls"), name="accounts"),
path(route="admin/", view=admin.site.urls),
path(route="", view=index, name="index"),
path(
route="games/",
view=game_view,
name="games",
),
path(
route="reward_campaigns/",
view=reward_campaign_view,
name="reward_campaigns",
),
path(route="webhooks/", view=WebhooksView.as_view(), name="webhooks"),
path(route="games/", view=game_view, name="games"),
path(route="reward_campaigns/", view=reward_campaign_view, name="reward_campaigns"),
*debug_toolbar_urls(),
]

135
core/views.py
View File

@ -4,16 +4,19 @@ import logging
from dataclasses import dataclass
from typing import TYPE_CHECKING
import requests_cache
from django.db.models import Prefetch
from django.db.models.manager import BaseManager
from django.http import HttpRequest, HttpResponse
from django.template.response import TemplateResponse
from django.utils import timezone # type: ignore # noqa: PGH003
from django.utils import timezone
from django.views import View
from core.models import DropCampaign, Game, RewardCampaign
from core.models import DropCampaign, Game, RewardCampaign, Webhook
if TYPE_CHECKING:
from django.db.models.manager import BaseManager
from django.http import HttpRequest, HttpResponse
from django.http import HttpRequest
logger: logging.Logger = logging.getLogger(__name__)
@ -114,3 +117,129 @@ def reward_campaign_view(request: HttpRequest) -> HttpResponse:
reward_campaigns: BaseManager[RewardCampaign] = RewardCampaign.objects.all()
context: dict[str, BaseManager[RewardCampaign]] = {"reward_campaigns": reward_campaigns}
return TemplateResponse(request=request, template="reward_campaigns.html", context=context)
def get_webhook_data(webhook_url: str) -> dict[str, str]:
"""Get the webhook data from the URL.
Args:
webhook_url (str): The webhook URL.
Returns:
dict[str, str]: The webhook data.
"""
session = requests_cache.CachedSession("webhook_cache")
response: requests_cache.OriginalResponse | requests_cache.CachedResponse = session.get(webhook_url)
return response.json()
def split_webhook_url(webhook_url: str) -> tuple[str, str]:
"""Split the webhook URL into its components.
Webhooks are in the format:
https://discord.com/api/webhooks/{id}/{token}
Args:
webhook_url (str): The webhook URL.
Returns:
tuple[str, str]: The ID and token.
"""
webhook_id: str = webhook_url.split("/")[-2]
webhook_token: str = webhook_url.split("/")[-1]
return webhook_id, webhook_token
class WebhooksView(View):
"""Render the webhook view page."""
@staticmethod
def post(request: HttpRequest) -> HttpResponse:
"""Add a webhook to the list of webhooks.
Args:
request (HttpRequest): The request object.
Returns:
HttpResponse: The response object.
"""
webhook_url: str | None = request.POST.get("webhook_url")
if not webhook_url:
return HttpResponse(content="No webhook URL provided.", status=400)
# Read webhooks from cookie.
webhooks_cookies: str | None = request.COOKIES.get("webhooks")
webhooks_list: list[str] = webhooks_cookies.split(",") if webhooks_cookies else []
# Get webhook data.
webhook_id, webhook_token = split_webhook_url(webhook_url)
webhook_data: dict[str, str] = get_webhook_data(webhook_url)
list_of_json_keys: list[str] = ["avatar", "channel_id", "guild_id", "name", "type", "url"]
defaults: dict[str, str | None] = {key: webhook_data.get(key) for key in list_of_json_keys}
# Warn if JSON has more keys than expected.
if len(webhook_data.keys()) > len(list_of_json_keys):
logger.warning("Unexpected keys in JSON: %s", webhook_data.keys())
# Add the webhook to the database.
new_webhook, created = Webhook.objects.update_or_create(
id=webhook_id,
token=webhook_token,
defaults=defaults,
)
if created:
logger.info("Created webhook '%s'.", new_webhook)
# Add the new webhook to the list.
webhooks_list.append(webhook_url)
# Remove duplicates.
webhooks_list = list(set(webhooks_list))
# Save the new list of webhooks to the cookie.
response: HttpResponse = HttpResponse()
response.set_cookie("webhooks", ",".join(webhooks_list))
# Redirect to the webhooks page.
response["Location"] = "/webhooks/"
response.status_code = 302
return response
@staticmethod
def get(request: HttpRequest) -> HttpResponse:
# Read webhooks from cookie.
webhooks_cookies: str | None = request.COOKIES.get("webhooks")
webhooks_list: list[str] = webhooks_cookies.split(",") if webhooks_cookies else []
webhooks_from_db: list[Webhook] = []
# Get the webhooks from the database.
for webhook_url in webhooks_list:
webhook_id, webhook_token = split_webhook_url(webhook_url)
# Check if the webhook is in the database.
if not Webhook.objects.filter(id=webhook_id, token=webhook_token).exists():
webhook_data: dict[str, str] = get_webhook_data(webhook_url)
list_of_json_keys: list[str] = ["avatar", "channel_id", "guild_id", "name", "type", "url"]
defaults: dict[str, str | None] = {key: webhook_data.get(key) for key in list_of_json_keys}
# Warn if JSON has more keys than expected.
if len(webhook_data.keys()) > len(list_of_json_keys):
logger.warning("Unexpected keys in JSON: %s", webhook_data.keys())
new_webhook, created = Webhook.objects.update_or_create(
id=webhook_id,
token=webhook_token,
defaults=defaults,
)
if created:
logger.info("Created webhook '%s'.", new_webhook)
webhooks_from_db.append(new_webhook)
# If the webhook is in the database, get it from there.
else:
existing_webhook: Webhook = Webhook.objects.get(id=webhook_id, token=webhook_token)
webhooks_from_db.append(existing_webhook)
context: dict[str, list[Webhook]] = {"webhooks": webhooks_from_db}
return TemplateResponse(request=request, template="webhooks.html", context=context)