[Перевод] Создаем бота для несложной отчетности при помощи Python и Google BigQuery
Есть ли у вас задачи, которые повторяются изо дня в день, из недели в неделю? Например, написание отчетов. Вы запрашиваете данные, проводите анализ, визуализируете (делаете графики, диаграммы), а затем отправляете начальнику. Но что, если все это автоматизировать?
В этом туториале мы создадим бота для Telegram, который поможет автоматизировать отчетность. А самое классное — вся программа будет состоять всего из 50 строк кода! Если вы создаете бота для Telegram впервые, то стоит прочитать еще вот этот пост.
Skillbox рекомендует: Практический курс Python-разработчик с нуля.
Напоминаем: для всех читателей «Хабра» — скидка 10 000 рублей при записи на любой курс Skillbox по промокоду «Хабр».
Приступаем
Устанавливаем библиотеки
Мы будем использовать google-cloud-bigquery для получения данных из Google BigQuery. matplotlib, numpy и pandas помогут визуализировать данные. python-telegram-bot отправит готовые данные в Telegram.
pip3 install google-cloud-bigquery matplotlib numpy pandas python-telegram-bot
Подключаем Google BigQuery API
Если мы хотим использовать сервис, нужно подключить Google BigQuery API. Для этого идем в Google Developers Console и создаем новый проект (или же выбираем существующий).
В контрольной панели выбираем ENABLE APIS AND SERVICES и ищем BigQuery API.
Выбираем Enable для подключения API.
Создаем ключ аккаунта
Снова отправляемся в Google Developers Console, выбираем вкладку Credentials, Create credentials и Service account key.
Затем — New service account, и в поле Service account name вводим имя.
Из выпадающего списка Role выбираем Project > Owner, затем Create.
Файл, который будет автоматически загружаться, называем creds.json.
Выставляем GOOGLE_APPLICATION_CREDENTIALS, указав путь к creds.json в терминале.
export GOOGLE_APPLICATION_CREDENTIALS='[PATH_TO_CREDS.JSON]'
Если все прошло хорошо, самое время начать писать программу.
Создание приложения
Для туториала мы будем использовать данные из bigquery-public-data.stackoverflow, для нашего отчета выберем количество ежедневных публикаций.
Все достаточно просто.
Query the table → Visualize the data → Save the visualization → Send the image
Давайте создадим одну функцию для определения каждого потока.
Query to BigQuery
Сначала импортируем библиотеку.
from google.cloud import bigquery
Создаем функцию с названием query_to_bigquery, где параметр — query.
def query_to_bigquery(query):
client = bigquery.Client()
query_job = client.query(query)
result = query_job.result()
dataframe = result.to_dataframe()
return dataframe
Эта функция вернет запрос в виде фрейма данных.
Визуализируем данные
Для решения этой задачи выбираем matplotlib.
import matplotlib.pyplot as plt
Нам нужно пять параметров, где x — данные по оси х, x_label — название для оси, y — данные оси y, y_label — название для оси и title — заголовок всей визуализации.
def visualize_bar_chart(x, x_label, y, y_label, title):
plt.title(title)
plt.xlabel(x_label)
plt.ylabel(y_label)
index = np.arange(len(x))
plt.xticks(index, x, fontsize=5, rotation=30)
plt.bar(index, y)
return plt
Сохраняем изображение
Теперь давайте воспользуемся двумя функциями, чтобы создать визуализацию и сохранить ее.
Будем отправлять число ежедневно публикуемых постов. Сначала пишем запрос.
query = """
SELECT DATE(creation_date) date, COUNT(*) total_posts
FROM `bigquery-public-data.stackoverflow.post_history`
GROUP BY 1
HAVING date > DATE_SUB('2018-12-02', INTERVAL 14 DAY)
ORDER BY 1
"""
Запрос помогает собрать данные за две недели, начиная с 2 декабря 2018 года.
Мы используем эту дату, потому что 2018–12–02 — последние данные, записанные в bigquery-public-data.stackoverflow.post_history, в других случаях вы можете использовать CURRENT_DATE () для получения самых новых данных.
Вызываем функцию query_to_bigquery для получения данных.
dataframe = query_to_bigquery (query)
Затем используем колонку данных date для оси х, а колонку total_posts для оси у.
x = dataframe['date'].tolist ()
y = dataframe['total_posts'].tolist ()
Визуализируем при помощи функции visualize_bar_chart и сохраняем в виде изображения.
plt = visualize_bar_chart (x=x, x_label='Date', y=y, y_label='Total Posts', title='Daily Posts')
plt.savefig ('viz.png')
Оборачиваем этот код в функцию под названием get_and_save_image.
def get_and_save_image():
query = """
SELECT DATE(creation_date) date, COUNT(*) total_posts
FROM `bigquery-public-data.stackoverflow.post_history`
GROUP BY 1
HAVING date > DATE_SUB('2018-12-02', INTERVAL 14 DAY)
ORDER BY 1
"""
dataframe = query_to_bigquery(query)
x = dataframe['date'].tolist()
y = dataframe['total_posts'].tolist()
plt = visualize_bar_chart(x=x, x_label='Date', y=y, y_label='Total Posts', title='Daily Posts')
plt.savefig('viz.png')
Отправляем изображение
Для того, чтобы отправить отчет адресату, нужно знать параметр chat_id.
Используем userinfobot и набираем /start. Бот отвечает нужной информацией, chat_id содержится в поле id.
Теперь создаем функцию send_image. Она будет задействовать функцию get_and_save_image для получения и сохранения изображения. А затем уже отправляем все правильному контакту.
def send_image(bot, update):
get_and_save_image()
chat_id = 'CHAT_ID_RECEIVER'
bot.send_photo(chat_id=chat_id, photo=open('viz.png','rb'))
Главная программа
Наконец, создаем еще одну функцию, main, для запуска приложения. Не забывайте изменить YOUR_TOKEN для бота.
Запомните: эта программа отправит изображение автоматически в указанное вам время. К примеру, мы будем отправлять отчет в девять утра ежедневно.
def main():
updater = Updater('YOUR_TOKEN')
updater.job_queue.run_daily(send_image, time=datetime.datetime.strptime('9:00AM', '%I:%M%p').time(), days=(0,1,2,3,4,5,6))
updater.start_polling()
updater.idle()
if __name__ == '__main__':
main()
В итоге наше приложение будет выглядеть вот так:
from google.cloud import bigquery
from telegram.ext import Updater
import matplotlib.pyplot as plt
import numpy as np
import datetime
def query_to_bigquery(query):
client = bigquery.Client()
query_job = client.query(query)
result = query_job.result()
dataframe = result.to_dataframe()
return dataframe
def visualize_bar_chart(x, x_label, y, y_label, title):
plt.title(title)
plt.xlabel(x_label)
plt.ylabel(y_label)
index = np.arange(len(x))
plt.xticks(index, x, fontsize=5, rotation=30)
plt.bar(index, y)
return plt
def get_and_save_image():
query = """
SELECT DATE(creation_date) date, COUNT(*) total_posts
FROM `bigquery-public-data.stackoverflow.post_history`
GROUP BY 1
HAVING date > DATE_SUB('2018-12-02', INTERVAL 14 DAY)
ORDER BY 1
"""
dataframe = query_to_bigquery(query)
x = dataframe['date'].tolist()
y = dataframe['total_posts'].tolist()
plt = visualize_bar_chart(x=x, x_label='Date', y=y, y_label='Total Posts', title='Daily Posts')
plt.savefig('viz.png')
def send_image(bot, update):
get_and_save_image()
chat_id = 'CHAT_ID_RECEIVER'
bot.send_photo(chat_id=chat_id, photo=open('viz.png', 'rb'))
def main():
updater = Updater('YOUR_TOKEN')
updater.job_queue.run_daily(send_image, time=datetime.datetime.strptime('9:00AM', '%I:%M%p').time(), days=(0,1,2,3,4,5,6))
updater.start_polling()
updater.idle()
if __name__ == '__main__':
main()
Сохраняем файл и называем его main.py.
Запускаем приложение, вводя в терминале команду:
python3 main.py
Все готово. Теперь у нас есть робот, состоящий из 50 строк кода, который формирует отчетность без нашего вмешательства.
Проверим бота отсюда, выбрав команду /send.
Готовый код вы можете получить в моем GitHub.
Skillbox рекомендует:
