Рассылка email-оповещений

Эта статья демонстрирует, как функциональность, доступная в режиме Подготовка данных и ML может быть использована для отправки электронных письем с текстовым содержимым и (опционально) файлом-вложением через произвольный SMTP-сервер. Продемонстрированные блокноты можно запускать вручную или по расписанию через планировщик, переопределяя параметры письма (получателя, тему, текст и путь к вложению) без изменения кода.

• Начало работы
• Отправка сообщений
• Извлечение данных аналитического сервиса
• Генерация вложений
• Отправка сообщений по расписанию

1. Для начала работы, перейдите к редактированию рабочей области, в рамках которой будет вестись работа с данными. Нажмите на блок Подготовка данных и ML.

   

2. В режиме Подготовка данных и ML, нажмите (новый лаунчер) и выберите Блокнот — Python 3 (ipykernel) в отображённой вкладке для создания нового блокнота.

   

1. В первой ячейке блокнота определите параметры письма. Присвойте ячейке тэг parameters, чтобы при запуске по расписанию значения этих переменных могли быть переопределены параметрами планировщика:

    
   # Эти переменные могут быть переопределены из планировщикаrecipient = "receiver@example.com"           # Комуsubject = "Тестовое письмо с вложением"      # Темаbody_text = "Здравствуйте!\n\nОтправляем вам отчёт."  # Текст письмаattachment_path = "report.html"              # Путь к файлу вложения (относительный или абсолютный)# Если файла нет, вложение не добавится

2. Определите учётные данные и адрес SMTP-сервера. В примере эти настройки считываются из переменных окружения.

    
   import osfrom dotenv import load_dotenvload_dotenv()# === НАСТРОЙКИ SMTP (из переменных окружения) ===SMTP_SERVER = os.environ.get("SMTP_SERVER", "smtp.gmail.com")SMTP_PORT   = int(os.environ.get("SMTP_PORT", 587))SMTP_USER   = os.environ["SMTP_USER"] # обязательноSMTP_PASS   = os.environ["SMTP_PASS"] # обязательноSENDER      = SMTP_USER   # можно задать отдельную переменную, если адрес отправителя отличается

3. Функция send_email_with_attachment формирует MIME-сообщение из стандартной библиотеки Python: создаёт контейнер MIMEMultipart, добавляет текстовое тело письма и, если по указанному пути существует файл, прикрепляет его как вложение в кодировке base64. Если файл не найден, письмо отправляется без вложения. Отправка выполняется через smtplib с использованием защищённого соединения STARTTLS:

    
   import smtplibfrom email.mime.multipart import MIMEMultipartfrom email.mime.text import MIMETextfrom email.mime.base import MIMEBasefrom email import encodersfrom pathlib import Pathdef send_email_with_attachment(to, subject, body_text, attachment_path=None):    """    Отправляет письмо с текстовым телом и (опционально) вложением.    Если файл по указанному пути не существует, вложение игнорируется.    """    # Создаём контейнер сообщения    msg = MIMEMultipart()    msg['From'] = SENDER    msg['To'] = to    msg['Subject'] = subject    # Текстовое тело письма    msg.attach(MIMEText(body_text, 'plain'))    # Вложение    if attachment_path:        p = Path(attachment_path)        if p.is_file():            with open(p, 'rb') as f:                part = MIMEBase('application', 'octet-stream')                part.set_payload(f.read())            encoders.encode_base64(part)            part.add_header(                'Content-Disposition',                f'attachment; filename={p.name}'            )            msg.attach(part)            print(f"✔ Вложение добавлено: {p}")        else:            print(f"⚠ Файл-вложение не найден: {p}. Письмо будет отправлено без вложения.")    # Отправка    with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:        server.starttls()        server.login(SMTP_USER, SMTP_PASS)        server.send_message(msg)    print(f"✅ Письмо успешно отправлено на {to}")

4. Последняя ячейка вызывает функцию send_email_with_attachment для отправки письма с параметрами, заданными на первом шаге:

    
   # Отправка сообщения (значения переменных могут быть переопределены параметрами планировщика)send_email_with_attachment(    to=recipient,    subject=subject,    body_text=body_text,    attachment_path=attachment_path)

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

Для получения данных аналитических данных, используйте REST API административного сервиса VK Business Analytics.

1. Задайте параметры подключения. Удостоверьтесь, что используемая учётная запись имеет доступ на чтение к запрашиваемым данным.

    
   # === НАСТРОЙКИ API ===# Укажите данные для авторизацииLOGIN_PAYLOAD = {    "username": os.environ["API_USER"], # Имя пользователя VK Business Analytics    "password": os.environ["API_PASS"]  # Пароль}# Конфигурации APIBASE_URL = "http://admin:8000" # Базовый URL VK Business Analytics LOGIN_URL = f"{BASE_URL}/auth/login" # URL для выполнения аутентификацииPROJECT_UID = "09df9244-2227-4071-b04f-5e8b01c456cc" # Укажите актуальный UID проектаEXECUTE_SQL_URL = f"{BASE_URL}/api/projects/{PROJECT_UID}/query" # Эндпоинт для отправки запросов к аналитическим данным

2. Создайте HTTP-сессию requests.Session, выполните аутентификация в административном сервисе (полученные cookie сохраняются в сессии). Реализуйте логику отправки SQL-запросов на эндпоинт, заданный переменной EXECUTE_SQL_URL (функция sql в примере).

    
   # Запрос аналитических данных VK Business Analyticsimport requestssession = requests.Session()# Аутентификация в административном сервисеtry:    login_response = session.post(LOGIN_URL, json=LOGIN_PAYLOAD)    login_response.raise_for_status()    print("Авторизация прошла успешно.")except requests.exceptions.HTTPError as err:    print(f"Ошибка HTTP: {err}")def sql(text: str = "select 1 as test"):    """    Отправляет заданный SQL-запрос и возвращает результирующие данные в JSON    """    try:        sql_payload = {            "method": "sql",            "params": {                "query": text            }        }        sql_response = session.post(EXECUTE_SQL_URL, json=sql_payload)        sql_response.raise_for_status()        result = sql_response.json()        return result    except requests.exceptions.HTTPError as err:        print(f"Ошибка HTTP: {err}")

3. Реализованную функцию можно использовать для выполнения запросов к модели данных с использованием расширенного диалекта SQL платформы VK Business Analytics.

   Следующий пример кода демонстрирует, как получить общее количество экземпляров процесса в модели.

    
   # Запрос количества экземпляров процесса res = sql('''     select count(*) from "case_log" ''')

   Результат выполнения запроса представляет из себя словарь (десериализованный JSON-объект) и доступен через поле res['result']. Для доступа к заголовкам и типам колонок, используйте поле res['result']['metadata']. Для доступа к данным, используйте поле ['result']['rows'].

   В примере запрос возвращает одно значение. Пример кода ниже демонстрирует, как добавить его к тексту сообщения:

    
   # Отправка сообщения (значения переменных могут быть переопределены параметрами планировщика)send_email_with_attachment(    to=recipient,    subject=subject,    body_text=f"Всего экземпляров процесса: {res['result']['rows'][0][0]}",    attachment_path=attachment_path)

Этот раздел демонстрирует, как на основе данных, полученных из аналитического сервиса, сгенерировать HTML-отчёт и отправить его как вложение к письму.

1. Подготовьте HTML-файл, который будет использоваться в качестве шаблона для генерации HTML-отчёта, например:

    
   <!-- ./report_template.html --><!DOCTYPE html><html lang="ru"><head>    <meta charset="UTF-8">    <title>Отчёт</title>    <style>        body {            font-family: Arial, sans-serif;            color: #333333;        }        table {            border-collapse: collapse;            width: 100%;        }        th, td {            border: 1px solid #cccccc;            padding: 8px 12px;            text-align: left;        }        th {            background-color: #f2f2f2;        }        tr:nth-child(even) {            background-color: #fafafa;        }    </style></head><body>    <h2>Отчёт</h2>    <h3>Наиболее длительные завершённые экземпляры процесса:</h3>    {{TABLE}} <!-- плейсхолдер для подстановки данных отчёта --></body></html>

2. Сформируйте SQL-запрос для получения табличных данных. Например, запрос ниже получает пять наиболее длительных законченных экземпляров процесса.

    
   res = sql('''   select       case_id as "Экз. процесса",       round("activity_log@duration_fact" / 3600.0, 1) as "Общая длительность (ч)",       first_event(activity_log.activity_start) as "Дата начала",       last_event(activity_log.activity, activity_log.duration = max_event(activity_log.duration)) as "Самое продолжительное событие"   from case_log   where last_event(activity_log.activity) in ('Оплата счета', 'Выплата компенсации сотруднику')   order by "Общая длительность (ч)" desc   limit 5''')

3. Реализуйте логику, преобразующую результат SQL-запроса в HTML-разметку.

   Код ниже извлекает из результата запроса названия колонок и строки данных, форматирует значения дат к виду ДД.ММ.ГГГГ ЧЧ:ММ:СС (поддерживаются как timestamp в миллисекундах, так и строки в формате ISO 8601), собирает HTML-таблицу и подставляет её вместо плейсхолдера {{TABLE}} в шаблоне report_template.html. Результат сохраняется в файл report.html, путь к которому записывается в переменную attachment_path для последующей отправки:

    
   # Генерация HTML-отчёта на основе данных в переменной resfrom datetime import datetime, timezonecolumns = [col["name"] if isinstance(col, dict) else col for col in res["result"]["metadata"]]rows = res["result"]["rows"]DATE_COLUMNS = {"Дата начала"}def format_cell_value(column, value):    """Форматирует значение ячейки; для колонок с датами приводит к виду ДД.ММ.ГГГГ ЧЧ:ММ:СС."""    if column not in DATE_COLUMNS or value is None:        return value    try:        if isinstance(value, (int, float)):            # Похоже на timestamp в миллисекундах            ts = value / 1000 if value > 1e12 else value            dt = datetime.fromtimestamp(ts, tz=timezone.utc)        else:            text = str(value).strip()            if text.endswith("Z"):                text = text[:-1] + "+00:00"            dt = datetime.fromisoformat(text)        return dt.strftime("%d.%m.%Y %H:%M:%S")    except (ValueError, TypeError):        return valuedef build_html_table(columns, rows):    thead = "<tr>" + "".join(f"<th>{col}</th>" for col in columns) + "</tr>"    tbody_rows = []    for row in rows:        cells = "".join(            f"<td>{format_cell_value(col, cell)}</td>" for col, cell in zip(columns, row)        )        tbody_rows.append(f"<tr>{cells}</tr>")    tbody = "".join(tbody_rows)    return f"<table>{thead}{tbody}</table>"html_table = build_html_table(columns, rows)# Подстановка таблицы в шаблон отчёта и сохранение результата в report.htmltemplate_path = Path("report_template.html")report_path = Path("report.html")template_html = template_path.read_text(encoding="utf-8")report_html = template_html.replace("{{TABLE}}", html_table)report_path.write_text(report_html, encoding="utf-8")print(f"✔ Отчёт сохранён в {report_path.resolve()}")attachment_path = str(report_path)

После генерации отчёта письмо с вложением отправляется той же функцией send_email_with_attachment — получатель увидит в письме оформленный HTML-отчёт с таблицей данных.

Режим Подготовка данных и ML поставляется со встроенным планировщиком, позволяющим запускать блокноты на выполнение по расписанию.

Для получения подробной информации о функциональности планировщика, обратитесь к документации jupyterlab-scheduler.

1. Для того чтобы создать задачу на выполнение блокнота, откройте этот блокнот и нажмите .

   

2. Параметры задачи позволяют переопределять переменные блокнота, заданные в ячейке с тегом parameters. Для того чтобы задать значение параметра, нажмите в блоке parameters и укажите желаемые имя и значение параметра.

   

3. Для запуска по расписанию, выберите опцию Run on a schedule в блоке Schedule. Задайте необходимую периодичность и нажмите Create.

   

1. Для редактирования расписания, нажмите (новый лаунчер) и выберите Notebook Jobs в отображённой вкладке.

   

2. На вкладке Notebook Jobs перейдите в раздел Notebook Job Definitions и нажмите на имя задачи, которую необходимо отредактировать.

   

3. В отображенной вкладке нажмите Edit job definition.

   

4. Внесите необходимые правки в настройки задачи и нажмите Save Changes для сохранения изменений.

1. Нажмите (новый лаунчер) и выберите Notebook Jobs в отображённой вкладке.

   

2. На вкладке Notebook Jobs перейдите в раздел Notebook Job Definitions и нажмите в строке, отображающей задачу, которую необходимо удалить.