Kako delegirati jednostavna izvješća robotu. Pisanje bota u Pythonu i Google BigQueryju

Imate li zadatke koji se ponavljaju dan za danom, tjedan za tjednom? Na primjer, pisanje izvješća. Tražite podatke, analizirate ih, vizualizirate (napravite grafikone, dijagrame), a zatim pošaljete svom šefu. Ali što ako je sve ovo automatizirano?

U ovom tutorijalu izradit ćemo bota za Telegram koji će pomoći automatizirati izvješćivanje. A najkul je što će se cijeli program sastojati od samo 50 redaka koda! Ako prvi put izrađujete bota za Telegram, trebali biste pročitati i ovo pošta.

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

Podsjećamo: za sve čitatelje "Habra" - popust od 10 000 rubalja pri upisu na bilo koji tečaj Skillbox koristeći promotivni kod "Habr".

Započnimo

Instaliranje biblioteka

Koristit ćemo google-cloud-bigquery za dobivanje podataka iz Google BigQueryja. matplotlib, kvrgav и pande pomoći će vam da vizualizirate svoje podatke. python-telegram-bot poslat će 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 učinili idemo na Google konzola za razvojne programere i izradite novi projekt (ili odaberite postojeći).

Na upravljačkoj ploči odaberite OMOGUĆI APIS I USLUGE i potražite BigQuery API.

Odaberite Omogući za povezivanje API-ja.

Izradite ključ računa

Idemo ponovno Google konzola za razvojne programere, odaberite karticu Vjerodajnice, Stvori vjerodajnice i Ključ računa usluge.

Zatim - Novi račun usluge i unesite ime u polje Naziv računa usluge.

S padajućeg popisa Uloga odaberite Projekt > Vlasnik, a zatim Stvori.

Datoteka koja će se automatski preuzeti zove se creds.json.

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

izvoz GOOGLE_APPLICATION_CREDENTIALS='[PATH_TO_CREDS.JSON]'

Ako je sve prošlo kako treba, vrijeme je za početak pisanja programa.

Izrada aplikacije

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

Prilično je jednostavno.

Upitajte tablicu -> Vizualizirajte podatke -> Spremite vizualizaciju -> Pošaljite sliku

Kreirajmo jednu funkciju za definiranje svake niti.

Upit u BigQuery

Prvo uvozimo knjižnicu.

iz google.cloud import bigquery

Stvaramo funkciju koja se zove query_to_bigquery, gdje je parametar upit.

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 podatkovni okvir.

Vizualizacija podataka

Za rješavanje ovog problema odaberite matplotlib.

uvesti matplotlib.pyplot kao plt

Potrebno nam je pet parametara, gdje je x podatak osi x, oznaka osi x naslov osi, oznaka osi y, oznaka osi y naslov osi i naziv 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

Spremite sliku

Sada upotrijebimo dvije funkcije za izradu vizualizacije i njezino spremanje.

Poslat ćemo vam broj objavljenih objava dnevno. Prvo napiš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 dva tjedna počevši od 2. prosinca 2018.

Koristimo ovaj datum jer je 2018-12-02 najnoviji podatak zabilježen 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.

dataframe = query_to_bigquery(upit)

Zatim koristimo stupac podataka o datumu za x-os i stupac total_posts za y-os.

x = dataframe['date'].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='Ukupni postovi', 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

Kako biste poslali izvješće primatelju, morate znati parametar chat_id.

Koristimo userinfobot i upišite /start. Bot odgovara potrebnim informacijama, chat_id se nalazi u polju id.

Kreirajmo sada funkciju send_image. Koristit će funkciju get_and_save_image za dohvaćanje i spremanje slike. I onda sve šaljemo 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 kraju, stvaramo još jednu funkciju, glavnu, za pokretanje aplikacije. Ne zaboravite promijeniti YOUR_TOKEN za bota.

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

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 toga, naša će aplikacija 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()

Spremite datoteku i nazovite je main.py.

Aplikaciju pokrećemo unosom naredbe u terminal:

python3 main.py

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

Provjerimo bota stogaodabirom naredbe /send.

Gotov kod možete dobiti na moj GitHub.

Skillbox preporučuje:

• Dvogodišnji praktični tečaj "Ja sam PRO web programer".
• Mrežni tečaj "C# programer od 0".
• Praktični godišnji tečaj "PHP programer od 0 do PRO".

Izvor: www.habr.com