ETL процес за получаване на данни от имейл в Apache Airflow

Колкото и да се развиват технологиите, поредица от остарели подходи винаги изостават от развитието. Това може да се дължи на плавен преход, човешки фактор, технологични нужди или нещо друго. В областта на обработката на данни източниците на данни са най-показателни в тази част. Колкото и да мечтаем да се отървем от това, но засега част от данните се изпращат в месинджъри и имейли, да не говорим за по-архаични формати. Каня ви да разглобите една от опциите за Apache Airflow, илюстрирайки как можете да вземете данни от имейли.

праистория

Много данни все още се прехвърлят чрез електронна поща, от междуличностни комуникации до стандарти за взаимодействие между компаниите. Добре е, ако е възможно да напишете интерфейс за получаване на данни или да поставите хора в офиса, които ще въвеждат тази информация в по-удобни източници, но често това може просто да не е възможно. Конкретната задача, с която се сблъсках, беше свързването на прословутата CRM система към хранилището на данни, а след това и към OLAP системата. Исторически се случи така, че за нашата компания използването на тази система беше удобно в определена област на бизнеса. Следователно всички наистина искаха да могат да работят и с данни от тази система на трета страна. На първо място, разбира се, беше проучена възможността за получаване на данни от отворен API. За съжаление API не покриваше получаването на всички необходими данни и, с прости думи, беше в много отношения крив и техническата поддръжка не искаше или не можеше да се срещне наполовина, за да предостави по-изчерпателна функционалност. Но тази система предостави възможност за периодично получаване на липсващите данни по пощата под формата на връзка за разтоварване на архива.

Трябва да се отбележи, че това не е единственият случай, в който бизнесът иска да събира данни от имейли или месинджъри. В този случай обаче не бихме могли да повлияем на трета компания, която предоставя част от данните само по този начин.

въздушен поток Apache

За изграждане на ETL процеси най-често използваме Apache Airflow. За да може читателят, който не е запознат с тази технология, да разбере по-добре как изглежда в контекста и като цяло, ще опиша няколко уводни.

Apache Airflow е безплатна платформа, която се използва за изграждане, изпълнение и наблюдение на ETL (Extract-Transform-Loading) процеси в Python. Основната концепция в Airflow е насочен ацикличен граф, където върховете на графиката са специфични процеси, а ръбовете на графиката са поток от контрол или информация. Един процес може просто да извика всяка функция на Python или може да има по-сложна логика от последователно извикване на няколко функции в контекста на клас. За най-честите операции вече има много готови разработки, които могат да се използват като процеси. Такива разработки включват:

• оператори - за прехвърляне на данни от едно място на друго, например от таблица на база данни към хранилище за данни;
• сензори - за изчакване на настъпването на определено събитие и насочване на потока на управление към следващите върхове на графа;
• кукички - за операции от по-ниско ниво, например за получаване на данни от таблица на база данни (използва се в изрази);
• и т.н.

Би било неуместно да описваме Apache Airflow подробно в тази статия. Можете да видите кратки въведения тук или тук.

Кука за получаване на данни

Първо, за да решим проблема, трябва да напишем кука, с която можем:

• свържете се с имейл
• намери правилната буква
• получават данни от писмото.

from airflow.hooks.base_hook import BaseHook
import imaplib
import logging

class IMAPHook(BaseHook):
    def __init__(self, imap_conn_id):
        """
           IMAP hook для получения данных с электронной почты

           :param imap_conn_id:       Идентификатор подключения к почте
           :type imap_conn_id:        string
        """
        self.connection = self.get_connection(imap_conn_id)
        self.mail = None

    def authenticate(self):
        """ 
            Подключаемся к почте
        """
        mail = imaplib.IMAP4_SSL(self.connection.host)
        response, detail = mail.login(user=self.connection.login, password=self.connection.password)
        if response != "OK":
            raise AirflowException("Sign in failed")
        else:
            self.mail = mail

    def get_last_mail(self, check_seen=True, box="INBOX", condition="(UNSEEN)"):
        """
            Метод для получения идентификатора последнего письма, 
            удовлетвораяющего условиям поиска

            :param check_seen:      Отмечать последнее письмо как прочитанное
            :type check_seen:       bool
            :param box:             Наименования ящика
            :type box:              string
            :param condition:       Условия поиска писем
            :type condition:        string
        """
        self.authenticate()
        self.mail.select(mailbox=box)
        response, data = self.mail.search(None, condition)
        mail_ids = data[0].split()
        logging.info("В ящике найдены следующие письма: " + str(mail_ids))

        if not mail_ids:
            logging.info("Не найдено новых писем")
            return None

        mail_id = mail_ids[0]

        # если таких писем несколько
        if len(mail_ids) > 1:
            # отмечаем остальные прочитанными
            for id in mail_ids:
                self.mail.store(id, "+FLAGS", "\Seen")

            # возвращаем последнее
            mail_id = mail_ids[-1]

        # нужно ли отметить последнее прочитанным
        if not check_seen:
            self.mail.store(mail_id, "-FLAGS", "\Seen")

        return mail_id

Логиката е следната: свързваме се, намираме последната най-подходяща буква, ако има други, ги игнорираме. Тази функция се използва, защото по-късните писма съдържат всички данни от по-ранните. Ако това не е така, тогава можете да върнете масив от всички букви или да обработите първата, а останалите при следващото преминаване. Като цяло всичко, както винаги, зависи от задачата.

Добавяме две помощни функции към куката: за изтегляне на файл и за изтегляне на файл чрез връзка от имейл. Между другото, те могат да бъдат преместени на оператора, зависи от честотата на използване на тази функционалност. Какво друго да добавите към куката, отново зависи от задачата: ако файловете са получени веднага в писмото, тогава можете да изтеглите прикачени файлове към писмото, ако данните са получени в писмото, тогава трябва да анализирате писмото, и т.н. В моя случай писмото идва с една връзка към архива, която трябва да поставя на определено място и да започна по-нататъшната обработка.

    def download_from_url(self, url, path, chunk_size=128):
        """
            Метод для скачивания файла

            :param url:              Адрес загрузки
            :type url:               string
            :param path:             Куда положить файл
            :type path:              string
            :param chunk_size:       По сколько байтов писать
            :type chunk_size:        int
        """
        r = requests.get(url, stream=True)
        with open(path, "wb") as fd:
            for chunk in r.iter_content(chunk_size=chunk_size):
                fd.write(chunk)

    def download_mail_href_attachment(self, mail_id, path):
        """
            Метод для скачивания файла по ссылке из письма

            :param mail_id:         Идентификатор письма
            :type mail_id:          string
            :param path:            Куда положить файл
            :type path:             string
        """
        response, data = self.mail.fetch(mail_id, "(RFC822)")
        raw_email = data[0][1]
        raw_soup = raw_email.decode().replace("r", "").replace("n", "")
        parse_soup = BeautifulSoup(raw_soup, "html.parser")
        link_text = ""

        for a in parse_soup.find_all("a", href=True, text=True):
            link_text = a["href"]

        self.download_from_url(link_text, path)

Кодът е прост, така че едва ли се нуждае от допълнително обяснение. Просто ще ви разкажа за магическия ред imap_conn_id. Apache Airflow съхранява параметри на връзката (влизане, парола, адрес и други параметри), които могат да бъдат достъпни чрез идентификатор на низ. Визуално управлението на връзката изглежда така

Сензор за изчакване на данни

Тъй като вече знаем как да се свързваме и да получаваме данни от пощата, сега можем да напишем сензор, който да ги чака. В моя случай не се получи веднага да напиша оператор, който ще обработва данните, ако има такива, защото други процеси работят въз основа на данните, получени от пощата, включително тези, които вземат свързани данни от други източници (API, телефония , уеб показатели и т.н.) и т.н.). Ще ви дам пример. В CRM системата се появи нов потребител и все още не знаем за неговия UUID. След това, когато се опитваме да получим данни от SIP телефония, ще получаваме обаждания, свързани с неговия UUID, но няма да можем да ги запазим и използваме правилно. При такива въпроси е важно да се има предвид зависимостта на данните, особено ако са от различни източници. Това, разбира се, са недостатъчни мерки за запазване целостта на данните, но в някои случаи са необходими. Да, и бездействието за заемане на ресурси също е ирационално.

По този начин нашият сензор ще стартира следващите върхове на графиката, ако има свежа информация в пощата, а също така ще маркира предишната информация като неуместна.

from airflow.sensors.base_sensor_operator import BaseSensorOperator
from airflow.utils.decorators import apply_defaults
from my_plugin.hooks.imap_hook import IMAPHook

class MailSensor(BaseSensorOperator):
    @apply_defaults
    def __init__(self, conn_id, check_seen=True, box="Inbox", condition="(UNSEEN)", *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.conn_id = conn_id
        self.check_seen = check_seen
        self.box = box
        self.condition = condition

    def poke(self, context):
        conn = IMAPHook(self.conn_id)
        mail_id = conn.get_last_mail(check_seen=self.check_seen, box=self.box, condition=self.condition)

        if mail_id is None:
            return False
        else:
            return True

Ние получаваме и използваме данни

За да получавате и обработвате данни, можете да напишете отделен оператор, можете да използвате готови. Тъй като логиката все още е тривиална - да вземем данни от писмото, например, предлагам стандартния PythonOperator

from airflow.models import DAG

from airflow.operators.python_operator import PythonOperator
from airflow.sensors.my_plugin import MailSensor
from my_plugin.hooks.imap_hook import IMAPHook

start_date = datetime(2020, 4, 4)

# Стандартное конфигурирование графа
args = {
    "owner": "example",
    "start_date": start_date,
    "email": ["[email protected]"],
    "email_on_failure": False,
    "email_on_retry": False,
    "retry_delay": timedelta(minutes=15),
    "provide_context": False,
}

dag = DAG(
    dag_id="test_etl",
    default_args=args,
    schedule_interval="@hourly",
)

# Определяем сенсор
mail_check_sensor = MailSensor(
    task_id="check_new_emails",
    poke_interval=10,
    conn_id="mail_conn_id",
    timeout=10,
    soft_fail=True,
    box="my_box",
    dag=dag,
    mode="poke",
)

# Функция для получения данных из письма
def prepare_mail():
    imap_hook = IMAPHook("mail_conn_id")
    mail_id = imap_hook.get_last_mail(check_seen=True, box="my_box")
    if mail_id is None:
        raise AirflowException("Empty mailbox")

    conn.download_mail_href_attachment(mail_id, "./path.zip")

prepare_mail_data = PythonOperator(task_id="prepare_mail_data", default_args=args, dag=dag, python_callable= prepare_mail)

# Описание остальных вершин графа
...

# Задаем связь на графе
mail_check_sensor >> prepare_mail_data
prepare_data >> ...
# Описание остальных потоков управления

Между другото, ако вашата корпоративна поща също е на mail.ru, тогава няма да можете да търсите писма по тема, подател и т.н. Още през 2016 г. обещаха да го въведат, но явно са размислили. Реших този проблем, като създадох отделна папка за необходимите писма и настроих филтър за необходимите писма в уеб интерфейса на пощата. По този начин в тази папка влизат само необходимите букви и условия за търсене, в моя случай просто (НЕВИДИМО).

Обобщавайки, имаме следната последователност: проверяваме дали има нови писма, които отговарят на условията, ако има, тогава изтегляме архива, като използваме връзката от последното писмо.
Под последните точки е пропуснато, че този архив ще бъде разопакован, данните от архива ще бъдат изчистени и обработени и в резултат на това всичко ще отиде по-нататък в конвейера на ETL процеса, но това вече е отвъд обхвата на статията. Ако се оказа интересно и полезно, тогава с удоволствие ще продължа да описвам ETL решения и техните части за Apache Airflow.

Източник: www.habr.com