Cum să delegați rapoarte simple unui robot. Scrierea unui bot în Python și Google BigQuery

Ai sarcini care se repetă zi de zi, săptămână după săptămână? De exemplu, redactarea rapoartelor. Soliciți date, le analizezi, le vizualizezi (realizezi grafice, diagrame) și apoi le trimiți șefului tău. Dar dacă toate acestea ar fi automatizate?

În acest tutorial vom crea un bot pentru Telegram care va ajuta la automatizarea raportării. Și cel mai tare lucru este că întregul program va fi format din doar 50 de linii de cod! Dacă creați un bot pentru Telegram pentru prima dată, atunci ar trebui să îl citiți și pe acesta rapid.

Skillbox recomandă: Curs practic Dezvoltator Python de la zero.

Amintim: pentru toți cititorii „Habr” - o reducere de 10 de ruble la înscrierea la orice curs Skillbox folosind codul promoțional „Habr”.

Să începem

Instalarea bibliotecilor

Noi vom folosi google-cloud-bigquery pentru a obține date de la Google BigQuery. matplotlib, NumPy и panda vă va ajuta să vă vizualizați datele. python-telegram-bot va trimite datele finalizate către Telegram.

pip3 instalează google-cloud-bigquery matplotlib numpy panda python-telegram-bot

Se conectează API-ul Google BigQuery

Dacă dorim să folosim serviciul, trebuie să conectăm API-ul Google BigQuery. Pentru a face asta mergem la Google Developers Console și creați un proiect nou (sau selectați unul existent).

În panoul de control, selectați ACTIVARE API ȘI SERVICII și căutați API-ul BigQuery.

Selectați Activare pentru a conecta API-ul.

Creați o cheie de cont

Să mergem din nou la Google Developers Console, selectați fila Acreditări, Creare acreditări și Cheia contului de serviciu.

Apoi - Cont de serviciu nou și introduceți numele în câmpul Nume cont de serviciu.

Din lista derulantă Rol, selectați Proiect > Proprietar, apoi Creare.

Fișierul care va fi descărcat automat se numește creds.json.

Setați GOOGLE_APPLICATION_CREDENTIALS, specificând calea către creds.json în terminal.

exportați GOOGLE_APPLICATION_CREDENTIALS='[PATH_TO_CREDS.JSON]'

Dacă totul a mers bine, este timpul să începeți să scrieți programul.

Creați o aplicație

Pentru tutorial vom folosi date din bigquery-public-data.stackoverflow, pentru raportul nostru vom selecta numărul de publicații zilnice.

Totul este destul de simplu.

Interogați tabelul -> Vizualizați datele -> Salvați vizualizarea -> Trimiteți imaginea

Să creăm o funcție pentru a defini fiecare fir.

Interogați la BigQuery

Mai întâi importăm biblioteca.

din google.cloud import bigquery

Creăm o funcție numită query_to_bigquery, unde parametrul este query.

def query_to_bigquery(query):
    client = bigquery.Client()
    query_job = client.query(query)
    result = query_job.result()
    dataframe = result.to_dataframe()
    return dataframe

Această funcție va returna cererea ca un cadru de date.

Vizualizarea datelor

Pentru a rezolva această problemă, alegeți matplotlib.

import matplotlib.pyplot ca plt

Avem nevoie de cinci parametri, unde x este datele axei x, x_label este titlul axei, y este datele axei y, y_label este titlul axei și titlul este titlul întregii vizualizări.

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

Salvați imaginea

Acum să folosim două funcții pentru a crea o vizualizare și a o salva.

Vom trimite numărul de postări publicate zilnic. Mai întâi scriem o cerere.

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
        """

Interogarea ajută la colectarea datelor timp de două săptămâni începând cu 2 decembrie 2018.

Folosim această dată deoarece 2018-12-02 sunt cele mai recente date înregistrate în bigquery-public-data.stackoverflow.post_history, în alte cazuri puteți folosi CURRENT_DATE() pentru a obține cele mai noi date.

Apelați funcția query_to_bigquery pentru a obține datele.

cadru de date = query_to_bigquery(interogare)

Apoi folosim coloana de date date pentru axa x și coloana total_posts pentru axa y.

x = dataframe['data'].tolist()
y = dataframe['total_posts'].tolist()

Îl vizualizăm folosind funcția visualize_bar_chart și îl salvăm ca imagine.

plt = visualize_bar_chart(x=x, x_label='Data', y=y, y_label='Total Postări', title='Daily Posts')
plt.savefig('viz.png')

Încapsulăm acest cod într-o funcție numită 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')

Trimite o imagine

Pentru a trimite un raport destinatarului, trebuie să cunoașteți parametrul chat_id.

Folosim userinfobot și tastați /start. Botul răspunde cu informațiile necesare, chat_id este conținut în câmpul id.

Acum să creăm funcția send_image. Va folosi funcția get_and_save_image pentru a prelua și salva imaginea. Și apoi trimitem totul contactului corect.

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'))

Programul principal

În final, creăm o altă funcție, principală, pentru a lansa aplicația. Nu uitați să schimbați YOUR_TOKEN pentru bot.

Amintiți-vă: acest program va trimite imaginea automat la ora specificată. De exemplu, vom trimite un raport la nouă dimineața în fiecare zi.

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()

Ca rezultat, aplicația noastră va arăta astfel:

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()

Salvați fișierul și numiți-l main.py.

Lansăm aplicația introducând comanda în terminal:

python3 main.py

Totul este gata. Acum avem un robot format din 50 de linii de cod care generează rapoarte fără intervenția noastră.

Să verificăm botul prin urmareprin selectarea comenzii /send.

Puteți obține codul final la GitHub-ul meu.

Skillbox recomandă:

• Curs practic de doi ani „Sunt un dezvoltator web PRO”.
• Curs online „Dezvoltator C# de la 0”.
• Curs practic anual „Dezvoltator PHP de la 0 la PRO”.

Sursa: www.habr.com