24.02.2020 13:57
laniminel 1 4300 Источник

image


Нет, это не одна из сотен статей о том, как написать свой первый 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 позволила нам написать быстрого бота, при этом, не выходя за рамки красивого кода. Такая структура оформления проекта позволит в будущем легко пополнять функционал. Потыкать бота можно здесь (или тут если ресурс заблокирован).