Hai compiti che si ripetono giorno dopo giorno, settimana dopo settimana? Ad esempio, scrivere rapporti. Richiedi dati, li analizzi, li visualizzi (crea grafici, diagrammi) e poi li invii al tuo capo. Ma cosa succederebbe se tutto questo fosse automatizzato?
In questo tutorial creeremo un bot per Telegram che aiuterà ad automatizzare il reporting. E la cosa più bella è che l'intero programma sarà composto da sole 50 righe di codice! Se stai creando un bot per Telegram per la prima volta, dovresti leggere anche questo .
Skillbox consiglia: Corso pratico .
Ti ricordiamo: per tutti i lettori di "Habr" - uno sconto di 10 rubli al momento dell'iscrizione a qualsiasi corso Skillbox utilizzando il codice promozionale "Habr".
Scendendo
Installazione delle librerie
Noi useremo per ottenere dati da Google BigQuery. , и ti aiuterà a visualizzare i tuoi dati. invierà i dati finiti a Telegram.
pip3 installa google-cloud-bigquery matplotlib numpy pandas python-telegram-bot
Connessione dell'API Google BigQuery
Se vogliamo utilizzare il servizio, dobbiamo connettere l'API Google BigQuery. Per fare questo andiamo a e creare un nuovo progetto (o selezionarne uno esistente).
Nel pannello di controllo, seleziona ABILITA API E SERVIZI e cerca API BigQuery.
Seleziona Abilita per connettere l'API.
Crea una chiave account
Andiamo di nuovo , seleziona la scheda Credenziali, Crea credenziali e Chiave dell'account di servizio.
Quindi: Nuovo account di servizio e inserisci il nome nel campo Nome account di servizio.
Dall'elenco a discesa Ruolo, seleziona Progetto > Proprietario, quindi Crea.
Il file che verrà scaricato automaticamente si chiama creds.json.
Imposta GOOGLE_APPLICATION_CREDENTIALS, specificando il percorso di creds.json nel terminale.
esporta GOOGLE_APPLICATION_CREDENTIALS='[PATH_TO_CREDS.JSON]'
Se tutto è andato bene, è ora di iniziare a scrivere il programma.
Creare un'applicazione
Per il tutorial utilizzeremo i dati di bigquery-public-data.stackoverflow, per il nostro report selezioneremo il numero di pubblicazioni giornaliere.
Tutto è abbastanza semplice.
Interroga la tabella -> Visualizza i dati -> Salva la visualizzazione -> Invia l'immagine
Creiamo una funzione per definire ciascun thread.
Effettua una query su BigQuery
Per prima cosa importiamo la libreria.
da google.cloud importa bigquery
Creiamo una funzione chiamata query_to_bigquery, dove il parametro è query.
def query_to_bigquery(query):
client = bigquery.Client()
query_job = client.query(query)
result = query_job.result()
dataframe = result.to_dataframe()
return dataframeQuesta funzione restituirà la richiesta come frame di dati.
Visualizzazione dei dati
Per risolvere questo problema, scegli matplotlib.
import matplotlib.pyplot come plt
Sono necessari cinque parametri, dove x sono i dati dell'asse x, x_label è il titolo dell'asse, y sono i dati dell'asse y, y_label è il titolo dell'asse e title è il titolo dell'intera visualizzazione.
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 pltSalva l'immagine
Ora utilizziamo due funzioni per creare una visualizzazione e salvarla.
Ti invieremo il numero di post pubblicati giornalmente. Per prima cosa scriviamo una richiesta.
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
"""La query consente di raccogliere dati per due settimane a partire dal 2 dicembre 2018.
Usiamo questa data perché 2018-12-02 è l'ultimo dato registrato in bigquery-public-data.stackoverflow.post_history, in altri casi puoi utilizzare CURRENT_DATE() per ottenere i dati più recenti.
Chiama la funzione query_to_bigquery per ottenere i dati.
dataframe = query_to_bigquery(query)
Quindi utilizziamo la colonna dei dati della data per l'asse x e la colonna total_posts per l'asse y.
x = dataframe['data'].tolist()
y = dataframe['total_posts'].tolist()
Lo visualizziamo utilizzando la funzione visualize_bar_chart e lo salviamo come immagine.
plt = visualize_bar_chart(x=x, x_label='Data', y=y, y_label='Messaggi totali', titolo='Messaggi giornalieri')
plt.savefig('viz.png')
Inseriamo questo codice in una funzione chiamata 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')Invia un'immagine
Per inviare un report al destinatario, è necessario conoscere il parametro chat_id.
Noi usiamo e digitare /inizio. Il bot risponde con le informazioni necessarie, chat_id è contenuto nel campo id.
Ora creiamo la funzione send_image. Utilizzerà la funzione get_and_save_image per recuperare e salvare l'immagine. E poi inviamo tutto al contatto corretto.
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'))Programma principale
Infine, creiamo un'altra funzione, main, per avviare l'applicazione. Non dimenticare di modificare YOUR_TOKEN per il bot.
Ricorda: questo programma invierà l'immagine automaticamente all'ora specificata. Ad esempio, invieremo un rapporto ogni giorno alle nove del mattino.
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()Di conseguenza, la nostra applicazione sarà simile a questa:
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()Salva il file e chiamalo main.py.
Lanciamo l'applicazione inserendo il comando nel terminale:
python3 principale.py
Tutto è pronto. Ora abbiamo un robot composto da 50 righe di codice che genera report senza il nostro intervento.
Controlliamo il bot selezionando il comando /send.
Puoi ottenere il codice finito su .
Skillbox consiglia:
- Corso pratico di due anni .
- Corso online .
- Corso pratico annuale .
Fonte: habr.com