diff --git a/src/__main__.py b/src/__main__.py index af6c4cf..35a7fa1 100644 --- a/src/__main__.py +++ b/src/__main__.py @@ -16,6 +16,7 @@ handler_start, handler_unknown_command, ) +from src.errors_solver import native_error_handler from src.settings import Settings @@ -45,4 +46,5 @@ application.add_handler(MessageHandler(filters.Document.ALL, handler_mismatch_doctype)) application.add_handler(MessageHandler(filters.UpdateType.MESSAGE & filters.TEXT, handler_register)) application.add_handler(MessageHandler(filters.UpdateType.MESSAGE, handler_other_messages)) + application.add_error_handler(native_error_handler) application.run_polling() diff --git a/src/answers.py b/src/answers.py index 2ef0c37..c598a91 100644 --- a/src/answers.py +++ b/src/answers.py @@ -1,57 +1,67 @@ # Marakulin Andrey https://github.com/Annndruha # 2023 -ans = { - "auth": "🔑 Авторизация", - "about": "📄 Описание", - "back": "◀️ Назад", - "qr": "📷 Печать по QR", - "kb_print": "⚙️ Настройки печати", - "kb_print_copies": "📑 Копий:", - "kb_print_side": "📎 Односторонняя печать", - "kb_print_two_side": "🖇 Двухсторонняя печать", - "hello": "👋🏻 Привет! Я телеграм-бот бесплатного принтера.\n" "Отправьте PDF файл и получите PIN для печати.", - "help": "Я телеграм-бот бесплатного принтера профкома студентов физического факультета МГУ!\n\n" - "❔ Отправьте PDF файл и получите PIN для печати. " - "Поддерживаются только .pdf файлы размером не более 3МБ.\n" - "С этим PIN необходимо подойти к принтеру и ввести его в терминал печати. " - "Либо отсканировать QR-код на принтере с помощью кнопки. После этого начнётся печать." - "\n\n⚙️ Настройки печати можно изменять после отправки файла, они сохраняются автоматически. " - "В момент печати используются самые последние настройки." - "\n\n❗️ Файлы, которые вы отправляете через бота, будут храниться в течение нескольких месяцев на сервере " - "в Москве, а также в этом чате Telegram.\nДоступ к файлам имеет узкий круг лиц, " - "ответственных за работоспособность сервиса печати.\n" - "Мы НЕ рекомендуем использовать данный сервис для печати конфиденциальных документов!" - "\n\n💻 Бот разработан группой программистов профкома, как и приложение " - 'Твой ФФ! ' - "В приложении вы сможете найти больше настроек печати, расписание и много других возможностей.\n" - 'Так же есть бот для печати ВКонтакте.', - "val_fail": "⚠️ Проверка не пройдена. Удостоверьтесь что вы состоите в профсоюзе и правильно ввели данные." - "\n\nВведите фамилию и номер профсоюзного билета в формате:\n\nИванов\n1234567", - "val_pass": "🥳 Поздравляю! Проверка пройдена и данные сохранены для этого телеграм-аккаунта. Можете присылать pdf.", - "val_need": "👤 Для использования принтера необходимо авторизоваться.\n" - "Отправьте фамилию и номер профсоюзного билета в формате:\n\nИванов\n1234567", - "val_update_fail": "Сообщение не распознано.\nЧтобы открыть инструкцию введите: /help\nДля того чтобы обновить " - "данные авторизации введите фамилию и номер" - "профсоюзного билета в формате:\n\nИванов\n1234567", - "val_update_pass": "🥳 Поздравляю! Проверка пройдена и данные обновлены.", - "val_addition": "\n\nНо для начала нужно авторизоваться. Нажмите на кнопку ниже:", - "val_info": "Вы авторизованы!\nВаш id в телеграм: {}\nФамилия: {}\nНомер профсоюзного " - "билета: {}", - "unknown_command": "Неизвестная команда.\nУ бота лишь три команды: /start /help /auth", - "only_pdf": "Документы на печать принимаются только в формате PDF", - "doc_not_accepted": "⚠️ Документ не принят, сначала авторизуйтесь.\n" - "Отправьте фамилию и номер профсоюзного билета в формате:\n\nИванов\n1234567", - "file_size_error": "⚠️ Принимаются только файлы размером меньше 3 MB.\nФайл {} не принят.", - "send_to_print": "✅ Файл {} успешно загружен. " - "Для печати подойдите к принтеру и введите PIN:\n\n{}\n\n" - "Для быстрой печати отсканируйте QR код на экране принтера.", - "qr_print": "{}{}", - "settings_warning": "\nНастройки сохраняются автоматически.", - "settings_change_fail": "Что-то сломалось, настройки печати не изменены, попробуйте через пару минут.", - "unknown_keyboard_payload": "Видимо бот обновился, выполните команду /start", - "im_broken": "Глубоко внутри меня что-то сломалось...\nПопробуйте через пару минут.", - "download_error": "Ошибка получения файла, попробуйте позже.", - "print_err": "😵 Ошибка сервера печати. Попробуйте позже.", - "db_err": "😵 Ошибка базы данных. Попробуйте ещё раз, если не получилось, то попробуйте позже.", -} +from dataclasses import dataclass + + +@dataclass +class Answers: + auth = '🔑 Авторизация' + about = '📄 Описание' + back = '◀️ Назад' + qr = '📷 Печать по QR' + kb_print = '⚙️ Настройки печати' + kb_print_copies = '📑 Копий:' + kb_print_side = '📎 Односторонняя печать' + kb_print_two_side = '🖇 Двухсторонняя печать' + hello = '👋🏻 Привет! Я телеграм-бот бесплатного принтера.\n' 'Отправьте PDF файл и получите PIN для печати.' + help = ('Я телеграм-бот бесплатного принтера профкома студентов физического факультета МГУ!\n\n' + '❔ Отправьте PDF файл и получите PIN для печати. ' + 'Поддерживаются только .pdf файлы не более 3МБ.\n' + 'С этим PIN необходимо подойти к принтеру и ввести его в терминал печати. ' + 'Либо отсканировать QR-код на принтере с помощью кнопки. После этого начнётся печать.' + '\n\n' + '⚙️ Настройки печати можно изменять после отправки файла, они сохраняются автоматически. ' + 'В момент печати используются самые последние настройки.\n\n' + '❗️ Файлы, которые вы отправляете через бота, будут храниться в течение нескольких месяцев' + ' на сервере в Москве, а также в этом чате Telegram.\n' + 'Доступ к файлам имеет узкий круг лиц, ответственных за работоспособность сервиса печати.\n' + 'Мы НЕ рекомендуем использовать данный сервис для печати конфиденциальных документов!\n\n' + '💻 Бот разработан группой программистов профкома, ' + 'как и приложение Твой ФФ! ' + 'В приложении вы сможете найти больше настроек печати, расписание и много других возможностей.\n' + 'Так же есть бот для печати ВКонтакте.') + + val_fail = ('⚠️ Проверка не пройдена. Удостоверьтесь что вы состоите в профсоюзе и правильно ввели данные.\n\n' + 'Введите фамилию и номер профсоюзного билета в формате:\n\nИванов\n1234567') + val_pass = '🥳 Поздравляю! Проверка пройдена и данные сохранены для этого телеграм-аккаунта. Можете присылать pdf.' + val_need = ('👤 Для использования принтера необходимо авторизоваться.\n' + 'Отправьте фамилию и номер профсоюзного билета в формате:\n\nИванов\n1234567') + val_update_fail = ('Сообщение не распознано.\nЧтобы открыть инструкцию введите: /help\n' + 'Для того чтобы обновить данные авторизации введите фамилию и номер' + 'профсоюзного билета в формате:\n\nИванов\n1234567') + val_update_pass = '🥳 Поздравляю! Проверка пройдена и данные обновлены.' + val_addition = '\n\nНо для начала нужно авторизоваться. Нажмите на кнопку ниже:' + val_info = ('Вы авторизованы!\n' + 'Ваш id в телеграм: {}\n' + 'Фамилия: {}\n' + 'Номер профсоюзного билета: {}') + unknown_command = ('Неизвестная команда.\n' + 'У бота лишь три команды: /start /help /auth') + only_pdf = 'Документы на печать принимаются только в формате PDF' + doc_not_accepted = ('⚠️ Документ не принят, сначала авторизуйтесь.\n' + 'Отправьте фамилию и номер профсоюзного билета в формате:\n\nИванов\n1234567') + file_size_error = ('⚠️ Принимаются только файлы размером меньше 3 MB.\n' + 'Файл {} не принят.') + send_to_print = ('✅ Файл {} успешно загружен. Для печати подойдите к принтеру и введите PIN:\n\n' + '{}\n\n' + 'Для быстрой печати отсканируйте QR код на экране принтера.') + qr_print = '{}{}' + settings_warning = 'Настройки сохраняются автоматически.' + settings_change_fail = 'Что-то сломалось, настройки печати не изменены, попробуйте через пару минут.' + unknown_keyboard_payload = 'Видимо бот обновился, выполните команду /start' + im_broken = 'Глубоко внутри меня что-то сломалось...\nПопробуйте через пару минут.' + download_error = 'Ошибка получения файла, попробуйте позже.' + print_err = '😵 Ошибка сервера печати. Попробуйте позже.' + db_err = '😵 Ошибка базы данных. Попробуйте ещё раз, если не получилось, то попробуйте позже.' + err_message_type = 'Сообщение не распознано.\nЧтобы открыть инструкцию введите: /help' diff --git a/src/errors_solver.py b/src/errors_solver.py index 70c1b3c..4026c5c 100644 --- a/src/errors_solver.py +++ b/src/errors_solver.py @@ -10,7 +10,13 @@ from telegram.error import TelegramError from telegram.ext import ContextTypes -from src.answers import ans +from src.answers import Answers + +ans = Answers() + + +async def native_error_handler(update: object, context: ContextTypes.DEFAULT_TYPE): + pass def errors_solver(func): @@ -28,7 +34,7 @@ async def wrapper(update: Update, context: ContextTypes.DEFAULT_TYPE): except (SQLAlchemyError, psycopg2.Error) as err: logging.error(err) traceback.print_tb(err.__traceback__) - await context.bot.send_message(chat_id=update.message.chat.id, text=ans["db_err"]) + await context.bot.send_message(chat_id=update.message.chat.id, text=ans.db_err) except Exception as err: logging.error(err) traceback.print_tb(err.__traceback__) diff --git a/src/handlers.py b/src/handlers.py index 5cc4c90..83685f2 100644 --- a/src/handlers.py +++ b/src/handlers.py @@ -13,13 +13,13 @@ from telegram.ext import CallbackContext, ContextTypes from src import marketing -from src.answers import ans +from src.answers import Answers from src.db import TgUser from src.errors_solver import errors_solver from src.log_formatter import log_actor, log_formatter from src.settings import Settings - +ans = Answers() settings = Settings() engine = create_engine(url=str(settings.DB_DSN), pool_pre_ping=True, isolation_level="AUTOCOMMIT") Session = sessionmaker(bind=engine) @@ -28,15 +28,15 @@ @errors_solver @log_formatter async def handler_start(update: Update, context: ContextTypes.DEFAULT_TYPE): - keyboard_base = [[InlineKeyboardButton(ans["about"], callback_data="to_about")]] - text, reply_markup = __change_message_by_auth(update, ans["hello"], keyboard_base) + keyboard_base = [[InlineKeyboardButton(ans.about, callback_data="to_about")]] + text, reply_markup = __change_message_by_auth(update, ans.hello, keyboard_base) await update.message.reply_text(text=text, reply_markup=reply_markup, disable_web_page_preview=True) @errors_solver @log_formatter async def handler_help(update: Update, context: ContextTypes.DEFAULT_TYPE): - await update.message.reply_text(ans["help"], disable_web_page_preview=True, parse_mode=ParseMode("HTML")) + await update.message.reply_text(ans.help, disable_web_page_preview=True, parse_mode=ParseMode("HTML")) @errors_solver @@ -44,32 +44,32 @@ async def handler_help(update: Update, context: ContextTypes.DEFAULT_TYPE): async def handler_auth(update: Update, context: ContextTypes.DEFAULT_TYPE): requisites = __auth(update) if requisites is None: - await update.message.reply_text(ans["val_need"]) + await update.message.reply_text(ans.val_need) else: - await update.message.reply_text(ans["val_info"].format(*requisites), parse_mode=ParseMode("HTML")) + await update.message.reply_text(ans.val_info.format(*requisites), parse_mode=ParseMode("HTML")) @errors_solver @log_formatter async def handler_button_browser(update: Update, context: CallbackContext) -> None: if update.callback_query.data == "to_hello": - keyboard_base = [[InlineKeyboardButton(ans["about"], callback_data="to_about")]] - text, reply_markup = __change_message_by_auth(update, ans["hello"], keyboard_base) + keyboard_base = [[InlineKeyboardButton(ans.about, callback_data="to_about")]] + text, reply_markup = __change_message_by_auth(update, ans.hello, keyboard_base) elif update.callback_query.data == "to_about": - keyboard_base = [[InlineKeyboardButton(ans["back"], callback_data="to_hello")]] - text, reply_markup = __change_message_by_auth(update, ans["help"], keyboard_base) + keyboard_base = [[InlineKeyboardButton(ans.back, callback_data="to_hello")]] + text, reply_markup = __change_message_by_auth(update, ans.help, keyboard_base) elif update.callback_query.data == "to_auth": - keyboard_base = [[InlineKeyboardButton(ans["back"], callback_data="to_hello")]] - text, reply_markup = ans["val_need"], InlineKeyboardMarkup(keyboard_base) + keyboard_base = [[InlineKeyboardButton(ans.back, callback_data="to_hello")]] + text, reply_markup = ans.val_need, InlineKeyboardMarkup(keyboard_base) elif update.callback_query.data.startswith("print_"): await __print_settings_solver(update, context) return else: - text, reply_markup = ans["unknown_keyboard_payload"], None + text, reply_markup = ans.unknown_keyboard_payload, None await update.callback_query.edit_message_text( text=text, @@ -82,7 +82,7 @@ async def handler_button_browser(update: Update, context: CallbackContext) -> No @errors_solver @log_formatter async def handler_unknown_command(update: Update, context: ContextTypes.DEFAULT_TYPE): - await update.message.reply_text(ans["unknown_command"]) + await update.message.reply_text(ans.unknown_command) @errors_solver @@ -90,7 +90,7 @@ async def handler_unknown_command(update: Update, context: ContextTypes.DEFAULT_ async def handler_print(update: Update, context: ContextTypes.DEFAULT_TYPE): requisites = __auth(update) if requisites is None: - await context.bot.send_message(chat_id=update.message.chat.id, text=ans["doc_not_accepted"]) + await context.bot.send_message(chat_id=update.message.chat.id, text=ans.doc_not_accepted) logging.warning(f"{log_actor(update)} try print with no auth") return try: @@ -98,14 +98,14 @@ async def handler_print(update: Update, context: ContextTypes.DEFAULT_TYPE): logging.info(f"{log_actor(update)} get attachments OK") except FileSizeError: await update.message.reply_text( - text=ans["file_size_error"].format(update.message.document.file_name), + text=ans.file_size_error.format(update.message.document.file_name), reply_to_message_id=update.message.id, parse_mode=ParseMode("HTML"), ) logging.warning(f"{log_actor(update)} get attachments FileSizeError") return except TelegramError: - await update.message.reply_text(text=ans["download_error"], reply_to_message_id=update.message.id) + await update.message.reply_text(text=ans.download_error, reply_to_message_id=update.message.id) logging.warning(f"{log_actor(update)} get attachments download_error") return @@ -129,15 +129,15 @@ async def handler_print(update: Update, context: ContextTypes.DEFAULT_TYPE): [ [ InlineKeyboardButton( - text=ans["qr"], - web_app=WebAppInfo(ans["qr_print"].format(settings.PRINT_URL_QR, pin)), + text=ans.qr, + web_app=WebAppInfo(ans.qr_print.format(settings.PRINT_URL_QR, pin)), ) ], - [InlineKeyboardButton(ans["kb_print"], callback_data=f"print_settings_{pin}")], + [InlineKeyboardButton(ans.kb_print, callback_data=f"print_settings_{pin}")], ] ) await update.message.reply_text( - text=ans["send_to_print"].format(update.message.document.file_name, pin), + text=ans.send_to_print.format(update.message.document.file_name, pin), reply_markup=reply_markup, reply_to_message_id=update.message.id, disable_web_page_preview=True, @@ -153,7 +153,7 @@ async def handler_print(update: Update, context: ContextTypes.DEFAULT_TYPE): elif r.status_code == 413: await update.message.reply_text( - text=ans["file_size_error"].format(update.message.document.file_name), + text=ans.file_size_error.format(update.message.document.file_name), reply_to_message_id=update.message.id, parse_mode=ParseMode("HTML"), ) @@ -161,7 +161,7 @@ async def handler_print(update: Update, context: ContextTypes.DEFAULT_TYPE): return await context.bot.send_message( chat_id=update.effective_user.id, - text=ans["print_err"], + text=ans.print_err, parse_mode=ParseMode("HTML"), ) logging.warning(f"{log_actor(update)} print unknown error") @@ -170,14 +170,14 @@ async def handler_print(update: Update, context: ContextTypes.DEFAULT_TYPE): @errors_solver @log_formatter async def handler_mismatch_doctype(update: Update, context: ContextTypes.DEFAULT_TYPE): - await update.message.reply_text(ans["only_pdf"]) + await update.message.reply_text(ans.only_pdf) marketing.print_exc_format(tg_id=update.message.chat_id) @errors_solver @log_formatter async def handler_other_messages(update: Update, context: ContextTypes.DEFAULT_TYPE): - await update.message.reply_text('Сообщение не распознано.\nЧтобы открыть инструкцию введите: /help') + await update.message.reply_text(ans.err_message_type) @errors_solver @@ -189,10 +189,10 @@ async def handler_register(update: Update, context: ContextTypes.DEFAULT_TYPE): if text is None or len(text.split("\n")) != 2: with Session() as session: if session.query(TgUser).filter(TgUser.tg_id == chat_id).one_or_none() is None: - await context.bot.send_message(chat_id=chat_id, text=ans["val_need"]) + await context.bot.send_message(chat_id=chat_id, text=ans.val_need) logging.warning(f"{log_actor(update)} val_need") else: - await context.bot.send_message(chat_id=chat_id, text=ans["val_update_fail"]) + await context.bot.send_message(chat_id=chat_id, text=ans.val_update_fail) logging.warning(f"{log_actor(update)} val_update_fail") return @@ -209,19 +209,19 @@ async def handler_register(update: Update, context: ContextTypes.DEFAULT_TYPE): if r.json() and data is None: session.add(TgUser(tg_id=chat_id, surname=surname, number=number)) session.commit() - await context.bot.send_message(chat_id=chat_id, text=ans["val_pass"]) + await context.bot.send_message(chat_id=chat_id, text=ans.val_pass) marketing.register(tg_id=chat_id, surname=surname, number=number) logging.info(f"{log_actor(update)} register OK: {surname} {number}") return True elif r.json() and data is not None: data.surname = surname data.number = number - await context.bot.send_message(chat_id=chat_id, text=ans["val_update_pass"]) + await context.bot.send_message(chat_id=chat_id, text=ans.val_update_pass) marketing.re_register(tg_id=chat_id, surname=surname, number=number) logging.info(f"{log_actor(update)} register repeat OK: {surname} {number}") return True elif r.json() is False: - await context.bot.send_message(chat_id=chat_id, text=ans["val_fail"]) + await context.bot.send_message(chat_id=chat_id, text=ans.val_fail) marketing.register_exc_wrong(tg_id=chat_id, surname=surname, number=number) logging.info(f"{log_actor(update)} register val_fail: {surname} {number}") @@ -233,7 +233,7 @@ async def __print_settings_solver(update: Update, context: CallbackContext): if r.status_code == 200: options = r.json()["options"] else: - await update.callback_query.message.reply_text(ans["settings_change_fail"]) + await update.callback_query.message.reply_text(ans.settings_change_fail) return if button == "copies": @@ -241,29 +241,29 @@ async def __print_settings_solver(update: Update, context: CallbackContext): elif button == "twosided": options["two_sided"] = not options["two_sided"] else: - await context.bot.answer_callback_query(update.callback_query.id, ans["settings_warning"]) + await context.bot.answer_callback_query(update.callback_query.id, ans.settings_warning) r = requests.patch(settings.PRINT_URL + f"""/file/{pin}""", json={"options": options}) if r.status_code != 200: - await update.callback_query.message.reply_text(ans["settings_change_fail"]) + await update.callback_query.message.reply_text(ans.settings_change_fail) return keyboard = [ [ InlineKeyboardButton( - text=ans["qr"], - web_app=WebAppInfo(ans["qr_print"].format(settings.PRINT_URL_QR, pin)), + text=ans.qr, + web_app=WebAppInfo(ans.qr_print.format(settings.PRINT_URL_QR, pin)), ) ], [ InlineKeyboardButton( - f'{ans["kb_print_copies"]} {options["copies"]}', + f'{ans.kb_print_copies} {options["copies"]}', callback_data=f"print_copies_{pin}", ) ], [ InlineKeyboardButton( - ans["kb_print_two_side"] if options["two_sided"] else ans["kb_print_side"], + ans.kb_print_two_side if options["two_sided"] else ans.kb_print_side, callback_data=f"print_twosided_{pin}", ) ], @@ -286,8 +286,8 @@ def __auth(update): def __change_message_by_auth(update, text, keyboard): if __auth(update) is None: - text += ans["val_addition"] - keyboard.append([InlineKeyboardButton(ans["auth"], callback_data="to_auth")]) + text += ans.val_addition + keyboard.append([InlineKeyboardButton(ans.auth, callback_data="to_auth")]) return text, InlineKeyboardMarkup(keyboard)