Нет, это не одна из сотен статей о том, как написать свой первый Hello World бот на Python. Здесь вы не найдете подробной инструкции, как получить API-токен в BotFather или запустить бота в облаке. Взамен я вам покажу как раскрутить всю мощь Python на максимум, чтобы добиться максимально эстетичного и красивого кода. Исполним песню про обращение сложных структур — станцуем и спляшем. Под катом асинхронность, собственная система сейвов, куча полезных декораторов и много красивого кода.
Дисклеймер: люди с ООП головного мозга и адепты "правильных" паттернов могут игнорировать эту статью.
Идея
Чтобы понять, какого это, не знать английский в современном обществе, представьте себе, что вы дворянин XVIII века, который не знает французский. Даже если вы не очень хорошо разбираетесь в истории, то все равно представляете, насколько тяжело было бы жить при таких обстоятельствах. В современном мире английский стал уже необходимостью, а не уделом привилегированных, особенно если вы крутитесь в IT-индустрии.
В основе проекта заложен катехизис будущего: развитие нейронной сети, как обособленной единицы, и образование, в основе которого лежат игры и спортивный дух. Изоморфные парадигмы висят в воздухе еще с давних времен, но, кажется, со временем люди начали забывать, что самые простые решения — наиболее эффективные.
Вот краткий список базовых вещей, которые я хочу собрать воедино:
- Уметь работать с тремя пользовательскими словарями
- Возможность парсить youtube-видео/текст, а затем добавлять новые слова в словарь пользователя
- Два базовых режима тренировки навыков
- Гибкая настройка: полный контроль над пользовательскими словарями и средой в целом
- Встроенная админ панель
Естественно все должно работать быстро, с возможностью в будущем легко пополнять имеющийся функционал. Собрав все вместе, я подумал, что наилучшим воплощением моей идеи в реальность будет Telegram-бот.
Мой сказ не о том, как правильно написать обработчиков для бота — таких статей десятки и это несложная механическая работа. Я хочу, чтобы читатель научился абстрагироваться от типичных догм программирования. Используйте то, что выгодно и эффективно, здесь и сейчас.
«Я научился пропускать возгласы неверующих мимо ушей, поскольку подавить их было невозможно»
Базовая структура
Бот будет базироваться на библиотеке python-telegram-bot (ptb). В качестве логгера я использую loguru, правда здесь есть одна небольшая загвоздка. Дело в том, что ptb по умолчанию использует другой логгер (стандартный logging) и не дает возможности подключить собственный. Конечно, можно было бы перехватить все сообщения журнала глобально и направить их в наш регистратор, но мы поступим несколько проще:
from loguru import logger import sys # Настраиваем двухпоточный вывод: в консоль и в файл config = { 'handlers': [ {'sink': sys.stdout, 'level': 'INFO'}, {'sink': 'logs.log', 'serialize': False, 'level': 'DEBUG'}, ], } logger.configure(**config) # ... updater = Updater('YOUR_TOKEN') dp = updater.dispatcher # Грубо прикручиваем свой логгер. Дешево и сердито updater.logger = logger dp.logger = loggerК сожалению, у меня нет возможности развернуть своего бота на стабильных дата-центрах, поэтому сохранность данных — первоочередная задача. Для этих целей я реализовал собственную систему сейвов. Она предоставляет гибкую и удобную работу с данными — сбор статистики, как пример легкого пополнения функционала в будущем.
from __future__ import annotations # В дальнейшем я буду опускать этот импорт from loguru import logger import inspect import functools import os def file_is_empty(path: str) -> bool: return os.stat(path).st_size == 0 def clear_file(path: str) -> None: with open(path, 'w'): pass def cache_decorator(method): @functools.wraps(method) def wrapper(self, *args, **kwargs): res = method(self, *args, **kwargs) Cache.link.recess(self, {'method_name': method.__name__}) # (1) # В процессе тестирования множественные вызовы могут замедлять # работу программы; opt позволяет избежать этого logger.opt(lazy=True).debug(f'Decorator for {method.__name__} was end') return res return wrapper class Cache: """ + cache_size - Шаг, через который будет происходить сохранение всех данных + cache_files - Файл дампов, в котором сохраняются все промежуточные операции над данными """ link = None def __init__(self, cache_size=10): # Сохраняем все прикрученные классы. Это позволяет гибко работать с данными self._classes = [] # Файлы, соответствующие классам self._cache_files = [] # (1): Небольшой хак, который позволяет вызвать конкретный экземпляр через общий класс # Это работает, потому что у нас есть всего один экземпляр класса, который # реализует всю логику работы с данными. К тому же, это удобно и позволяет # значительно расширить функционал в будущем self.__class__.link = self self._counter = 0 self.CACHE_SIZE = cache_size def add(self, cls: class, file: str) -> None: """ Позволяет прикрутить класс к сейверу + cls - Экземпляр класса + file - Файл, с которым работает экземпляр """ self._cache_files.append(file) self._classes.append(cls) if file_is_empty(file): return None logger.opt(lazy=True).debug(f'For {cls.__class__.__name__} file {file} is not empty') for data in self.load(file): cls.save_non_caching(data) clear_file(file) self._counter = 0 def recess(self, cls: class, data: dict) -> None: """ Основной метод, выполняющий основную логику сейвов """ if self._counter + 1 >= self.CACHE_SIZE: self.save_all() else: self._counter += 1 filename = self._cache_files[self._classes.index(cls)] self.save(data, filename=filename) # ... # Для простоты методы save_all, save, load опущены # ... Теперь мы можем создавать любые методы, которые могут изменять данные, без боязни потерять важные данные:
@cache_decorator def add_smth_important(*args, **kwargs) -> Any: # ... # Производим какие-то важные действия над данными... # ...Теперь, когда мы разобрались с основной структурой, остается главный вопрос: каким образом собрать все воедино. Я реализовал главный класс — EnglishBot, который собирает вместе всю базовую структуру: ptb, работа с БД, система сейвов, и который будет управлять всей бизнес-логикой бота. Если бы реализация Telegram-команд была простой, мы бы легко смогли добавить их в этот же класс. Но, к сожалению, их организация занимает большую часть кода всего приложения, поэтому добавление их в этот же класс было бы безумством. Создавать новые классы/подклассы мне тоже не хотелось, потому я предлагаю использовать очень простую структуру:
# Импортируем основной класс from modules import EnglishBot # Импортируем классы, которые реализуют Telegram-команды from modules.module import start # ... if __name__ == '__main__': # Инициализируем бота tbot = EnglishBot( # ... ) # Добавляем обработчики в стек tbot.add_command_handler(start, 'start') # ...Каким образом модули команд получают доступ к основному классу, мы рассмотрим далее.
Все из ничего
У ptb обработчиков есть два аргумента — update и context, в которых хранится весь необходимый стек информации. У context есть замечательный аргумент chat_data, который может использоваться как словарь хранения данных для чата. Но я не хочу постоянно обращаться к нему в формате context.chat_data['data']. Хотелось бы чего-нибудь легкого и красивого, скажем context.data. Впрочем, это не проблема.
from telegram.ext import CommandHandler def a(self, key: str): # Сначала проверяем нет ли необходимого значения у класса # Если нет, то пытаемся вернуть его из chat_data try: return object.__getattribute__(self, key) except: return self.chat_data[key] def b(self, key: str, data=None, replace=True): # Небольшой хак: если replace=False и данные существуют, то перезапись не происходит # При этом, если данные не указаны, то они ставятся в None if replace or not self.chat_data.get(key, None): self.chat_data[key] = data # Биндим context, чтобы получать данные в формате context.data CallbackContext.__getattribute__ = a # А также удобный сеттер для своих нужд CallbackContext.set = bПродолжаем упрощать себе жизнь. Теперь я хочу, чтобы вся необходимая информация для конкретного пользователя была в быстром доступе context.
def bind_context(func): def wrapper(update, context): context._bot.bind_user_data(update, context) # (2) return func(update, context) return wrapper class EnglishBot: # ... def bind_user_data(self, update, context) -> dict: context.set('t_id', update.message.chat_id, replace=False) context.set('t_ln', update.message.from_user.language_code, replace=False) # ... # Устанавливаем всю необходимую информацию, к которой хотим иметь быстрый доступ из context # Например что-нибудь из базы данных # ...Теперь совсем обнаглеем и прикрутим экземпляр нашего бота к context:
class EnglishBot: # ... def __init__(self, *args, **kwargs): # ... # (2): Теперь мы можем обращаться к экземпляру в формате context._bot CallbackContext._bot = selfСобираем все в месте и получаем цитадель комфорта и удобства всего в одном вызове.
from EnglishBot import bind_context @bind_context def start(update, context): # Теперь мы имеем доступ ко всему из одного места # Например мы можем с легкостью добавить нового пользователя в БД # Проверяем есть ли пользователь в нашей БД if not context._bot.user_exist(context.t_id): # Например добавим какое-нибудь важное уведомления context.set('push_notification', True) # А затем добавим пользователя в БД context._bot.new_user(context.t_id, context.t_ln) return update.message.reply_text('Добро пожаловать') # ...Декораторы — наше все
Чаще всего оказывается так, что имя функции совпадает с именем команды, к которой мы хотим добавить обработчик. Почему бы не использовать данную статистическую особенность в своих корыстных целях.
class EnglishBot: # ... def add_command_handler(self, func: function, name=None) -> None: """ Функция, которая добавляет обработчика команд """ name = name or func.__name__ self.dp.add_handler(CommandHandler(name, func)) # ... # В основном файле: tbot.add_command_handler(start) # Вместо tbot.add_command_handler(start, 'start')Это выглядит круто, но это не работает. Все дело в декораторе bind_context, который всегда будет возвращать имя функции wrapper. Исправим это недоразумение.
import functools def bind_context(func): # functools.wraps из stdlib сохраняет подписи начиная с Python 3.4 @functools.wraps(func) def wrapper(update, context): context._bot.bind_user_data(update, context) return func(update, context) return wrapperВ боте имеется много обработчиков сообщений, которые по задумке должны отменять команду при вводе нуля. Также мне нужно отбросить все отредактированные сообщения.
import functools END = -1 def zero_exiter(func): @functools.wraps(func) def wrapper(update, context): if update.to_dict()['message'].get('text', None) == '0': update.message.reply_text('Отправляем какое-то сообщение') return END return func(update, context) return wrapper def skip_edited(func): @functools.wraps(func) def wrapper(update, context): # Это работает во всех случаях, потому что None, возвращенный # в стеке conversation_handler, оставляет функцию на текущем состоянии if not update.to_dict().get('edited_message', None): return func(update, context) return wrapper Не забываем при этом про самый главный декоратор — @run_async, на котором зиждется асинхронность. Теперь собираем тяжеловесную функцию.
from telegram.ext.dispatcher import run_async from EnglishBot import skip_edited @run_async @skip_edited def heavy_function(update, context): # ... # Тяжелая вычислительная функция, которая нуждается в асинхронности # Имеет защиту от редактирования сообщения # ...Помните, что асинхронность — джедайский меч, но этим мечом вы можете с легкостью себя убить.
Порой программы зависают, ничего не передавая в журнал. @logger.catch, последний декоратор в нашем списке, гарантирует, что любая ошибка будет правильно передана в logger.
from loguru import logger @logger.catch def heavy_function2(update, context): # ... # Еще одна тяжелая вычислительная функция, на которой программа может зависнуть # ...Админ панель
Давайте реализуем админ панель с возможностью получать/удалять логи и отсылать сообщение всем пользователям.
from EnglishBot import bind_context from Cache import file_is_empty from telegram import ReplyKeyboardMarkup from loguru import logger LOG_FILE = 'logs.log' SENDING, MAIN, END = range(1, -2, -1) buttons = ReplyKeyboardMarkup( [('Получить логи', 'logs'), ('Очистить логи', 'clear')], [('Отправить сообщение', 'send')], # [...], ) @bind_context def admin_panel(update, context): # Проверяем права администратора у пользователя if not context._bot.acess_check(context.t_id): # Хорошей практикой безопасности будет показать пользователю, что такой команды не существует return update.message.reply_text(f'Неизвестная команда {update.message.text}') update.message.reply_text('Выбери опцию:', reply_markup=buttons) return MAIN # # Logs methods # def get_logs(update, context): if file_is_empty(LOG_FILE): update.callback_query.reply_text('Логи пусты') else: # Показываем загрузку документа context.bot.send_chat_action(chat_id=context.t_id, action='upload_document') context.bot.sendDocument(chat_id=context.t_id, document=open(LOG_FILE, 'rb'), name=LOG_FILE, timeout=1000) # Поскольку мы не закрываем админку, необходимо убрать значок загрузки на кнопке update.callback_query.answer(text='') # В принципе, это излишне. Как я говорил ранее, None не меняет положение обработчика # Но так код выглядит гораздо читабельнее return MAIN def logs_clear(update, context): with open(LOG_FILE, 'w') as file: update.callback_query.reply_text('Очищено') update.callback_query.answer(text='') return MAIN # # Send methods # def take_message(update, context): update.callback_query.reply_text('Отправь сообщение') update.callback_query.answer(text='') return SENDING @zero_exiter def send_all(update, context): count = 0 # Получаем айдишники пользователей из БД for id in list(context._bot.get_user_ids()): # Может получиться так, что какой-то пользователь добавил бота в ЧС # Тогда при попытке отослать ему сообщение будет поймана ошибка try: # Сообщение самому себе не отправляем if id == context.t_id: continue # Обязательно используем Markdown, чтобы сохранить оформление сообщений context.bot.send_message(context.t_id, text=update.message.text_markdown, parse_mode='Markdown') count += 1 except: pass update.callback_query.reply_text(f'Отправлено {count} людям') update.callback_query.answer(text='') return MAIN # ... # В основном файле: tbot.add_conversation_handler( entry_points = [('admin', admin)], # Регулярка для send_alk позволяет обработать любое сообщение, которое начинается не с / states = [[(logs, '^logs$'), (logs_clear, '^clear$'), (send, '^send$')], [(send_alk, '@^((?!.*((^/)+)).*)(.+)$')]] )Функция add_conversation_handler позволяет минималистично добавить обработчик разговора:
class EnglishBot: # ... def add_conversation_handler(self, entry_points: list, states: list, fallbacks: list) -> None: fallbacks = [CommandHandler(name, func) for name, func in fallbacks] entry_points = [CommandHandler(name, func) for name, func in entry_points] r_states = {} for i in range(len(states)): r_states[i] = [] # Каждый массив описывает функции одного состояния for func, pattern in states[i]: # Если регулярка начинается с символа @, то мы добавляем обработчик сообщений # Иначе - обычный обработчик для кнопок if pattern[0] == '@': r_states[i].append(MessageHandler(Filters.regex(pattern[1:]), func)) else: r_states[i].append(CallbackQueryHandler(func, pattern=pattern)) conv_handler = ConversationHandler(entry_points=entry_points, states=r_states, fallbacks=fallbacks) dp.add_handler(conv_handler)Основной функционал
Давайте научим нашего бота добавлять новые слова в пользовательские словарики.
from EnglishBot import bind_context, skip_edited, zero_exiter from youtube_transcript_api import YouTubeTranscriptApi from telegram.ext.dispatcher import run_async START_OVER, ADDING, END = range(1, -2, -1) re_youtube = re.compile('^(http(s)?://)?((w){3}.)?youtu(be|.be)?(.com)?/.+') re_text = re.compile('^[a-z]{3,20}$') def is_youtube_link(link: str) -> bool: if re_youtube.match(link) is not None: return True def clear_text(text: str) -> str: bad_symbols = '!@#%$^&*()_+1234567890-=/|?><.,":;`~[]{}' for s in bad_symbols: text = text.replace(s, '') return text.strip().lower() @skip_edited @bind_context def add_words(update, context): update.message.reply_text('Введите текст:') return ADDING @run_async @skip_edited @zero_exiter def parse_text(update, context): # Загрузка занимает какое-то время, поэтому необходимо уведомить пользователя, что все нормально message = update.message.reply_text('Загрузка...') if is_youtube_link(update.message.text): # Если видео невалидное или в нем нет субтитров, то мы поймаем ошибку try: transcript_list = YouTubeTranscriptApi.list_transcripts(get_video_id(update.message.text)) t = transcript_list.find_transcript(['en']) _text = clear_text('. '.join([i['text'] for i in t.fetch()])).split() except: message.edit_text('Невалидное видео. Попробуйте еще раз: ') return ADDING else: _text = clear_text(update.message.text).split() # Получаем ссылку на уже имеющиеся у пользователя слова _words = context._bot.get_dict_words(context.t_id) # Слова, которые будут добавлены good_words = [] # Отброшенные слова bad_words = [] # Первым делом отбрасываем дубли for word in set(_text): # Потом проверяем корректность слова регуляркой z = re_text.match(word) if z: # Добавляем слово только если его еще нет в словаре юзера if z.group() not in _words: good_words.append(word) else: bad_words.append(word) # Осталось дело за малым - предложить пользователю добавить слова # А затем добавить их в словарь # ... # ... # В основном файле: tbot.add_conversation_handler( entry_points = [('add_words', add_words)], states = [ [(parse_text, '@^((?!.*((^/)+)).*)(.+)$')], # ... ])Упаковываем бота
Прежде чем мы запакуем нашего бота, добавим поддержку прокси и изменения уровня логов из консоли.
# В основном файле: import argparse parser = argparse.ArgumentParser() parser.add_argument('-l', '--level', default='INFO', choices=['TRACE', 'DEBUG', 'INFO', 'SUCCESS', 'WARNING', 'ERROR', 'CRITICAL'], help='Позволяет включить бота в заданном режиме') parser.add_argument('-p', '--proxy', help='Позволяет включить бота с прокси') args = parser.parse_args() config = { 'handlers': [ {'sink': sys.stdout, 'level': args.level}, # Это не ошибка. В файле я собираю логи уровня DEBUG и выше {'sink': 'logs.log', 'serialize': False, 'level': 'DEBUG'}, ], } logger.configure(**config) if args.proxy: t_proxy = {'proxy_url': args.proxy, 'read_timeout': 1000, 'connect_timeout': 1000} # ... # Прокси для других сервисов # ... else: t_proxy = None Python 3.5+ поддерживает возможность упаковать каталог в один исполняемый файл. Давайте воспользуемся этой возможностью, чтобы иметь возможность легко развернуть рабочую среду на любом VPS. Для начала получим файл зависимостей. Если вы используете виртуальную среду, то это можно сделать одной командой: pip freeze > requirements.txt. Если в проекте не развернута виртуальная среда, то здесь придется немного повозиться. Можно попробовать использовать pip freeze и вручную вычленить все необходимые пакеты, однако если в системе установлено слишком много пакетов, то такой способ явно не подойдет. Второй вариант — воспользоваться готовыми решениями, например pipreqs.
Теперь, когда наш файл с зависимостями готов, мы можем упаковать наш каталог в .pyz файл. Для этого необходимо ввести команду py -m zipapp "ПУТЬ_К_КАТАЛОГУ" -m "ИМЯ_ГЛАВНОГО_ФАЙЛА:ГЛАВНАЯ_ФУНКЦИЯ" -o bot.pyz, она создаст файл bot.pyz в папке с проектом. Учтите, что код в __init__.py должен быть завернут в какую-то функцию, иначе исполняемый файл будет невозможно скомпилировать.
# Пример __init__.py файла # py -m zipapp "ПУТЬ_К_КАТАЛОГУ" -m "__init__.py:main" -o bot.pyz def main(): # ... if __name__ == '__main__': main() Заворачиваем файлы в архив zip bot.zip requirements.txt bot.pyz и отправляем его на наш VPS.
Итоги
Ссылочная мощь Python позволила нам написать быстрого бота, при этом, не выходя за рамки красивого кода. Такая структура оформления проекта позволит в будущем легко пополнять функционал. Потыкать бота можно здесь (или тут если ресурс заблокирован).