This repository has been archived by the owner on Mar 6, 2024. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.py
314 lines (262 loc) · 12.6 KB
/
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
"""_summary_.
Returns:
_type_: _description_
"""
import asyncio
import logging
import os
import discord
from discord.ext import commands
from discord.ext import tasks
from kayo import instance
from kayo.alert import create_league_alert
from kayo.alert import create_team_alert
from kayo.alert import delete_alert
from kayo.alert import get_alerts_by_channel_id
from kayo.alert import get_alerts_league
from kayo.alert import get_alerts_teams
from kayo.league import fetch_leagues
from kayo.league import get_league_by_id
from kayo.league import get_league_by_name
from kayo.league import get_league_names
from kayo.league import get_leagues
from kayo.lib import fetch_events_and_teams
from kayo.lib import send_match_alert
from kayo.match import get_matches
from kayo.match import get_upcoming_matches
from kayo.model import Base
from kayo.team import get_team_by_name
from kayo.team import get_team_names
from kayo.team import get_teams
Base.metadata.create_all(instance.engine)
# BOT LOGIC
@instance.bot.event
async def on_ready():
"""Executed when the Discord bot boots up."""
checkForMatches.start()
updateDatabase.start()
logging.info(f"{instance.bot.user} is online! 🚀")
@instance.bot.event
async def on_disconnect():
"""What happens when the bot is disconnected from Discord."""
logging.error(f"{instance.bot.user} is disconnected ! 💣")
@instance.bot.slash_command(name="help", description="Prints out a help message")
async def help(ctx):
"""Displays help.
Args:
ctx (discord.ApplicationContext): Information about the current message.
"""
await ctx.respond("""Hello ! My name is KAY/O and my mission is to send messages when pro matches are about to start. Here is a list of my commands :
`/list_alerts` will list all the alerts currently configured in the channel.
`/subscribe all_leagues` will give you alerts for every single matches happening.
`/subscribe league [name]` will give you alerts for a specific league (for example, VCT EMEA)
`/subscribe team [name]` will give you alerts for a specific team (for example, FNATIC)
`/unsubscribe league [name]` will give you alerts for a specific league (for example, VCT EMEA)
`/unsubscribe team [name]` will give you alerts for a specific team (for example, FNATIC)
`/ping` to check if I'm still working. I'm kinda new to this so sometimes I die.
Please note that the commands are channel-specific. Meaning the alerts will be sent in the channel you sent the command in.
Except the `ping` command, you need 'Manage Messages' permission to add / remove alerts in a public server.
You can still DM me if you want the alerts in your DMs though !
By the way, I'm open source, so check out my code if you want. Link is in my profile description.""")
@instance.bot.slash_command(name="ping", description="Sends the bot's latency.")
async def ping(ctx):
"""Simple ping command.
Args:
ctx (discord.ApplicationContext): Information about the current message.
"""
latency_ms = round(instance.bot.latency * 1000)
await ctx.respond(f"Pong! `{latency_ms}` ms")
@instance.bot.slash_command(name="list_alerts", description="Lists the alerts on this channel")
async def list_alerts(ctx):
"""Lists the alerts configured for the current channel.
Args:
ctx (discord.ApplicationContext): Information about the current message.
"""
await ctx.respond("Fetching alerts...")
list_of_alerts = get_alerts_by_channel_id(ctx.channel_id)
league_alerts = [x for x in list_of_alerts if x.league_id is not None]
team_alerts = [x for x in list_of_alerts if x.team_name is not None]
if not list_of_alerts:
await ctx.respond("There is no alerts configured for this channel.")
else:
if team_alerts:
answer = "List of team alerts : "
for alert in team_alerts:
if len(f"{answer}\r- {alert.team_name}") > 1500:
await ctx.respond(answer)
answer = ""
answer = f"{answer}\r- {alert.team_name}"
await ctx.respond(answer)
if league_alerts:
answer = "List of league alerts :"
for alert in league_alerts:
league = get_league_by_id(alert.league_id)
if len(f"{answer} \r - {league.name}") > 1500:
await ctx.respond({answer})
answer = ""
answer = f"{answer}\r- {league.name}"
await ctx.respond(answer)
@instance.subscribe.command(name="league", description="Subscribe to league alerts")
@commands.has_permissions(manage_messages=True)
async def subscribe_league(
ctx: discord.ApplicationContext,
league: discord.Option(
discord.SlashCommandOptionType.string,
autocomplete=discord.utils.basic_autocomplete(get_league_names),
),
):
"""Subscribes the channel to a league.
Args:
ctx (discord.ApplicationContext): Information about the current message.
league (discord.Option): Name of the League to follow.
Defaults to discord.utils.basic_autocomplete(get_league_names)).
"""
try:
alert = create_league_alert(get_league_by_name(league), ctx.channel_id)
instance.logger.info(f"Created alert {str(alert)}")
await ctx.respond(f"Successfully created an alert for {league} !")
except discord.ext.commands.errors.MissingPermissions:
await ctx.respond("You need to have the 'Manage Messages' permission to run this command in a server. Feel free to send me a DM !")
@instance.subscribe.command(name="team", description="Subscribe to team alerts")
@commands.has_permissions(manage_messages=True)
async def subscribe_team(
ctx: discord.ApplicationContext,
team: discord.Option(
discord.SlashCommandOptionType.string,
autocomplete=discord.utils.basic_autocomplete(get_team_names),
),
):
"""Subscribe the Discord channel to a Team.
Args:
ctx (discord.ApplicationContext): Information about the current message.
team (discord.Option, optional): Autocomplete.
Defaults to discord.utils.basic_autocomplete(get_league_names) ).
"""
try:
alert = create_team_alert(get_team_by_name(team), ctx.channel_id)
instance.logger.info(f"Created alert {str(alert)}")
await ctx.respond(f"Successfully created an alert for {team} !")
except discord.ext.commands.errors.MissingPermissions:
await ctx.respond("You need to have the 'Manage Messages' permission to run this command in a server. Feel free to send me a DM !")
@instance.unsubscribe.command(name="league", description="Delete a league alert for this channel")
@commands.has_permissions(manage_messages=True)
async def unsubscribe_league(
ctx: discord.ApplicationContext,
league: discord.Option(
discord.SlashCommandOptionType.string,
autocomplete=discord.utils.basic_autocomplete(get_league_names),
),
):
"""Subscribes the channel to a league.
Args:
ctx (discord.ApplicationContext): Information about the current message.
league (discord.Option): Name of the League to follow.
Defaults to discord.utils.basic_autocomplete(get_league_names)).
"""
try:
league_obj = [x for x in get_leagues() if x.name == league][0]
delete_alert(ctx.channel_id, league=league_obj)
await ctx.respond(f"Successfully deleted an alert for {league} !")
except discord.ext.commands.errors.MissingPermissions:
await ctx.respond("You need to have the 'Manage Messages' permission to run this command in a server. Feel free to send me a DM !")
@instance.unsubscribe.command(name="team", description="Delete a team alert for this channel")
@commands.has_permissions(manage_messages=True)
async def unsubscribe_team(
ctx: discord.ApplicationContext,
team: discord.Option(
discord.SlashCommandOptionType.string,
autocomplete=discord.utils.basic_autocomplete(get_team_names),
),
):
"""Subscribe the Discord channel to a Team.
Args:
ctx (discord.ApplicationContext): Information about the current message.
team (discord.Option, optional): Autocomplete.
Defaults to discord.utils.basic_autocomplete(get_league_names) ).
"""
try:
team_obj = [x for x in get_teams() if x.name == team][0]
delete_alert(ctx.channel_id, team=team_obj)
await ctx.respond(f"Successfully deleted an alert for {team} !")
except discord.ext.commands.errors.MissingPermissions:
await ctx.respond("You need to have the 'Manage Messages' permission to run this command in a server. Feel free to send me a DM !")
@instance.subscribe.command(name="all_leagues", description="Subscribe to league alerts")
@commands.has_permissions(manage_messages=True)
async def subscribe_all_leagues(ctx: discord.ApplicationContext):
"""Subscribe the channel to all the different leagues.
Args:
ctx (discord.ApplicationContext): Information about the current message.
"""
instance.logger.info('Creating alert...')
try:
for league in get_leagues():
create_league_alert(league, ctx.channel_id)
await ctx.respond("Subscribed to all the different leagues !")
except discord.ext.commands.errors.MissingPermissions as e:
instance.logger.error(str(e))
@instance.bot.event
async def on_application_command_error(ctx: discord.ApplicationContext, error: discord.DiscordException):
"""Handler for command errors inside Discord.
Args:
ctx (discord.ApplicationContext): Information about the current message.
error (discord.DiscordException): The Discord exception data.
Raises:
error: Just here to handle an exception-ception.
"""
if isinstance(error, discord.ext.commands.errors.MissingPermissions):
await ctx.respond("You need to have the 'Manage Messages' permission to run this command in a server.")
else:
instance.logger.exception(error)
raise error # Here we raise other errors to ensure they aren't ignored
@tasks.loop(seconds=300)
async def checkForMatches(prepared_matches=None):
"""Checks if there is new upcoming matches."""
instance.logger.info("Checking for alerts to send...")
if prepared_matches is None:
prepared_matches = get_upcoming_matches()
async with asyncio.TaskGroup() as tg:
for match in prepared_matches:
try:
team_alerts = get_alerts_teams(match.team_a, match.team_b)
league_alerts = get_alerts_league(match.league)
for alert in team_alerts:
tg.create_task(send_match_alert(alert.channel_id, match))
for alert in league_alerts:
tg.create_task(send_match_alert(alert.channel_id, match))
except* Exception as ex:
instance.logger.exception(f'Got an exception while handling alerts : {ex}')
instance.logger.info('Finished updating Matches and Teams !')
@tasks.loop(minutes=30)
async def updateDatabase():
"""Checks if there is new upcoming matches."""
instance.logger.info("Updating the database periodically...")
fetch_leagues()
await fetch_events_and_teams()
if os.environ.get('DEPLOYED').upper() != "PRODUCTION":
@instance.bot.slash_command(name="debug_alert", description="DO NOT USE ON YOUR SERVER !")
@commands.has_permissions(manage_roles=True, ban_members=True)
async def debug_alerts(ctx):
"""Sends a shit ton of alerts for debugging the format.
Args:
ctx (discord.ApplicationContext): Information about the current message.
"""
try:
await checkForMatches(prepared_matches=sorted(get_matches(), key=lambda x: x.startTime, reverse=True)[0:10])
except discord.ext.commands.errors.MissingPermissions as e:
instance.logger.error(str(e))
@instance.subscribe.command(name="all_teams", description="Subscribe to team alerts")
@commands.has_permissions(manage_roles=True, ban_members=True)
async def subscribe_all_teams(ctx: discord.ApplicationContext):
"""Subscribe the channel to all the different leagues.
Args:
ctx (discord.ApplicationContext): Information about the current message.
"""
instance.logger.info('Creating alert...')
try:
await ctx.respond("Subscribing you to all teams...")
for team in get_teams():
create_team_alert(team, ctx.channel_id)
await ctx.respond("Subscribed to all the different teams !")
except discord.ext.commands.errors.MissingPermissions as e:
instance.logger.error(str(e))
instance.bot.run(os.getenv("DISCORD_TOKEN"))