linly/Discord-Twitch-Live-Notifier
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Catch errors when calling Discord API and re-try

Browse Source
This commit is contained in:
Deko
2023-12-31 19:32:38 +01:00
parent 3432edbb55
commit d5aff81e56
3 changed files with 139 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files

212
app/discord_client.py
View File

@@ -1,7 +1,9 @@
import os
from time import sleep
import requests
from loguru import logger
from requests import exceptions
from app.twitch_client import StreamInformation
@@ -13,91 +15,116 @@ class DiscordClient:
self._webhook_url = os.environ["DISCORD_WEBHOOK_URL"]
def send_information_to_discord(
self, stream: StreamInformation, profile_image: str
self,
stream: StreamInformation,
profile_image: str,
retry_count: int = 0,
) -> None:
logger.info("Sending a message with an embed to the webhook...")
streamer_url = f"https://www.twitch.tv/{stream.user_login}"
response = requests.post(
url=f"{self._webhook_url}?wait=true",
json={
"username": "Oak Tree",
"avatar_url": "https://i.imgur.com/DBOuwjx.png",
"content": f"@everyone {stream.user_name} went live!",
"embeds": [
{
"title": stream.title,
"color": 8388863,
"timestamp": stream.started_at,
"url": streamer_url,
"author": {
"name": stream.user_name,
try:
response = requests.post(
url=f"{self._webhook_url}?wait=true",
json={
"username": "Oak Tree",
"avatar_url": "https://i.imgur.com/DBOuwjx.png",
"content": f"@everyone {stream.user_name} went live!",
"embeds": [
{
"title": stream.title,
"color": 8388863,
"timestamp": stream.started_at,
"url": streamer_url,
"icon_url": profile_image,
},
"image": {"url": stream.thumbnail_url},
"fields": [
{
"name": "Game",
"value": stream.game_name,
"inline": True,
"author": {
"name": stream.user_name,
"url": streamer_url,
"icon_url": profile_image,
},
{
"name": "Viewers",
"value": stream.viewer_count,
"inline": True,
},
],
}
],
},
)
"image": {"url": stream.thumbnail_url},
"fields": [
{
"name": "Game",
"value": stream.game_name,
"inline": True,
},
{
"name": "Viewers",
"value": stream.viewer_count,
"inline": True,
},
],
}
],
},
)
response.raise_for_status()
response.raise_for_status()
self._notification_msg_id = response.json()["id"]
logger.info("Stream information sent with ping to Discord.")
self._notification_msg_id = response.json()["id"]
logger.info("Stream information sent with ping to Discord.")
except (exceptions.ConnectionError, exceptions.HTTPError) as err:
logger.opt(exception=err).warning(
"Could not send embed to Discord."
)
if retry_count > 5:
logger.warning("Aborted sending the embed to Discord.")
return
retry_count += 1
logger.info(f"Retrying finalize in {retry_count * 5} seconds.")
sleep(retry_count * 5)
self.send_information_to_discord(
stream=stream,
profile_image=profile_image,
retry_count=retry_count,
)
def update_information_on_discord(
self, stream: StreamInformation, profile_image: str
) -> None:
logger.info("Updating stream information on Discord...")
streamer_url = f"https://www.twitch.tv/{stream.user_login}"
response = requests.patch(
url=f"{self._webhook_url}/messages/{self._notification_msg_id}",
json={
"embeds": [
{
"title": stream.title,
"color": 8388863,
"timestamp": stream.started_at,
"url": streamer_url,
"author": {
"name": f"{stream.user_name}",
try:
response = requests.patch(
url=f"{self._webhook_url}/messages/{self._notification_msg_id}",
json={
"embeds": [
{
"title": stream.title,
"color": 8388863,
"timestamp": stream.started_at,
"url": streamer_url,
"icon_url": profile_image,
},
"image": {"url": stream.thumbnail_url},
"fields": [
{
"name": "Game",
"value": stream.game_name,
"inline": True,
"author": {
"name": f"{stream.user_name}",
"url": streamer_url,
"icon_url": profile_image,
},
{
"name": "Viewers",
"value": stream.viewer_count,
"inline": True,
},
],
}
],
},
)
response.raise_for_status()
logger.info("Message embed content updated.")
"image": {"url": stream.thumbnail_url},
"fields": [
{
"name": "Game",
"value": stream.game_name,
"inline": True,
},
{
"name": "Viewers",
"value": stream.viewer_count,
"inline": True,
},
],
}
],
},
)
response.raise_for_status()
logger.info("Message embed content updated.")
except (exceptions.ConnectionError, exceptions.HTTPError) as err:
logger.opt(exception=err).warning(
"Could not update embed content due to connection error. "
"Not retrying due to this not being important."
)
def finalize_information_on_discord(
self, streamer_name, vod_url: str | None
self, streamer_name, vod_url: str | None, retry_count: int = 0
) -> None:
logger.info("Finalizing stream information on Discord...")
if not self._notification_msg_id:
@@ -105,19 +132,38 @@ class DiscordClient:
return
if not vod_url:
vod_url = "None available. Please contact the developer."
vod_url = "None available."
response = requests.patch(
url=f"{self._webhook_url}/messages/{self._notification_msg_id}",
json={
"username": "Oak Tree",
"avatar_url": "https://i.imgur.com/DBOuwjx.png",
"content": (
f"{streamer_name} stopped the stream. Check out the VOD!"
f"\n{vod_url}"
),
"embeds": [],
},
)
response.raise_for_status()
logger.info("Message updated with VOD.")
try:
response = requests.patch(
url=f"{self._webhook_url}/messages/{self._notification_msg_id}",
json={
"username": "Oak Tree",
"avatar_url": "https://i.imgur.com/DBOuwjx.png",
"content": (
f"{streamer_name} stopped the stream. Check out the VOD!"
f"\n{vod_url}"
),
"embeds": [],
},
)
response.raise_for_status()
logger.info("Message updated with VOD.")
except (exceptions.ConnectionError, exceptions.HTTPError) as err:
logger.opt(exception=err).warning(
"Could not finalize embed on Discord."
)
if retry_count > 5:
logger.warning(
"Aborted finalizing the embed on Discord. "
"It will be stuck on the last stream update."
)
return
retry_count += 1
logger.info(f"Retrying finalize in {retry_count * 5} seconds.")
sleep(retry_count * 5)
self.finalize_information_on_discord(
streamer_name=streamer_name,
vod_url=vod_url,
retry_count=retry_count,
)

5
app/tests/conftest.py
View File

@@ -8,9 +8,10 @@ import pytest
def mock_loggers():
with (
mock.patch("loguru.logger.info") as info_logger,
mock.patch("loguru.logger.warning") as warning_logger,
mock.patch("loguru.logger.error") as error_logger,
):
mocked_loggers = namedtuple(
"mocked_loggers", ["info_logger", "error_logger"]
"mocked_loggers", ["info_logger", "warning_logger", "error_logger"]
)
yield mocked_loggers(info_logger, error_logger)
yield mocked_loggers(info_logger, warning_logger, error_logger)

12
app/tests/test_discord_client.py
View File

@@ -5,7 +5,6 @@ from unittest import mock
import pytest
import requests_mock
from requests import HTTPError
from app.discord_client import DiscordClient
from app.twitch_client import StreamInformation
@@ -74,12 +73,15 @@ def test_send_information_to_discord_fails(mock_loggers):
requests_mocker.post(url="https://test", status_code=400)
discord_client = DiscordClient()
with pytest.raises(HTTPError):
discord_client.send_information_to_discord(
stream=stream, profile_image=""
)
discord_client.send_information_to_discord(
stream=stream, profile_image="", retry_count=6
)
assert len(mock_loggers.info_logger.call_args_list) == 1
assert mock_loggers.info_logger.call_args.args[0] == (
"Sending a message with an embed to the webhook..."
)
assert len(mock_loggers.warning_logger.call_args_list) == 1
assert mock_loggers.warning_logger.call_args.args[0] == (
"Aborted sending the embed to Discord."
)