Как я учил Алису отправлять почту / Habr

Всем привет! Меня зовут Иван Чечиков, я QA-lead в МТС Digital, работаю над проектом стримингового сервиса WASD.TV. В этой статье я расскажу о своем пет-проекте по созданию навыка «Умный почтальон» для Алисы, ассистента Яндекса.

Немного истории:

В декабре прошлого года я приобрел умную колонку Яндекса, Станцию Лайт. Еще не успел насладиться приобретением, а уже заметил, что, к сожалению, некоторого функционала Алисе не хватает. Погуглил и узнал, что существует витрина навыков, в которой представлены продукты сторонних разработчиков, – навыки для Алисы, расширяющие ее возможности. Идея этой витрины показалась мне интересной и я решил создать для нее новый навык. Мне хотелось научить Алису отправлять почту по команде с колонки, так как по дефолту она этого не умеет.

На момент создания кода я не нашел реализованных аналогов, поэтому начал разработку своего решения.

Итак, приступим.

Чтобы реализовать эту задумку, мне нужно было:

поднять локальный сервер на Python с пробросом хоста на удаленную машину с https;
реализовать авторизацию в Яндекс.Почта по imap и smtp;
создать приложение в Яндекс OAuth и навык в Яндекс.Диалогах;
написать сценарии для Алисы: логику получения списка контактов, формирование темы и текста письма, отправки письма;
протестировать созданный функционал;
поместить проект на виртуальный хостинг.

Пойдем по порядку.

Я решил создать свое приложение на python3.8 с использованием веб фреймворка Bottle. Локальный сервер я поднял, используя метод run.

from bottle import route, run

@route('/')
def index():
    print("Тест")

run(host='127.0.0.1', port=8080)

Пробросил локальный хост через ngrok. Это достаточно простой и удобный в использовании сервис. Необходимо зарегистрироваться, cкачать zip-архив и выполнить ряд команд в терминале:

$ unzip /path/to/ngrok.zip
$ ngrok config add-authtoken {authtoken}
$ ngrok http 8080

Результат:

При отправке GET запроса на https://1cc0-213-24-134-136.eu.ngrok.io вернется 200 код ответа http.

Переходим к реализации авторизации в Яндекс.Почта по imap и smtp. Для начала нужно дать доступ в своей учетной записи на коннект по imap и smtp.

Заходим в свой аккаунт Яндекс.Почта. Почта->Все настройки.

Ставим галочку напротив Пароли приложений и OAuth-токены. Готово, идем дальше.

У Яндекса есть API для разработчиков, позволяющее реализовать свои мечты и идеи в экосистеме сервисов. Для OAuth2-авторизации по imap и smtp есть свой мануал. Регистрируем приложение и изучаем мануал.

Заходим в наше зарегистрированное приложение в Яндекс.OAuth. В созданном приложении указываем Callback URL – такие, как на скрине.

Авторизация в навыке может осуществляться напрямую через приложение Яндекса, либо через свой собственный сервис, который должен реализовывать логику получения: верификационный код -> access_token. Мне не захотелось писать OAuth-сервер, поэтому я выбрал первый вариант, так как это по сути готовое решение от самого Яндекса. Если кто-то выберет второй вариант, то можно тут посмотреть пример получения OAuth-токенa веб-сервисом на python.

Даем приложению доступ к сервисам API Яндекс ID и Яндекс.Почта:

Код подключения по imap

def get_email_address_list(access_token: str, sender_email: str) -> list:
    try:
        xoauth2_token = f"user={sender_email}\x01auth=Bearer {access_token}\x01\x01"
        imap = imaplib.IMAP4_SSL(host='imap.yandex.com')
        imap.authenticate("XOAUTH2", lambda x: xoauth2_token)
        status, messages = imap.select("INBOX")
        messages = int(messages[0])
        messages_count = 0
        if messages >= 30:
            messages_count = 30
        elif messages < 30:
            messages_count = messages
        while messages_count:
            messages_count -= 1
            res, msg = imap.fetch(str(messages - messages_count), "(RFC822)")
            for response in msg:
                if isinstance(response, tuple):
                    msg = email.message_from_bytes(response[1])
                    sender_data = decode_header(msg.get("From"))
                    if len(sender_data) == 2:
                        sender_name, encoding = sender_data[0]
                        sender_email, type_email = sender_data[1]
                        if isinstance(sender_name, bytes) and isinstance(sender_email, bytes):
                            sender_name = sender_name.decode(encoding)
                            sender_name = ''.join(sender_name.split()).lower()
                            sender_email = sender_email.decode(encoding).replace("<", "") \
                                .replace(">", "").replace(" ", "")
                            sender_email = sender_email.lower()
                            email_address_list.append(sender_name + ":" + sender_email)                    
            
        imap.close()
        imap.logout()
        
    except Exception:
        return []
    
    return email_address_list

Приложение подключается к серверу по imap, получает список всех писем пользователя, перебирает последние 30, делает проверки на корректность данных и кладет строки в список в формате «имя отправителя: адрес почты». Я указываю лимит в 30 сообщений, так как вебхук должен дать ответ серверу Яндекса в течении 3 секунд, иначе Алиса вернет ответ, что вебхук не отвечает. Код получился недостаточно оптимизированный, функция не успевает перебрать более 30 сообщений и вернуть ответ за 3 сек.

Код подключения по smtp

def get_email_data_list(email_data_str: str) -> list:
    email_data_list = email_data_str.split("@")
    return email_data_list
  
def send_email(access_token: str, sender_email_name: str, sender_email: str, recipient_email_name: str,
               recipient_email: str, subject: str, message: str) -> bool:
    try:
        just_a_str = f"user={sender_email}\x01auth=Bearer {access_token}\x01\x01"
        xoauth2_token = base64.b64encode(bytes(just_a_str, 'utf-8')).decode('utf-8')

        sender_email_login = get_email_data_list(sender_email)[0]
        sender_email_domain = get_email_data_list(sender_email)[1]
        recipient_email_login = get_email_data_list(recipient_email)[0]
        recipient_email_domain = get_email_data_list(recipient_email)[1]

        msg = EmailMessage()
        msg['Subject'] = subject
        msg['From'] = Address(sender_email_name, sender_email_login, sender_email_domain)
        msg['To'] = Address(recipient_email_name, recipient_email_login, recipient_email_domain)
        msg.set_content(message)

        smtp = smtplib.SMTP_SSL(host='smtp.yandex.ru', port=465)
        smtp.connect(host='smtp.yandex.ru', port=465)
        smtp.docmd("auth", f"XOAUTH2 {xoauth2_token}")
        smtp.sendmail(sender_email, recipient_email, msg.as_string())
        smtp.quit()
    except Exception:
        return False
    
    return True

Приложение подключается по smtp, формирует тему письма, email отправителя, email получателя и текст. Функция send_email отправляет письмо и возвращает True, если успешно вызвана, в противном случае она возвращает False.

Создаем навык в Яндекс.Диалогах.

Выбираем Создать диалог.

Выбираем Навык в Алисе

Заполняем поля. Указываем Имя навыка, Webhook URL, Тип доступа.

Webhook URL – это наш локальный сервер, проксирующий на ngrok. С него будут слаться запросы к Алисе. Тип доступа я оставляю приватным.

В связке аккаунтов надо указать идентификатор приложения и секрет приложения – это ClientID и Client secret в Яндекс.OAuth.

URL авторизации, URL для получения и обновления токена заполняем также, как на скрине. Эти данные нужны, чтобы пользователь смог авторизоваться через сервис Яндекса в навыке и наше приложение смогло получить access_token для выполнения своей логики.

Переходим к написанию сценариев для Алисы.

Код файла app.py

from bottle import request, post, default_app
from mail_tools import get_sender_email_data, send_email, get_recipient_email_text, get_email_obj_text, \
    get_email_address_list
import json
import re

response_texts  = []
response_ttss = []


@post('/')
def work():
    try:
        response = {
            "version": request.json["version"],
            "session": request.json["session"],
            "response": {
                "end_session": False
            }
        }
        req = request.json

        if req["session"]["new"] or req["request"]["original_utterance"].lower().strip() in ["привет", "хай", "дарова",
                                                                                             "ку",
                                                                                             "даров", "здарова",
                                                                                             "hello", "hi"]:
            response["response"]["text"] = "Привет, навык позволяет отправлять почту с помощью умного ассистента." \
                                           "Чтобы начать ответьте 'старт'. Для выхода из навыка " \
                                           "ответьте 'пока' или 'стоп'."
            response["response"]["tts"] = "Прив+ет, н+авык позвол+яет отправл+ять п+очту с п+омощью умного ассист+ента. " \
                                          "Чт+обы нач+ать отв+етьте старт. Для в+ыхода из н+авыка " \
                                          "отв+етьте пок+а или стоп. "
            response_texts.append(response["response"]["text"])
            response_ttss.append(response["response"]["tts"])
        else:
            try:
                if req["request"]["original_utterance"].lower().strip() in ["стоп", "закончили",
                                                                            "не надо", "все", "хватит",
                                                                            "пока", "до свидания", "конец",
                                                                            "нет", "отбой", "хватит"]:

                    response["response"]["text"] = "Всего хорошего, до свидания!"
                    response["response"]["tts"] = "Всег+о хор+ошего, до свид+ания!"
                    response_texts.append(response["response"]["text"])
                    response_ttss.append(response["response"]["tts"])
                    response["response"]["end_session"] = True
                elif req["session"]["user"]["access_token"]:
                    response["response"]["text"] = response_texts[len(response_texts) - 1]
                    response["response"]["tts"] = response_ttss[len(response_ttss) - 1]
                    if req["request"]["original_utterance"].lower().strip() == "помощь":
                        response["response"]["text"] = "Привет, навык позволяет отправлять почту с помощью умного ассистента. " \
                                                       "Чтобы начать ответьте 'старт'. Для выхода из навыка " \
                                                       "ответьте 'пока' или 'стоп'."
                        response["response"]["tts"] = "Прив+ет, н+авык позвол+яет отправл+ять п+очту " \
                                                      "с п+омощью умного ассист+ента. Чт+обы нач+ать отв+етьте старт. " \
                                                      "Для в+ыхода из н+авыка отв+етьте пок+а или стоп."
                        response_texts.append(response["response"]["text"])
                        response_ttss.append(response["response"]["tts"])
                    else:

                        if req["request"]["original_utterance"].lower().strip().strip() == "старт":
                            global access_token
                            access_token = req["session"]["user"]["access_token"]
                            if get_sender_email_data(access_token):
                                global sender_email_name
                                sender_email_name = get_sender_email_data(access_token)[0]
                                global sender_email
                                sender_email = get_sender_email_data(access_token)[1]
                                response["response"]["text"] = f"{sender_email_name}, идентификация вашего email завершена. " \
                                                              f"Чтобы продолжить, ответьте, 'список контактов'." 
                                response["response"]["tts"] = f"{sender_email_name}, идентифик+ация в+ашего email завершен+а. " \
                                                              f"Чт+обы прод+олжить отв+етьте сп+исок конт+актов."
                                response_texts.append(response["response"]["text"])
                                response_ttss.append(response["response"]["tts"])
                            else:
                                response["response"]["text"] = "Не удалось определить ваши почтовые данные. Есть проблема, " \
                                                               "начните сначала. Ответьте 'старт' или 'помощь'."

                                response["response"]["tts"] = "Не удал+ось определ+ить в+аши почт+овые д+анные. " \
                                                              "Есть пробл+ема, начн+ите снач+ала. " \
                                                              "Отв+етьте старт или п+омощь."
                                response_texts.append(response["response"]["text"])
                                response_ttss.append(response["response"]["tts"])

                        elif req["request"]["original_utterance"].lower().strip() == "список контактов":
                            global email_address_list
                            email_address_list = get_email_address_list(access_token, sender_email)
                            if email_address_list:
                                response["response"]["text"] = "Список контактов сформирован. " \
                                                               "Я могу начать писать письмо. Кому будем отправлять? " \
                                                               "Назовите получателя, ответив 'получатель', " \
                                                               "а дальше назовите его имя и фамилию, " \
                                                               "если это физ. лицо, " \
                                                               "или же название компании, если это юр. лицо."

                                response["response"]["tts"] = "Сп+исок конт+актов сформир+ован. Я мог+у нач+ать пис+ать письм+о. " \
                                                              "Ком+у б+удем отправл+ять? Назов+ите получ+ателя, " \
                                                              "отв+етив получ+атель, а д+альше назов+ите ег+о имя " \
                                                              "и фам+илию, если это физ лиц+о или же назв+ание" \
                                                              "комп+ании, если это юр лиц+о."
                                response_texts.append(response["response"]["text"])
                                response_ttss.append(response["response"]["tts"])

                            else:
                                response["response"]["text"] = "Список контактов не удалось сформировать или же он пуст."

                                response["response"]["tts"] = " Сп+исок конт+актов не удал+ось сформиров+ать или же он пуст."
                                response_texts.append(response["response"]["text"])
                                response_ttss.append(response["response"]["tts"])

                        elif re.match(r"^получатель [\S\sa-zA-Zа-яА-Я0-9_.+-]+$",
                                      req["request"]["original_utterance"].lower().strip()):
                            if get_recipient_email_text(req["request"]["original_utterance"]):
                                global recipient_email_name
                                recipient_email_name = get_recipient_email_text(req["request"]["original_utterance"])[0]
                                global recipient_email
                                recipient_email = get_recipient_email_text(req["request"]["original_utterance"])[1]
                                response["response"]["text"] = f"Email получателя {recipient_email}. Чтобы продолжить, " \
                                                               f"ответьте, 'тема письма'. " \
                                                               f"Если хотите изменить email получателя, " \
                                                               f"ответьте прошлую команду еще раз."
                                response["response"]["tts"] = f"Email получ+ателя {recipient_email}. " \
                                                              f"Чт+обы прод+олжить, отв+етьте, т+ема письм+а. " \
                                                              f"Если хот+ите измен+ить email получ+ателя, " \
                                                              f"отв+етьте пр+ошлую ком+анду ещ+е раз."
                                response_texts.append(response["response"]["text"])
                                response_ttss.append(response["response"]["tts"])
                            else:
                                response["response"]["text"] = "Не удалось определить email получателя. " \
                                                               "Возможно его нет в сформированном списке контактов. " \
                                                               "Попробуйте снова, ответьте прошлую команду еще раз."
                                response["response"]["tts"] = "Не удал+ось определ+ить email получ+а " \
                                                              "теля. Возм+ожно его нет в сформир+ованном " \
                                                              "сп+иске конт+актов. Попр+обуйте сн+ова, " \
                                                              "отв+етьте пр+ошлую ком+анду еще раз."
                                response_texts.append(response["response"]["text"])
                                response_ttss.append(response["response"]["tts"])

                        elif req["request"]["original_utterance"].lower().strip() == "тема письма":
                            response["response"]["text"] = "Придумайте тему письма, ответив, 'тема'..., " \
                                                           "а дальше только полёт вашей мысли."
                            response["response"]["tts"] = "Прид+умайте т+ему письм+а, отв+етив, т+ема..., " \
                                                          "а д+альше т+олько пол+ёт в+ашей м+ысли."
                            response_texts.append(response["response"]["text"])
                            response_ttss.append(response["response"]["tts"])

                        elif re.match(r"^тема (?!.*\bписьма\b)[\S\sa-zA-Zа-яА-Я0-9_.+-]+$",
                                      req["request"]["original_utterance"].lower().strip()):
                            global subject
                            subject = get_email_obj_text(req["request"]["original_utterance"].strip(), "тема")
                            subject = subject if re.match(r'[+.!?]', subject[len(subject) - 1]) else subject + '.'
                            response["response"]["text"] = f"Тема письма - {subject} Чтобы продолжить, " \
                                                           f"ответьте, 'текст письма'. " \
                                                           f"Если хотите изменить тему письма, " \
                                                           f"ответьте прошлую команду еще раз."
                            response["response"]["tts"] = f"Т+ема письм+а - {subject} Чт+обы прод+олжить, " \
                                                          f"отв+етьте, текст письм+а. " \
                                                          f"Если хот+ите измен+ить т+ему письм+а, " \
                                                          f"отв+етьте пр+ошлую ком+анду еще раз."
                            response_texts.append(response["response"]["text"])
                            response_ttss.append(response["response"]["tts"])

                        elif req["request"]["original_utterance"].lower().strip() == "текст письма":
                            response["response"]["text"] = "Придумайте текст письма, ответив, 'текст'..., " \
                                                           "а дальше только полёт вашей мысли."
                            response["response"]["tts"] = "Прид+умайте текст письм+а, отв+етив, текст... " \
                                                          "а д+альше т+олько пол+ёт в+ашей м+ысли. "
                            response_texts.append(response["response"]["text"])
                            response_ttss.append(response["response"]["tts"])

                        elif re.match(r"^текст (?!.*\bписьма\b)[\S\sa-zA-Zа-яА-Я0-9_.+-]+$",
                                      req["request"]["original_utterance"].lower().strip()):
                            global message
                            message = get_email_obj_text(req["request"]["original_utterance"].strip(), "текст")
                            message = message if re.match(r'[+.!?]', message[len(message) - 1]) else message + '.'
                            response["response"]["text"] = f"Текст письма - {message} Чтобы продолжить, " \
                                                           f"ответьте, 'отправка письма'. " \
                                                           f"Если хотите изменить текст письма, " \
                                                           f"ответьте прошлую команду еще раз."
                            response["response"]["tts"] = f"Текст письм+а - {message} Чт+обы прод+олжить, " \
                                                          f"отв+етьте, отпр+авка письм+а. " \
                                                          f"Если хот+ите измен+ить текст письм+а, " \
                                                          f"отв+етьте пр+ошлую ком+анду еще раз."
                            response_texts.append(response["response"]["text"])
                            response_ttss.append(response["response"]["tts"])

                        elif req["request"]["original_utterance"].lower().strip() == "отправка письма":
                            response["response"]["text"] = "Отправляю письмо, подтвердите, " \
                                                           "ответив, 'подтверждаю', либо вы можете " \
                                                           "вернуться назад и изменить текст, " \
                                                           "тему или получателя письма."
                            response["response"]["tts"] = "Отправл+яю письм+о, подтверд+ите, " \
                                                          "отв+етив, подтвержд+аю, л+ибо вы м+ожете " \
                                                          "верн+уться наз+ад и измен+ить текст,т+ему или " \
                                                          "получ+ателя письм+а."
                            response_texts.append(response["response"]["text"])
                            response_ttss.append(response["response"]["tts"])

                        elif req["request"]["original_utterance"].lower().strip() == "подтверждаю":
                            if send_email(access_token=access_token, sender_email_name=sender_email_name,
                                          sender_email=sender_email, recipient_email_name=recipient_email_name,
                                          recipient_email=recipient_email, subject=subject, message=message):
                                response["response"]["text"] = "Письмо отправлено."
                                response["response"]["tts"] = "Письм+о отпр+авлено."
                                response_texts.append(response["response"]["text"])
                                response_ttss.append(response["response"]["tts"])
                            else:
                                response["response"]["text"] = "Письмо уже отправлено, либо на " \
                                                               "сервере какие то проблемы."
                                response["response"]["tts"] = "Письм+о уж+е отпр+авлено, л+ибо на " \
                                                              "с+ервере как+ие то пробл+емы."
                                response_texts.append(response["response"]["text"])
                                response_ttss.append(response["response"]["tts"])
                            sender_email_name = sender_email = recipient_email_name = \
                                recipient_email = subject = message = ""
                        elif req["request"]["command"].lower().strip() in ["спасибо", "большое спасибо"]:
                            response["response"]["text"] = "Пожалуйста, всегда к вашим услугам!"
                            response["response"]["tts"] = "Пож+алуйста, всегд+а к в+ашим усл+угам!"

            except KeyError:
                response["start_account_linking"] = {}
                response["response"][
                    "text"] = "Привет, отправим почту кому-нибудь? Только для начала надо авторизоваться. " \
                              "Чтобы начать ответьте 'старт'. " \
                              "Для выхода из навыка ответьте 'пока' или 'стоп'."

                response["response"][
                    "tts"] = "Прив+ет, отпр+авим п+очту ком+у ниб+удь? Только для начала надо авторизоваться. " \
                             "Чт+обы нач+ать отв+етьте старт. Для в+ыхода из н+авыка отв+етьте пок+а " \
                             "или стоп."
    except Exception :
        response["response"]["text"] = "Потеряна связь с космосом, повторите предыдущую команду или начните сначала."
        response["response"]["tts"] = "Пот+еряна связь с космосом, повтор+ите предыд+ущую ком+анду или начн+ите снач+ала."
    return json.dumps(response)

run(host='127.0.0.1', port=8080)

В ветвях сценариев я использовал оптимально нужные условия для работы навыка. Логика проста: пользователь начинает работу с навыком командой 'старт', если в запросе приложения отсутствует access_token, Алиса просит пользователя авторизоваться в навыке. Если пользователь авторизовался, приложение получает имя и email пользователя, формирует список контактов, помогает пользователю создать тему и текст письма, выполняет его отправку. В противном случае пользоваться функционалом нельзя.

Авторизация выглядит так:

Верхний уровень покрыт блоком try->except. При возникновении какой-либо ошибки в любом месте кода пользователь увидит сообщение: «Потеряна связь с космосом, повторите предыдущую команду или начните сначала». По-хорошему надо добавить логирование с записью в файл, но я обошелся без этого, ~~it is bad practiсe~~.

В каждом условии я сохраняю response Алисы для того, чтобы в случае отправки незнакомой ей команды, она возвращала последний отправленный ответ.


...
response_texts.append(response["response"]["text"])
response_ttss.append(response["response"]["tts"])
...

elif req["session"]["user"]["access_token"]:
  response["response"]["text"] = response_texts[len(response_texts) - 1]
  response["response"]["tts"] = response_ttss[len(response_ttss) - 1]

Добавляю код из файла mail_tools.py. В итоге у меня получилось bottle-приложение из двух файлов.

Код файла mail_tools.py

import base64
import imaplib
import re
import smtplib
import email
import email.header
from email.header import decode_header
from email.headerregistry import Address
from email.message import EmailMessage

import requests

login_info_url = "https://login.yandex.ru/info?oauth_token={}"
email_address_list = []


def get_sender_email_data(access_token: str) -> list:
  try:
    response = requests.get(login_info_url.format(access_token))
    user_data = response.json()
    sender_name = user_data["display_name"]
    sender_email = user_data["default_email"]
  except Exception:
    return []
  return [sender_name, sender_email]


def send_email(access_token: str, sender_email_name: str, sender_email: str, recipient_email_name: str,
               recipient_email: str, subject: str, message: str) -> bool:
    try:
        just_a_str = f"user={sender_email}\x01auth=Bearer {access_token}\x01\x01"
        xoauth2_token = base64.b64encode(bytes(just_a_str, 'utf-8')).decode('utf-8')

        sender_email_login = get_email_data_list(sender_email)[0]
        sender_email_domain = get_email_data_list(sender_email)[1]
        recipient_email_login = get_email_data_list(recipient_email)[0]
        recipient_email_domain = get_email_data_list(recipient_email)[1]

        msg = EmailMessage()
        msg['Subject'] = subject
        msg['From'] = Address(sender_email_name, sender_email_login, sender_email_domain)
        msg['To'] = Address(recipient_email_name, recipient_email_login, recipient_email_domain)
        msg.set_content(message)

        smtp = smtplib.SMTP_SSL(host='smtp.yandex.ru', port=465)
        smtp.connect(host='smtp.yandex.ru', port=465)
        smtp.docmd("auth", f"XOAUTH2 {xoauth2_token}")
        smtp.sendmail(sender_email, recipient_email, msg.as_string())
        smtp.quit()
    except Exception:
        return False
    
    return True


def get_email_data_list(email_data_str: str) -> list:
    email_data_list = email_data_str.split("@")
    return email_data_list


def get_email_address_list(access_token: str, sender_email: str) -> list:
    try:
        xoauth2_token = f"user={sender_email}\x01auth=Bearer {access_token}\x01\x01"
    
        imap = imaplib.IMAP4_SSL(host='imap.yandex.com')
        imap.debug = 4
        imap.authenticate("XOAUTH2", lambda x: xoauth2_token)
        status, messages = imap.select("INBOX")
        messages = int(messages[0])
        messages_count = 0
        if messages >= 30:
            messages_count = 30
        elif messages < 30:
            messages_count = messages
        while messages_count:
            messages_count -= 1
            res, msg = imap.fetch(str(messages - messages_count), "(RFC822)")
            for response in msg:
                if isinstance(response, tuple):
                    msg = email.message_from_bytes(response[1])

                    sender_data = decode_header(msg.get("From"))

                    if len(sender_data) == 2:
                        sender_name, encoding = sender_data[0]
                        sender_email, type_email = sender_data[1]
                        if isinstance(sender_name, bytes) and isinstance(sender_email, bytes):
                            sender_name = sender_name.decode(encoding)
                            sender_name = ''.join(sender_name.split()).lower()
                            sender_email = sender_email.decode(encoding).replace("<", "") \
                                .replace(">", "").replace(" ", "")
                            sender_email = sender_email.lower()
                            email_address_list.append(sender_name + ":" + sender_email)
            
        imap.close()
        imap.logout()
        
    except Exception:
        return []
    
    return email_address_list


def get_recipient_email_text(text: str) -> list:
    try:
        email_text = ''.join(text.split()).lower()
        email_text = email_text.split("получатель")[1]
        for email_recipient in email_address_list:
            if email_text in email_recipient:
                recipient_name = email_recipient.split(":")[0]
                recipient_email = email_recipient.split(":")[1]
                return [recipient_name, recipient_email]

    except Exception:
        return []
    return []


def get_email_obj_text(email_obj_text: str, email_obj_type: str) -> str:
    email_obj_text_list = email_obj_text.split(email_obj_type + " ") \
        if len(email_obj_text.split(email_obj_type + " ")) > 1 \
        else email_obj_text.split(email_obj_type.title() + " ")
    email_obj_text = email_obj_text_list[1]
    email_obj_text = email_obj_text.replace(" точка с запятой", ";")
    email_obj_text = email_obj_text.replace(" точка", ".")
    email_obj_text = email_obj_text.replace(" знак восклицания", "!")
    email_obj_text = email_obj_text.replace(" восклицательный знак", "!")
    email_obj_text = email_obj_text.replace(" знак вопроса", "?")
    email_obj_text = email_obj_text.replace(" вопросительный знак", "?")
    email_obj_text = email_obj_text.replace("тире", "-")
    email_obj_text = email_obj_text.replace(" двоеточие", ":")
    email_obj_text = email_obj_text.replace(" запятая", ",")

    email_obj_text_list = re.findall(r"[\sА-Яа-я-,;:_-]*!?.", email_obj_text)
    text_list = []
    for phrase in email_obj_text_list:
        phrase = phrase.strip()
        phrase = phrase.capitalize()
        text_list.append(phrase)
    email_obj_text = " ".join(text_list)
    return email_obj_text

Актуальные имя и email пользователя приложение получает в функции get_sender_email_data отправляя GET запрос на https://login.yandex.ru/info с передачей access_token.

login_info_url = "https://login.yandex.ru/info?oauth_token={}"
email_address_list = []


def get_sender_email_data(access_token: str) -> list:
  try:
    response = requests.get(login_info_url.format(access_token))
    user_data = response.json()
    sender_name = user_data["display_name"]
    sender_email = user_data["default_email"]
  except Exception:
    return []
  return [sender_name, sender_email]

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

def get_email_obj_text(email_obj_text: str, email_obj_type: str) -> str:
    email_obj_text_list = email_obj_text.split(email_obj_type + " ") \
        if len(email_obj_text.split(email_obj_type + " ")) > 1 \
        else email_obj_text.split(email_obj_type.title() + " ")
    email_obj_text = email_obj_text_list[1]
    email_obj_text = email_obj_text.replace(" точка с запятой", ";")
    email_obj_text = email_obj_text.replace(" точка", ".")
    email_obj_text = email_obj_text.replace(" знак восклицания", "!")
    email_obj_text = email_obj_text.replace(" восклицательный знак", "!")
    email_obj_text = email_obj_text.replace(" знак вопроса", "?")
    email_obj_text = email_obj_text.replace(" вопросительный знак", "?")
    email_obj_text = email_obj_text.replace("тире", "-")
    email_obj_text = email_obj_text.replace(" двоеточие", ":")
    email_obj_text = email_obj_text.replace(" запятая", ",")

    email_obj_text_list = re.findall(r"[\sА-Яа-я-,;:_-]*!?.", email_obj_text)
    text_list = []
    for phrase in email_obj_text_list:
        phrase = phrase.strip()
        phrase = phrase.capitalize()
        text_list.append(phrase)
    email_obj_text = " ".join(text_list)
    return email_obj_text

Приложение получает тему/текст письма, использует регулярку для разделения текста на список предложений со знаками окончания. Список склеивает в строку. Код не идеален, но свою функцию он выполняет.

Можно протестировать сборку приложения в самом навыке, во вкладке Тестирование, либо в приложении Алиса на Windows/iOS/Android или на умной колонке. Для этого надо опубликовать приватный навык и сгенерировать одноразовую ссылку.

Как это работает у меня на колонке:

Под приложение я использовал сервер Beget. Виртуальный хостинг позволяет использовать код на Python последних версий. Мануал по деплою bottle приложения простой: создаете директории, закидываете файлы, немного магии и все работает. Я решил оставить навык приватным. Он не идеален и порой может отваливаться. Возможно, в дальнейшем я оптимизирую и доработаю код, прикручу БД, но это будет уже другая история.

Спасибо за уделенное время, если у вас есть вопросы – буду рад на них ответить в комментариях.

Как я учил Алису отправлять почту

{{ titleHtml }}

{{ titleHtml }}