From 0e0682b49c6699871b8d5a1e9799a0395a85afbd Mon Sep 17 00:00:00 2001 From: Andrey Marakulin Date: Sat, 9 Sep 2023 05:42:45 +0300 Subject: [PATCH] Black, isort and replace " to ' --- src/__main__.py | 2 +- src/answers.py | 84 +++++++++++++++++++-------------- src/errors_solver.py | 1 + src/handlers.py | 109 ++++++++++++++++++++++--------------------- 4 files changed, 105 insertions(+), 91 deletions(-) diff --git a/src/__main__.py b/src/__main__.py index 35a7fa1..ff7f86c 100644 --- a/src/__main__.py +++ b/src/__main__.py @@ -5,6 +5,7 @@ from telegram.ext import ApplicationBuilder, CallbackQueryHandler, CommandHandler, MessageHandler, filters +from src.errors_solver import native_error_handler from src.handlers import ( handler_auth, handler_button_browser, @@ -16,7 +17,6 @@ handler_start, handler_unknown_command, ) -from src.errors_solver import native_error_handler from src.settings import Settings diff --git a/src/answers.py b/src/answers.py index c598a91..87d9aba 100644 --- a/src/answers.py +++ b/src/answers.py @@ -15,47 +15,59 @@ class Answers: 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' - 'Так же есть бот для печати ВКонтакте.') + 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_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_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') + 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 код на экране принтера.') + 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 = 'Что-то сломалось, настройки печати не изменены, попробуйте через пару минут.' diff --git a/src/errors_solver.py b/src/errors_solver.py index 4026c5c..958d70f 100644 --- a/src/errors_solver.py +++ b/src/errors_solver.py @@ -12,6 +12,7 @@ from src.answers import Answers + ans = Answers() diff --git a/src/handlers.py b/src/handlers.py index 83685f2..46bdc5a 100644 --- a/src/handlers.py +++ b/src/handlers.py @@ -19,16 +19,17 @@ 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") +engine = create_engine(url=str(settings.DB_DSN), pool_pre_ping=True, isolation_level='AUTOCOMMIT') Session = sessionmaker(bind=engine) @errors_solver @log_formatter async def handler_start(update: Update, context: ContextTypes.DEFAULT_TYPE): - keyboard_base = [[InlineKeyboardButton(ans.about, callback_data="to_about")]] + 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) @@ -36,7 +37,7 @@ async def handler_start(update: Update, context: ContextTypes.DEFAULT_TYPE): @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 @@ -46,25 +47,25 @@ async def handler_auth(update: Update, context: ContextTypes.DEFAULT_TYPE): if requisites is None: 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")]] + 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) - elif update.callback_query.data == "to_about": - keyboard_base = [[InlineKeyboardButton(ans.back, callback_data="to_hello")]] + 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) - elif update.callback_query.data == "to_auth": - keyboard_base = [[InlineKeyboardButton(ans.back, callback_data="to_hello")]] + 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) - elif update.callback_query.data.startswith("print_"): + elif update.callback_query.data.startswith('print_'): await __print_settings_solver(update, context) return @@ -75,7 +76,7 @@ async def handler_button_browser(update: Update, context: CallbackContext) -> No text=text, reply_markup=reply_markup, disable_web_page_preview=True, - parse_mode=ParseMode("HTML"), + parse_mode=ParseMode('HTML'), ) @@ -91,39 +92,39 @@ 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) - logging.warning(f"{log_actor(update)} try print with no auth") + logging.warning(f'{log_actor(update)} try print with no auth') return try: filebytes, filename = await __get_attachments(update, context) - logging.info(f"{log_actor(update)} get attachments OK") + 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), reply_to_message_id=update.message.id, - parse_mode=ParseMode("HTML"), + parse_mode=ParseMode('HTML'), ) - logging.warning(f"{log_actor(update)} get attachments FileSizeError") + 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) - logging.warning(f"{log_actor(update)} get attachments download_error") + logging.warning(f'{log_actor(update)} get attachments download_error') return r = requests.post( - settings.PRINT_URL + "/file", - json={"surname": requisites[1], "number": requisites[2], "filename": filename, "source": "tgbot"}, + settings.PRINT_URL + '/file', + json={'surname': requisites[1], 'number': requisites[2], 'filename': filename, 'source': 'tgbot'}, ) if r.status_code == 200: - pin = r.json()["pin"] + pin = r.json()['pin'] files = { - "file": ( + 'file': ( filename, filebytes.getvalue(), - "application/pdf", - {"Expires": "0"}, + 'application/pdf', + {'Expires': '0'}, ) } - rfile = requests.post(settings.PRINT_URL + "/file/" + pin, files=files) + rfile = requests.post(settings.PRINT_URL + '/file/' + pin, files=files) if rfile.status_code == 200: reply_markup = InlineKeyboardMarkup( [ @@ -133,7 +134,7 @@ async def handler_print(update: Update, context: ContextTypes.DEFAULT_TYPE): 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( @@ -141,9 +142,9 @@ async def handler_print(update: Update, context: ContextTypes.DEFAULT_TYPE): reply_markup=reply_markup, reply_to_message_id=update.message.id, disable_web_page_preview=True, - parse_mode=ParseMode("HTML"), + parse_mode=ParseMode('HTML'), ) - logging.info(f"{log_actor(update)} print success") + logging.info(f'{log_actor(update)} print success') marketing.print_success( tg_id=update.message.chat.id, surname=requisites[1], @@ -155,16 +156,16 @@ async def handler_print(update: Update, context: ContextTypes.DEFAULT_TYPE): await update.message.reply_text( text=ans.file_size_error.format(update.message.document.file_name), reply_to_message_id=update.message.id, - parse_mode=ParseMode("HTML"), + parse_mode=ParseMode('HTML'), ) - logging.warning(f"{log_actor(update)} print api 413 SizeErr") + logging.warning(f'{log_actor(update)} print api 413 SizeErr') return await context.bot.send_message( chat_id=update.effective_user.id, text=ans.print_err, - parse_mode=ParseMode("HTML"), + parse_mode=ParseMode('HTML'), ) - logging.warning(f"{log_actor(update)} print unknown error") + logging.warning(f'{log_actor(update)} print unknown error') @errors_solver @@ -186,22 +187,22 @@ async def handler_register(update: Update, context: ContextTypes.DEFAULT_TYPE): text = update.message.text chat_id = update.message.chat.id - if text is None or len(text.split("\n")) != 2: + 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) - logging.warning(f"{log_actor(update)} 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) - logging.warning(f"{log_actor(update)} val_update_fail") + logging.warning(f'{log_actor(update)} val_update_fail') return - if len(text.split("\n")) == 2: - surname = text.split("\n")[0].strip() - number = text.split("\n")[1].strip() + if len(text.split('\n')) == 2: + surname = text.split('\n')[0].strip() + number = text.split('\n')[1].strip() r = requests.get( - settings.PRINT_URL + "/is_union_member", + settings.PRINT_URL + '/is_union_member', params=dict(surname=surname, v=1, number=number), ) with Session() as session: @@ -211,39 +212,39 @@ async def handler_register(update: Update, context: ContextTypes.DEFAULT_TYPE): session.commit() 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}") + 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) marketing.re_register(tg_id=chat_id, surname=surname, number=number) - logging.info(f"{log_actor(update)} register repeat OK: {surname} {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) marketing.register_exc_wrong(tg_id=chat_id, surname=surname, number=number) - logging.info(f"{log_actor(update)} register val_fail: {surname} {number}") + logging.info(f'{log_actor(update)} register val_fail: {surname} {number}') async def __print_settings_solver(update: Update, context: CallbackContext): - _, button, pin = update.callback_query.data.split("_") + _, button, pin = update.callback_query.data.split('_') - r = requests.get(settings.PRINT_URL + f"""/file/{pin}""") + r = requests.get(settings.PRINT_URL + f'''/file/{pin}''') if r.status_code == 200: - options = r.json()["options"] + options = r.json()['options'] else: await update.callback_query.message.reply_text(ans.settings_change_fail) return - if button == "copies": - options["copies"] = options["copies"] % 5 + 1 - elif button == "twosided": - options["two_sided"] = not options["two_sided"] + if button == 'copies': + options['copies'] = options['copies'] % 5 + 1 + elif button == 'twosided': + options['two_sided'] = not options['two_sided'] else: 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}) + 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) return @@ -258,13 +259,13 @@ async def __print_settings_solver(update: Update, context: CallbackContext): [ InlineKeyboardButton( f'{ans.kb_print_copies} {options["copies"]}', - callback_data=f"print_copies_{pin}", + callback_data=f'print_copies_{pin}', ) ], [ InlineKeyboardButton( - ans.kb_print_two_side if options["two_sided"] else ans.kb_print_side, - callback_data=f"print_twosided_{pin}", + ans.kb_print_two_side if options['two_sided'] else ans.kb_print_side, + callback_data=f'print_twosided_{pin}', ) ], ] @@ -277,7 +278,7 @@ def __auth(update): tg_user: TgUser | None = session.query(TgUser).filter(TgUser.tg_id == update.effective_user.id).one_or_none() if tg_user is not None: r = requests.get( - settings.PRINT_URL + "/is_union_member", + settings.PRINT_URL + '/is_union_member', params=dict(surname=tg_user.surname, number=tg_user.number, v=1), ) if r.json(): @@ -287,7 +288,7 @@ 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")]) + keyboard.append([InlineKeyboardButton(ans.auth, callback_data='to_auth')]) return text, InlineKeyboardMarkup(keyboard)