Перейти к содержанию

Документация по использованию основных воркеров

Middleware

Middleware (мидлвари далее) используются как первичная обработка, с помощью них можно, например:

  • Не обрабатывать сообщения от ботов
  • Регистрировать пользователя и возвращать в хендлер как обязательный аргумент
  • Записывать статистику

Все примеры будут разобраны ниже

ВНИМАНИЕ! С версии 2.7.5 Middleware должен имплементить методы pre и post, а не middleware, как было раньше. Если вы пользуетесь версией ниже 2.7.5 замените pre на middleware, а от post придется воздержаться

Не обрабатывать сообщения от ботов

Создадим простой мидлварь, который будет возвращать False если источник сообщения - бот, а не пользователь:

from vkbottle.bot import Bot, Message
from vkbottle.ext import Middleware

bot = Bot("token")

@bot.middleware.middleware_handler()
class NoBotMiddleware(Middleware):
    async def pre(self, message: Message, *args):
        if not message.from_user:
            return False

@bot.on.message_handler(text="тест")
async def wrapper(ans: Message):
    await ans("Работаю только для юзеров")

bot.run_polling()

Если сообщение будет отправлено не юзером то бот не будет его обрабатывать

Регистрировать пользователя и возвращать в хендлер как обязательный аргумент

from vkbottle.bot import Bot, Message
from vkbottle.ext import Middleware
import typing

bot = Bot("token")
database: typing.Dict[int, str] = {}  # Наш прототип базы данных

@bot.middleware.middleware_handler()
class RegistrationMiddleware(Middleware):
    async def pre(self, message: Message, *args):
        if message.from_id not in database:
            database[message.from_id] = message.text
        return database[message.from_id]

@bot.on.message_handler(text="привет", lower=True)
async def wrapper(ans: Message, first_message: str):
    await ans(f"Я все помню, твое первое сообщение: <<{first_message}>>")

bot.run_polling()

Записывать статистику

from vkbottle.bot import Bot, Message
from vkbottle.ext import Middleware
import typing

bot = Bot("token")
poor_statistics: typing.List[int] = []

@bot.middleware.middleware_handler()
class StatisticsMiddleware(Middleware):
    async def pre(self, message: Message, *args):
        user = (await bot.api.users.get(user_ids=message.from_id))[0]
        poor_statistics.append(user.sex)

@bot.on.message_handler(text="стат")
async def wrapper(ans: Message):
    stat = (sum(poor_statistics) / len(poor_statistics)) - 1
    await ans(f"{round(stat * 100, 2)} пользователей бота - девочки")

bot.run_polling()

Грызем провода

Если вам по какой-то причине нужно возвращать boolean тип из хендлеров вы можете отключить это:

bot.status.middleware_expressions = False

Error Handler

Swear

Вы можете хендлить все исключения, которые выбрасываются хендлером с помощью декоратора swear:

from vkbottle.framework import swear
from vkbottle import VKError

async def exc_kick(e: VKError, message: Message):
    await message("Не могу кикнуть =(")

@bot.on.chat_message(commands=["самобан"])
@swear(VKError, exception_handler=exc_kick)
async def self_ban(ans: Message):
    await bot.api.messages.remove_chat_user(ans.chat_id, ans.from_id)
    return "хаха самобан"

Error Handler

Для того чтобы хендлить ошибки вк без выбрасывания исключения можно воспользоваться error_handler. Он отлично подойдет для, например, решения капчи или delay для методов которые привысили лимит (частая проблема для пользователей User Longpoll)

Для того чтобы добавить простейший хендлер ошибки можно воспользоваться декоратором:

from vkbottle.user import User, Message
from vkbottle.exceptions import VKError
from asyncio import sleep

user = User("token")

@user.error_handler.error_handler(6)
async def rps_handler(e: VKError):
    await sleep(1)
    await e.method_requested(**e.params_requested)

Для того чтобы добавить решение капчи существует специальный хендлер captcha_handler, он должен возвращать код от решенной капчи:

@user.error_handler.captcha_handler
async def solve_captcha(e: VKError):
    key = "Здесь вы решаете капчу и возвращаете код"
    return key

TaskManager

Если вам нужно начать работу с vkbottle "с нуля" вам может понадобится TaskManager

from vkbottle import TaskManager
from vkbottle.api import API

api = API("token")

# Создайте вашу асинхронную функцию для запуска в таск менеджере
# для начала работы с апи
async def main():
    await api.status.set("Я люблю котиков")

task_manager = TaskManager()
# Добавьте нужные таски
task_manager.add_task(main())
task_manager.run()

Если таск один, как в прошлом примере, оптимальнее будет воспользоваться методом run_task