Marc Ahlgrim
331 lines
13 KiB
Python
331 lines
13 KiB
Python
import discord
|
|
import random
|
|
import numpy as np
|
|
import utils
|
|
import datetime
|
|
import config
|
|
|
|
from glob import glob
|
|
from dateutil import tz
|
|
from sys import exit
|
|
|
|
from discord import app_commands
|
|
from discord.ext import tasks
|
|
|
|
from classes.dis_events import DiscordEvents
|
|
from classes.subscribe import Subscribe
|
|
|
|
|
|
# #######################################
|
|
# The OMFClient class
|
|
# #######################################
|
|
|
|
|
|
class OMFClient(discord.Client):
|
|
|
|
channel_idle_timer: int
|
|
lastNotifyTimeStamp = None
|
|
theGuild : discord.Guild = None
|
|
|
|
lastSentMessage:discord.Message=None
|
|
|
|
guildEventsList = None
|
|
guildEventsClass: DiscordEvents = None
|
|
|
|
# #######################################
|
|
# init constructor
|
|
# #######################################
|
|
|
|
def __init__(self) -> None:
|
|
|
|
print('Init')
|
|
|
|
# Try to set all intents
|
|
|
|
intents = discord.Intents.all()
|
|
super().__init__(intents=intents)
|
|
|
|
# We need a `discord.app_commands.CommandTree` instance
|
|
# to register application commands (slash commands in this case)
|
|
|
|
self.tree = app_commands.CommandTree(self)
|
|
|
|
# The subscribe command will add/remove the notification roles
|
|
# based on the scheduled events
|
|
|
|
@self.tree.command(name="subscribe", description="(un)subscribe to Events)")
|
|
async def subscribe(interaction: discord.Interaction):
|
|
|
|
# preload the menu items with the roles that the user has already
|
|
# we might move this to the init of the modal
|
|
|
|
x: Subscribe
|
|
role: discord.Role
|
|
member: discord.Member
|
|
|
|
x=Subscribe()
|
|
member = interaction.user
|
|
|
|
for option in x.Menu.options:
|
|
role = option.value
|
|
if not (member.get_role(role) is None):
|
|
option.default=True
|
|
await interaction.response.send_modal(x)
|
|
|
|
self.channel_idle_timer = 0
|
|
self.idle_channel = self.get_channel(config.CONFIG["IDLE_MESSAGE_CHANNEL_ID"])
|
|
|
|
# #########################
|
|
# setup_hook waits for the
|
|
# command tree to sync
|
|
# #########################
|
|
|
|
async def setup_hook(self) -> None:
|
|
# Sync the application command with Discord.
|
|
await self.tree.sync()
|
|
|
|
# ######################################################
|
|
# send_random_message is called when the server is idle
|
|
# and posts a random message to the server
|
|
# ######################################################
|
|
|
|
async def send_random_message(self):
|
|
print("Sending random message")
|
|
if self.idle_channel == None:
|
|
self.idle_channel = self.get_channel(config.CONFIG["IDLE_MESSAGE_CHANNEL_ID"])
|
|
print (f'The idle channel is {config.CONFIG["IDLE_MESSAGE_CHANNEL_ID"]} - {self.idle_channel}')
|
|
await self.idle_channel.send(f"{random.choice(self.idle_messages)}")
|
|
|
|
# ######################################################
|
|
# on_ready is called once the client is initialized
|
|
# it then reads in the files in the config.IDLE_MESSAGE_DIR
|
|
# directory and starts the schedulers
|
|
# ######################################################
|
|
|
|
async def on_ready(self):
|
|
print('Logged on as', self.user)
|
|
|
|
# read in the random message files
|
|
# the idle_messages array holds one element per message
|
|
# every file is read in as a whole into one element of the array
|
|
|
|
self.idle_messages = []
|
|
|
|
for filename in glob(config.CONFIG["IDLE_MESSAGE_DIR"] + '/*.txt'):
|
|
print ("read {}",filename)
|
|
with open(filename) as f:
|
|
self.idle_messages.append(f.read())
|
|
|
|
self.idle_messages = np.array(self.idle_messages)
|
|
|
|
# store the guild for further use
|
|
guild: discord.Guild
|
|
for guild in self.guilds:
|
|
if (int(guild.id) == int(config.SECRETS["GUILD_ID"])):
|
|
print (f"GUILD MATCHES {guild.id}")
|
|
self.theGuild = guild
|
|
|
|
if (self.theGuild is None):
|
|
print("the guild (Server ID)could not be found - please check all config data")
|
|
exit()
|
|
|
|
self.guildEventsClass = DiscordEvents(
|
|
discord_token=config.SECRETS["BOT_TOKEN"],
|
|
client_id=config.SECRETS["CLIENT_ID"],
|
|
bot_permissions=8,
|
|
api_version=10,
|
|
guild_id=config.SECRETS["GUILD_ID"]
|
|
)
|
|
|
|
# start the schedulers
|
|
|
|
self.task_scheduler.start()
|
|
self.daily_tasks.start()
|
|
|
|
|
|
|
|
# ######################################################
|
|
# create_events will create the Sunday Funday events
|
|
# for the next sunday
|
|
# ######################################################
|
|
|
|
async def create_events (self):
|
|
|
|
print("Create Events")
|
|
|
|
for theEvent in config.AUTO_EVENTS:
|
|
|
|
# calculate the date of the future event
|
|
theDate:datetime.datetime = utils.onDay(datetime.date.today(),theEvent['day_of_week'])
|
|
# we need the offset from local time to UTC for the date of the event,
|
|
# not the date when we create the event. The event is on a Sunday and might be the
|
|
# first daylight saving day
|
|
utcOffset=tz.tzlocal().utcoffset(datetime.datetime.strptime(f"{theDate}","%Y-%m-%d"))
|
|
# the times are stored in local time in the dictionary but will
|
|
# be UTC in the discord API
|
|
utcStartTime=format(datetime.datetime.strptime(theEvent['start_time'],"%H:%M:%S")-utcOffset,"%H:%M:%S")
|
|
utcEndTime=format(datetime.datetime.strptime(theEvent['end_time'],"%H:%M:%S")-utcOffset,"%H:%M:%S")
|
|
# Now we just add the date portion and the time portion
|
|
# of course this will not work for events where the UTCOffset is bigger
|
|
# than the start time - for Germany this is OK as long as the start time is
|
|
# after 2 AM
|
|
strStart=theDate.strftime(f"%Y-%m-%dT{utcStartTime}")
|
|
strEnd=theDate.strftime(f"%Y-%m-%dT{utcEndTime}")
|
|
await self.guildEventsClass.create_guild_event(
|
|
event_name=theEvent['title'],
|
|
event_description=theEvent['description'],
|
|
event_start_time=f"{strStart}",
|
|
event_end_time=f"{strEnd}",
|
|
event_metadata={},
|
|
event_privacy_level=2,
|
|
channel_id=theEvent['channel'])
|
|
# once we have created the event, we let everyone know
|
|
channel = self.get_channel(config.CONFIG["IDLE_MESSAGE_CHANNEL_ID"])
|
|
await channel.send(f'Hi - I have created the scheduled Event {theEvent["title"]}')
|
|
|
|
|
|
# ######################################################
|
|
# get_event_list gives a list of scheduled events
|
|
# ######################################################
|
|
|
|
async def get_events_list (self):
|
|
eventList = await self.guildEventsClass.list_guild_events()
|
|
self.guildEventsList = eventList
|
|
return eventList
|
|
|
|
# ######################################################
|
|
# on_message scans for message contents and takes
|
|
# corresponding actions.
|
|
# User sends ping - bot replies with pong
|
|
# User asks a question - bot checks if question has been
|
|
# answered
|
|
# ######################################################
|
|
|
|
async def on_message(self, message : discord.Message ):
|
|
|
|
# ignore ephemeral messages
|
|
if message.flags.ephemeral:
|
|
return
|
|
|
|
|
|
print("{} has just sent {}".format(message.author, message.content))
|
|
# if the author of the previously last sent message and
|
|
# the new message is ourselves, then delete the
|
|
# previous message
|
|
|
|
if (int(f'{config.CONFIG["AVOID_SPAM"]}') == 1) and (self.lastSentMessage is not None):
|
|
if ((message.author == self.user) and
|
|
(self.lastSentMessage.author == self. user) and
|
|
(int(f"{self.lastSentMessage.channel.id}") == (int(config.CONFIG["IDLE_MESSAGE_CHANNEL_ID"])))):
|
|
try:
|
|
await self.lastSentMessage.delete()
|
|
except Exception as e:
|
|
print(f"delete lastmessage error: {e}")
|
|
|
|
|
|
self.lastSentMessage = message
|
|
|
|
# don't respond to ourselves
|
|
if message.author == self.user:
|
|
return
|
|
|
|
# reset the idle timer if a message has been sent or received
|
|
self.channel_idle_timer = 0
|
|
|
|
|
|
# ######################################################
|
|
# on_typing detects if a user types.
|
|
# We might use this one day to have users agree to policies etc.
|
|
# before they are allowed to speak
|
|
# or we might launch the Support() Modal if a user starts
|
|
# to type in the support channel
|
|
# ######################################################
|
|
|
|
async def on_typing(self, channel, user, _):
|
|
# we do not want the bot to reply to itself
|
|
if user.id == self.user.id:
|
|
return
|
|
print(f"{user} is typing in {channel}")
|
|
self.channel_idle_timer = 0
|
|
#await channel.trigger_typing()
|
|
|
|
# ######################################################
|
|
# daily_tasks runs once a day and checks the following:
|
|
# - does an event need to be created ?
|
|
# ######################################################
|
|
|
|
@tasks.loop(hours=24)
|
|
async def daily_tasks(self):
|
|
print("DAILY TASKS")
|
|
|
|
# Every Monday (weekday 0) we want to create the
|
|
# scheduled events for the next Sunday
|
|
|
|
if datetime.date.today().weekday() == 0:
|
|
print("create events")
|
|
try:
|
|
await self.create_events()
|
|
except Exception as e:
|
|
print(f"Daily Task create Events failed: {e}")
|
|
|
|
|
|
# ######################################################
|
|
# task_scheduler is the main supervisor task
|
|
# it runs every 10 minutes and checks the following:
|
|
# - has a question been asked that has not been answered ?
|
|
# - do reminders need to be sent out ?
|
|
# - does a random message need to be sent out ?
|
|
# ######################################################
|
|
|
|
|
|
@tasks.loop(minutes=10)
|
|
async def task_scheduler(self):
|
|
|
|
self.channel_idle_timer += 1
|
|
print("SCHEDULE")
|
|
|
|
# #####################################
|
|
# see if we need to send a random message
|
|
# if the counter is greater than CHANNEL_IDLE_INTERVAL
|
|
# then send a random message into the idle_channel
|
|
# #####################################
|
|
try:
|
|
if self.channel_idle_timer >= config.CONFIG["CHANNEL_IDLE_INTERVAL"]:
|
|
self.channel_idle_timer = 0
|
|
await self.send_random_message()
|
|
except Exception as e:
|
|
print(f"Scheduler random_message failed: {e}")
|
|
|
|
# see if we need to send out notifications for events
|
|
# The Event details are stored in config.
|
|
eventList=None
|
|
for theEvent in config.AUTO_EVENTS:
|
|
# first let's convert the String dates to datetime:
|
|
theDate=utils.onDay(datetime.date.today(),theEvent['day_of_week'])
|
|
startTime=theEvent['start_time']
|
|
eventScheduledTimeDate=datetime.datetime.fromisoformat (theDate.strftime(f"%Y-%m-%dT{startTime}"))
|
|
# now let's figure out the time deltas:
|
|
timeUntilEvent=eventScheduledTimeDate - datetime.datetime.today()
|
|
earliestPing=eventScheduledTimeDate - datetime.timedelta(minutes=theEvent["notify_minutes"]) - datetime.timedelta(minutes=10)
|
|
latestPing=eventScheduledTimeDate - datetime.timedelta(minutes=theEvent["notify_minutes"])
|
|
#print("found scheduled event")
|
|
#print(f"The event is on {theDate} at {startTime} - that is in {timeUntilEvent}")
|
|
#print (f"we ping between {earliestPing} and {latestPing}")
|
|
# If we are in the interval then let's initiate the reminder
|
|
if (earliestPing < datetime.datetime.today() <= latestPing):
|
|
# let's first check if the event is still on
|
|
# it may have been deleted or modified on the server
|
|
# we don't want to alert for non-existing events
|
|
# we'll just use the title to compare for the time being.
|
|
print("Let me check if the event is still on")
|
|
try:
|
|
if eventList == None:
|
|
eventList = await self.get_events_list()
|
|
for theScheduledEvent in eventList:
|
|
if theScheduledEvent["name"] == theEvent["title"]:
|
|
channel = self.get_channel(config.CONFIG["IDLE_MESSAGE_CHANNEL_ID"])
|
|
theMessageText=f'Hi <@&{theEvent["subscription_role_num"]}>, the event *** {theEvent["title"]} *** will start in roughly {theEvent["notify_minutes"]} minutes in the <#{theEvent["channel"]}> channel. {theEvent["description"]}'
|
|
await channel.send(f"{theMessageText}")
|
|
except Exception as e:
|
|
print(f"Scheduler event_reminder failed: {e}")
|