#!/usr/bin/python3 # -*- coding: utf-8 -*- # since 3.5: type hints # def ( : import asyncio # write concurrent code (not threading. not multiprocessing) # it's cooperative multitasking, no parallelism # coroutines: suspend execution before return and pass control to another coroutine # use await only in async functions import datetime import json import signal import sys import traceback import urllib.request import xml.etree.ElementTree as ET from enum import Enum from re import sub from time import time from typing import Any from urllib.error import HTTPError import telepot from babel.dates import format_date from googletrans import Translator from telepot.aio import DelegatorBot from telepot.aio.delegate import per_chat_id, create_open, pave_event_space from telepot.aio.loop import MessageLoop from telepot.delegate import include_callback_query_chat_id from telepot.exception import BotWasBlockedError from telepot.namedtuple import InlineKeyboardMarkup, InlineKeyboardButton class Essen: """Class which represents a Essen-Object""" def __init__(self, name, preis, kategorie) -> None: """ Initialize a Essen-Object with given values Parameters: name (string): name of the meal preis (float): price of the meal kategorie (string): category of the meal Returns: None """ self.name = name self.preis = preis self.kategorie = kategorie return None def __str__(self) -> str: """ Formats the attribute of the Essen-Object to a string Parameters: None Returns: _ (string): formatted String of attributes """ if self.preis > 0: return str("*%s*: `%s` (%.2f €)" % (self.kategorie, self.name, self.preis)) else: return str("*%s*: `%s`" % (self.kategorie, self.name)) class Language(Enum): GERMAN = "de" ENGLISH = "en" GERMAN_ENGLISH = "de+en" config_filename = "config.json" info_str = ''' *Inoffizieller Mensa-Bot der Hochschule Mittweida*\nDieser Bot versendet jeden Tag um 10 Uhr den aktuellen Mensa-Speiseplan. Er wird über /start für den aktuellen Chat oder die aktuelle Gruppe gestartet, /stop beendet ihn wieder. Mit /essen, /mensa und /speiseplan (optional gefolgt von _en_ oder _de_) kann der aktuelle Speiseplan manuell abgerufen werden. Mit /settings kann (von Gruppenadmins) die Sprache verändert werden.\n\n_Haftungsausschluss: Dieser Bot steht in keiner Verbindung mit der Hochschule Mittweida oder dem Studentenwerk Freiberg. Alle Angaben ohne Gewähr._\n\nGrafik bereitgestellt von [ vecteezy.com](https://de.vecteezy.com) ''' status = "" essen = [] essen_eng = [] var = True datum = None last_updated = 0.0 botID = -1 class EnumEncode(json.JSONEncoder): def default(self, obj): if type(obj) is Language: return str(obj.name) return json.JSONEncoder.default(self, obj) class HSMensaW(telepot.aio.helper.ChatHandler): async def on_chat_message(self, msg: dict) -> None: global config, essen, status, var, logging_enabled content_type, chat_type, chat_id = telepot.glance(msg) if content_type == "text": text = str(msg["text"]).lower() if text.startswith("/start"): if chat_id not in ids: ids[chat_id] = Language.GERMAN config['ids'] = ids await send_message(bot_obj=bot, chat_id=chat_id, msg="Bot wurde aktiviert") chat = get_chat_name(msg) await send_status("Bot aktiviert für Chat %s (ID: %i)" % (chat, chat_id)) write_config() else: await send_message(bot_obj=bot, chat_id=chat_id, msg="Bot war bereits aktiviert") elif text.startswith("/stop"): if ids.pop(chat_id, None) is not None: config['ids'] = ids await send_message(bot_obj=bot, chat_id=chat_id, msg="Bot wurde deaktiviert") chat = get_chat_name(msg) await send_status("Bot deaktiviert für Chat %s (ID: %i)" % (chat, chat_id)) write_config() else: await send_message(bot_obj=bot, chat_id=chat_id, msg="Bot war nicht aktiv") elif text.startswith("/essen") or text.startswith("/mensa") or text.startswith("/speiseplan"): chat = get_chat_name(msg) await send_status("Essen angefordert für Chat %s (ID: %i)" % (chat, chat_id)) await get_essen(False) if len(essen) == 0: if var: await send_message(bot_obj=bot, chat_id=chat_id, msg="Es ist ein Fehler aufgetreten. Bitte später erneut versuchen.") else: await send_message(bot_obj=bot, chat_id=chat_id, msg="Für heute ist leider kein Speiseplan verfügbar.") else: if text.endswith(" en"): language = Language.ENGLISH elif text.endswith(" de"): language = Language.GERMAN else: if chat_id in ids.keys(): language = ids[chat_id] else: language = Language.GERMAN await send_essen(chat_id, language) await send_status("Essen versendet für Chat %s (ID: %i)" % (chat, chat_id)) elif text.startswith("/help"): await send_message(bot_obj=bot, chat_id=chat_id, msg=info_str, parse_mode="markdown") elif text.startswith("/settings"): if chat_id in ids: callback_chat_id = "::" + str(chat_id) keyboard = InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="Deutsch", callback_data=Language.GERMAN.value + callback_chat_id)], [InlineKeyboardButton(text="English", callback_data=Language.ENGLISH.value + callback_chat_id)], [InlineKeyboardButton(text="Deutsch + English", callback_data=Language.GERMAN_ENGLISH.value + callback_chat_id)], ]) await send_message(bot_obj=bot, chat_id=chat_id, msg="Bitte Sprache(n) auswählen/Please select " "language(s)", reply_markup=keyboard) else: await send_message(bot_obj=bot, chat_id=chat_id, msg="Bitte starten Sie zuerst den Bot/Please start the bot first.\n/start") elif text.startswith("/status") and chat_id in config_ids: await send_message(bot_obj=bot, chat_id=chat_id, msg=status, parse_mode="markdown") elif text.startswith("/logging") and chat_id in config_ids: if text.endswith("on") or text.endswith("1") or text.lower().endswith("true"): logging_enabled = True config["logging_enabled"] = True write_config() await send_message(bot_obj=bot, chat_id=chat_id, msg="logging enabled", parse_mode="markdown") elif text.endswith("off") or text.endswith("0") or text.lower().endswith("false"): logging_enabled = False config["logging_enabled"] = False write_config() await send_message(bot_obj=bot, chat_id=chat_id, msg="logging disabled", parse_mode="markdown") else: message = "logging enabled: %s (change with on/off or true/false or 1/0)" % logging_enabled await send_message(bot_obj=bot, chat_id=chat_id, msg=message, parse_mode="markdown") elif content_type == "new_chat_member": if msg["new_chat_participant"]["id"] == get_bot_id(): await send_message(bot_obj=bot, chat_id=chat_id, msg=info_str, parse_mode="markdown") elif content_type == "left_chat_member": if msg["left_chat_participant"]["id"] == get_bot_id(): if ids.pop(chat_id, None) is not None: config['ids'] = ids write_config() async def on_callback_query(self, msg): global config, ids query_id, from_id, query_data = telepot.glance(msg, flavor='callback_query') lang = query_data[:query_data.index("::")] chat_id = int(query_data[query_data.index("::") + 2:]) chat_member = await bot.getChatMember(chat_id=chat_id, user_id=from_id) member_status = chat_member["status"] if msg["message"]["chat"]["type"] != "private" and member_status != "creator" and \ member_status != "administrator": await bot.answerCallbackQuery(callback_query_id=query_id, text="Keine Berechtigung. Bitten Sie einen " "Admin, die Sprache zu ändern.") return name = "" user_id = msg["from"]["id"] if "first_name" in msg["from"].keys(): name = msg["from"]["first_name"] + " " if "last_name" in msg["from"].keys(): name += msg["from"]["last_name"] if len(name) == 0: name = "User " + user_id msg = "[" + name + "](tg://user?id=" + str(user_id) + ") hat die Sprache geändert nach " message = "Fehler beim Setzen der Sprache" if lang == Language.GERMAN.value: message = msg + "_Deutsch_." elif lang == Language.ENGLISH.value: message = msg + "_English_." elif lang == Language.GERMAN_ENGLISH.value: message = msg + "_Deutsch + Englisch_." if chat_id in ids: ids[chat_id] = Language(lang) config['ids'] = ids write_config() else: message = "Fehler: Bot nicht aktiviert für diesen Chat!" await send_message(bot_obj=bot, chat_id=chat_id, msg=message, parse_mode="markdown") async def send_essen(chat_id: int, sprache: Language = Language.GERMAN) -> None: global datum, essen, essen_eng if sprache == Language.GERMAN or sprache == Language.GERMAN_ENGLISH: nachricht = "Speiseplan am %s:\n" % format_date(datum, format="full", locale="de_DE") for i in essen: nachricht += "- " + str(i).replace(".", ",") + "\n\n" await send_message(bot_obj=bot, chat_id=chat_id, msg=nachricht, parse_mode="markdown") if sprache == Language.ENGLISH or sprache == Language.GERMAN_ENGLISH: nachricht = "Menu on %s:\n" % format_date(datum, format="full", locale="en") for i in essen_eng: nachricht += "- " + str(i) + "\n\n" await send_message(bot_obj=bot, chat_id=chat_id, msg=nachricht, parse_mode="markdown") async def send_status(text: str, ignore_flag: bool = False) -> None: """ Sends a status message to all chats specified in config_ids :param text: Message :param ignore_flag: Should the logging_enabled flag be ignored (should a message be sent even if it is set to False)? :return: """ global config_ids message_log.write("[%s] %s\n" % (get_now(), text)) if ignore_flag or logging_enabled: for chat_id in config_ids: await send_message(bot_obj=bot, chat_id=chat_id, msg=text) async def send_message(bot_obj: DelegatorBot, chat_id: int, msg: str, parse_mode: Any = None, disable_web_page_preview: bool = None, disable_notification: bool = None, reply_to_message_id: int = None, reply_markup: Any = None) -> None: """Sends a message with bot_obj to chat_id. See https://core.telegram.org/bots/api#sendmessage for details.""" try: await bot_obj.sendMessage(chat_id=chat_id, text=msg, parse_mode=parse_mode, disable_web_page_preview=disable_web_page_preview, disable_notification=disable_notification, reply_to_message_id=reply_to_message_id, reply_markup=reply_markup) except BotWasBlockedError: if ids.pop(chat_id, None) is not None: config['ids'] = ids await send_status("Bot wurde blockiert für Chat ID %i" % chat_id) write_config() def write_config() -> None: with open(config_filename, 'w') as outfile: json.dump(config, outfile, indent=4, sort_keys=True, cls=EnumEncode) async def get_essen(only_today: bool) -> None: """ Downloads the plan for the next day and parses it :param only_today: only current day; no download if the mensa is closed :return: """ global datum, essen, essen_eng, var, last_updated # , ctx if time() < last_updated + 30 or (len(essen) > 0 and time() < last_updated + 900): return last_updated = time() essen = [] essen_eng = [] try: # response = urllib.request.urlopen(url, context=ctx) response = urllib.request.urlopen(url) except HTTPError: var = False return except Exception as expt: await send_status("Es ist ein Fehler aufgetreten:\n[%s] %s" % (sys.exc_info()[0], str(expt)), ignore_flag=True) print("[%s] Fehler:\n%s" % (get_now(), traceback.format_exc()), file=sys.stderr) var = True return data = response.read() response.close() text = data.decode('utf-8') et = ET.fromstring(text) dates = et.findall("./menus/day/date") if datetime.datetime.now().hour > 13: day = datetime.date.today() + datetime.timedelta(1) else: day = datetime.date.today() if len(dates) == 0: return date_dt = None for i in dates: date_xml = i.text date_dt = datetime.date(int(date_xml[:4]), int(date_xml[5:7]), int(date_xml[8:10])) if date_dt >= day: break if date_dt < day or (only_today and date_dt != day): return datum = date_dt menus = et.findall("./menus/day[date='" + datum.isoformat() + "']/menu") essen_transl = [] preise = [] for i in menus: kategorie = i.findall("type")[0].text essen_transl.append(kategorie) name = sub("\\s*\\([0-9A-Z, ]*\\)", "", i.findall("description")[0].text) essen_transl.append(name) try: preis = float(i.findall("./prices/price[label='Studenten']/value")[0].text.replace(",", ".")) except ValueError: preis = -1 preise.append(preis) essen.append(Essen(name, preis, kategorie)) translations = translator.translate(essen_transl) for i in range(int(len(translations) / 2)): essen_eng.append(Essen(translations[i * 2 + 1].text, preise[i], translations[i * 2].text)) def get_bot_id() -> int: global botID if botID == -1: api_url = "https://api.telegram.org/bot" + telegram_bot_token + "/getMe" with urllib.request.urlopen(api_url) as response: data = json.loads(response.read().decode('utf-8')) botID = data["result"]["id"] return botID def get_chat_name(msg: dict) -> str: chat = "" if msg["chat"]["type"] == "group": chat = msg["chat"]["title"] + " (g)" elif msg["chat"]["type"] == "private": if "first_name" in msg["chat"].keys(): chat = msg["chat"]["first_name"] + " " if "last_name" in msg["chat"].keys(): chat += msg["chat"]["last_name"] + " " chat += "(p)" else: chat = "null" return chat def shutdown(signum, frame): print('Signal handler called with signal', signum) ml.close() loop.stop() loop.close() message_log.flush() message_log.close() sys.stdout.flush() sys.stdout.close() sys.stderr.flush() sys.stderr.close() sys.exit(-1) async def essen_loop() -> None: global status, essen, ids, var while True: now = datetime.datetime.today() next_day = datetime.datetime(now.year, now.month, now.day) + datetime.timedelta(1, 36000) await send_status("Wartezeit: %i Sekunden" % (next_day - now).seconds) status = "Warten bis %s" % next_day await asyncio.sleep((next_day - now).seconds) await send_status("Aufwachen um 10 Uhr") status = "Essen abrufen" await get_essen(True) # try again if error if len(essen) == 0 and var: await send_status("Fehler, warte 30 Sekunden") await asyncio.sleep(30) await get_essen(True) await send_status("%i Essen gefunden" % len(essen)) status = "Essen senden" if len(essen) > 0: await send_status("Essen werden gesendet") for i in ids.keys(): await send_essen(i, ids[i]) await send_status("Abgeschlossen, warte 30 Sekunden") status = "Warte 30 Sekunden" await asyncio.sleep(30) def get_now() -> str: return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") def process_json(d): if isinstance(d, dict): try: return {int(k): getattr(Language, v) for k, v in d.items()} except (ValueError, TypeError, AttributeError): return d return d sys.stdout = open("out.log", "a") sys.stderr = open("err.log", "a") message_log = open("msg.log", "a") signal.signal(signal.SIGTERM, shutdown) signal.signal(signal.SIGINT, shutdown) translator = Translator() try: with open(config_filename, 'r') as config_file: config = json.load(config_file, object_hook=process_json) except FileNotFoundError as e: print('Error: config file "{}" not found: {}'.format(config_filename, e)) sys.exit(-1) except ValueError as e: print('Error: invalid config file "{}": {}'.format(config_filename, e)) sys.exit(-1) telegram_bot_token = config.get("telegram_bot_token") url = config.get("url") ids = config.get("ids") config_ids = config.get("config_ids") logging_enabled = config.get("logging_enabled") bot = telepot.aio.DelegatorBot(telegram_bot_token, [ include_callback_query_chat_id( pave_event_space())( per_chat_id(), create_open, HSMensaW, timeout=10 ), ]) loop = asyncio.get_event_loop() # ml = MessageLoop(bot, handle).run_forever() ml = MessageLoop(bot).run_forever() loop.create_task(ml) loop.create_task(essen_loop()) loop.run_forever() # ctx = ssl.create_default_context() # ctx.check_hostname = False # ctx.verify_mode = ssl.CERT_NONE