Release 0.3

Signed-off-by: Marc Ahlgrim <marc@onemarcfifty.com>
This commit is contained in:
Marc Ahlgrim
2022-07-22 12:50:06 +02:00
parent e58ba3a745
commit 6e35d9e064
15 changed files with 497 additions and 173 deletions
+354
View File
@@ -0,0 +1,354 @@
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.support import Support
from classes.subscribe import Subscribe
# #######################################
# The OMFClient class
# #######################################
class OMFClient(discord.Client):
channel_idle_timer: int
asked_question = False
last_question: discord.Message = None
lastNotifyTimeStamp = None
theGuild : discord.Guild = 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 support command will ask for a thread title and description
# and create a support thread for us
@self.tree.command(name="support", description="Create a support thread")
async def support(interaction: discord.Interaction):
x : Support
x= Support()
await interaction.response.send_modal(x)
# 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()
# ######################################################
# handle_ping is called when a user sends ping
# (case sensitive, exact phrase)
# it just replies with pong
# ######################################################
async def handle_ping (self,message : discord.Message):
await message.channel.send('pong', reference=message)
# ######################################################
# 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 ):
print("{} has just sent {}".format(message.author, message.content))
# 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
# reply to ping
if message.content == 'ping':
await self.handle_ping(message)
# check if there is a question
if "?" in message.content:
self.asked_question = True
self.last_question = message
else:
self.asked_question = False
self.last_question = None
# ######################################################
# 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(seconds=10)
async def task_scheduler(self):
self.channel_idle_timer += 1
print("SCHEDULE")
# #####################################
# See if there are unanswered questions
# #####################################
try:
if(self.asked_question):
print("scheduler: Question")
print(self.last_question.created_at)
# TODO - we need to send out a message to the @here role
# asking users to help if the question had not been answered
# Also - if the message is a reply then we should not post into the channel
if self.channel_idle_timer > config.CONFIG["QUESTION_SLEEPING_TIME"]:
print("QUESTION WITHOUT REPLY")
self.asked_question = False
except Exception as e:
print(f"Scheduler question failed: {e}")
# #####################################
# 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}")
+67
View File
@@ -0,0 +1,67 @@
import json
from classes.restfulapi import DiscordAPI
# #########################################################
# The class for Discord scheduled events
# as of 2022-07, discord.py does not support
# scheduled Events. Therefore we need to use the Rest API
# #########################################################
class DiscordEvents(DiscordAPI):
def __init__(self,
discord_token: str,
client_id: str,
bot_permissions: int,
api_version: int,
guild_id:str) -> None:
super().__init__(discord_token,client_id,bot_permissions,api_version)
self.guild_id = guild_id
print (f" DiscordEvent Client {client_id} guild {guild_id}")
async def list_guild_events(self) -> list:
# Returns a list of upcoming events for the supplied guild ID
# Format of return is a list of one dictionary per event containing information.
event_retrieve_url = f'{self.base_api_url}/guilds/{self.guild_id}/scheduled-events'
response_list = await self.get_api(event_retrieve_url)
return response_list
async def create_guild_event(
self,
event_name: str,
event_description: str,
event_start_time: str,
event_end_time: str,
event_metadata: dict,
event_privacy_level=2,
channel_id=None
) -> None:
# Creates a guild event using the supplied arguments
# The expected event_metadata format is event_metadata={'location': 'YOUR_LOCATION_NAME'}
# We hard code Entity type to "2" which is Voice channel
# hence we need no Event Metadata
# The required time format is %Y-%m-%dT%H:%M:%S
# Event times can use UTC Offsets! - if you omit, then it will be
# undefined (i.e. UTC / GMT+0)
event_create_url = f'{self.base_api_url}/guilds/{self.guild_id}/scheduled-events'
event_data = json.dumps({
'name': event_name,
'privacy_level': event_privacy_level,
'scheduled_start_time': event_start_time,
'scheduled_end_time': event_end_time,
'description': event_description,
'channel_id': channel_id,
'entity_metadata': event_metadata,
'entity_type': 2
})
try:
await self.post_api(event_create_url,event_data)
except Exception as e:
print(f'DiscordEvents.createguildevent : {e}')
+61
View File
@@ -0,0 +1,61 @@
import json
import aiohttp
# #########################################################
# The class for Discord API integration
# this can be used if certain functions are not supported
# by discord.py
# #########################################################
class DiscordAPI:
# the init constructor stores the authentication-relevant data
# such as the token and the bot url
# guild ID, Client ID, Token, Bot permissions, API Version
def __init__(self,
discord_token: str,
client_id: str,
bot_permissions: int,
api_version: int) -> None:
self.base_api_url = f'https://discord.com/api/v{api_version}'
self.bot_url = f'https://discord.com/api/oauth2/authorize?client_id={client_id}&permissions={bot_permissions}&scope=bot'
self.auth_headers = {
'Authorization':f'Bot {discord_token}',
'User-Agent':f'DiscordBot ({self.bot_url}) Python/3.9 aiohttp/3.8.1',
'Content-Type':'application/json'
}
# get_api does an https get on the api
# it expects json data as result
async def get_api(self, api_url: str) -> list:
async with aiohttp.ClientSession(headers=self.auth_headers) as session:
try:
async with session.get(api_url) as response:
response.raise_for_status()
assert response.status == 200
response_list = json.loads(await response.read())
except Exception as e:
print(f'get_api EXCEPTION: {e}')
finally:
await session.close()
return response_list
# post_api does an https put to the api url
# it expects json data as var which it posts
async def post_api(self, api_url: str, dumped_jsondata: str) -> None:
async with aiohttp.ClientSession(headers=self.auth_headers) as session:
try:
async with session.post(api_url, data=dumped_jsondata) as response:
response.raise_for_status()
assert response.status == 200
except Exception as e:
print(f'post_api EXCEPTION: {e}')
finally:
await session.close()
+73
View File
@@ -0,0 +1,73 @@
import discord
import traceback
import config
# ############################################
# the Subscribe() class is a modal ui dialog
# that let's you select one or multiple
# roles that will be assigned to you.
# The roles are set in the config.AUTO_EVENTS
# variable
# ############################################
class Subscribe(discord.ui.Modal, title='(un)subscribe to Event-Notification'):
# We need to make sure that the config is read at object definition time
# because AUTO_EVENTS might be empty otherwise
if config.AUTO_EVENTS == []:
config.readConfig()
# define the menu options (label=text, vaue=role_id)
menu_options = []
for menu_option in config.AUTO_EVENTS:
menu_options.append(discord.SelectOption(label= menu_option["notify_hint"],
value=menu_option["subscription_role_num"]))
# define the UI dropdown menu Element
Menu = discord.ui.Select(
options = menu_options,
max_values=len(config.AUTO_EVENTS),
min_values=0)
# #####################################
# on_submit is called when the user submits
# the modal. It will then go through
# the menu options and revoke / assign
# the corresponding roles
# #####################################
async def on_submit(self, interaction: discord.Interaction):
role: discord.Role
roles = interaction.user.guild.roles
member: discord.Member
member = interaction.user
# first we remove all roles
for option in self.Menu.options:
role=member.get_role(option.value)
if not (role is None):
await member.remove_roles(role)
# then we assign all selected roles
# unfortunately we need to loop through all rolles
# maybe there is a better solution ?
for option in self.Menu._selected_values:
for role in roles:
if int(option) == int(role.id):
await member.add_roles(role)
await interaction.response.send_message(f'Thanks for using my services !', ephemeral=True)
async def on_error(self, interaction: discord.Interaction, error: Exception) -> None:
await interaction.response.send_message('Oops! Something went wrong.', ephemeral=True)
# Make sure we know what the error actually is
traceback.print_tb(error.__traceback__)
+79
View File
@@ -0,0 +1,79 @@
import discord
import traceback
import config
# ############################################
# the Support() class is a modal ui dialog
# that helps you create a thread in a
# selected channel. It asks for a title
# and a description and then creates
# a Thread in the config.CONFIG["SUPPORT_CHANNEL_ID"]
# channel. It also sends a message to the
# config.CONFIG["IDLE_MESSAGE_CHANNEL_ID"]
# in order to notify everyone that a
# new support message has been created
# ############################################
class Support(discord.ui.Modal, title='Open a support thread'):
# This will be a short input, where the user can enter a title
# for the new thread
theTitle = discord.ui.TextInput(
label='Title',
placeholder='a catchy title for the issue',
)
# This is a longer, paragraph style input, where user can submit
# a description of the problem
theDescription = discord.ui.TextInput(
label='Describe the problem',
style=discord.TextStyle.long,
placeholder='Type in what the problem is...',
required=False,
max_length=300,
)
# ############################################
# on_submit is called when the user submits the
# Modal. This is where we create the thread
# and send all related messages
# ############################################
async def on_submit(self, interaction: discord.Interaction):
# first let's find out which channel we will create the thread in
theGuild = interaction.guild
theChannel : discord.TextChannel
theChannel = theGuild.get_channel(config.CONFIG["SUPPORT_CHANNEL_ID"])
if not (theChannel is None):
try:
# we send a message into that channel that serves as "hook" for the thread
# (if we didn't have a message to hook then the thread would be created
# as private which requires a boost level)
xMsg= await theChannel.send (f"Support Thread for <@{interaction.user.id}>")
newThread=await theChannel.create_thread(name=f"{self.theTitle.value}",message=xMsg,auto_archive_duration=1440)
# next we want to post about the new "ticket" in the IDLE_MESSAGE_CHANNEL
theChannel = theGuild.get_channel(config.CONFIG["IDLE_MESSAGE_CHANNEL_ID"])
if (not (theChannel is None)) and (not (newThread is None)):
xMsg= await theChannel.send (f'I have created a **Support Thread** on behalf of <@{interaction.user.id}> :\n\n <#{newThread.id}> in the <#{config.CONFIG["SUPPORT_CHANNEL_ID"]}>\n\nMaybe you could check in and see if **you** can help ??? \nMany thanks !')
xMsg= await newThread.send (f'<@{interaction.user.id}> describes the problem as follows: \n{self.theDescription.value} \n \n please tag the user on your reply - thank you!' )
except Exception as e:
print(f"Support Error: {e}")
# last but not least we send an ephemeral message to the user
# linking to the created thread
await interaction.response.send_message(f'Your Support Thread has been created here: <#{newThread.id}> Please check if everything is correct.\nThank you for using my services!', ephemeral=True)
async def on_error(self, interaction: discord.Interaction, error: Exception) -> None:
await interaction.response.send_message('Oops! Something went wrong.', ephemeral=True)
# Make sure we know what the error actually is
traceback.print_tb(error.__traceback__)