Как делегировать простые отчеты роботу. Пишем бота на 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 рекомендует:

• Двухлетний практический курс «Я — веб-разработчик PRO».
• Онлайн-курс «С#-разработчик с 0».
• Практический годовой курс «PHP-разработчик с 0 до PRO».

Источник: habr.com