Ar turite užduočių, kurios kartojasi diena iš dienos, savaitė po savaitės? Pavyzdžiui, ataskaitų rašymas. Jūs paprašote duomenų, juos analizuojate, vizualizuojate (sudarote grafikus, diagramas) ir siunčiate juos savo viršininkui. Bet kas, jei visa tai būtų automatizuota?
Šioje pamokoje mes sukursime „Telegram“ robotą, kuris padės automatizuoti ataskaitų teikimą. Ir šauniausia, kad visą programą sudarys tik 50 kodo eilučių! Jei „Telegram“ robotą kuriate pirmą kartą, turėtumėte perskaityti ir šį .
„Skillbox“ rekomenduoja: Praktinis kursas .
Primename: visiems „Habr“ skaitytojams – 10 000 rublių nuolaida užsiregistravus į bet kurį „Skillbox“ kursą naudojant „Habr“ reklamos kodą.
Pradėkime
Bibliotekų diegimas
Mes naudosime gauti duomenis iš „Google BigQuery“. , и padės vizualizuoti duomenis. išsiųs baigtus duomenis į „Telegram“.
pip3 įdiegti google-cloud-bigquery matplotlib numpy pandas python-telegram-bot
„Google BigQuery“ API prijungimas
Jei norime naudotis paslauga, turime prisijungti prie Google BigQuery API. Norėdami tai padaryti, einame į ir sukurti naują projektą (arba pasirinkti esamą).
Valdymo skydelyje pasirinkite ĮJUNGTI API IR PASLAUGAS ir ieškokite BigQuery API.
Pasirinkite Įgalinti, kad prijungtumėte API.
Sukurkite paskyros raktą
Eikime dar kartą , pasirinkite skirtuką Kredencialai, Sukurti kredencialus ir paslaugos paskyros raktą.
Tada - Nauja paslaugos paskyra ir įveskite pavadinimą lauke Paslaugos paskyros pavadinimas.
Išskleidžiamajame sąraše Vaidmuo pasirinkite Projektas > Savininkas, tada Kurti.
Failas, kuris bus automatiškai atsisiunčiamas, vadinamas creds.json.
Nustatykite GOOGLE_APPLICATION_CREDENTIALS, terminale nurodydami kelią į creds.json.
eksportuoti GOOGLE_APPLICATION_CREDENTIALS='[PATH_TO_CREDS.JSON]'
Jei viskas klostėsi gerai, laikas pradėti rašyti programą.
Aplikacijos kūrimas
Mokomojoje programoje naudosime duomenis iš bigquery-public-data.stackoverflow, o ataskaitoje pasirinksime kasdienių publikacijų skaičių.
Viskas gana paprasta.
Pateikite lentelės užklausą -> Vizualizuokite duomenis -> Išsaugokite vizualizaciją -> Siųsti vaizdą
Sukurkime vieną funkciją kiekvienai gijai apibrėžti.
Užklausa „BigQuery“.
Pirmiausia importuojame biblioteką.
iš google.cloud import bigquery
Sukuriame funkciją query_to_bigquery, kur parametras yra užklausa.
def query_to_bigquery(query):
client = bigquery.Client()
query_job = client.query(query)
result = query_job.result()
dataframe = result.to_dataframe()
return dataframeŠi funkcija grąžins užklausą kaip duomenų rėmelį.
Duomenų vizualizavimas
Norėdami išspręsti šią problemą, pasirinkite matplotlib.
importuoti matplotlib.pyplot kaip plt
Mums reikia penkių parametrų, kur x yra x ašies duomenys, x_label yra ašies pavadinimas, y yra y ašies duomenys, y_label yra ašies pavadinimas, o pavadinimas yra visos vizualizacijos pavadinimas.
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 pltIšsaugokite vaizdą
Dabar sukurkime vizualizaciją ir išsaugokime ją naudodami dvi funkcijas.
Kiekvieną dieną siųsime paskelbtų pranešimų skaičių. Pirmiausia rašome prašymą.
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
"""Užklausa padeda rinkti dviejų savaičių duomenis nuo 2 m. gruodžio 2018 d.
Naudojame šią datą, nes 2018-12-02 yra naujausi duomenys, įrašyti bigquery-public-data.stackoverflow.post_history, kitais atvejais galite naudoti CURRENT_DATE(), kad gautumėte naujausius duomenis.
Norėdami gauti duomenis, iškvieskite funkciją query_to_bigquery.
duomenų rėmelis = query_to_bigquery(query)
Tada x ašiai naudojame datos duomenų stulpelį, o y ašiai – total_posts.
x = duomenų rėmelis['data'].tolist()
y = duomenų rėmelis['total_posts'].tolist()
Ją vizualizuojame naudodami funkciją visualize_bar_chart ir išsaugome kaip vaizdą.
plt = visualize_bar_chart(x=x, x_label='Data', y=y, y_label='Iš viso įrašų', title='Dienos įrašai')
plt.savefig('viz.png')
Šį kodą suvyniojame į funkciją, vadinamą 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')Siųsti paveikslėlį
Norėdami išsiųsti ataskaitą gavėjui, turite žinoti parametrą chat_id.
Mes naudojame ir įveskite /start. Botas atsako pateikdamas reikiamą informaciją, id lauke yra chat_id.
Dabar sukurkime funkciją send_image. Vaizdui nuskaityti ir išsaugoti bus naudojama funkcija get_and_save_image. Ir tada mes siunčiame viską teisingam kontaktui.
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'))Pagrindinė programa
Galiausiai sukuriame kitą funkciją, pagrindinę, kad paleistume programą. Nepamirškite pakeisti roboto YOUR_TOKEN.
Atminkite: ši programa automatiškai išsiųs vaizdą jūsų nurodytu laiku. Pavyzdžiui, kiekvieną dieną devintą ryto išsiųsime ataskaitą.
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()Dėl to mūsų programa atrodys taip:
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()Išsaugokite failą ir pavadinkite jį main.py.
Programą paleidžiame įvesdami komandą terminale:
python3 main.py
Viskas paruošta. Dabar turime robotą, susidedantį iš 50 kodo eilučių, kuris generuoja ataskaitas be mūsų įsikišimo.
Patikrinkime botą pasirinkdami komandą /send.
Baigtą kodą galite gauti adresu .
„Skillbox“ rekomenduoja:
- Dvejų metų praktinis kursas .
- Internetinis kursas .
- Praktinių metų kursas .
Šaltinis: www.habr.com