743 lines
41 KiB
Python
743 lines
41 KiB
Python
#!/usr/bin/python3
|
|
|
|
__author__ = "blueShard (ByteDream)"
|
|
__license__ = "MPL-2.0"
|
|
__version__ = "1.1"
|
|
|
|
# Startscript um zu check, ob python3, pip3 + alle benötigten externen python3 libraries installiert sind (wenn nicht wird das benötigte nachinstalliert), das danach die main.py startet:
|
|
"""
|
|
#!/bin/bash
|
|
|
|
which python3 &> /dev/null
|
|
[ $? -eq 0 ] || apt-get -y install python3
|
|
|
|
which pip3 &> /dev/null
|
|
[ $? -eq 0 ] || apt-get -y install python3-pip
|
|
|
|
python3 -c "import aiogram" &> /dev/null
|
|
[ $? -eq 0 ] || yes | pip3 install aiogram 1> /dev/null
|
|
|
|
python3 -c "import discord" &> /dev/null
|
|
[ $? -eq 0 ] || yes | pip3 install discord 1> /dev/null
|
|
|
|
python3 -c "import webuntis" &> /dev/null
|
|
[ $? -eq 0 ] || yes | pip3 install webuntis 1> /dev/null
|
|
|
|
python3 main.py <discord api token> <telegram api token> <webuntis username> <webuntis password>
|
|
"""
|
|
|
|
import asyncio
|
|
import discord # https://github.com/Rapptz/discord.py
|
|
import logging
|
|
import xml.etree.ElementTree as ET
|
|
from aiogram import Bot, Dispatcher, types # https://github.com/aiogram/aiogram
|
|
from datetime import date, datetime, time, timedelta
|
|
from math import ceil
|
|
from random import choice
|
|
from sys import argv
|
|
from traceback import format_exc
|
|
from webuntis import Session # https://github.com/python-webuntis/python-webuntis
|
|
from xml.dom import minidom
|
|
|
|
# logging.basicConfig(format="[%(asctime)s] %(levelname)s: %(message)s", level=logging.INFO)
|
|
logging.basicConfig(handlers=[logging.StreamHandler(), logging.FileHandler("/var/log/ScheduleAndMoreBot.log", "a+")], format="[%(asctime)s] %(levelname)s: %(message)s", level=logging.INFO)
|
|
|
|
logging.info("Start logging")
|
|
|
|
|
|
class ScheduleAnMoreBot(discord.Client):
|
|
telegram_bot = Bot(token=argv[2])
|
|
dispatcher = Dispatcher(telegram_bot)
|
|
|
|
def __init__(self, ignore_discord: bool = False, **options) -> None:
|
|
super().__init__(**options)
|
|
self.ignore_discord = ignore_discord
|
|
|
|
self.info_file = "infos.txt"
|
|
|
|
self.discord_utils = DiscordUtils()
|
|
|
|
self.discord_channel = None
|
|
|
|
self.telegram_utils = TelegramUtils()
|
|
|
|
self.dispatcher.register_message_handler(self.telegram_private, commands=["private"])
|
|
self.dispatcher.register_message_handler(self.telegram_example, commands=["example"])
|
|
self.dispatcher.register_message_handler(self.telegram_help, commands=["help"])
|
|
self.dispatcher.register_message_handler(self.telegram_add_info, commands=["add_info"])
|
|
self.dispatcher.register_message_handler(self.telegram_info, commands=["info"])
|
|
self.dispatcher.register_message_handler(self.telegram_source, commands=["src", "source"])
|
|
|
|
if self.ignore_discord:
|
|
loop = asyncio.get_event_loop()
|
|
loop.create_task(self.dispatcher.start_polling(self.dispatcher))
|
|
loop.create_task(Checker(None, self.telegram_bot, self.telegram_utils.group_id).main())
|
|
|
|
# ----- Discord ----- #
|
|
|
|
async def on_ready(self) -> None:
|
|
logging.info("Connected to Discord server")
|
|
|
|
async def on_message(self, message: discord.Message) -> None:
|
|
user_input = message.content.lower().strip()
|
|
if not user_input.startswith("$"):
|
|
return
|
|
elif self.discord_channel is None:
|
|
if message.content.lower().strip() == "$start" and message.channel.id == self.discord_utils.channel_id:
|
|
self.discord_channel = message.channel
|
|
await message.channel.send("Der Bot wurde aktiviert")
|
|
if not self.ignore_discord:
|
|
loop = asyncio.get_event_loop()
|
|
loop.create_task(self.dispatcher.start_polling(self.dispatcher))
|
|
loop.create_task(Checker(self.discord_channel, self.telegram_bot, self.telegram_utils.group_id).main())
|
|
else:
|
|
await message.channel.send("Tippe '$start' im richtigen Channel um den Bot zu aktivieren")
|
|
else:
|
|
user_input = user_input[1:]
|
|
# switch-case wäre schon :p
|
|
if user_input == "help":
|
|
await self.discord_help(message)
|
|
elif user_input == "example":
|
|
await self.discord_example(message)
|
|
elif user_input.startswith("add_info"):
|
|
await self.discord_add_info(message)
|
|
elif user_input == "info":
|
|
await self.discord_info(message)
|
|
elif user_input in ["src", "source"]:
|
|
await self.discord_source(message)
|
|
else:
|
|
await message.channel.send("Tippe '$help' für Hilfe")
|
|
|
|
async def discord_help(self, message: discord.Message) -> None:
|
|
"""Zeigt alle Discord Befehle + Information was diese tuhen an"""
|
|
if self.discord_utils.is_valid_channel(message.channel) or self.discord_utils.is_valid_user(self.discord_channel, message.author):
|
|
await message.channel.send(self.discord_utils.help())
|
|
else:
|
|
await message.channel.send(self.discord_utils.not_permitted())
|
|
|
|
async def discord_example(self, message: discord.Message) -> None:
|
|
"""Zeigt Beispiele zu allen Discord Befehlen, wie man diese nutzt"""
|
|
if self.discord_utils.is_valid_channel(message.channel) or self.discord_utils.is_valid_user(self.discord_channel, message.author):
|
|
await message.channel.send(self.discord_utils.example())
|
|
else:
|
|
await message.channel.send(self.discord_utils.not_permitted())
|
|
|
|
async def discord_add_info(self, message: discord.Message) -> None:
|
|
"""Fügt eine neue Info hinzu"""
|
|
if self.discord_utils.is_valid_channel(message.channel) or self.discord_utils.is_valid_user(self.discord_channel, message.author):
|
|
command_no_space = message.content.replace(" ", "")
|
|
infos = Infos()
|
|
full_date = datetime.today()
|
|
today = datetime(full_date.year, full_date.month, full_date.day) # hier wird auf die genau Uhrzeit verzichtet, damit man noch anträge für den selben Tag erstellen kann
|
|
date_for_info = command_no_space[9:19].split("-")
|
|
|
|
for index, x in enumerate(date_for_info):
|
|
if x.startswith("0"):
|
|
date_for_info[index] = x[1:]
|
|
|
|
try:
|
|
if today > datetime(int(date_for_info[2]), int(date_for_info[1]), int(date_for_info[0])):
|
|
await message.channel.send("Das Datum liegt in der Vergangenheit")
|
|
return
|
|
else:
|
|
date = command_no_space[9:19]
|
|
information = message.content.replace("$add_info", "", 1).replace(command_no_space[9:19], "", 1).strip()
|
|
infos.addappend(date, information)
|
|
for embed in self.discord_utils.embed_info(date, information):
|
|
await self.discord_channel.send(embed=embed)
|
|
await self.telegram_bot.send_message(self.telegram_utils.group_id, "Eine neue Info für " + date + " wurde hinzugefügt: " + information)
|
|
logging.info("New entry for date " + date + " was added: " + information)
|
|
except (IndexError, SyntaxError, ValueError):
|
|
await message.channel.send("Es wurde kein richtiges Datum angegeben")
|
|
logging.warning("An error occurred while trying to add a new information:\n" + format_exc())
|
|
|
|
async def discord_info(self, message: discord.Message) -> None:
|
|
"""Zeigt alle Infos an"""
|
|
if self.discord_utils.is_valid_channel(message.channel) or self.discord_utils.is_valid_user(self.discord_channel, message.author):
|
|
infos = Infos()
|
|
|
|
all_colors = [discord.Color.blue(),
|
|
discord.Color.blurple(),
|
|
discord.Color.dark_blue(),
|
|
discord.Color.dark_gold(),
|
|
discord.Color.darker_grey(),
|
|
discord.Color.dark_green(),
|
|
discord.Color.dark_grey(),
|
|
discord.Color.dark_magenta(),
|
|
discord.Color.dark_orange(),
|
|
discord.Color.dark_purple(),
|
|
discord.Color.dark_red(),
|
|
discord.Color.dark_teal(),
|
|
discord.Color.default()]
|
|
choosed_colors = []
|
|
|
|
for child in infos.root:
|
|
info = infos.get(child.tag)
|
|
separator = info.split("~", 1)[0]
|
|
day_infos = info.replace("~", "", 1).split(separator)[1:]
|
|
|
|
if len(choosed_colors) >= len(all_colors):
|
|
choosed_colors = []
|
|
color = choice(all_colors)
|
|
while color in choosed_colors:
|
|
color = choice(all_colors)
|
|
|
|
discord_info = discord.Embed(title="Infos für " + child.tag[1:], color=color)
|
|
# discord_info.set_image(url="https://upload.wikimedia.org/wikipedia/commons/thumb/e/e4/Infobox_info_icon.svg/2000px-Infobox_info_icon.svg.png")
|
|
discord_info.set_thumbnail(url="https://upload.wikimedia.org/wikipedia/commons/thumb/e/e4/Infobox_info_icon.svg/2000px-Infobox_info_icon.svg.png")
|
|
for index, day_info in enumerate(day_infos):
|
|
if len(day_info) > 1000:
|
|
for x in range(0, ceil(len(day_info) / 1000)):
|
|
if x % 6:
|
|
await message.channel.send(embed=discord_info)
|
|
discord_info.clear_fields()
|
|
discord_info.add_field(name=str(index + 1) + "/" + str(x), value=day_info[x * 1000:(x + 1) * 1000], inline=False)
|
|
else:
|
|
discord_info.add_field(name=str(index + 1), value=day_info, inline=False)
|
|
|
|
await message.channel.send(embed=discord_info)
|
|
|
|
async def discord_source(self, message: discord.Message) -> None:
|
|
"""Stellt den Source Code zu Verfügung"""
|
|
await message.channel.send(file=discord.File("main.py", filename="main.py"))
|
|
|
|
# ----- Telegram ----- #
|
|
|
|
async def telegram_private(self, message: types.Message) -> None:
|
|
"""Fügt einen Telegram Nutzer zur liste hinzu, damit dieser per DM mit dem Bot interagieren"""
|
|
if self.telegram_utils.is_valid_group(message.chat):
|
|
if not self.telegram_utils.is_private_user(message.from_user):
|
|
user_id = message.from_user.id
|
|
self.telegram_utils.private_users_id.append(user_id)
|
|
with open(self.telegram_utils.private_users_file, "a+") as file:
|
|
file.write(str(user_id) + ";")
|
|
file.close()
|
|
await message.answer("Neuer Nutzer wurde eingetragen")
|
|
logging.info("New private telegram user registered")
|
|
else:
|
|
await message.answer(self.telegram_utils.not_permitted())
|
|
|
|
async def telegram_help(self, message: types.Message) -> None:
|
|
"""Zeigt alle Telegram Befehle + Information was diese tuhen an"""
|
|
if self.telegram_utils.is_valid_group(message.chat) or self.telegram_utils.is_private_user(message.from_user):
|
|
await message.answer(self.telegram_utils.help(), parse_mode="MarkdownV2")
|
|
else:
|
|
await message.answer(self.telegram_utils.not_permitted())
|
|
|
|
async def telegram_example(self, message: types.Message) -> None:
|
|
"""Zeigt Beispiele zu allen Telegram Befehlen, wie man diese nutzt"""
|
|
if self.telegram_utils.is_valid_group(message.chat) or self.telegram_utils.is_private_user(message.from_user):
|
|
await message.answer(self.telegram_utils.example(), parse_mode="MarkdownV2")
|
|
else:
|
|
await message.answer(self.telegram_utils.not_permitted())
|
|
|
|
async def telegram_add_info(self, message: types.Message) -> None:
|
|
"""Fügt eine neue Info hinzu"""
|
|
if self.telegram_utils.is_valid_group(message.chat) or self.telegram_utils.is_private_user(message.from_user):
|
|
infos = Infos()
|
|
message_no_space = message.text.replace(" ", "")
|
|
full_date = datetime.today()
|
|
today = datetime(full_date.year, full_date.month, full_date.day) # hier wird auf die genau Uhrzeit verzichtet, damit man noch anträge für den selben Tag erstellen kann
|
|
date_for_info = message_no_space[9:19].split("-")
|
|
for index, x in enumerate(date_for_info):
|
|
if x.startswith("0"):
|
|
date_for_info[index] = x[1:]
|
|
|
|
try:
|
|
if today > datetime(int(date_for_info[2]), int(date_for_info[1]), int(date_for_info[0])):
|
|
await message.answer("Das Datum liegt in der Vergangenheit")
|
|
return
|
|
else:
|
|
date = message_no_space[9:19]
|
|
information = message.text.replace("/add_info", "", 1).replace(date, "", 1).strip()
|
|
infos.addappend(date, information)
|
|
await self.telegram_bot.send_message(self.telegram_utils.group_id, "Eine neue Info für " + date + " wurde hinzugefügt: " + information)
|
|
for embed in self.discord_utils.embed_info(date, information):
|
|
await self.discord_channel.send(embed=embed)
|
|
logging.info("New entry for date " + date + " was added: " + information)
|
|
except (IndexError, SyntaxError, ValueError):
|
|
await message.answer("Es wurde kein richtiges Datum angegeben")
|
|
else:
|
|
await message.answer(self.telegram_utils.not_permitted())
|
|
|
|
async def telegram_info(self, message: types.Message) -> None:
|
|
"""Zeigt alle Infos an"""
|
|
if self.telegram_utils.is_valid_group(message.chat) or self.telegram_utils.is_private_user(message.from_user):
|
|
infos = Infos()
|
|
information = ""
|
|
|
|
for child in infos.root:
|
|
info = infos.get(child.tag)
|
|
info.replace(info.split("~", 1)[0], "\n\n")
|
|
information = information + child.tag[1:] + ": " + info.split("~", 1)[1]
|
|
await message.answer(information)
|
|
information = ""
|
|
else:
|
|
await message.answer(self.telegram_utils.not_permitted())
|
|
|
|
async def telegram_source(self, message: types.Message) -> None:
|
|
"""Stellt den Source Code zu Verfügung"""
|
|
if self.telegram_utils.is_valid_group(message.chat) or self.telegram_utils.is_private_user(message.from_user):
|
|
await message.answer_document(document=open("main.py", "rb"))
|
|
else:
|
|
await message.answer(self.telegram_utils.not_permitted())
|
|
|
|
|
|
class DiscordUtils:
|
|
|
|
def __init__(self) -> None:
|
|
self.channel_id = 746369803941576784
|
|
# Test: 746477001237594174
|
|
|
|
def embed_info(self, date, info) -> list:
|
|
"""Erstellt discord embeds für die gegeben info"""
|
|
return_list = []
|
|
all_colors = [discord.Color.blue(),
|
|
discord.Color.blurple(),
|
|
discord.Color.dark_blue(),
|
|
discord.Color.dark_gold(),
|
|
discord.Color.darker_grey(),
|
|
discord.Color.dark_green(),
|
|
discord.Color.dark_grey(),
|
|
discord.Color.dark_magenta(),
|
|
discord.Color.dark_orange(),
|
|
discord.Color.dark_purple(),
|
|
discord.Color.dark_red(),
|
|
discord.Color.dark_teal(),
|
|
discord.Color.default()]
|
|
choosed_colors = []
|
|
|
|
if len(choosed_colors) >= len(all_colors):
|
|
choosed_colors = []
|
|
color = choice(all_colors)
|
|
while color in choosed_colors:
|
|
color = choice(all_colors)
|
|
|
|
discord_info = discord.Embed(title="Eine neue Info für " + date + " wurde hinzugefügt", color=color)
|
|
# discord_info.set_image(url="https://upload.wikimedia.org/wikipedia/commons/thumb/e/e4/Infobox_info_icon.svg/2000px-Infobox_info_icon.svg.png")
|
|
discord_info.set_thumbnail(url="https://upload.wikimedia.org/wikipedia/commons/thumb/e/e4/Infobox_info_icon.svg/2000px-Infobox_info_icon.svg.png")
|
|
if len(info) > 1000:
|
|
for x in range(0, ceil(len(info) / 1000)):
|
|
if x % 6:
|
|
return_list.append(discord_info)
|
|
discord_info.clear_fields()
|
|
discord_info.add_field(name="Info" + "/" + str(x), value=info[x * 1000:(x + 1) * 1000], inline=False)
|
|
else:
|
|
discord_info.add_field(name="Info", value=info, inline=False)
|
|
|
|
return_list.append(discord_info)
|
|
|
|
return return_list
|
|
|
|
def example(self) -> str:
|
|
"""Discord Text, der Beispiele zu allen Befehlen zeigt"""
|
|
example_text = "```\n" \
|
|
"$start $start\n" \
|
|
"$help $help\n" \
|
|
"$example $example\n" \
|
|
"$add_info [dd-mm-yyyy] [info] $add_info 01-01-2222 Eine einfache test Info\n" \
|
|
"$info $info\n" \
|
|
"$src / $source $src\n" \
|
|
"```"
|
|
return example_text
|
|
|
|
def help(self) -> str:
|
|
"""Discord Text, der Hilfe zu allen Befehlen zeigt"""
|
|
help_text = "```\n" \
|
|
"DM (direct message) = Nur per Direktnachticht ausführbar\n" \
|
|
"SC (source channel) = Nur vom Channel von dem aus der Bot gestartet wurde ausführbar\n" \
|
|
"EV (everywhere) = Von überall ausführbar\n\n" \
|
|
"Befehlsname Von wo ausführbar Beschreibung\n\n" \
|
|
"$start SC Startet den Bot\n\n" \
|
|
"$help EV Zeigt Hilfe zu den vorhanden Befehlen an\n" \
|
|
"$example EV Zeigt beispiele für jeden Befehl\n" \
|
|
"$add_info [dd-mm-yyyy] [info] EV Fügt neue Informationen zu einem bestimmten Tag hinzu\n" \
|
|
"$info EV Gibt eingetragene infos wieder\n" \
|
|
"$src / $source EV Stellt die Datei mit dem Quellcode zu Verfügung\n" \
|
|
"```"
|
|
return help_text
|
|
|
|
def is_valid_channel(self, channel: discord.TextChannel) -> bool:
|
|
"""Checkt, ob der gegebene Channel der Channel ist, auf dem der Bot aktiv sein soll"""
|
|
try:
|
|
if channel.id == self.channel_id:
|
|
return True
|
|
else:
|
|
return False
|
|
except AttributeError:
|
|
return False
|
|
|
|
def is_valid_user(self, channel: discord.TextChannel, user: discord.User) -> bool:
|
|
"""Überprüft, ob der Nutzer auf dem Discord Server Mitglied ist"""
|
|
print(user.id, channel.members)
|
|
try:
|
|
for member in channel.members:
|
|
if user.id == member.id:
|
|
return True
|
|
except AttributeError:
|
|
logging.warning("Attribute error occurred while trying to check if discord user is valid")
|
|
|
|
return False
|
|
|
|
def not_permitted(self) -> str:
|
|
"""Info, wenn eine nicht berechtigte Person einen Discord Befehl ausführt"""
|
|
return "Nur Personen, die Mitglieder auf dem Discord Server sind, haben Zugriff auf die Befehle"
|
|
|
|
|
|
class TelegramUtils:
|
|
|
|
def __init__(self) -> None:
|
|
self.group_id = -384078711
|
|
self.private_users_file = "private_users.scv"
|
|
self.private_users_id = open(self.private_users_file, "r+").readline().split(";")
|
|
|
|
def example(self) -> str:
|
|
"""Telegram Text, der Beispiele zu allen Befehlen zeigt"""
|
|
example_text = "```\n" \
|
|
"/start\n" \
|
|
"/start\n\n" \
|
|
"/help\n" \
|
|
"/help\n\n" \
|
|
"/example\n" \
|
|
"/example\n\n" \
|
|
"/add_info [dd-mm-yyyy] [info]\n" \
|
|
"/add_info 01-01-2222 Eine einfache test Info\n\n" \
|
|
"/info\n" \
|
|
"/info\n\n" \
|
|
"/src or /source\n" \
|
|
"/src\n" \
|
|
"```"
|
|
return example_text
|
|
|
|
def help(self) -> str:
|
|
"""Discord Text, der Hilfe zu allen Befehlen zeigt"""
|
|
help_text = "```\n" \
|
|
"DM (direct message) = Nur per Direktnachticht ausführbar\n" \
|
|
"GR (group) = Nur vom Channel von dem aus der Bot gestartet wurde ausführbar\n" \
|
|
"EV (everywhere) = Von überall ausführbar\n\n" \
|
|
"/private\n" \
|
|
"GR\n" \
|
|
"Nutzer bekommt Zugriff auf Befehle, die per DM ausgeführt werden können\n\n\n" \
|
|
"/help\n" \
|
|
"EV\n" \
|
|
"Zeigt Hilfe zu den vorhanden Befehlen an\n\n" \
|
|
"/example\n" \
|
|
"EV\n" \
|
|
"Zeigt Hilfe zu den vorhanden Befehlen an\n\n" \
|
|
"/add_info [dd-mm-yyyy] [info]\n" \
|
|
"EV\n" \
|
|
"Fügt neue Informationen zu einem bestimmten Tag hinzu\n\n" \
|
|
"/info\n" \
|
|
"EV\n" \
|
|
"Gibt eingetragene Infos wieder\n\n\n" \
|
|
"/src or /source\n" \
|
|
"EV\n" \
|
|
"Stellt die Datei mit dem Quellcode zu Verfügung\n" \
|
|
"```"
|
|
return help_text
|
|
|
|
def is_private_user(self, user: types.User) -> bool:
|
|
"""Überprüft, ob der Nutzer '/private' in der Gruppe eingegeben hat"""
|
|
if str(user.id) in self.private_users_id:
|
|
return True
|
|
else:
|
|
return False
|
|
|
|
def is_valid_group(self, chat: types.Chat) -> bool:
|
|
"""Checkt, ob die gegeben Gruppe die Gruppe ist, worin der Bot aktiv sein soll"""
|
|
if chat.id == self.group_id:
|
|
return True
|
|
else:
|
|
return False
|
|
|
|
def not_permitted(self) -> str:
|
|
"""Info, wenn eine nicht berechtigte Person einen Telegram Befehl ausführt"""
|
|
return "Gebe '/private' in der Gruppe ein, um Zugriff auf Befehle, die per DM ausgeführt werden können, zu erhalten"
|
|
|
|
|
|
class Infos: # wird eventuell in Zukunft durch ein lua programm ersetzt
|
|
|
|
def __init__(self, info_file: str = "infos.xml") -> None:
|
|
self.info_file = info_file
|
|
self.root = ET.fromstring("".join([item.replace("\n", "").strip() for item in [line for line in open(info_file, "r")]]))
|
|
|
|
def __create_separator(self, text: str) -> str:
|
|
"""Erstellt ein separator"""
|
|
indicator = "^|^"
|
|
choices = ("§", "!", "^")
|
|
|
|
while True:
|
|
if indicator in text:
|
|
list_choice = choice(choices)
|
|
splitted_indicator = indicator.split("|")
|
|
indicator = splitted_indicator[0] + list_choice + "|" + list_choice + splitted_indicator[1]
|
|
else:
|
|
return indicator
|
|
|
|
def _prettify(self, string: str = None) -> str:
|
|
"""Macht den XML Tree lesbarer für Menschis^^"""
|
|
if string is None:
|
|
reparsed = minidom.parseString(ET.tostring(self.root, "utf-8"))
|
|
else:
|
|
reparsed = minidom.parseString(string)
|
|
pre_output = reparsed.toprettyxml(indent=" ")
|
|
return "\n".join(pre_output.split("\n")[1:])
|
|
|
|
def addappend(self, date_: str, text: str) -> None:
|
|
"""Fügt einen neuen Eintrag zum gegebenen Datum hinzu"""
|
|
date_ = "_" + date_
|
|
for child in self.root:
|
|
if child.tag == date:
|
|
child_text = child.text
|
|
old_separator = child.attrib["separator"]
|
|
new_separator = self.__create_separator(child_text + text)
|
|
child.text = child.text.replace(old_separator, new_separator) + new_separator + text
|
|
child.attrib["separator"] = new_separator
|
|
self.write()
|
|
return
|
|
|
|
new_entry = ET.Element(date_)
|
|
new_entry.text = text
|
|
new_entry.attrib["separator"] = self.__create_separator(text)
|
|
|
|
self.root.append(new_entry)
|
|
self.write()
|
|
|
|
def delete(self, date_: str) -> None:
|
|
"""Löscht alle Einträge an dem gegeben Datum"""
|
|
for child in self.root:
|
|
if child.tag == date_:
|
|
self.root.remove(child)
|
|
self.write()
|
|
return
|
|
|
|
def get(self, date_: str) -> str:
|
|
"""Gibt alle Einträge an dem gegeben Datum zurück"""
|
|
for child in self.root:
|
|
if child.tag == date_:
|
|
return child.attrib["separator"] + "~" + child.text
|
|
|
|
return ""
|
|
|
|
def write(self) -> None:
|
|
"""Schreibt den XML Tree in die Datei"""
|
|
with open(self.info_file, "w+") as file:
|
|
file.write(self._prettify())
|
|
file.close()
|
|
|
|
|
|
class Checker:
|
|
|
|
def __init__(self, discord_channel: discord.TextChannel, telegram_bot: Bot, telegram_group_id: int):
|
|
self.discord_channel = discord_channel
|
|
self.telegram_bot = telegram_bot
|
|
self.telegram_group_id = telegram_group_id
|
|
|
|
self.lessons = {"1": [time(8, 0,), time(8, 45)],
|
|
"2": [time(8, 45), time(9, 30)],
|
|
"3": [time(9, 45), time(10, 30)],
|
|
"4": [time(10, 30), time(11, 15)],
|
|
"5": [time(11, 45), time(12, 30)],
|
|
"6": [time(12, 30), time(13, 15)],
|
|
"7": [time(13, 30), time(14, 15)],
|
|
"8": [time(14, 15), time(15, 0)]}
|
|
|
|
self.all_cancelled_lessons_thursday = {}
|
|
self.all_ignored_lessons_thursday = {}
|
|
self.which_thursday = date.today()
|
|
self.all_cancelled_lessons_friday = {}
|
|
self.all_ignored_lessons_friday = {}
|
|
self.which_friday = date.today()
|
|
|
|
self.session: Session = None
|
|
|
|
async def __check_and_send_cancelled_lessons(self, date_to_check: date) -> None: # die methode ist etwas schwer zu lesen
|
|
"""Überprüft, ob Stunden ausfallen / verlegt wurden und gibt das Ergebnis (wenn es eins gibts) in Discord und Telegram wieder"""
|
|
try:
|
|
embed = None
|
|
all_embed_fields = {}
|
|
all_telegram_messages = {}
|
|
telegram_message = ""
|
|
|
|
if date_to_check.weekday() == 3:
|
|
already_cancelled_lessons: dict = self.all_cancelled_lessons_thursday
|
|
all_ignored_lessons: dict = self.all_ignored_lessons_thursday
|
|
weekday_in_german = "Donnerstag"
|
|
elif date_to_check.weekday() == 4:
|
|
already_cancelled_lessons: dict = self.all_cancelled_lessons_friday
|
|
all_ignored_lessons: dict = self.all_ignored_lessons_friday
|
|
weekday_in_german = "Freitag"
|
|
else:
|
|
raise ValueError('date_to_check (datetime.date) must be thursday or friday')
|
|
timetable = self.session.timetable(start=date_to_check, end=date_to_check, klasse=2015)
|
|
|
|
for lesson in timetable:
|
|
lesson_number = str(lesson.start.time().strftime("%H:%M")) + " Uhr - " + str(lesson.end.time().strftime("%H:%M") + " Uhr")
|
|
for lesson_num, lesson_time in self.lessons.items():
|
|
if lesson_time[0] == lesson.start.time():
|
|
lesson_number = lesson_num
|
|
break
|
|
|
|
embed_title = "Stunden Ausfall Information für " + weekday_in_german + ", den " + date_to_check.strftime("%d.%m.%Y")
|
|
|
|
if lesson.code == "irregular" and lesson_number not in all_ignored_lessons.keys() and lesson.teachers not in all_ignored_lessons.values():
|
|
embed = discord.Embed(title=embed_title, color=discord.Color.from_rgb(77, 255, 77))
|
|
for lesson1 in timetable:
|
|
if lesson.teachers == lesson1.teachers and lesson.start is not lesson1.start and lesson1.code == "cancelled":
|
|
lesson1_number = str(lesson.start.time().strftime("%H:%M")) + " Uhr - " + str(lesson.end.time().strftime("%H:%M") + " Uhr")
|
|
for lesson_num, lesson_time in self.lessons.items():
|
|
if lesson_time[0] == lesson1.start.time():
|
|
lesson1_number = lesson_num
|
|
break
|
|
|
|
for number in list(all_embed_fields.keys()): # wenn es ohne list gemacht werden würde, würde ein RuntimeError kommen
|
|
if number in [lesson_number, lesson1_number]:
|
|
del all_embed_fields[number]
|
|
del all_telegram_messages[number]
|
|
|
|
if len(lesson1_number) == 1:
|
|
all_embed_fields[lesson_number] = {lesson1_number + ". Stunde wurde zur " + lesson_number +
|
|
". Stunde umverlegt": "Die " + lesson1_number + ". Stunde (" + lesson1.start.time().strftime("%H:%M") + " Uhr - " + lesson1.end.time().strftime("%H:%M") + " Uhr) bei " + \
|
|
", ".join([teacher.long_name for teacher in lesson.teachers]) + " wurde zur " + lesson_number + ". Stunde (" + lesson.start.time().strftime("%H:%M") + \
|
|
" Uhr - " + lesson.end.time().strftime("%H:%M") + " Uhr) umverlegt"}
|
|
all_telegram_messages[lesson_number] = "Die " + lesson1_number + ". Stunde (" + lesson1.start.time().strftime("%H:%M") + " Uhr - " + lesson1.end.time().strftime("%H:%M") + " Uhr) bei " + \
|
|
", ".join([teacher.long_name for teacher in lesson.teachers]) + " wurde zur " + lesson_number + ". Stunde (" + lesson.start.time().strftime("%H:%M") + \
|
|
" Uhr - " + lesson.end.time().strftime("%H:%M") + " Uhr) umverlegt"
|
|
else:
|
|
all_embed_fields[lesson_number] = {"Die Stunde " + lesson1_number + " wurde zur Stunde" + lesson_number +
|
|
" umverlegt": "Die Stunde " + lesson1_number + " bei " + ", ".join([teacher.long_name for teacher in lesson.teachers]) + " wurde zur Stunde " + lesson_number + " umverlegt"}
|
|
all_telegram_messages[lesson_number] = "Die Stunde " + lesson1_number + " bei " + ", ".join([teacher.long_name for teacher in lesson.teachers]) + " wurde zur Stunde " + lesson_number + " umverlegt"
|
|
|
|
all_ignored_lessons[lesson_number] = lesson.teachers
|
|
all_ignored_lessons[lesson1_number] = lesson.teachers
|
|
elif lesson.code == "cancelled":
|
|
embed = discord.Embed(title=embed_title, color=discord.Color.from_rgb(255, 0, 0))
|
|
if lesson_number not in already_cancelled_lessons.keys() and lesson_number not in all_ignored_lessons.keys():
|
|
already_cancelled_lessons[lesson_number] = lesson.teachers
|
|
if len(lesson_number) == 1:
|
|
all_embed_fields[lesson_number] = {"Ausfall " + str(lesson_number) + ". Stunde (" + lesson.start.time().strftime("%H:%M") + " Uhr - " +
|
|
lesson.end.time().strftime("%H:%M") + " Uhr)": "Ausfall bei " + ", ".join([teacher.long_name for teacher in lesson.teachers]) + " in " +
|
|
", ".join([subject.long_name for subject in lesson.subjects])}
|
|
all_telegram_messages[lesson_number] = "Ausfall am " + weekday_in_german + ", den " + date_to_check.strftime("%d.%m.%Y") + " in der " + lesson_number + " Stunde bei " +\
|
|
", ".join([teacher.long_name for teacher in lesson.teachers]) + " in " + ", ".join([subject.long_name for subject in lesson.subjects]) + "\n\n"
|
|
else:
|
|
all_embed_fields[lesson_number] = {"Ausfall " + lesson_number: "Ausfall bei " + ", ".join([teacher.long_name for teacher in lesson.teachers]) + " in " + ", ".join([subject.long_name for subject in lesson.subjects])}
|
|
all_telegram_messages[lesson_number] = "Ausfall " + lesson_number + " am " + weekday_in_german + ", den " + date_to_check.strftime("%d.%m.%Y") + " bei " +\
|
|
", ".join([teacher.long_name for teacher in lesson.teachers]) + " in " + ", ".join([subject.long_name for subject in lesson.subjects]) + "\n\n"
|
|
elif lesson_number in already_cancelled_lessons.keys():
|
|
embed = discord.Embed(title=embed_title, color=discord.Color.from_rgb(77, 255, 77))
|
|
if lesson.teachers in already_cancelled_lessons.values():
|
|
del already_cancelled_lessons[lesson_number]
|
|
if len(lesson_number) == 1:
|
|
all_embed_fields[lesson_number] = {"KEIN Ausfall " + str(lesson_number) + ". Stunde (" + lesson.start.time().strftime("%H:%M") + " Uhr - " +
|
|
lesson.end.time().strftime("%H:%M") + " Uhr)": "KEIN Ausfall bei " + ", ".join([teacher.long_name for teacher in lesson.teachers]) + " in " +
|
|
", ".join([subject.long_name for subject in lesson.subjects])}
|
|
all_telegram_messages[lesson_number] = "KEIN Ausfall am " + weekday_in_german + ", den " + date_to_check.strftime("%d.%m.%Y") + " in der " + lesson_number + " Stunde bei " + \
|
|
", ".join([teacher.long_name for teacher in lesson.teachers]) + " in " + ", ".join([subject.long_name for subject in lesson.subjects]) + "\n\n"
|
|
else:
|
|
all_embed_fields[lesson_number] = {"KEIN Ausfall " + lesson_number: "KEIN Ausfall bei " + ", ".join([teacher.long_name for teacher in lesson.teachers]) + " in " +
|
|
", ".join([subject.long_name for subject in lesson.subjects])}
|
|
all_telegram_messages[lesson_number] = "KEIN Ausfall " + lesson_number + " am " + weekday_in_german + ", den " + date_to_check.strftime("%d.%m.%Y") + " bei " +\
|
|
", ".join([teacher.long_name for teacher in lesson.teachers]) + " in " + ", ".join([subject.long_name for subject in lesson.subjects]) + "\n\n"
|
|
|
|
if date_to_check.weekday() == 3:
|
|
self.all_cancelled_lessons_thursday = already_cancelled_lessons
|
|
self.all_ignored_lessons_thursday = all_ignored_lessons
|
|
elif date_to_check.weekday() == 4:
|
|
self.all_cancelled_lessons_friday = already_cancelled_lessons
|
|
self.all_ignored_lessons_friday = all_ignored_lessons
|
|
|
|
if len(all_telegram_messages) != 0 and len(all_embed_fields) != 0:
|
|
for number, content in all_embed_fields.items():
|
|
embed.add_field(name=list(content.keys())[0], value=list(content.values())[0])
|
|
telegram_message += all_telegram_messages[number]
|
|
await self.discord_channel.send(embed=embed)
|
|
await self.telegram_bot.send_message(self.telegram_group_id, telegram_message)
|
|
logging.info("Send message(s) (content from telegram message): " + telegram_message.replace("\n\n", "\n"))
|
|
except Exception:
|
|
logging.warning("An unexpected error occured, while trying to check the schedule\n" + format_exc())
|
|
await self.discord_channel.send("Ein Fehler trat auf, während der Stundenplan auf Veränderungen überprüft wurde. Siehe Logs für Details")
|
|
await self.telegram_bot.send_message(self.telegram_group_id, "Ein Fehler trat auf, während der Stundenplan auf veränderungen überprüft wurde. Siehe Logs für Details")
|
|
|
|
async def main(self, check_time: int = 60 * 60) -> None:
|
|
"""Überprüft nach einer gewissen Zeit immer wieder, ob veraltete Infos exestieren"""
|
|
try:
|
|
self.session = Session(server="asopo.webuntis.com",
|
|
username=argv[3],
|
|
password=argv[4],
|
|
school="Konrad-Zuse-schule",
|
|
useragent="")
|
|
try:
|
|
self.session.login()
|
|
except Exception as e:
|
|
logging.warning("A login error occurred (" + "\n".join([arg for arg in e.args]) + ")")
|
|
await self.discord_channel.send("Ein (Web)Untis Loginfehler ist aufgetreten. Siehe Logs für Details")
|
|
await self.telegram_bot.send_message(self.telegram_group_id, "Ein (Web)Untis Loginfehler ist aufgetrten. Siehe Logs für Details")
|
|
except IndexError:
|
|
logging.warning("No username and / or password for webuntis is / are given")
|
|
await self.discord_channel.send("Ein (Web)Untis Loginfehler ist aufgetreten. Siehe Logs für Details")
|
|
await self.telegram_bot.send_message(self.telegram_group_id, "Ein (Web)Untis Loginfehler ist aufgetrten. Siehe Logs für Details")
|
|
except Exception:
|
|
logging.warning("An exception for the webuntis session occurred:\n" + format_exc())
|
|
await self.discord_channel.send("Ein (Web)Untis Loginfehler ist aufgetreten. Siehe Logs für Details")
|
|
await self.telegram_bot.send_message(self.telegram_group_id, "Ein (Web)Untis Loginfehler ist aufgetrten. Siehe Logs für Details")
|
|
|
|
while True:
|
|
if self.session is not None:
|
|
today = date.today()
|
|
today_weekday = today.weekday()
|
|
if today_weekday == 3: # donnerstag
|
|
await self.__check_and_send_cancelled_lessons(today + timedelta(days=1))
|
|
|
|
if datetime.now().hour > 12: # wenn es über 12 uhr ist, wird angefangen nach ausfall in der nächsten woche zu suchen
|
|
if self.which_thursday < today:
|
|
self.all_cancelled_lessons_thursday = {}
|
|
self.all_ignored_lessons_thursday = {}
|
|
await self.__check_and_send_cancelled_lessons(today + timedelta(days=7))
|
|
else:
|
|
await self.__check_and_send_cancelled_lessons(today + timedelta(days=7))
|
|
else:
|
|
await self.__check_and_send_cancelled_lessons(today)
|
|
|
|
elif today_weekday == 4: # freitag
|
|
await self.__check_and_send_cancelled_lessons(today + timedelta(days=6))
|
|
|
|
if datetime.now().hour > 12: # wenn es über 12 uhr ist, wird angefangen nach ausfall in der nächsten woche zu gucken
|
|
if self.which_friday < today:
|
|
self.all_cancelled_lessons_friday = {}
|
|
self.all_cancelled_lessons_friday = {}
|
|
await self.__check_and_send_cancelled_lessons(today + timedelta(days=7))
|
|
else:
|
|
await self.__check_and_send_cancelled_lessons(today + timedelta(days=7))
|
|
else:
|
|
await self.__check_and_send_cancelled_lessons(today)
|
|
|
|
else:
|
|
for day in range(1, 6):
|
|
new_day = today + timedelta(days=day)
|
|
if new_day.weekday() in [3, 4]:
|
|
await self.__check_and_send_cancelled_lessons(new_day)
|
|
|
|
try:
|
|
infos = Infos()
|
|
today = datetime.today()
|
|
for child in infos.root:
|
|
child_date = child.tag[1:].split("-")
|
|
for index, x in enumerate(child_date):
|
|
if x.startswith("0"):
|
|
child_date[index] = x[1:]
|
|
if today > datetime(int(child_date[2]), int(child_date[1]), int(child_date[0]) + 1):
|
|
infos.delete(child.tag)
|
|
logging.info("Removed informations for day " + child.tag)
|
|
logging.info("Checked for old informations")
|
|
except Exception:
|
|
logging.warning("An unexpected error occured, while trying to check the infos\n" + format_exc())
|
|
await self.discord_channel.send("Ein Fehler trat auf, während die Infos Datei auf alte Daten überprüft wurde. Siehe Logs für Details")
|
|
await self.telegram_bot.send_message(self.telegram_group_id, "Ein Fehler trat auf, während die Infos Datei auf alte Daten überprüft wurde. Siehe Logs für Details")
|
|
await asyncio.sleep(check_time) # schläft die gegebene Zeit und checkt dann wieder von neuem, ob sich was am Stundenplan geändert hat / ob Infos gelöscht werden können
|
|
|
|
|
|
if __name__ == '__main__':
|
|
schedule_and_more_bot = ScheduleAnMoreBot()
|
|
schedule_and_more_bot.run(argv[1])
|