Posthorn/bot.py

225 lines
7.5 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

import json
import re
# needs the following version of discord.py for recurring events to work: https://github.com/DA-344/d.py/tree/recurrent_events
# uncomment following lines
# import sys
# sys.path.insert(1, '/path/to/Posthorn/vendor/d.py')
import discord
from discord.ext import commands, tasks
from dotenv import load_dotenv
from os import getenv
from github_connector import create_or_update_file, delete_file, similar_exists
load_dotenv("./.env")
DISCORD_TOKEN = getenv("DISCORD_TOKEN")
# Intents are required to listen to events like messages
intents = discord.Intents.default()
intents.message_content = True
intents.members = True
intents.reactions = True
intents.dm_messages = True
# Create an instance of the Bot
bot = commands.Bot(command_prefix="!", intents=intents)
def getCommitMessage(event):
return f"{event.start_time.strftime('%Y-%m-%d')}_{event.name}"
def getFilePath(event):
return f"_events/{event.start_time.strftime('%Y')}/discord-event-{event.id}.md"
def getDay(day):
match day:
case 0:
return 'MO'
case 1:
return 'TU'
case 2:
return 'WE'
case 3:
return 'TH'
case 4:
return 'FR'
case 5:
return 'SA'
case 6:
return 'SU'
case _:
raise Exception(f'no matching day for int {day}')
def getRrule(event):
byDay = False
if hasattr(event.recurrence_rule, "n_weekdays"):
weekdays = event.recurrence_rule.n_weekdays
i = 0
for week, day in weekdays:
comma = ''
if i < len(weekdays) - 1:
comma = ','
byDay = f'{str(week)}{getDay(day)}{comma}'
i += 1
if byDay:
return f"FREQ={event.recurrence_rule.frequency.name.upper()};INTERVAL={event.recurrence_rule.interval};BYDAY={byDay}"
else:
return f"FREQ={event.recurrence_rule.frequency.name.upper()};INTERVAL={event.recurrence_rule.interval}"
def getEventYML(event):
location = "Leibnizstraße 32, 39104 Magdeburg"
if hasattr(event, 'location'):
location = event.location
if hasattr(event, "recurrence_rule") and hasattr(event.recurrence_rule, "frequency") and hasattr(event.recurrence_rule, "interval"):
return f'---\nlayout: event\ntitle: "{event.name}"\nauthor: "Netz39 e.V." \nevent:\n start: {event.start_time.strftime("%Y-%m-%d %H:%M:%S")} \n end: {event.end_time.strftime("%Y-%m-%d %H:%M:%S")} \n organizer: "Netz39 Team <kontakt@netz39.de>" \n location: "{location}"\n rrule: "{getRrule(event)}"\n---\n<!-- event imported from discord manual changes may be overwritten -->\n{event.description}'
else:
return f'---\nlayout: event\ntitle: "{event.name}"\nauthor: "Netz39 e.V." \nevent:\n start: {event.start_time.strftime("%Y-%m-%d %H:%M:%S")} \n end: {event.end_time.strftime("%Y-%m-%d %H:%M:%S")} \n organizer: "Netz39 Team <kontakt@netz39.de>" \n location: "{location}"\n---\n<!-- event imported from discord manual changes may be overwritten -->\n{event.description}'
def getAdmins(guild):
admins = []
for member in guild.members:
if member.guild_permissions.administrator:
admins.append(member.id)
return admins
@tasks.loop(hours=24) # Runs every 7 days
async def check_events():
await bot.wait_until_ready()
for guild in bot.guilds:
scheduledEvents = await guild.fetch_scheduled_events()
for event in scheduledEvents:
await handleEvent(event, True)
async def handleEvent(event, isUpdate = False):
guild = event.guild
channel = guild.system_channel
admins = getAdmins(guild)
eventYML = getEventYML(event)
file_path = getFilePath(event)
commit_message = getCommitMessage(event)
if event.creator_id in admins or similar_exists(event.name):
create_or_update_file(
file_path,
eventYML,
f"new event: {commit_message}",
)
if not isUpdate:
embed = discord.Embed(
title=f"📅 New event on {event.start_time.strftime('%Y-%m-%d')}",
description=f"### {event.name}\n\n{event.description}\n\n",
color=discord.Color.light_embed()
)
if hasattr(event, "cover_image") and event.cover_image:
embed.set_image(url=event.cover_image.url)
embed.add_field(name="📅 Time", value=event.start_time.strftime("%Y-%m-%d %HUhr"), inline=True)
if hasattr(event, 'location'):
embed.add_field(name="📍 Location", value=event.location, inline=True)
eventLink = f"[🔗 Netz39](https://www.netz39.de/events/2025/discord-event-{event.id})";
embed.add_field(value=eventLink, name="", inline=False)
embed.set_footer(text="Dont miss it! ")
await channel.send(embed=embed)
else:
eventJSON = json.dumps(
{
"yml": eventYML,
"path": file_path,
"fileName": commit_message,
"name": event.name,
"date": event.start_time.strftime("%Y-%m-%d"),
},
indent=2,
)
for admin in admins:
user = await bot.fetch_user(event.creator_id)
await sendDm(
admin,
f"{user.name} added a new Event on {event.start_time.strftime('%Y')}: {event.name}. Like this message to approve.\n\n\n```json\n{eventJSON}```",
)
# Event: When the bot is ready
@bot.event
async def on_ready():
print(f"Logged in as {discord.user}")
# Command: !hello
@bot.command()
async def hello(ctx):
await ctx.send(
"Hello, I am a bot to add events from discord to the Netz39 calendar on the website!"
)
@bot.event
async def on_ready():
print(f"Logged in as {bot.user}")
for guild in bot.guilds:
system_channel = guild.system_channel
if system_channel:
await system_channel.send("Hello everyone! I'm online and ready to go! 📅")
await check_events()
@bot.event
async def on_scheduled_event_update(before, after):
if after.status != "canceled":
await handleEvent(after, True)
@bot.event
async def on_scheduled_event_create(event):
await handleEvent(event)
async def sendDm(userID, message):
user = await bot.fetch_user(userID)
await user.send(message)
@bot.event
async def on_reaction_add(reaction, user):
message = reaction.message
channel = message.channel
guild = message.guild
admins = getAdmins(guild)
if not user.id in admins:
return # Ignore non-maintainers
if user.bot:
return # Ignore bot reactions
if (
isinstance(message.channel, discord.DMChannel) and message.author.bot
): # Check if it's a DM and from the bot
if reaction.emoji == "👍":
messageJSON = json.loads(
re.search(
"(?s)(?<=```json).*?(?=```)", reaction.message.content
).group()
)
create_or_update_file(
messageJSON["path"],
messageJSON["yml"],
f"new event: {messageJSON['fileName']}",
)
await channel.send(
f"📅 New event '{messageJSON['name']}' on {messageJSON['date']}."
)
@bot.event
async def on_scheduled_event_delete(event):
delete_file(getFilePath(event))
guild = event.guild
channel = guild.system_channel
await channel.send(f"❌ Event has been canceled: {event.name}")
bot.run(DISCORD_TOKEN)