Geekly articles weekly

Como delegar relatórios simples ao robô. Escrevendo um bot no Python e no Google BigQuery



Você tem tarefas repetidas dia após dia, de semana para semana? Por exemplo, escrevendo relatórios. Você solicita dados, realiza análises, visualiza (cria gráficos, tabelas) e os envia ao chefe. Mas e se tudo isso for automatizado?

Neste tutorial, criaremos um bot para o Telegram que ajudará a automatizar os relatórios. E o mais legal é que todo o programa consistirá em apenas 50 linhas de código! Se você estiver criando um bot para o Telegram pela primeira vez, leia este post aqui .

A Skillbox recomenda: Um desenvolvedor prático de Python do zero .

Lembramos que: para todos os leitores de "Habr" - um desconto de 10.000 rublos ao se inscrever em qualquer curso Skillbox usando o código promocional "Habr".

Descer


Instalar bibliotecas

Usaremos o google-cloud-bigquery para recuperar dados do Google BigQuery. matplotlib , numpy e pandas ajudarão a visualizar os dados. O python-telegram-bot enviará os dados finalizados para o Telegram.

O pip3 instala o google-cloud-bigquery matplotlib numpy pandas python-telegram-bot

Conecte a API do Google BigQuery

Se quisermos usar o serviço, precisamos conectar a API do Google BigQuery. Para fazer isso, acesse o Google Developers Console e crie um novo projeto (ou selecione um existente).

No painel de controle, selecione ATIVAR APIS E SERVIÇOS e procure a API do BigQuery.



Selecione Ativar para conectar a API.



Crie uma chave de conta

Novamente, vamos ao Google Developers Console , selecione a guia Credenciais, Criar credenciais e Chave da conta de serviço.

Então - Nova conta de serviço e, no campo Nome da conta de serviço, digite o nome.

Na lista suspensa Função, selecione Projeto> Proprietário e, em seguida, Criar.



Um arquivo que será carregado automaticamente é chamado creds.json.

Nós expomos o GOOGLE_APPLICATION_CREDENTIALS, indicando o caminho para creds.json no terminal.

exportar GOOGLE_APPLICATION_CREDENTIALS = '[PATH_TO_CREDS.JSON]'

Se tudo correu bem, é hora de começar a escrever um programa.

Criação de aplicativo


Para o tutorial, usaremos os dados de bigquery-public-data.stackoverflow; em nosso relatório, escolheremos o número de publicações diárias.

Tudo é bem simples.

Consulte a tabela -> Visualize os dados -> Salve a visualização -> Envie a imagem

Vamos criar uma função para definir cada thread.

Consulta ao BigQuery

Primeiro, importe a biblioteca.

do google.cloud import bigquery

Crie uma função chamada query_to_bigquery, em que o parâmetro é 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 função retornará a solicitação como um quadro de dados.

Visualize os dados

Para resolver esse problema, selecione matplotlib.

importar matplotlib.pyplot como plt

Precisamos de cinco parâmetros, onde x são os dados no eixo x, x_label é o nome do eixo, y são os dados do eixo y, y_label é o nome do eixo e title é o título de toda a visualização.

 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

Salvar imagem

Agora vamos usar duas funções para criar uma visualização e salvá-la.

Enviaremos o número de postagens diárias publicadas. Primeiro, escrevemos uma solicitação.

 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 """

A consulta ajuda a coletar dados por duas semanas a partir de 2 de dezembro de 2018.

Usamos essa data porque 02/12/2018 é os dados mais recentes registrados em bigquery-public-data.stackoverflow.post_history; em outros casos, você pode usar CURRENT_DATE () para obter os dados mais recentes.

Chame a função query_to_bigquery para obter os dados.

dataframe = query_to_bigquery (query)

Em seguida, use a coluna de data para o eixo xe a coluna total_posts para o eixo y.

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

Nós visualizamos usando a função visualize_bar_chart e a salvamos como uma imagem.

plt = visualizar_bar_chart (x = x, x_label = 'Data', y = y, y_label = 'Total de postagens', title = 'Postagens diárias')
plt.savefig ('viz.png')

Envolvemos esse código em uma função chamada 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 imagem

Para enviar um relatório ao destinatário, você precisa conhecer o parâmetro chat_id.

Usamos userinfobot e digite / start. O bot responde com as informações necessárias, chat_id está contido no campo id.

Agora crie a função send_image. Ele usará a função get_and_save_image para recuperar e salvar a imagem. E então enviamos tudo para o contato correto.

 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

Por fim, crie outra função, principal, para iniciar o aplicativo. Não se esqueça de alterar YOUR_TOKEN pelo bot.

Lembre-se: este programa enviará a imagem automaticamente na hora especificada. Por exemplo, enviaremos um relatório às nove da manhã todos os dias.

 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, nosso aplicativo ficará assim:

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

Salve o arquivo e nomeie-o como main.py.

Iniciamos o aplicativo digitando o comando no terminal:

python3 main.py

Está tudo pronto. Agora, temos um robô composto por 50 linhas de código que gera relatórios sem nossa intervenção.

Vamos verificar o bot daqui selecionando o comando / send.



Você pode obter o código finalizado no meu GitHub .

A Skillbox recomenda:

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

More articles:


All Articles