Як делегувати простий звіт роботи. Пишемо бота на Python і Google BigQuery

Чи є у вас завдання, які повторюються день за днем, з тижня на тиждень? Наприклад, написання звітів. Ви запитуєте дані, проводите аналіз, візуалізуєте (робите графіки, діаграми), потім відправляєте начальнику. Але якщо все це автоматизувати?

У цьому туторіалі ми створимо робота для Telegram, який допоможе автоматизувати звітність. А найкласніше — вся програма складатиметься лише з 50 рядків коду! Якщо ви створюєте бота для Telegram вперше, то варто прочитати ще ось цей пост.

Skillbox рекомендує: Практичний курс Python-розробник з нуля.

Нагадуємо: для всіх читачів "Хабра" - знижка 10 000 рублів при записі на будь-який курс Skillbox за промокодом "Хабр".

приступаємо

Встановлюємо бібліотеки

Ми будемо використовувати google-cloud-bigquery для отримання даних із Google BigQuery. matplotlib, нумпі и панди допоможуть візуалізувати дані. python-telegram-бот відправить готові дані до Telegram.

pip3 install google-cloud-bigquery matplotlib numpy pandas python-telegram-bot

Підключаємо Google BigQuery API

Якщо ми бажаємо використовувати сервіс, потрібно підключити Google BigQuery API. Для цього йдемо в Консоль розробників Google і створюємо новий проект (або вибираємо існуючий).

У контрольній панелі вибираємо ENABLE APIS AND SERVICES та шукаємо BigQuery API.

Вибираємо Enable для підключення API.

Створюємо ключ облікового запису

Знову вирушаємо в Консоль розробників Google, вибираємо вкладку 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.

імпортувати matplotlib.pyplot як 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