Ai sarcini care se repetă zi de zi, săptămână după săptămână? De exemplu, redactarea rapoartelor. Soliciți date, le analizezi, le vizualizezi (realizezi grafice, diagrame) și apoi le trimiți șefului tău. Dar dacă toate acestea ar fi automatizate?
În acest tutorial vom crea un bot pentru Telegram care va ajuta la automatizarea raportării. Și cel mai tare lucru este că întregul program va fi format din doar 50 de linii de cod! Dacă creați un bot pentru Telegram pentru prima dată, atunci ar trebui să îl citiți și pe acesta .
Skillbox recomandă: Curs practic .
Amintim: pentru toți cititorii „Habr” - o reducere de 10 de ruble la înscrierea la orice curs Skillbox folosind codul promoțional „Habr”.
Să începem
Instalarea bibliotecilor
Noi vom folosi pentru a obține date de la Google BigQuery. , и vă va ajuta să vă vizualizați datele. va trimite datele finalizate către Telegram.
pip3 instalează google-cloud-bigquery matplotlib numpy panda python-telegram-bot
Se conectează API-ul Google BigQuery
Dacă dorim să folosim serviciul, trebuie să conectăm API-ul Google BigQuery. Pentru a face asta mergem la și creați un proiect nou (sau selectați unul existent).
În panoul de control, selectați ACTIVARE API ȘI SERVICII și căutați API-ul BigQuery.
Selectați Activare pentru a conecta API-ul.
Creați o cheie de cont
Să mergem din nou la , selectați fila Acreditări, Creare acreditări și Cheia contului de serviciu.
Apoi - Cont de serviciu nou și introduceți numele în câmpul Nume cont de serviciu.
Din lista derulantă Rol, selectați Proiect > Proprietar, apoi Creare.
Fișierul care va fi descărcat automat se numește creds.json.
Setați GOOGLE_APPLICATION_CREDENTIALS, specificând calea către creds.json în terminal.
exportați GOOGLE_APPLICATION_CREDENTIALS='[PATH_TO_CREDS.JSON]'
Dacă totul a mers bine, este timpul să începeți să scrieți programul.
Creați o aplicație
Pentru tutorial vom folosi date din bigquery-public-data.stackoverflow, pentru raportul nostru vom selecta numărul de publicații zilnice.
Totul este destul de simplu.
Interogați tabelul -> Vizualizați datele -> Salvați vizualizarea -> Trimiteți imaginea
Să creăm o funcție pentru a defini fiecare fir.
Interogați la BigQuery
Mai întâi importăm biblioteca.
din google.cloud import bigquery
Creăm o funcție numită query_to_bigquery, unde parametrul este query.
def query_to_bigquery(query):
client = bigquery.Client()
query_job = client.query(query)
result = query_job.result()
dataframe = result.to_dataframe()
return dataframeAceastă funcție va returna cererea ca un cadru de date.
Vizualizarea datelor
Pentru a rezolva această problemă, alegeți matplotlib.
import matplotlib.pyplot ca plt
Avem nevoie de cinci parametri, unde x este datele axei x, x_label este titlul axei, y este datele axei y, y_label este titlul axei și titlul este titlul întregii vizualizări.
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ți imaginea
Acum să folosim două funcții pentru a crea o vizualizare și a o salva.
Vom trimite numărul de postări publicate zilnic. Mai întâi scriem o cerere.
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
"""Interogarea ajută la colectarea datelor timp de două săptămâni începând cu 2 decembrie 2018.
Folosim această dată deoarece 2018-12-02 sunt cele mai recente date înregistrate în bigquery-public-data.stackoverflow.post_history, în alte cazuri puteți folosi CURRENT_DATE() pentru a obține cele mai noi date.
Apelați funcția query_to_bigquery pentru a obține datele.
cadru de date = query_to_bigquery(interogare)
Apoi folosim coloana de date date pentru axa x și coloana total_posts pentru axa y.
x = dataframe['data'].tolist()
y = dataframe['total_posts'].tolist()
Îl vizualizăm folosind funcția visualize_bar_chart și îl salvăm ca imagine.
plt = visualize_bar_chart(x=x, x_label='Data', y=y, y_label='Total Postări', title='Daily Posts')
plt.savefig('viz.png')
Încapsulăm acest cod într-o funcție numită 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')Trimite o imagine
Pentru a trimite un raport destinatarului, trebuie să cunoașteți parametrul chat_id.
Folosim și tastați /start. Botul răspunde cu informațiile necesare, chat_id este conținut în câmpul id.
Acum să creăm funcția send_image. Va folosi funcția get_and_save_image pentru a prelua și salva imaginea. Și apoi trimitem totul contactului corect.
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'))Programul principal
În final, creăm o altă funcție, principală, pentru a lansa aplicația. Nu uitați să schimbați YOUR_TOKEN pentru bot.
Amintiți-vă: acest program va trimite imaginea automat la ora specificată. De exemplu, vom trimite un raport la nouă dimineața în fiecare zi.
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()Ca rezultat, aplicația noastră va arăta astfel:
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ți fișierul și numiți-l main.py.
Lansăm aplicația introducând comanda în terminal:
python3 main.py
Totul este gata. Acum avem un robot format din 50 de linii de cod care generează rapoarte fără intervenția noastră.
Să verificăm botul prin selectarea comenzii /send.
Puteți obține codul final la .
Skillbox recomandă:
- Curs practic de doi ani .
- Curs online .
- Curs practic anual .
Sursa: www.habr.com