Jak delegovat jednoduché reporty na robota. Psaní robota v Pythonu a Google BigQuery

Máte úkoly, které se opakují den po dni, týden po týdnu? Například psaní zpráv. Vyžádáte si data, analyzujete je, vizualizujete (vytvářejte grafy, tabulky) a poté je pošlete svému šéfovi. Ale co když to všechno bylo automatizované?

V tomto tutoriálu vytvoříme robota pro Telegram, který pomůže automatizovat hlášení. A nejlepší na tom je, že celý program se bude skládat pouze z 50 řádků kódu! Pokud vytváříte robota pro Telegram poprvé, měli byste si také přečíst tento zveřejnit.

Skillbox doporučuje: Praktický kurz Vývojář Pythonu od nuly.

Připomínáme: pro všechny čtenáře "Habr" - sleva 10 000 rublů při zápisu do jakéhokoli kurzu Skillbox pomocí propagačního kódu "Habr".

Dostat se dolů

Instalace knihoven

Budeme používat google-cloud-bigquery získat data z Google BigQuery. matplotlib, otupělý и pandy vám pomůže vizualizovat vaše data. python-telegram-bot odešle hotová data do telegramu.

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

Připojování Google BigQuery API

Pokud chceme službu využívat, potřebujeme připojit Google BigQuery API. K tomu jdeme do Google Developers Console a vytvořte nový projekt (nebo vyberte existující).

Na ovládacím panelu vyberte ENABLE APIS AND SERVICES a vyhledejte BigQuery API.

Chcete-li rozhraní API připojit, vyberte možnost Povolit.

Vytvořte klíč k účtu

Pojďme znovu Google Developers Console, vyberte kartu Pověření, Vytvořit přihlašovací údaje a Klíč servisního účtu.

Poté - Nový účet služby a zadejte název do pole Název účtu služby.

Z rozevíracího seznamu Role vyberte Projekt > Vlastník a poté Vytvořit.

Soubor, který bude automaticky stažen, se nazývá creds.json.

Nastavte GOOGLE_APPLICATION_CREDENTIALS a zadejte cestu k creds.json v terminálu.

exportovat GOOGLE_APPLICATION_CREDENTIALS='[PATH_TO_CREDS.JSON]'

Pokud vše proběhlo v pořádku, je čas začít psát program.

Vytvořte aplikaci

Pro tutoriál použijeme data z bigquery-public-data.stackoverflow, pro náš report vybereme počet denních publikací.

Všechno je velmi jednoduché.

Dotaz na tabulku -> Vizualizovat data -> Uložit vizualizaci -> Odeslat obrázek

Vytvořme jednu funkci pro definování každého vlákna.

Dotaz do BigQuery

Nejprve importujeme knihovnu.

z google.cloud importovat bigquery

Vytvoříme funkci nazvanou query_to_bigquery, kde parametrem je dotaz.

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

Tato funkce vrátí požadavek jako datový rámec.

Vizualizace dat

Chcete-li tento problém vyřešit, vyberte matplotlib.

importovat matplotlib.pyplot jako plt

Potřebujeme pět parametrů, kde x jsou data na ose x, x_label je název pro osu, y je data na ose y, y_label je název pro osu a title je název celé vizualizace.

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

Uložte obrázek

Nyní pomocí dvou funkcí vytvoříme vizualizaci a uložíme ji.

Počet zveřejněných příspěvků budeme zasílat denně. Nejprve napíšeme žádost.

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

Dotaz pomáhá shromažďovat data po dobu dvou týdnů od 2. prosince 2018.

Toto datum používáme, protože 2018-12-02 jsou nejnovější data zaznamenaná v bigquery-public-data.stackoverflow.post_history, v ostatních případech můžete k získání nejnovějších dat použít CURRENT_DATE().

Chcete-li získat data, zavolejte funkci query_to_bigquery.

datový rámec = query_to_bigquery(dotaz)

Pak použijeme datový sloupec pro osu x a sloupec total_posts pro osu y.

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

Ten vizualizujeme pomocí funkce visualize_bar_chart a uložíme jako obrázek.

plt = visualize_bar_chart(x=x, x_label='Datum', y=y, y_label='Celkový počet příspěvků', title='Denní příspěvky')
plt.savefig('viz.png')

Tento kód zabalíme do funkce nazvané 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šlete obrázek

Abyste mohli odeslat zprávu příjemci, potřebujete znát parametr chat_id.

Používáme userinfobot a zadejte /start. Bot odpoví potřebnými informacemi, chat_id je obsaženo v poli id.

Nyní vytvoříme funkci send_image. K načtení a uložení obrázku použije funkci get_and_save_image. A pak vše zašleme správnému kontaktu.

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

Hlavní program

Nakonec vytvoříme další funkci, hlavní, pro spuštění aplikace. Nezapomeňte změnit YOUR_TOKEN pro robota.

Pamatujte: tento program automaticky odešle obrázek v čase, který určíte. Například každý den v devět ráno pošleme hlášení.

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

V důsledku toho bude naše aplikace vypadat takto:

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

Uložte soubor a nazvěte jej main.py.

Aplikaci spustíme zadáním příkazu do terminálu:

python3 main.py

Vše je připraveno. Nyní máme robota skládajícího se z 50 řádků kódu, který generuje zprávy bez našeho zásahu.

Pojďme zkontrolovat robota protovýběrem příkazu /send.

Hotový kód můžete získat na můj GitHub.

Skillbox doporučuje:

• Dvouletý praktický kurz "Jsem PRO webový vývojář".
• Online kurz "C# vývojář od 0".
• Praktický roční kurz "PHP vývojář od 0 do PRO".

Zdroj: www.habr.com