Всем привет! Зовут меня Виталий, автор тг канала Детектив данных про мой вкат в аналитику данных.
Сегодня я хочу поделиться с вами одной полезной функцией, которую когда-то придумал для себя и теперь активно использую в своей работе. Это такая мини-надстройка, которая помогает сделать регулярные отчёты немного приятнее и функциональнее, хоть это и не обязательное улучшение.
С помощью 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 или найти любую таблицу с кодами)
Может случиться необходимость когда сообщение с бота нужно отправить нескольким коллегам – всё просто отправляем коллегу писать сообщение бота и смотрим его чат айди, с помощью первого кода в статье, а дальше запускаем самый стандартный цикл отправки через for, теперь оба айди сохраняем в переменную
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()Результат еще раз:
Быстро.
Удобно.
Не требует вмешательств и ручных корректировок.
Делаем один раз и наслаждаемся всегда.
Спасибо за просмотр, пишите комментарии, заходите в гости в ТГ канал Детектив данных.
Сподвигла ли вас статья на какие-либо изменения в ваших проектах?