Files
Discord-Twitch-Live-Notifier/app/discord_client.py
2023-03-04 00:15:19 +01:00

123 lines
4.4 KiB
Python

import os
import requests
from loguru import logger
from app.twitch_client import StreamInformation
class DiscordClient:
_notification_msg_id: str = ""
def __init__(self):
self._webhook_url = os.environ["DISCORD_WEBHOOK_URL"]
def send_information_to_discord(
self, stream: StreamInformation, profile_image: str
) -> str:
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": "@everyone",
"embeds": [
{
"title": stream.title,
"color": 8388863,
"timestamp": stream.started_at,
"url": streamer_url,
"author": {
"name": stream.user_name,
"url": streamer_url,
"icon_url": profile_image,
},
"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()
self._notification_msg_id = response.json()["id"]
logger.info("Stream information sent with ping to Discord.")
return self._notification_msg_id
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}",
"url": streamer_url,
"icon_url": profile_image,
},
"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.")
def finalize_information_on_discord(
self, streamer_name, vod_url: str | None
) -> None:
logger.info("Finalizing stream information on Discord...")
if not self._notification_msg_id:
logger.info("Message ID not set, nothing to finalize.")
return
if not vod_url:
vod_url = "None available. Please contact the developer."
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. VOD: \n{vod_url}"
),
"embeds": [],
},
)
response.raise_for_status()
logger.info("Message updated with VOD.")