Создание простейшего бота для Delta Chat

Подготовка

Вы работаете на Windows, и у вас установлен Python >= 3.11. (под Linux все тоже самое, только команды создания и активации среды немного другие)

Я буду использовать пакеты deltachat-rpc-server и deltachat-rpc-client (они требуют python >= 3.10, но на windows и python 3.10 работать не будет).

Эти пакеты идут от разработчика deltachat

Если вы используете среду разработки (например pyCharm), которая умеет создавать virtualenv, то шаги 1-3 можно пропустить и создать проект с помощью среды.

  1. Установим последнюю версию virtualenv:
    py -m pip install --upgrade virtualenv
  2. Создадим виртуальную среду для нашего проекта, в папке проекта запускаем:
    py -m virtualenv .venv
  3. Активируем виртуальную среду
    .venv\Scripts\activate.bat
  4. Устанавливаем нужные пакеты
    pip3 install deltachat-rpc-server deltachat-rpc-client
  5. Также понадобится почта (mail.ru или yandex) и пароль для приложений (как его получить я рассказывал в статье )

Пишем скрипт для конфигурирования и проверки связи с IMAP и SMTP сервером.

Основная идея - запустить rpc server, сконфигурировать его (повторное конфигурирование не уничтожает базу данных) и дождаться события CONFIGURE_PROGRESS со значением progress 1000. И что бы при этом не появились ошибки (события ERROR)

Инициализатор класса Rpc на вход принимает путь к папке в которой будет база данных. (Что бы полностью обновить бота ее потом можно просто стереть. Так же ее можно перенести на другой компьютер)

Поддерживается несколько аккаунтов, мы будем использовать только один. (получить его можно get_all_accounts()[0])

При конфигурировании запуск осуществляется методом configure(), после можно получать события методом wait_for_event()

Надо подставить свою почту и пароль для приложений

MAIL = 'mail@yandex.ru' PW = 'password' import logging from pathlib import Path from deltachat_rpc_client import Rpc, DeltaChat, EventType def main(): with Rpc(str(Path(__file__).resolve().parent / 'instance')) as rpc: dc = DeltaChat(rpc) inf = dc.get_system_info() logging.info("Running deltachat core %s", inf["deltachat_core_version"]) acc = dc.get_all_accounts() if acc: logging.info('account already created') acc = acc[0] else: # create account acc = dc.add_account() acc.set_config('addr', MAIL) acc.set_config('mail_pw', PW) acc.configure() # wait for configure done while True: evt = acc.wait_for_event() match evt.kind: case EventType.CONFIGURE_PROGRESS: logging.info(f'progress: {evt.progress}/1000') if evt.progress == 1000: logging.info(f'Configure done') break case EventType.ERROR: logging.error(f'Error: {evt.msg}') break case _: logging.debug(repr(evt)) if __name__ == '__main__': logging.basicConfig(level=logging.INFO) main()

Пишем скрипт для получения ссылки для подключения к боту и просмотра контактов, которые общались с ботом.

Основная идея - запустить rpc server и только получить список контактов, затем у каждого контакта вытаскиваем почту из его VCARD.

В конце получаем ссылку (хоть метод и называется get_qr_code(), он выдает ссылку).

Естественно при первом запуске контактов у нас не будет, нам нужна только ссылка.

import logging from pathlib import Path from deltachat_rpc_client import Rpc, DeltaChat def main(): with Rpc(str(Path(__file__).resolve().parent / 'instance')) as rpc: dc = DeltaChat(rpc) inf = dc.get_system_info() logging.info("Running deltachat core %s", inf["deltachat_core_version"]) accounts = dc.get_all_accounts() if not accounts: logging.error('account not created') return acc = accounts[0] contacts = acc.get_contacts() for contact in contacts: vc = contact.make_vcard() mail = [s.split(':', 1)[1] for s in vc.splitlines() if s.startswith('EMAIL:')] print(f"ID: {contact.id} | Email: {mail}") dc_link = acc.get_qr_code() print(f"Ссылка для подключения: {dc_link}") if __name__ == '__main__': logging.basicConfig(level=logging.INFO) main()

Пишем самого бота.

Создадим простейший обработчик сообщений в файле handler.py, всего лишь одна функция, которая на вход принимает текст сообщения и возвращает текст ответа.

def handle_msg(text: str) -> str: return f'Answer: {text}'

И создадим самого бота.

Основная идея - отлавливать события INCOMING_MSG и по нему вытаскивать все сообщения (функция handle_msgs) и для подходящих сообщений вызывать handler.handle_msg

Перед циклом обработки сообщения надо вызвать метод start_io()

Так же бот при каждом запуске создает новый журнал и записывает туда события - это хороший источник данных для дальнейшей разработки - чтобы увидеть какие события и в каком порядке возникают.

import logging from pathlib import Path from deltachat_rpc_client import Rpc, DeltaChat, EventType, SpecialContactId import handler def main(): with Rpc(str(Path(__file__).resolve().parent / 'instance')) as rpc: dc = DeltaChat(rpc) inf = dc.get_system_info() logging.info("Running deltachat core %s", inf["deltachat_core_version"]) accounts = dc.get_all_accounts() if not accounts: logging.error('account not created') return acc = accounts[0] acc.start_io() def handle_msgs(): for msg in acc.get_next_messages(): data = msg.get_snapshot() logging.info(f'message received (msg: {repr(msg)}) (snapshot:{repr(data)})') if data.from_id != SpecialContactId.SELF and not data.is_bot and not data.is_info: answer = handler.handle_msg(data.text) if answer: data.chat.send_text(f'Answer {data.text}') data.message.mark_seen() while True: evt = acc.wait_for_event() match evt.kind: case EventType.ERROR: logging.error(f'Error: {evt.msg}') case EventType.WARNING: logging.warning(f'Warning: {evt.msg}') case EventType.INCOMING_MSG: handle_msgs() case _: logging.debug(f'Event: repr(evt)') if __name__ == '__main__': from datetime import datetime fn = 'LOG_' + datetime.isoformat(datetime.now(), '_', 'seconds') + '.log' fn = fn.replace(':', '_').replace('-', '_') logging.basicConfig(format='%(asctime)s %(message)s', level=logging.INFO, filename=fn) logging.getLogger().addHandler(logging.StreamHandler()) main()

Проблемы:

Если не происходит установка соединения с ботом - надо проверить папку спам и если письмо там, нажать не спам. (также в настройках фильтрации можно создать фильтр или белый список для определенных адресов).

Заключение.

Это самый простой бот.

Примеры эхоботов по разной технологии использования можно найти в документации:

К сожалению документация далеко не полная, там не описаны большинство атрибутов которые появляются у объектов.

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

Начать дискуссию