87 lines
3.7 KiB
Python
87 lines
3.7 KiB
Python
|
import json
|
||
|
|
from libs.Cog import Cog
|
||
|
|
from libs.Twitch import Twitch
|
||
|
|
from libs.Channels import Channels
|
||
|
|
from libs.Guilds import Guilds
|
||
|
|
from discord_bot.embeds.TwitchGameNotificationEmbed import TwitchGameNotificationEmbed
|
||
|
|
from discord.ext import commands, tasks
|
||
|
|
from data.models.TwitchLiveNotification import TwitchLiveNotification
|
||
|
|
from discord_bot.bot import EmperorFredDiscordBot
|
||
|
|
|
||
|
|
|
||
|
|
class TwitchNotificationsCog(Cog):
|
||
|
|
def __init__(self, bot: EmperorFredDiscordBot):
|
||
|
|
super().__init__(bot)
|
||
|
|
self.bot = bot
|
||
|
|
self.guild = None
|
||
|
|
self.twitch = None
|
||
|
|
self.online_users = {}
|
||
|
|
self.config = {}
|
||
|
|
|
||
|
|
@commands.Cog.listener()
|
||
|
|
async def on_ready(self):
|
||
|
|
await self.load_config()
|
||
|
|
self.guild = Guilds().get_guild(self.bot)
|
||
|
|
self.twitch = Twitch(self.config[self.guild.name]["twitch"]["client_id"], self.config[self.guild.name]["twitch"]["client_secret"])
|
||
|
|
await self.set_access_token_and_expires_date()
|
||
|
|
await self.set_notification_channel_and_purge_channel()
|
||
|
|
await self.write_config()
|
||
|
|
self.check_twitch_online_streamers.start()
|
||
|
|
self.check_twitch_access_token.start()
|
||
|
|
|
||
|
|
async def load_config(self):
|
||
|
|
with open("discord_bot/config.json") as config_file:
|
||
|
|
self.config = json.load(config_file)
|
||
|
|
|
||
|
|
async def write_config(self):
|
||
|
|
with open("discord_bot/config.json", "w") as config_file:
|
||
|
|
json.dump(self.config, config_file, indent=4)
|
||
|
|
|
||
|
|
async def set_notification_channel_and_purge_channel(self):
|
||
|
|
channel = await Channels().get_channel(self.guild, self.config[self.guild.name]['twitch']['channel_name'])
|
||
|
|
if self.config[self.guild.name]['twitch']['channel_id'] == 'xxx':
|
||
|
|
self.config[self.guild.name]["twitch"]["channel_id"] = channel.id
|
||
|
|
await channel.purge()
|
||
|
|
|
||
|
|
|
||
|
|
async def set_access_token_and_expires_date(self):
|
||
|
|
access_token, access_token_expires = await self.twitch.get_access_token()
|
||
|
|
self.config[self.guild.name]["twitch"]["access_token"] = access_token
|
||
|
|
self.config[self.guild.name]["twitch"]["expire_date"] = access_token_expires
|
||
|
|
|
||
|
|
@tasks.loop(seconds=60)
|
||
|
|
async def check_twitch_access_token(self):
|
||
|
|
print("Access token is checked")
|
||
|
|
access_token, access_token_expires = await self.twitch.check_access_token()
|
||
|
|
self.config[self.guild.name]["twitch"]["access_token"] = access_token
|
||
|
|
self.config[self.guild.name]["twitch"]["expire_date"] = access_token_expires
|
||
|
|
await self.write_config()
|
||
|
|
|
||
|
|
|
||
|
|
@tasks.loop(seconds=90)
|
||
|
|
async def check_twitch_online_streamers(self):
|
||
|
|
print("Online streamers are checked")
|
||
|
|
channel = await Channels().get_channel(self.guild, self.config[self.guild.name]["twitch"]["channel_name"])
|
||
|
|
if not channel:
|
||
|
|
return
|
||
|
|
|
||
|
|
watchlist_notifications = self.bot.db.session.query(TwitchLiveNotification).all()
|
||
|
|
watchlist = [notify.username for notify in watchlist_notifications]
|
||
|
|
print("Twitch Watchlist: ", watchlist)
|
||
|
|
|
||
|
|
online_notifications, offline_notifications = await self.twitch.get_notifications(watchlist)
|
||
|
|
print("online_notifications", online_notifications)
|
||
|
|
print("offline_notifications", offline_notifications)
|
||
|
|
|
||
|
|
for username in offline_notifications:
|
||
|
|
message = await channel.fetch_message(self.online_users[username])
|
||
|
|
await message.delete()
|
||
|
|
del self.online_users[username]
|
||
|
|
|
||
|
|
for notification in online_notifications:
|
||
|
|
embed = TwitchGameNotificationEmbed(notification)
|
||
|
|
message = await channel.send(embed=embed)
|
||
|
|
self.online_users[notification["user_login"]] = message.id
|
||
|
|
|
||
|
|
async def setup(bot):
|
||
|
|
await bot.add_cog(TwitchNotificationsCog(bot))
|