import json import io import logging import asyncio from os.path import exists, normpath from os import remove from pathlib import Path from threading import Event from typing import Dict, Union import backoff import urllib3 from aiogram import Bot, types from aiogram.types import Message, CallbackQuery from aiogram.types.inline_keyboard import InlineKeyboardButton, InlineKeyboardMarkup from aiogram.dispatcher.dispatcher import Dispatcher from aiogram.types.input_file import InputFile from aiogram.types.input_media import InputMediaAudio logging.basicConfig(level=logging.INFO) try: import extensions.telegram_bot.source.text_process as tp import extensions.telegram_bot.source.const as const import extensions.telegram_bot.source.utils as utils import extensions.telegram_bot.source.buttons as buttons from extensions.telegram_bot.source.conf import cfg from extensions.telegram_bot.source.user import User as User from extensions.telegram_bot.source.extension.silero import Silero as Silero from extensions.telegram_bot.source.extension.sd_api import SdApi as SdApi except ImportError: import source.text_process as tp import source.const as const import source.utils as utils import source.buttons as buttons from source.conf import cfg from source.user import User as User from source.extension.silero import Silero as Silero from source.extension.sd_api import SdApi as SdApi class AiogramLlmBot: # Set dummy obj for telegram updater bot: Union[Bot, None] = None dp: Union[Dispatcher, None] = None # dict of User data dicts, here stored all users' session info. users: Dict[int, User] = {} def __init__(self, config_file_path="configs/app_config.json"): """Init telegram bot class. Use run_telegram_bot() to initiate bot. Args config_file_path: path to config file """ logging.info(f"### TelegramBotWrapper INIT config_file_path: {config_file_path} ###") # Set&Load main config file self.config_file_path = config_file_path cfg.load(self.config_file_path) # Silero initiate self.silero = Silero() # SdApi initiate self.SdApi = SdApi(cfg.sd_api_url, cfg.sd_config_file_path) # Load user rules if exists(cfg.user_rules_file_path): with open(normpath(cfg.user_rules_file_path), "r") as user_rules_file: self.user_rules = json.loads(user_rules_file.read()) else: logging.error("Cant find user_rules_file_path: " + cfg.user_rules_file_path) self.user_rules = {} # initiate generator tp.generator.init( cfg.generator_script, cfg.llm_path, n_ctx=cfg.generation_params.get("chat_prompt_size", 1024), n_gpu_layers=cfg.generation_params.get("n_gpu_layers", 0), ) logging.info(f"### TelegramBotWrapper INIT DONE ###") logging.info(f"### !!! READY !!! ###") # ============================================================================= # Run bot with token! Initiate updater obj! async def run_telegram_bot(self, bot_token="", token_file_name=""): """ Start the Telegram bot. :param bot_token: (str) The Telegram bot tokens separated by ',' If not provided, try to read it from `token_file_name`. :param token_file_name: (str) The name of the file containing the bot token. Default is `None`. :return: None """ if not bot_token: token_file_name = token_file_name or cfg.token_file_path with open(normpath(token_file_name), "r", encoding="utf-8") as f: bot_token = f.read().strip() proxy_url = cfg.proxy_url if cfg.proxy_url else None self.bot = Bot(token=bot_token, proxy=proxy_url) self.dp = Dispatcher(self.bot) self.dp.register_message_handler(self.thread_welcome_message, commands=["start"]) self.dp.register_message_handler(self.thread_get_message) self.dp.register_message_handler(self.thread_get_json_document, content_types=types.ContentType.DOCUMENT) self.dp.register_callback_query_handler(self.thread_push_button) await self.dp.start_polling() # Thread(target=self.no_sleep_callback).start() # ============================================================================= # Additional telegram actions @staticmethod def get_user_profile_name(message) -> str: message = message or message.message user_name = cfg.user_name_template.replace("FIRSTNAME", message.from_user.first_name or "") user_name = user_name.replace("LASTNAME", message.from_user.last_name or "") user_name = user_name.replace("USERNAME", message.from_user.username or "") user_name = user_name.replace("ID", str(message.from_user.id) or "") return user_name async def make_template_message(self, request: str, chat_id: int, custom_string="") -> str: # create a message using default_messages_template or return # UNKNOWN_TEMPLATE if chat_id in self.users: user = self.users[chat_id] if request in const.DEFAULT_MESSAGE_TEMPLATE: msg = const.DEFAULT_MESSAGE_TEMPLATE[request] msg = msg.replace("_CHAT_ID_", str(chat_id)) msg = msg.replace("_NAME1_", user.name1) msg = msg.replace("_NAME2_", user.name2) msg = msg.replace("_CONTEXT_", user.context) msg = msg.replace( "_GREETING_", await utils.prepare_text(user.greeting, user, "to_user"), ) msg = msg.replace( "_CUSTOM_STRING_", await utils.prepare_text(custom_string, user, "to_user"), ) msg = msg.replace("_OPEN_TAG_", cfg.html_tag[0]) msg = msg.replace("_CLOSE_TAG_", cfg.html_tag[1]) return msg else: return const.UNKNOWN_TEMPLATE else: return const.UNKNOWN_USER # ============================================================================= # Work with history! Init/load/save functions async def thread_get_json_document(self, message: Message): chat_id = message.chat.id if not utils.check_user_permission(chat_id): return False utils.init_check_user(self.users, chat_id) user = self.users[chat_id] default_user_file_path = str(Path(f"{cfg.history_dir_path}/{str(chat_id)}.json")) file = await self.bot.get_file(file_id=message.document.file_id) file_path = file.file_path await self.bot.download_file(file_path=file_path, destination=default_user_file_path) user.load_user_history(default_user_file_path) if len(user.history) > 0: last_message = user.history_last_out else: last_message = "" send_text = await self.make_template_message("hist_loaded", chat_id, last_message) await self.bot.send_message( chat_id=chat_id, text=send_text, reply_markup=self.get_initial_keyboard(chat_id, user), parse_mode="HTML", ) async def start_send_typing_status(self, chat_id: int) -> Event: typing_active = Event() typing_active.set() asyncio.create_task(self.thread_typing_status(chat_id, typing_active)) return typing_active async def thread_typing_status(self, chat_id: int, typing_active: Event): limit_counter = int(cfg.generation_timeout / 5) while typing_active.is_set() and limit_counter > 0: await self.bot.send_chat_action(chat_id=chat_id, action="typing") await asyncio.sleep(5) limit_counter -= 1 @backoff.on_exception( backoff.expo, (urllib3.exceptions.HTTPError, urllib3.exceptions.ConnectTimeoutError), max_time=10, ) async def send_sd_image(self, message, answer: str, user_text: str): chat_id = message.chat.id try: file_list = await self.SdApi.get_image(answer) answer = answer.replace(cfg.sd_api_prompt_of.replace("OBJECT", user_text[1:].strip()), "") for char in ["[", "]", "{", "}", "(", ")", "*", '"', "'"]: answer = answer.replace(char, "") if len(answer) > 1023: answer = answer[:1023] if len(file_list) > 0: for image_path in file_list: if exists(image_path): await self.bot.send_photo(caption=answer, chat_id=chat_id, photo=InputFile(image_path)) remove(image_path) except Exception as e: logging.error("send_sd_image: " + str(e)) await self.bot.send_message(text=answer, chat_id=chat_id) @backoff.on_exception( backoff.expo, (urllib3.exceptions.HTTPError, urllib3.exceptions.ConnectTimeoutError), max_time=10, ) async def clean_last_message_markup(self, chat_id: int): if chat_id in self.users and len(self.users[chat_id].msg_id) > 0: last_msg = self.users[chat_id].msg_id[-1] try: await self.bot.edit_message_reply_markup(chat_id=chat_id, message_id=last_msg, reply_markup=None) except Exception as exception: logging.info("clean_last_message_markup: " + str(exception)) @backoff.on_exception( backoff.expo, (urllib3.exceptions.HTTPError, urllib3.exceptions.ConnectTimeoutError), max_time=10, ) async def send_message(self, chat_id: int, text: str) -> Message: user = self.users[chat_id] text = await utils.prepare_text(text, user, "to_user") if user.silero_speaker == "None" or user.silero_model_id == "None": message = await self.bot.send_message( text=text, chat_id=chat_id, parse_mode="HTML", reply_markup=self.get_chat_keyboard(chat_id, True), ) else: if ":" in text: audio_text = ":".join(text.split(":")[1:]) else: audio_text = text audio_path = await self.silero.get_audio(text=audio_text, user_id=chat_id, user=user) if audio_path is not None: message = await self.bot.send_audio( chat_id=chat_id, audio=InputFile(audio_path), caption=text, parse_mode="HTML", reply_markup=self.get_chat_keyboard(chat_id, True), ) else: message = await self.bot.send_message( text=text, chat_id=chat_id, parse_mode="HTML", reply_markup=self.get_chat_keyboard(chat_id, True), ) return message @backoff.on_exception( backoff.expo, (urllib3.exceptions.HTTPError, urllib3.exceptions.ConnectTimeoutError), max_time=10, ) async def edit_message( self, cbq, chat_id: int, text: str, message_id: int, ): user = self.users[chat_id] text = await utils.prepare_text(text, user, "to_user") if cbq.message.text is not None: await self.bot.edit_message_text( text=text, chat_id=chat_id, parse_mode="HTML", message_id=message_id, reply_markup=self.get_chat_keyboard(chat_id), ) if cbq.message.audio is not None and user.silero_speaker != "None" and user.silero_model_id != "None": if ":" in text: audio_text = ":".join(text.split(":")[1:]) else: audio_text = text audio_path = await self.silero.get_audio(text=audio_text, user_id=chat_id, user=user) if audio_path is not None: await self.bot.edit_message_media( chat_id=chat_id, media=InputMediaAudio(media=normpath(audio_path)), message_id=message_id, reply_markup=self.get_chat_keyboard(chat_id), ) if cbq.message.caption is not None: await self.bot.edit_message_caption( chat_id=chat_id, caption=text, parse_mode="HTML", message_id=message_id, reply_markup=self.get_chat_keyboard(chat_id), ) # ============================================================================= # Message handler async def thread_welcome_message(self, message: types.Message): chat_id = message.chat.id if not utils.check_user_permission(chat_id): return False utils.init_check_user(self.users, chat_id) send_text = await self.make_template_message("char_loaded", chat_id) await self.bot.send_message( chat_id=chat_id, text=send_text, reply_to_message_id=None, reply_markup=self.get_initial_keyboard(chat_id, self.users[chat_id] if chat_id in self.users else None), parse_mode="HTML", ) async def thread_get_message(self, message: types.Message): # Extract user input and chat ID user_text = message.text chat_id = message.chat.id utils.init_check_user(self.users, chat_id) user = self.users[chat_id] # check permission and flooding if not utils.check_user_permission(chat_id): return False if not user.check_flooding(cfg.flood_avoid_delay): return False if cfg.only_mention_in_chat and message.chat.type != "CHAT_PRIVATE": if "".join(["@", message.from_user.username]) in user_text: user_text = user_text.replace("".join(["@", message.from_user.username]), "") else: return # Send "typing" message typing = await self.start_send_typing_status(chat_id) try: if utils.check_user_rule(chat_id=chat_id, option=const.GET_MESSAGE) is not True: return False if not user_text.startswith(tuple(cfg.sd_api_prefixes)): user_text = await utils.prepare_text(user_text, user, "to_model") answer, system_message = await tp.aget_answer( text_in=user_text, user=user, bot_mode=cfg.bot_mode, generation_params=cfg.generation_params, name_in=self.get_user_profile_name(message), ) if system_message == const.MSG_SYSTEM: await message.reply(text=answer) elif system_message == const.MSG_SD_API: user.truncate_last_message() await self.send_sd_image(message, answer, user_text) else: if system_message == const.MSG_DEL_LAST: await message.delete() # message = self.send_message(text=answer, chat_id=chat_id, context=context) reply = await self.send_message(text=answer, chat_id=chat_id) # Clear buttons on last message (if they exist in current thread) await self.clean_last_message_markup(chat_id) # Add message ID to message history user.msg_id.append(reply.message_id) # Save user history user.save_user_history(chat_id, cfg.history_dir_path) except Exception as e: logging.error("thread_get_message" + str(e) + str(e.args)) finally: pass typing.clear() # ============================================================================= # button async def thread_push_button(self, cbq: types.CallbackQuery): chat_id = cbq.message.chat.id msg_id = cbq.message.message_id option = cbq.data if not utils.check_user_permission(chat_id): return False # Send "typing" message typing = await self.start_send_typing_status(chat_id) await self.bot.answer_callback_query(cbq.id) try: # if new user - init utils.init_check_user(self.users, chat_id) # if button - is chat-interactive button need to check msg consistency if option in [ const.BTN_IMPERSONATE, const.BTN_NEXT, const.BTN_CONTINUE, const.BTN_DEL_WORD, const.BTN_REGEN, const.BTN_CUTOFF, ]: # if lost message button - clear markup and return if len(self.users[chat_id].msg_id) == 0: await cbq.message.edit_reply_markup(reply_markup=None) return else: # if msg list not empty but lost button - clear markup and return if msg_id != self.users[chat_id].msg_id[-1]: await cbq.message.edit_reply_markup(reply_markup=None) return await self.handle_button_option(option, chat_id, cbq) self.users[chat_id].save_user_history(chat_id, cfg.history_dir_path) except Exception as e: logging.error("thread_push_button " + str(e) + str(e.args)) finally: pass typing.clear() async def handle_button_option(self, option, chat_id, cbq: types.CallbackQuery): if option == const.BTN_RESET and utils.check_user_rule(chat_id, option): await self.on_reset_history_button(cbq) elif option == const.BTN_CONTINUE and utils.check_user_rule(chat_id, option): await self.on_continue_message_button(cbq) elif option == const.BTN_IMPERSONATE and utils.check_user_rule(chat_id, option): await self.on_impersonate_button(cbq) elif option == const.BTN_NEXT and utils.check_user_rule(chat_id, option): await self.on_next_message_button(cbq) elif option == const.BTN_IMPERSONATE_INIT and utils.check_user_rule(chat_id, option): await self.on_impersonate_button(cbq, initial=True) elif option == const.BTN_NEXT_INIT and utils.check_user_rule(chat_id, option): await self.on_next_message_button(cbq, initial=True) elif option == const.BTN_DEL_WORD and utils.check_user_rule(chat_id, option): await self.on_delete_word_button(cbq) elif option == const.BTN_PREVIOUS and utils.check_user_rule(chat_id, option): await self.on_previous_message_button(cbq) elif option == const.BTN_REGEN and utils.check_user_rule(chat_id, option): await self.on_regenerate_message_button(cbq) elif option == const.BTN_CUTOFF and utils.check_user_rule(chat_id, option): await self.on_cutoff_message_button(cbq) elif option == const.BTN_DOWNLOAD and utils.check_user_rule(chat_id, option): await self.on_download_json_button(cbq) elif option == const.BTN_OPTION and utils.check_user_rule(chat_id, option): await self.show_options_button(cbq) elif option == const.BTN_DELETE and utils.check_user_rule(chat_id, option): await self.on_delete_pressed_button(cbq) elif option.startswith(const.BTN_CHAR_LIST) and utils.check_user_rule(chat_id, option): await self.keyboard_characters_button(cbq, option=option) elif option.startswith(const.BTN_CHAR_LOAD) and utils.check_user_rule(chat_id, option): await self.load_character_button(cbq, option=option) elif option.startswith(const.BTN_PRESET_LIST) and utils.check_user_rule(chat_id, option): await self.keyboard_presets_button(cbq, option=option) elif option.startswith(const.BTN_PRESET_LOAD) and utils.check_user_rule(chat_id, option): await self.load_presets_button(cbq, option=option) elif option.startswith(const.BTN_MODEL_LIST) and utils.check_user_rule(chat_id, option): await self.on_keyboard_models_button(cbq, option=option) elif option.startswith(const.BTN_MODEL_LOAD) and utils.check_user_rule(chat_id, option): await self.on_load_model_button(cbq, option=option) elif option.startswith(const.BTN_LANG_LIST) and utils.check_user_rule(chat_id, option): await self.on_keyboard_language_button(cbq, option=option) elif option.startswith(const.BTN_LANG_LOAD) and utils.check_user_rule(chat_id, option): await self.on_load_language_button(cbq, option=option) elif option.startswith(const.BTN_VOICE_LIST) and utils.check_user_rule(chat_id, option): await self.on_keyboard_voice_button(cbq, option=option) elif option.startswith(const.BTN_VOICE_LOAD) and utils.check_user_rule(chat_id, option): await self.on_load_voice_button(cbq, option=option) async def show_options_button(self, cbq: CallbackQuery): chat_id = cbq.message.chat.id user = self.users[chat_id] send_text = utils.get_conversation_info(user) await self.bot.send_message( text=send_text, chat_id=chat_id, reply_markup=self.get_options_keyboard(chat_id, user), parse_mode="HTML", ) async def on_delete_pressed_button(self, cbq): chat_id = cbq.message.chat.id message_id = cbq.message.message_id await self.bot.delete_message(chat_id=chat_id, message_id=message_id) async def on_impersonate_button(self, cbq, initial=False): chat_id = cbq.message.chat.id message_id = cbq.message.message_id user = self.users[chat_id] if initial: await self.bot.edit_message_reply_markup( chat_id=chat_id, message_id=message_id, reply_markup=self.get_options_keyboard(chat_id, user), ) else: await self.clean_last_message_markup(chat_id) answer, _ = await tp.aget_answer( text_in=const.GENERATOR_MODE_IMPERSONATE, user=user, bot_mode=cfg.bot_mode, generation_params=cfg.generation_params, name_in=self.get_user_profile_name(cbq), ) message = await self.send_message(text=answer, chat_id=chat_id) user.msg_id.append(message.message_id) user.save_user_history(chat_id, cfg.history_dir_path) async def on_next_message_button(self, cbq, initial=False): chat_id = cbq.message.chat.id message_id = cbq.message.message_id user = self.users[chat_id] if initial: await self.bot.edit_message_reply_markup( chat_id=chat_id, message_id=message_id, reply_markup=self.get_options_keyboard(chat_id, user), ) else: await self.clean_last_message_markup(chat_id) answer, _ = await tp.aget_answer( text_in=const.GENERATOR_MODE_NEXT, user=user, bot_mode=cfg.bot_mode, generation_params=cfg.generation_params, name_in=self.get_user_profile_name(cbq), ) message = await self.send_message(text=answer, chat_id=chat_id) user.msg_id.append(message.message_id) user.save_user_history(chat_id, cfg.history_dir_path) async def on_continue_message_button(self, cbq): chat_id = cbq.message.chat.id message = cbq.message user = self.users[chat_id] # get answer and replace message text! answer, _ = await tp.aget_answer( text_in=const.GENERATOR_MODE_CONTINUE, user=user, bot_mode=cfg.bot_mode, generation_params=cfg.generation_params, name_in=self.get_user_profile_name(cbq), ) await self.edit_message( text=answer, chat_id=chat_id, message_id=message.message_id, ) user.save_user_history(chat_id, cfg.history_dir_path) async def on_previous_message_button(self, cbq): chat_id = cbq.message.chat.id message = cbq.message user = self.users[chat_id] # get answer and replace message text! answer = user.back_to_previous_out(msg_id=message.message_id) if answer is not None: await self.edit_message(text=answer, chat_id=chat_id, message_id=message.message_id, cbq=cbq) user.save_user_history(chat_id, cfg.history_dir_path) async def on_delete_word_button(self, cbq): chat_id = cbq.message.chat.id user = self.users[chat_id] answer, return_msg_action = await tp.aget_answer( text_in=const.GENERATOR_MODE_DEL_WORD, user=user, bot_mode=cfg.bot_mode, generation_params=cfg.generation_params, name_in=self.get_user_profile_name(cbq), ) if return_msg_action != const.MSG_NOTHING_TO_DO: await self.edit_message(text=answer, chat_id=chat_id, message_id=user.msg_id[-1], cbq=cbq) user.save_user_history(chat_id, cfg.history_dir_path) async def on_regenerate_message_button(self, cbq): chat_id = cbq.message.chat.id msg = cbq.message user = self.users[chat_id] await self.clean_last_message_markup(chat_id) # get answer and replace message text! answer, _ = await tp.aget_answer( text_in=const.GENERATOR_MODE_REGENERATE, user=user, bot_mode=cfg.bot_mode, generation_params=cfg.generation_params, name_in=self.get_user_profile_name(cbq), ) await self.edit_message( text=answer, chat_id=chat_id, message_id=msg.message_id, cbq=cbq, ) user.save_user_history(chat_id, cfg.history_dir_path) async def on_cutoff_message_button(self, cbq): chat_id = cbq.message.chat.id user = self.users[chat_id] # Edit or delete last message ID (strict lines) last_msg_id = user.msg_id[-1] await self.bot.delete_message(chat_id=chat_id, message_id=last_msg_id) # Remove last message and bot answer from history user.truncate_last_message() # If there is previous message - add buttons to previous message if user.msg_id: message_id = user.msg_id[-1] await self.bot.edit_message_reply_markup( chat_id=chat_id, message_id=message_id, reply_markup=self.get_chat_keyboard(chat_id), ) user.save_user_history(chat_id, cfg.history_dir_path) async def on_download_json_button(self, cbq): chat_id = cbq.message.chat.id user = self.users[chat_id] if chat_id not in self.users: return user_file = io.StringIO(self.users[chat_id].to_json()) send_caption = await self.make_template_message("hist_to_chat", chat_id) await self.bot.send_document( chat_id=chat_id, caption=send_caption, document=InputFile(user_file, filename=user.name2 + ".json"), ) async def on_reset_history_button(self, cbq): # check if it is a callback_query or a command if cbq: chat_id = cbq.message.chat.id else: chat_id = cbq.chat.id if chat_id not in self.users: return user = self.users[chat_id] if user.msg_id: await self.clean_last_message_markup(chat_id) user.reset() user.load_character_file(cfg.characters_dir_path, user.char_file) send_text = await self.make_template_message("mem_reset", chat_id) await self.bot.send_message( chat_id=chat_id, text=send_text, reply_markup=self.get_initial_keyboard(chat_id, user), parse_mode="HTML", ) # ============================================================================= # switching keyboard async def on_load_model_button(self, cbq, option: str): if tp.generator.get_model_list is not None: model_list = tp.generator.get_model_list() model_file = model_list[int(option.replace(const.BTN_MODEL_LOAD, ""))] chat_id = cbq.effective_chat.id send_text = "Loading " + model_file + ". ๐Ÿช„" message_id = cbq.message.message_id await self.bot.edit_message_text( text=send_text, chat_id=chat_id, message_id=message_id, parse_mode="HTML", ) try: tp.generator.load_model(model_file) send_text = await self.make_template_message( request="model_loaded", chat_id=chat_id, custom_string=model_file ) await self.bot.edit_message_text( chat_id=chat_id, message_id=message_id, text=send_text, parse_mode="HTML", reply_markup=self.get_options_keyboard( chat_id, self.users[chat_id] if chat_id in self.users else None ), ) except Exception as e: logging.error("model button error: " + str(e)) await self.bot.edit_message_text( chat_id=chat_id, message_id=message_id, text="Error during " + model_file + " loading. โ›”", parse_mode="HTML", reply_markup=self.get_options_keyboard( chat_id, self.users[chat_id] if chat_id in self.users else None ), ) raise e async def on_keyboard_models_button(self, cbq, option: str): if tp.generator.get_model_list() is not None: chat_id = cbq.message.chat.id msg = cbq.message model_list = tp.generator.get_model_list() if option == const.BTN_MODEL_LIST + const.BTN_OPTION: await self.bot.edit_message_reply_markup( chat_id=chat_id, message_id=msg.message_id, reply_markup=self.get_options_keyboard( chat_id, self.users[chat_id] if chat_id in self.users else None ), ) return shift = int(option.replace(const.BTN_MODEL_LIST, "")) characters_buttons = self.get_switch_keyboard( opt_list=model_list, shift=shift, data_list=const.BTN_MODEL_LIST, data_load=const.BTN_MODEL_LOAD, ) await self.bot.edit_message_reply_markup( chat_id=chat_id, message_id=msg.message_id, reply_markup=characters_buttons, ) async def load_presets_button(self, cbq, option: str): chat_id = cbq.message.chat.id preset_char_num = int(option.replace(const.BTN_PRESET_LOAD, "")) cfg.preset_file = utils.parse_presets_dir()[preset_char_num] cfg.load_preset(preset_file=cfg.preset_file) user = self.users[chat_id] send_text = utils.get_conversation_info(user) message_id = cbq.message.message_id await self.bot.edit_message_text( text=send_text, message_id=message_id, chat_id=chat_id, parse_mode="HTML", reply_markup=self.get_options_keyboard(chat_id, user), ) async def keyboard_presets_button(self, cbq, option: str): chat_id = cbq.message.chat.id msg = cbq.message # if "return character_file markup" button - clear markup if option == const.BTN_PRESET_LIST + const.BTN_OPTION: await self.bot.edit_message_reply_markup( chat_id=chat_id, message_id=msg.message_id, reply_markup=self.get_options_keyboard(chat_id, self.users[chat_id] if chat_id in self.users else None), ) return # get keyboard list shift shift = int(option.replace(const.BTN_PRESET_LIST, "")) preset_list = utils.parse_presets_dir() characters_buttons = self.get_switch_keyboard( opt_list=preset_list, shift=shift, data_list=const.BTN_PRESET_LIST, data_load=const.BTN_PRESET_LOAD, keyboard_column=3, ) await self.bot.edit_message_reply_markup( chat_id=chat_id, message_id=msg.message_id, reply_markup=characters_buttons ) async def load_character_button(self, cbq, option: str): chat_id = cbq.message.chat.id char_num = int(option.replace(const.BTN_CHAR_LOAD, "")) char_list = utils.parse_characters_dir() await self.clean_last_message_markup(chat_id) utils.init_check_user(self.users, chat_id) user = self.users[chat_id] char_file = char_list[char_num] user.load_character_file(characters_dir_path=cfg.characters_dir_path, char_file=char_file) # If there was conversation with this character_file - load history user.find_and_load_user_char_history(chat_id, cfg.history_dir_path) if len(user.history) > 0: send_text = await self.make_template_message("hist_loaded", chat_id, user.history_last_out) else: send_text = await self.make_template_message("char_loaded", chat_id) await self.bot.send_message( text=send_text, chat_id=chat_id, parse_mode="HTML", reply_markup=self.get_initial_keyboard(chat_id, self.users[chat_id] if chat_id in self.users else None), ) async def keyboard_characters_button(self, cbq, option: str): chat_id = cbq.message.chat.id msg = cbq.message # if "return character_file markup" button - clear markup if option == const.BTN_CHAR_LIST + const.BTN_OPTION: await self.bot.edit_message_reply_markup( chat_id=chat_id, message_id=msg.message_id, reply_markup=self.get_options_keyboard(chat_id, self.users[chat_id] if chat_id in self.users else None), ) return # get keyboard list shift shift = int(option.replace(const.BTN_CHAR_LIST, "")) char_list = utils.parse_characters_dir() if shift == -9999 and self.users[chat_id].char_file in char_list: shift = char_list.index(self.users[chat_id].char_file) # create chars list characters_buttons = self.get_switch_keyboard( opt_list=char_list, shift=shift, data_list=const.BTN_CHAR_LIST, data_load=const.BTN_CHAR_LOAD, ) await self.bot.edit_message_reply_markup( chat_id=chat_id, message_id=msg.message_id, reply_markup=characters_buttons ) async def on_load_language_button(self, cbq, option: str): chat_id = cbq.message.chat.id message_id = cbq.message.message_id user = self.users[chat_id] lang_num = int(option.replace(const.BTN_LANG_LOAD, "")) language = list(cfg.language_dict.keys())[lang_num] self.users[chat_id].language = language send_text = utils.get_conversation_info(user) await self.bot.edit_message_text( text=send_text, message_id=message_id, chat_id=chat_id, parse_mode="HTML", reply_markup=self.get_options_keyboard(chat_id, user), ) async def on_keyboard_language_button(self, cbq, option: str): chat_id = cbq.message.chat.id msg = cbq.message # if "return character_file markup" button - clear markup if option == const.BTN_LANG_LIST + const.BTN_OPTION: await self.bot.edit_message_reply_markup( chat_id=chat_id, message_id=msg.message_id, reply_markup=self.get_options_keyboard(chat_id, self.users[chat_id] if chat_id in self.users else None), ) return # get keyboard list shift shift = int(option.replace(const.BTN_LANG_LIST, "")) # create list lang_buttons = self.get_switch_keyboard( opt_list=list(cfg.language_dict.keys()), shift=shift, data_list=const.BTN_LANG_LIST, data_load=const.BTN_LANG_LOAD, keyboard_column=4, ) await self.bot.edit_message_reply_markup(chat_id=chat_id, message_id=msg.message_id, reply_markup=lang_buttons) async def on_load_voice_button(self, cbq, option: str): chat_id = cbq.message.chat.id user = self.users[chat_id] male = Silero.voices[user.language]["male"] female = Silero.voices[user.language]["female"] voice_dict = ["None"] + male + female voice_num = int(option.replace(const.BTN_VOICE_LOAD, "")) user.silero_speaker = voice_dict[voice_num] user.silero_model_id = Silero.voices[user.language]["model"] send_text = utils.get_conversation_info(user) message_id = cbq.message.message_id await self.bot.edit_message_text( text=send_text, message_id=message_id, chat_id=chat_id, parse_mode="HTML", reply_markup=self.get_options_keyboard(chat_id, user), ) async def on_keyboard_voice_button(self, cbq, option: str): chat_id = cbq.message.chat.id msg = cbq.message # if "return character_file markup" button - clear markup if option == const.BTN_VOICE_LIST + const.BTN_OPTION: await self.bot.edit_message_reply_markup( chat_id=chat_id, message_id=msg.message_id, reply_markup=self.get_options_keyboard(chat_id, self.users[chat_id] if chat_id in self.users else None), ) return # get keyboard list shift shift = int(option.replace(const.BTN_VOICE_LIST, "")) # create list user = self.users[chat_id] male = list(map(lambda x: x + "๐Ÿšน", Silero.voices[user.language]["male"])) female = list(map(lambda x: x + "๐Ÿšบ", Silero.voices[user.language]["female"])) voice_dict = ["๐Ÿ”‡None"] + male + female voice_buttons = self.get_switch_keyboard( opt_list=list(voice_dict), shift=shift, data_list=const.BTN_VOICE_LIST, data_load=const.BTN_VOICE_LOAD, keyboard_column=4, ) await self.bot.edit_message_reply_markup(chat_id=chat_id, message_id=msg.message_id, reply_markup=voice_buttons) # ============================================================================= # load characters char_file from ./characters def get_initial_keyboard(self, chat_id, user: User): options = buttons.get_options_keyboard(chat_id=chat_id, user=user) chat_actions = buttons.get_chat_init_keyboard(chat_id=chat_id) return self.keyboard_raw_to_keyboard_tg([options[0], chat_actions[0]]) def get_options_keyboard(self, chat_id, user: User): return self.keyboard_raw_to_keyboard_tg(buttons.get_options_keyboard(chat_id=chat_id, user=user)) def get_chat_keyboard(self, chat_id=0, no_previous=False): if chat_id in self.users: user = self.users[chat_id] else: user = None return self.keyboard_raw_to_keyboard_tg(buttons.get_chat_keyboard(chat_id, user, no_previous)) def get_switch_keyboard( self, opt_list: list, shift: int, data_list: str, data_load: str, keyboard_rows=6, keyboard_column=2, ): return self.keyboard_raw_to_keyboard_tg( buttons.get_switch_keyboard( opt_list=opt_list, shift=shift, data_list=data_list, data_load=data_load, keyboard_rows=keyboard_rows, keyboard_column=keyboard_column, ) ) @staticmethod def keyboard_raw_to_keyboard_tg(keyboard_raw): keyboard_tg = [] for buttons_row in keyboard_raw: keyboard_tg.append([]) for button_dict in buttons_row: keyboard_tg[-1].append(InlineKeyboardButton(**button_dict)) return InlineKeyboardMarkup(inline_keyboard=keyboard_tg)