Отправляем себе сообщения в телеграмм через питон. Как это может быть полезно в работе аналитика?

Всем привет! Зовут меня Виталий, автор тг канала Детектив данных про мой вкат в аналитику данных.

‍Сегодня я хочу поделиться с вами одной полезной функцией, которую когда-то придумал для себя и теперь активно использую в своей работе. Это такая мини-надстройка, которая помогает сделать регулярные отчёты немного приятнее и функциональнее, хоть это и не обязательное улучшение.

С помощью Python мы научимся отправлять себе сообщения в Telegram от имени нашего собственного бота. Причём это будут не просто сообщения, а уведомления с информацией о времени загрузки отчёта и ещё и с графиком для анализа. Пошагово разберём, как это сделать, обсудим, зачем это нужно и какие возможности для улучшения есть.

Эта статья будет полезна как начинающим в мире Python, так и продвинутым пользователям, которые проходили мимо работы с телеграммом

Немного предыстории: есть у меня один регулярный ежедневный отчёт, который грузится самым первым, еще до начала рабочего дня. Отчёт состоит из нескольких тяжелых SQL запросов, и по ним я обычно с помощью команды %%time отслеживал скорость загрузки данных. В запросы могут быть внесены изменения, и хочется отследить повлияло ли это на быстродействие выгрузки. И вообще понять — а как сегодня работает сервер, будем ли мы летать или пол дня выгружать «select * from table».

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

Ты можешь быть на созвоне, есть, спать в конце концов — а тут сообщение на часах «отчёт Х загружен». Отлично, сейчас поставлю следующую выгрузку а с этими данными можно начинать работать.

Или что поинтереснее «запрос Y выполнен за пятнадцать минут» Хм… А так то он обычно выгружается по сорок — значит пора бить тревогу и идти смотреть — а что там произошло? а что-то да произошло раз время сократилось. А ты вовремя пришёл, поправил и поставил заново — не потеряв драгоценные часы работы, и поиск места ошибок постфактум окончания выгрузки с неполными данными.

Порядок статьи такой:

• создаём своего бота, берём с него всю инфу

• пишем код на отправку сообщений

• нюансы и доделки

• график

Создаём бота в Телеграме

Открываем поиск и ищем 

@BotFather

Придумываем название бота (должно оканчиваться на bot), пусть будет detective_test_report_bot

Вам будет выдан токен примерно такого вида, сохраняем его

1234567894:HUJRhjh_sPoO135eQz4EwbFBKJkTcIGBMCM66

В целом больше ничего тут не нужно, кроме пожалуй аватарки бота, которую можно загрузить прямо тут через команду

/setuserpic

Переходим в уже наш созданный бот и жмём/вводим команду

/start

И обязательно пишем боту любой текст

hello world

С телегой закончили, начинаем питонить:

import requests TOKEN = "В КАВЫЧКИ ВСТАВЛЯЕМ СВОЙ ТОКЕН" url = f"https://api.telegram.org/bot{TOKEN}/getUpdates" print(requests.get(url).json())

В результате выполнения кода вы увидите информацию о последнем отправленном сообщении боту, среди всей информации вы увидите кусок

…’chat’: {‘id’4815162342,…. Вот  это число id нам и нужно это наш чат айди

 Сохраняем ваш чат айди в переменную

Готово. Уже сейчас мы можем отправить себе первое сообщение: 

import requests report_name = 'Отчёт' bot_token = '1234567894:HUJRhjh_sPoO135eQz4EwbFBKJkTcIGBMCM66' chat_id = '4815162342'  send_message_url = f'https://api.telegram.org/bot{bot_token}/sendMessage' message_text = f'Отчёт {report_name} - загрузка начата'  payload = {    'chat_id': chat_id,    'text': message_text    } response = requests.post(send_message_url, data=payload) message_text

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

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

• Время загрузки отчетов

• Небольшое условное форматирование, смайлики

• Отправка сообщений нескольким пользователям

• Графики

Поехали сначала — время запроса

Перед запросом, или отчётом фиксируем текущее время

import time start_time = time.time() 

после отчета или его части фиксируем время окончания, и округляем получившуюся разницу :

end_time = time.time() elapsed_time = end_time - start_time elapsed_minutes_end = round(elapsed_time / 60)

Теперь мы можем добавить в сообщение время выполнения запроса, или загрузки отчета в целом. Просто редактируем наш message_text заодно выполнив два переноса строки с помощью «\n\n» не забыв взять сообщение в скобки

message_text = (f'Запрос 3 - данные загружены за {elapsed_minutes_3} минут\n\n' f'Отчёт {report_name} загружен за {elapsed_minutes_end} минут ')

Супер. Отчет стал полезнее, давайте поколдуем что-бы он стал немного нагляднее:

Добавим новый атбирут в payload:

‘parse_mode’: ‘Markdown’ – теперь мы можем просто поставить ** в тексте и шрифт выделится жирным, ну или

‘parse_mode’: ‘HTML’ если вы как и я (используем свои знаниями из 2005 года) хотите использовать теги <b> </b>

Что-то не хватает – смайликов! Помогут отделить один вид сообщений от других – как ни странно ничего выдумывать не надо, просто копируете понравившийся смайлик хоть тут в телеге и вставляете в свой текст в питоне. Можно вставить напрямую, а можно использовать юникод код например «\u2198\ufe0f» для стрелки (юникод можно посмотреть в теле ссылки например telegram.org/a/img-apple-64/2198-fe0f.png или найти любую таблицу с кодами)

chat_ids = [‘151675936’, ‘463459322’]

и сам цикл

for chat_id in chat_ids:    payload = {'chat_id': chat_id,  'text': message_text, 'parse_mode': 'Markdown' }    response = requests.post(send_message_url, data=payload) message_text -- в конце (уже вне цикла) я вывожу себе одно сообщение для проверки.

Ну вот и всё, ничего сложного и довольно удобно. Но отправлять можно не только сообщения, но и небольшие файлы и фото, например графики (только аккуратно, данные составляющие коммерческую тайну через телегу я бы не советовал никуда отправлять без предварительных согласований), а что-то нейтральное — вроде времени отработки запроса — почему бы и нет? В следующем посте — создадим мини базу для хранения времени загрузки отчётов, настроим её обновление, отрисуем график с помощью библиотеки matplotlib, покопаемся в мелочах и отправим себе удобную картинку с анализом нашей загрузки. Жду ваших эмоций к посту, комментариев — и до скорой встречи в новых статьях!

Ну а теперь самое интересное — отрисуем график загрузки

Давайте договоримся о переменных. Почти все они уже были определены в прошлый раз, и сейчас мы их просто обозначим для независимости первой части стати от второй статей друг от друга.

• наименование отчёта — это нужно для фильтрации в дальнейшем

• токен ТГ бота

• наш чат айди с ботом

• и время загрузки отчета

report_name = 'Daily report' bot_token = '1234567894:HUJRhjh_sPoO135eQz4EwbFBKJkTcIGBMCM66' chat_id = '4815162342' last_value= 87

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

import pandas as pd data = {  'дата': ["04.06", "05.06", "07.06", "08.06", "09.06", "10.06", "11.06", "12.06", "13.06", "14.06", "15.06", "16.06"],  'значение': [74, 80,  77,  86,  103,  70,  93,  81,  81,  71,  80,  78],  'отчёт': ["Daily report"] * 12 # } df = pd.DataFrame(data) df.to_csv('data.csv', index=False)

Далее создаём таблицу с сегодняшними значениями и записываем её в csv, не переписывая а добавляя его в базу (mode=’a’)

from datetime import datetime data = {   'дата': [datetime.now().strftime('%-d.%m')],   'значение': [last_value],   'отчёт': [report_name] } df_time = pd.DataFrame(data) df_time.to_csv('data.csv', mode='a', header=False, index=False)

Затем читаем наш файл, фильтруем по текущему отчёту и выбираем последние 10 записей

loaded_df = pd.read_csv(csv_file) filtered_df = loaded_df[loaded_df['отчёт'] == report_name].tail(10) filtered_df

Далее минутка или даже ячейка душноты — мини функция которая определяет правильную форму слова «Минута» в зависимости от числа. Да, вроде и не неважно, но кто я чтобы сопротивляется своему внутреннему перфекционизму?

def get_minute_word(number):     if 11 <= number % 100 <= 19:         return 'минут'     else:         last_digit = number % 10         if last_digit == 1:             return 'минуту'         elif 2 <= last_digit <= 4:             return 'минуты'         else:             return 'минут'

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

• Оси,

• Среднее время загрузки,

• Минимальное и максимальное значение,

• Последнее (текущее) значение времени загрузки — перезапишем last_value (хоть можно и оставить, но для надёжности, я всё-таки решил взять последнее значение из таблицы)

• Правильные формы «Минуты» для последнего и среднего значения

import matplotlib.pyplot as plt import requests from io import BytesIO # Подготовка данных для графика x_data = filtered_df['дата'].astype(str).tolist() y_data = filtered_df['значение'].tolist() average_value = round(sum(y_data) / len(y_data)) min_value = min(y_data) max_value = max(y_data) last_value = y_data[-1]  minute_last = get_minute_word(last_value) minute_avg = get_minute_word(average_value)

Ну и сам график (важно поместить весь код ниже — в одну ячейку)

# Параметры графика # Размер графика, основная линия, линяя среднего + легенда, в которой, мы сразу укажем среднее время загрузки графика и заголовок plt.figure(figsize=(12, 6)) plt.plot(x_data, y_data, linestyle='-', linewidth=3, color='teal') plt.axhline(y=average_value, color='red', linestyle='--', linewidth=2, label=f'В среднем: {average_value} {minute_avg}') plt.title(f'Cкорость загрузки отчёта "{report_name}" за последние десять дней, мин.', fontsize=15, fontweight='bold')  # Далее отображаем на графике только значения минимума, максимума и последнего времени загрузки.  # Учитываем среднее время, для понимания - будет ли последнее значение выше или ниже относительно своей точки графика.  # Добавляем цвет (сейчас  интуитивно непонятно что лучше - высокое или низкое значение) - пусть минимальное время загрузки будет зеленым, и наоборот самая долгая загрузка (и выше средней последнее значение) - будут красными. for i, (x, y) in enumerate(zip(x_data, y_data)):   if y == min_value:     plt.text(x, y - 3, f'{y}', ha='center', fontsize=15, color='green') # Мин значение снизу   elif y == max_value:     plt.text(x, y + 2, f'{y}', ha='center', fontsize=15, color='red')  # Макс значение сверху # Выводим последнее значение только для последней даты   elif i == len(y_data) - 1:     if y < average_value:       plt.text(x, y - 3, f'{y}', ha='center', fontsize=15, color='green') # Last value снизу и зеленым     else:       plt.text(x, y + 2, f'{y}', ha='center', fontsize=15, color='red')  # Last value сверху и красным  # Настройка осей и легенды. Подписи на временной оси X оставим, а Y будто и не нужна, ведь есть значения.  # Определяем мин и макс оси Y - мы строим не от нуля, нам важно видеть мелкие колебания - опытным путем на этих данных оптимально +-10% plt.xticks(x_data) plt.yticks([]) plt.ylim(min(y_data)  0.9, max(y_data)  1.1)  # убираем рамку вокруг графика for spine in plt.gca().spines.values():   spine.set_visible(False)  # показываем легенду plt.legend(frameon=False, fontsize=12)  # Сохранение в буфере и отправка графика. Будто и нет смысла сохранять график в виде картинки в памяти, так как всё равно этот график придёт вам в ТГ. buf = BytesIO() plt.savefig(buf, format='png') buf.seek(0)  send_photo_url = f'https://api.telegram.org/bot{bot_token}/sendPhoto' files = {'photo': buf} payload = {     'chat_id': chat_id,     'caption': f'🏁 Отчёт "{report_name}" загружен за {last_value} {minute_last}!\nСреднее время загрузки составило {average_value} {minute_avg}',     'parse_mode': 'Markdown'} response = requests.post(send_photo_url, files=files, data=payload)  buf.close()

Результат еще раз:

Быстро.

Удобно.

Не требует вмешательств и ручных корректировок.

Делаем один раз и наслаждаемся всегда.

Спасибо за просмотр, пишите комментарии, заходите в гости в ТГ канал Детектив данных.

Сподвигла ли вас статья на какие-либо изменения в ваших проектах?