Geekly articles weekly

Cómo delegar informes simples al robot. Escribir un bot en Python y Google BigQuery



¿Tiene tareas que se repiten día tras día, semana a semana? Por ejemplo, escribir informes. Solicita datos, realiza análisis, visualiza (hace gráficos, cuadros) y luego los envía al jefe. Pero, ¿y si todo esto está automatizado?

En este tutorial, crearemos un bot para Telegram que ayudará a automatizar los informes. ¡Y lo mejor es que todo el programa constará de solo 50 líneas de código! Si está creando un bot para Telegram por primera vez, entonces debería leer esta publicación aquí .

Skillbox recomienda: Un desarrollador de Python de curso práctico desde cero .

Le recordamos: para todos los lectores de "Habr": un descuento de 10.000 rublos al registrarse en cualquier curso de Skillbox con el código de promoción "Habr".

Bajando


Instalar bibliotecas

Utilizaremos google-cloud-bigquery para recuperar datos de Google BigQuery. matplotlib , numpy y pandas ayudarán a visualizar los datos. python-telegram-bot enviará los datos terminados a Telegram.

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

Conecte la API de Google BigQuery

Si queremos usar el servicio, necesitamos conectar la API de Google BigQuery. Para hacer esto, vaya a la Consola de desarrolladores de Google y cree un nuevo proyecto (o seleccione uno existente).

En el panel de control, seleccione HABILITAR APIS Y SERVICIOS y busque la API BigQuery.



Seleccione Habilitar para conectar la API.



Crear una clave de cuenta

Volvemos a la Consola de desarrolladores de Google , seleccionamos las pestañas Credenciales, Crear credenciales y Clave de cuenta de servicio.

Luego - Nueva cuenta de servicio, y en el campo Nombre de la cuenta de servicio, ingrese el nombre.

En la lista desplegable Rol, seleccione Proyecto> Propietario, luego Crear.



Un archivo que se cargará automáticamente se llama creds.json.

Exponemos GOOGLE_APPLICATION_CREDENTIALS, indicando la ruta a creds.json en la terminal.

export GOOGLE_APPLICATION_CREDENTIALS = '[PATH_TO_CREDS.JSON]'

Si todo salió bien, es hora de comenzar a escribir un programa.

Creación de aplicaciones


Para el tutorial usaremos los datos de bigquery-public-data.stackoverflow, para nuestro informe elegiremos el número de publicaciones diarias.

Todo es bastante simple.

Consultar la tabla -> Visualizar los datos -> Guardar la visualización -> Enviar la imagen

Creemos una función para definir cada hilo.

Consulta a BigQuery

Primero, importe la biblioteca.

de google.cloud import bigquery

Cree una función llamada query_to_bigquery, donde el parámetro es query.

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

Esta función devolverá la solicitud como un marco de datos.

Visualiza los datos

Para resolver este problema, seleccione matplotlib.

importar matplotlib.pyplot como plt

Necesitamos cinco parámetros, donde x son los datos en el eje x, x_label es el nombre del eje, y es los datos para el eje y, y_label es el nombre del eje y title es el título de toda la visualización.

 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

Guardar imagen

Ahora usemos dos funciones para crear una visualización y guardarla.

Le enviaremos el número de publicaciones diarias publicadas. Primero escribimos una solicitud.

 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 consulta ayuda a recopilar datos durante dos semanas a partir del 2 de diciembre de 2018.

Usamos esta fecha porque 2018-12-02 es la última información registrada en bigquery-public-data.stackoverflow.post_history, en otros casos puede usar CURRENT_DATE () para obtener la información más reciente.

Llame a la función query_to_bigquery para obtener los datos.

dataframe = query_to_bigquery (consulta)

Luego use la columna de fecha para el eje x, y la columna total_posts para el eje y.

x = marco de datos ['fecha']. tolist ()
y = dataframe ['total_posts']. tolist ()

Visualizamos usando la función visualize_bar_chart y la guardamos como una imagen.

plt = visualize_bar_chart (x = x, x_label = 'Date', y = y, y_label = 'Total de publicaciones', title = 'Publicaciones diarias')
plt.savefig ('viz.png')

Envolvemos este código en una función llamada 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')

Enviar imagen

Para enviar un informe al destinatario, debe conocer el parámetro chat_id.

Usamos userinfobot y escribimos / start. El bot responde con la información necesaria, chat_id está contenido en el campo id.

Ahora cree la función send_image. Utilizará la función get_and_save_image para recuperar y guardar la imagen. Y luego enviamos todo al contacto correcto.

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

Programa principal

Finalmente, cree otra función, principal, para iniciar la aplicación. No olvides cambiar YOUR_TOKEN por el bot.

Recuerde: este programa enviará la imagen automáticamente a la hora especificada. Por ejemplo, enviaremos un informe a las nueve de la mañana todos los días.

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

Como resultado, nuestra aplicación se verá así:

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

Guarde el archivo y asígnele el nombre main.py.

Iniciamos la aplicación ingresando el comando en la terminal:

python3 main.py

Todo esta listo. Ahora tenemos un robot que consta de 50 líneas de código que genera informes sin nuestra intervención.

Verifiquemos el bot desde aquí seleccionando el comando / send.



Puede obtener el código terminado en mi GitHub .

Skillbox recomienda:

Source: https://habr.com/ru/post/443702/

More articles:


All Articles