Ako delegovať jednoduché správy na robota. Písanie robota v Pythone a Google BigQuery

Máte úlohy, ktoré sa opakujú deň čo deň, týždeň čo týždeň? Napríklad písanie správ. Vyžiadate si údaje, analyzujete ich, vizualizujete (vytvoríte grafy, tabuľky) a potom ich pošlete svojmu šéfovi. Ale čo ak je toto všetko automatizované?

V tomto návode vytvoríme robota pre telegram, ktorý pomôže automatizovať hlásenie. A najlepšie na tom je, že celý program bude pozostávať iba z 50 riadkov kódu! Ak vytvárate robota pre telegram prvýkrát, mali by ste si prečítať aj tento pošta.

Skillbox odporúča: Praktický kurz Vývojár Pythonu od nuly.

Pripomíname vám: pre všetkých čitateľov „Habr“ - zľava 10 000 rubľov pri registrácii do akéhokoľvek kurzu Skillbox pomocou propagačného kódu „Habr“.

Začnime

Inštalácia knižníc

Použijeme google-cloud-bigquery na získanie údajov zo služby Google BigQuery. matplotlib, otupený и pandy vám pomôže vizualizovať vaše údaje. python-telegram-bot odošle hotové údaje do telegramu.

pip3 nainštalovať google-cloud-bigquery matplotlib numpy pandas python-telegram-bot

Pripája sa Google BigQuery API

Ak chceme službu využívať, potrebujeme pripojiť Google BigQuery API. Aby sme to urobili, ideme do Google Developers Console a vytvorte nový projekt (alebo vyberte existujúci).

Na ovládacom paneli vyberte POVOLIŤ API A SLUŽBY a vyhľadajte BigQuery API.

Ak chcete pripojiť API, vyberte možnosť Povoliť.

Vytvorte kľúč účtu

Poďme znova Google Developers Console, vyberte kartu Poverenia, Vytvoriť poverenia a Kľúč servisného účtu.

Potom - Nové konto služby a zadajte názov do poľa Názov konta služby.

Z rozbaľovacieho zoznamu Rola vyberte Projekt > Vlastník a potom Vytvoriť.

Súbor, ktorý sa automaticky stiahne, sa volá creds.json.

Nastavte GOOGLE_APPLICATION_CREDENTIALS a zadajte cestu k súboru creds.json v termináli.

exportovať GOOGLE_APPLICATION_CREDENTIALS='[PATH_TO_CREDS.JSON]'

Ak všetko prebehlo dobre, je čas začať písať program.

Vytvorenie aplikácie

Pre tutoriál použijeme údaje z bigquery-public-data.stackoverflow, pre náš prehľad vyberieme počet denných publikácií.

Všetko je celkom jednoduché.

Dotaz na tabuľku -> Vizualizácia údajov -> Uložiť vizualizáciu -> Odoslať obrázok

Vytvorme jednu funkciu na definovanie každého vlákna.

Dopyt do BigQuery

Najprv importujeme knižnicu.

z google.cloud importujte bigquery

Vytvoríme funkciu s názvom query_to_bigquery, kde parametrom 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

Táto funkcia vráti požiadavku ako dátový rámec.

Vizualizácia údajov

Ak chcete tento problém vyriešiť, vyberte matplotlib.

import matplotlib.pyplot as plt

Potrebujeme päť parametrov, kde x sú údaje osi x, x_label je názov osi, y je údaj osi y, y_label je názov osi a title je názov celej vizualizácie.

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ázok

Teraz pomocou dvoch funkcií vytvoríme vizualizáciu a uložíme ju.

Počet uverejnených príspevkov budeme posielať denne. Najprv napíšeme žiadosť.

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áha zhromažďovať údaje počas dvoch týždňov od 2. decembra 2018.

Tento dátum používame, pretože 2018-12-02 sú najnovšie údaje zaznamenané v bigquery-public-data.stackoverflow.post_history, v ostatných prípadoch môžete na získanie najnovších údajov použiť CURRENT_DATE().

Ak chcete získať údaje, zavolajte funkciu query_to_bigquery.

dataframe = query_to_bigquery(query)

Potom použijeme stĺpec s údajmi o dátume pre os x a stĺpec total_posts pre os y.

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

Vizualizujeme ho pomocou funkcie visualize_bar_chart a uložíme ako obrázok.

plt = visualize_bar_chart(x=x, x_label='Dátum', y=y, y_label='Celkový počet príspevkov', title='Denné príspevky')
plt.savefig('viz.png')

Tento kód zabalíme do funkcie s názvom 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šlite obrázok

Aby ste mohli odoslať správu príjemcovi, potrebujete poznať parameter chat_id.

Používame userinfobot a zadajte /start. Robot odpovie potrebnými informáciami, chat_id sa nachádza v poli id.

Teraz vytvoríme funkciu send_image. Na získanie a uloženie obrázka použije funkciu get_and_save_image. A potom všetko pošleme na správny 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'))

Hlavný program

Nakoniec vytvoríme ďalšiu funkciu, hlavnú, na spustenie aplikácie. Nezabudnite zmeniť YOUR_TOKEN pre robota.

Pamätajte: tento program odošle obrázok automaticky v čase, ktorý určíte. Napríklad každý deň o deviatej ráno pošleme hlásenie.

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ša aplikácia vyzerať 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 súbor a nazvite ho main.py.

Aplikáciu spustíme zadaním príkazu do terminálu:

python3 main.py

Všetko je pripravené. Teraz máme robota pozostávajúceho z 50 riadkov kódu, ktorý generuje správy bez nášho zásahu.

Skontrolujeme robota pretovýberom príkazu /send.

Hotový kód môžete získať na môj GitHub.

Skillbox odporúča:

• Dvojročný praktický kurz "Som PRO web developer".
• Online kurz "C# vývojár od 0".
• Praktický ročníkový kurz "PHP vývojár od 0 po PRO".

Zdroj: hab.com