Kako prenesti preprosta poročila na robota. Pisanje bota v Pythonu in Google BigQuery

Ali imate naloge, ki se ponavljajo dan za dnem, teden za tednom? Na primer pisanje poročil. Zahtevate podatke, jih analizirate, vizualizirate (naredite grafe, grafikone) in jih nato pošljete svojemu šefu. Kaj pa, če bi bilo vse to avtomatizirano?

V tej vadnici bomo ustvarili bota za Telegram, ki bo pomagal avtomatizirati poročanje. In najbolj kul je, da bo celoten program sestavljen iz samo 50 vrstic kode! Če prvič ustvarjate bota za Telegram, potem morate prebrati tudi to post.

Skillbox priporoča: Praktični tečaj Python razvijalec iz nič.

Spomnimo: za vse bralce "Habr" - popust v višini 10 rubljev ob vpisu v kateri koli tečaj Skillbox s promocijsko kodo "Habr".

Začnimo

Namestitev knjižnic

Uporabili bomo google-cloud-bigquery za pridobivanje podatkov iz Google BigQuery. matplotlib, otopeli и pand vam bo pomagal vizualizirati vaše podatke. python-telegram-bot bo končane podatke poslal Telegramu.

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

Povezovanje Google BigQuery API

Če želimo uporabljati storitev, moramo povezati Google BigQuery API. Da bi to naredili, gremo na Google Developers Console in ustvarite nov projekt (ali izberite obstoječega).

Na nadzorni plošči izberite OMOGOČI APIS IN STORITVE in poiščite BigQuery API.

Izberite Omogoči, da povežete API.

Ustvarite ključ računa

Pojdimo znova Google Developers Console, izberite zavihek Poverilnice, Ustvari poverilnice in Ključ računa storitve.

Nato - Nov račun storitve in vnesite ime v polje Ime računa storitve.

Na spustnem seznamu Vloga izberite Projekt > Lastnik in nato Ustvari.

Datoteka, ki bo samodejno prenesena, se imenuje creds.json.

Nastavite GOOGLE_APPLICATION_CREDENTIALS in navedite pot do creds.json v terminalu.

izvoz GOOGLE_APPLICATION_CREDENTIALS='[PATH_TO_CREDS.JSON]'

Če je šlo vse v redu, je čas, da začnemo pisati program.

Ustvarjanje aplikacije

Za vadnico bomo uporabili podatke iz bigquery-public-data.stackoverflow, za naše poročilo bomo izbrali število dnevnih objav.

Vse je zelo preprosto.

Poizvedi po tabeli -> Vizualiziraj podatke -> Shrani vizualizacijo -> Pošlji sliko

Ustvarimo eno funkcijo za definiranje vsake niti.

Poizvedba v BigQuery

Najprej uvozimo knjižnico.

iz google.cloud uvozite bigquery

Ustvarimo funkcijo, imenovano query_to_bigquery, kjer je parameter poizvedba.

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

Ta funkcija vrne zahtevo kot podatkovni okvir.

Vizualizacija podatkov

Za rešitev te težave izberite matplotlib.

uvoz matplotlib.pyplot kot plt

Potrebujemo pet parametrov, kjer so x podatki o osi x, x_label je naslov za os, y so podatki o y-osi, y_label je naslov za os in naslov je naslov celotne 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

Shranite sliko

Zdaj pa uporabimo dve funkciji za ustvarjanje vizualizacije in njeno shranjevanje.

Pošiljali vam bomo dnevno število objavljenih objav. Najprej napišemo prošnjo.

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

Poizvedba pomaga zbirati podatke za dva tedna od 2. decembra 2018.

Ta datum uporabljamo, ker je 2018-12-02 zadnji podatek, zabeležen v bigquery-public-data.stackoverflow.post_history, v drugih primerih pa lahko uporabite CURRENT_DATE(), da dobite najnovejše podatke.

Za pridobitev podatkov pokličite funkcijo query_to_bigquery.

dataframe = query_to_bigquery(poizvedba)

Nato uporabimo stolpec datumskih podatkov za os x in stolpec total_posts za os y.

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

Vizualiziramo ga s funkcijo visualize_bar_chart in shranimo kot sliko.

plt = visualize_bar_chart(x=x, x_label='Datum', y=y, y_label='Skupno število objav', title='Dnevne objave')
plt.savefig('viz.png')

To kodo zavijemo v funkcijo, imenovano 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šlji sliko

Če želite poslati poročilo prejemniku, morate poznati parameter chat_id.

Uporaba userinfobot in vnesite /start. Bot odgovori s potrebnimi informacijami, chat_id je v polju id.

Zdaj pa ustvarimo funkcijo send_image. Za pridobitev in shranjevanje slike bo uporabil funkcijo get_and_save_image. In potem vse pošljemo 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

Na koncu ustvarimo še eno funkcijo, glavno, za zagon aplikacije. Ne pozabite spremeniti YOUR_TOKEN za bota.

Ne pozabite: ta program bo samodejno poslal sliko ob času, ki ga določite. Vsak dan bomo na primer poslali poročilo ob devetih zjutraj.

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

Posledično bo naša aplikacija videti takole:

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

Shranite datoteko in jo poimenujte main.py.

Aplikacijo zaženemo tako, da v terminal vnesemo ukaz:

python3 main.py

Vse je pripravljeno. Zdaj imamo robota, sestavljenega iz 50 vrstic kode, ki ustvarja poročila brez našega posredovanja.

Preverimo bota zatoz izbiro ukaza /send.

Končano kodo lahko dobite na moj GitHub.

Skillbox priporoča:

• Dvoletni praktični tečaj "Sem PRO spletni razvijalec".
• Spletni tečaj "C# razvijalec od 0".
• Praktični enoletni tečaj "PHP razvijalec od 0 do PRO".

Vir: www.habr.com