From a1d76d3badb215160a033738dd04c9432eb55ff0 Mon Sep 17 00:00:00 2001 From: dymik739 Date: Sat, 5 Nov 2022 18:44:07 +0200 Subject: [PATCH] added the ability to test modules locally without using dedicated telegram bot --- module-testing.py | 240 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 240 insertions(+) create mode 100644 module-testing.py diff --git a/module-testing.py b/module-testing.py new file mode 100644 index 0000000..bb98311 --- /dev/null +++ b/module-testing.py @@ -0,0 +1,240 @@ +import datetime +import codecs +import time +import json +import sys +import os +import threading + +# global variables +STOP_REQUESTED = False + +# some functions that increase readability of the code +def readfile(filename): + try: + return codecs.open(filename, encoding = "utf-8").read() + except FileNotFoundError: + return False + except Exception as e: + print( "[ERROR] Unexpected error occured in readfile() ({0})".format(e) ) + return False + +# module object classes +class ModuleV1: + def __init__(self, path, code, enabled, alias, predefine): + self.version = 1 + self.enabled = enabled + self.code = code + self.alias = alias + self.path = path + self.predefine = predefine + + if self.predefine: + self.set_predefine() + + # set environmental variables + def set_env(self): + self.RESPONCE = "" + + def set_predefine(self): + try: + exec(self.predefine) + except Exception as e: + print("[ERROR] module: module \"{}\" ({}) raised exception \"{}\" during predefine stage, disabling it...".format(self.path, self.alias, e)) + + # running the module + def process(self, msg): + self.set_env() + + self.MESSAGE = msg + try: + exec(self.code) + return self.RESPONCE + except Exception as e: + print("[ERROR] module: module \"{}\" ({}) raised exception \"{}\"".format(self.path, self.alias, e)) + return "" + + +# module control unit +class ModuleControlUnit: + def __init__(self): + self.modules = [] + self.reload_modules() + + print("[INFO] ModuleControlUnit: initialized successfully") + + def reload_modules(self): + for folder in os.listdir("modules/"): + try: + meta_raw = readfile("modules/{}/meta.json".format(folder)) + if not meta_raw: + print("[WARN] module_loader: no meta.json found in module folder \"{}\"".format(folder)) + continue + + meta = json.loads( meta_raw ) + if "version" in meta: + if meta["version"] == 1: + if "index_file" in meta: + index_file = meta["index_file"] + else: + index_file = "index.py" + + code = readfile( "modules/{}/{}".format(folder, index_file) ) + if not code: # False both when readfile() returns False and when the code string is empty + print("[WARN] reload_modules: module {} does not have any code, skipping...".format(folder)) + continue + + if "start_on_boot" in meta: + enabled = meta["start_on_boot"] + else: + enabled = False + + if "alias" in meta: + alias = meta["alias"] + else: + alias = None + + if "predefine" in meta: + predefine = readfile("modules/{}/{}".format(folder, meta["predefine"])) + else: + predefine = False + + self.modules.append( ModuleV1( "modules/{}/".format(folder), code, enabled, alias, predefine ) ) + + print("[INFO] reload_modules: successfully loaded {} as {} (start_on_boot: {})".format(folder, alias, enabled)) + else: + print("[WARN] reload_modules: module {} requires unsupported version ({} > 1), skipping...".format(folder, meta["version"])) + except Exception as e: + print("[ERROR] module_loader: error while loading module \"{}\" ({})".format(folder, e)) + + def reload_module(self, folder_name): + try: + meta_raw = readfile("modules/{}/meta.json".format(folder)) + if not meta_raw: + print("[WARN] module_loader: no meta.json found in module folder \"{}\"".format(folder_name)) + return 1 + + meta = json.loads( readfile("modules/{}/meta.json".format(folder_name)) ) + if "version" in meta: + if meta["version"] == 1: + if "index_file" in meta: + index_file = meta["index_file"] + else: + index_file = "index.py" + + code = readfile( "modules/{}/{}".format(folder_name, index_file) ) + if not code: # False both when readfile() returns False and when the code string is empty + print("[WARN] reload_modules: module {} does not have any code, skipping...".format(folder_name)) + + if "start_on_boot" in meta: + enabled = meta["start_on_boot"] + else: + enabled = False + + if "alias" in meta: + alias = meta["alias"] + else: + alias = None + + self.modules.append( ModuleV1( code, enabled, alias ) ) + + print("[INFO] reload_modules: successfully loaded {} as {} (start_on_boot: {})".format(folder_name, alias, enabled)) + else: + print("[WARN] reload_modules: module {} requires unsupported version ({} > 1), skipping...".format(folder_name, meta["version"])) + except Exception as e: + print("[ERROR] module_loader: error while loading module \"{}\" ({})".format(folder_name, e)) + +# message queue object to go back to synchronous message processing +#class MessageQueue: +# def __init__(self): +# print("[INFO] Initializing the message queue...") +# self.queue = [] + + +# synchronous message processor +def queue_processor(): + while True: + if len(message_queue) > 0: + msg = message_queue[0] + print("[DEBUG] queue_processor: {}".format(msg)) # debug + + # check for control commands + if msg["chat"]["id"] == 575246355: + if msg["text"][0] == "$": + command = msg["text"][1:].split(" ") + + if len(command) >= 2 and command[0] == "module": + if command[1] == "reload": + print("[INFO] Module reloading triggered by a command") + del mcu.modules[:] + mcu.reload_modules() + + del message_queue[0] + continue + + # modules are used in here + for mod in mcu.modules: + if mod.enabled: + if mod.version == 1: + responce = mod.process(msg) + if len(responce) > 0: + #updater.bot.send_message(chat_id = msg.chat.id, text = responce) + print("Responded using module {} ({}) with text: {}".format(mod.path, mod.alias, responce)) + break + + del message_queue[0] + else: + if STOP_REQUESTED: + break + else: + time.sleep(1) + + print("[INFO] queue_processor thread stops successfully") + + +# telegram bot processor +def message_handler(update, context): + print("[DEBUG] Received new message") # just for testing + message_queue.append(update.message) + + +# --- Final stage --- + + +# initializing services and queues + +message_queue = [] + +mcu = ModuleControlUnit() + +processor_thread = threading.Thread( target = queue_processor, args = [] ) +processor_thread.start() + + +# connecting to Telegram servers and listening for messages +''' +TOKEN = readfile("config/token") +if not TOKEN: + print("[CRIT] Token has not been defined, quitting") + sys.exit(1) + +# connect to Telegram servers +updater = Updater(TOKEN, use_context = True) +dispatcher = updater.dispatcher + +# assign the handler for messages +dispatcher.add_handler(MessageHandler(Filters.text, message_handler)) + +# run the bot +updater.start_polling() +updater.idle() +''' + +print("Enter testing messages one by one, end with an empty line") + +while True: + new_msg = input() + if len(new_msg) == 0: + break + + message_queue.append({'text': new_msg, 'new_chat_photo': [], 'message_id': 74, 'new_chat_members': [], 'delete_chat_photo': False, 'group_chat_created': False, 'supergroup_chat_created': False, 'chat': {'first_name': 'Дмитро', 'username': 'dmytrofiot23', 'id': 575246355, 'type': 'private'}, 'caption_entities': [], 'channel_chat_created': False, 'date': 1667585646, 'photo': [], 'entities': [], 'from': {'first_name': 'Дмитро', 'id': 575246355, 'is_bot': False, 'username': 'dmytrofiot23', 'language_code': 'en'}})