Kako delegirati jednostavne izvještaje robotu. Pisanje bota u Pythonu i Google BigQueryju

Imate li zadatke koji se ponavljaju iz dana u dan, iz sedmice u sedmicu? Na primjer, pisanje izvještaja. Zahtijevate podatke, analizirate ih, vizualizirate (napravite grafikone, grafikone), a zatim ih šaljete svom šefu. Ali šta ako je sve ovo automatizovano?

U ovom vodiču ćemo kreirati bota za Telegram koji će pomoći u automatizaciji izvještavanja. A najkul stvar je što će se cijeli program sastojati od samo 50 linija koda! Ako prvi put kreirate bota za Telegram, pročitajte i ovaj post.

Skillbox preporučuje: Praktični kurs Python programer od nule.

Podsećamo: za sve čitaoce "Habra" - popust od 10 rubalja pri upisu na bilo koji Skillbox kurs koristeći "Habr" promotivni kod.

Počinjemo

Instaliranje biblioteka

Koristićemo google-cloud-bigquery da dobijete podatke sa Google BigQueryja. matplotlib, numpy и pande pomoći će vam da vizualizirate svoje podatke. python-telegram-bot će poslati gotove podatke Telegramu.

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

Povezivanje Google BigQuery API-ja

Ako želimo koristiti uslugu, moramo povezati Google BigQuery API. Da bismo to uradili idemo na Google Developers Console i kreirajte novi projekat (ili odaberite postojeći).

Na kontrolnoj tabli odaberite OMOGUĆI API-JE I USLUGE i potražite BigQuery API.

Odaberite Omogući da povežete API.

Kreirajte ključ računa

Idemo ponovo Google Developers Console, odaberite karticu Akreditivi, Kreirajte vjerodajnice i ključ računa usluge.

Zatim - Novi nalog usluge i unesite ime u polje Naziv naloga usluge.

Na padajućoj listi Uloga odaberite Projekt > Vlasnik, a zatim Kreiraj.

Fajl koji će se automatski preuzeti zove se creds.json.

Postavite GOOGLE_APPLICATION_CREDENTIALS, navodeći stazu do creds.json u terminalu.

izvoz GOOGLE_APPLICATION_CREDENTIALS='[PATH_TO_CREDS.JSON]'

Ako je sve prošlo kako treba, vrijeme je da počnete pisati program.

Kreiranje aplikacije

Za tutorijal ćemo koristiti podatke iz bigquery-public-data.stackoverflow, za naš izvještaj ćemo odabrati broj dnevnih publikacija.

Prilično je jednostavno.

Upitajte tabelu -> Vizualizirajte podatke -> Sačuvajte vizualizaciju -> Pošaljite sliku

Kreirajmo jednu funkciju za definiranje svake niti.

Upit za BigQuery

Prvo uvozimo biblioteku.

sa google.cloud import bigquery

Kreiramo funkciju pod nazivom query_to_bigquery, gdje je parametar query.

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

Ova funkcija će vratiti zahtjev kao okvir podataka.

Vizualizacija podataka

Da biste riješili ovaj problem, odaberite matplotlib.

uvozite matplotlib.pyplot kao plt

Treba nam pet parametara, gdje je x podaci osi x, x_label je naslov za os, y je podatak y ose, y_label je naslov za os, a title je naslov cijele vizualizacije.

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

Sačuvajte sliku

Sada koristimo dvije funkcije da kreiramo vizualizaciju i spremimo je.

Dnevno ćemo slati broj objava. Prvo pišemo zahtjev.

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

Upit pomaže u prikupljanju podataka za dvije sedmice počevši od 2. decembra 2018.

Koristimo ovaj datum jer su 2018-12-02 najnoviji podaci zabilježeni u bigquery-public-data.stackoverflow.post_history, u drugim slučajevima možete koristiti CURRENT_DATE() da dobijete najnovije podatke.

Pozovite funkciju query_to_bigquery da dobijete podatke.

okvir podataka = query_to_bigquery(upit)

Zatim koristimo stupac podataka datuma za x-osu i kolonu total_posts za y-os.

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

Vizualiziramo ga pomoću funkcije visualize_bar_chart i spremamo ga kao sliku.

plt = visualize_bar_chart(x=x, x_label='Datum', y=y, y_label='Ukupno postova', title='Dnevni postovi')
plt.savefig('viz.png')

Ovaj kod umotavamo u funkciju koja se zove 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')

Pošalji sliku

Da biste poslali izvještaj primaocu, morate znati parametar chat_id.

Koristimo userinfobot i otkucajte /start. Bot odgovara sa potrebnim informacijama, chat_id se nalazi u polju id.

Sada kreirajmo funkciju send_image. Koristit će funkciju get_and_save_image za preuzimanje i spremanje slike. A onda šaljemo sve na pravi kontakt.

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

Glavni program

Konačno, kreiramo još jednu funkciju, glavnu, za pokretanje aplikacije. Ne zaboravite promijeniti YOUR_TOKEN za bota.

Zapamtite: ovaj program će automatski poslati sliku u vrijeme koje odredite. Na primjer, svaki dan ćemo poslati izvještaj u devet ujutro.

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

Kao rezultat, naša aplikacija će izgledati ovako:

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

Sačuvajte datoteku i nazovite je main.py.

Aplikaciju pokrećemo unosom naredbe u terminalu:

python3 main.py

Sve je spremno. Sada imamo robota koji se sastoji od 50 linija koda koji generira izvještaje bez naše intervencije.

Hajde da proverimo bota odavdeodabirom naredbe /send.

Gotov kod možete dobiti na moj GitHub.

Skillbox preporučuje:

• Dvogodišnji praktični kurs "Ja sam PRO web developer".
• Online kurs "C# programer od 0".
• Praktična godina kursa "PHP programer od 0 do PRO".

izvor: www.habr.com