API

[rosinaldopinto1221]

Bom dia a todos,

Sou iniciante neste universo da programação, porém um entusiasta da linguagem python. Bom, tenho visto que agora as informações referentes as arboviroses (Dengue, Zika e Chikungunya) possuem api's para acesso aos dados. Gostaria de saber se a comunidade possuem scripts que possam compartilhar para acesso a esses dados, que em sua maioria utilizam a biblioteca Elasticsearch para conexão as bases de dados. Tenho um que utilizei mas que está apresentando erro de conexão como resposta. Abaixo segue detalhes do código que gostaria de ajuda para melhorá-lo.

Estou usando um script chamado baixaDengue.py abaixo o fonte, porém apresenta um erro de conexão:

=================ESTE É O ERRO ==================
File "/home/suporte/baixaDengue.py", line 101, in job
dados = getDadosAPI(query, proxies)
File "/home/suporte/baixaDengue.py", line 49, in getDadosAPI
resultado = es.search(index=index, body=query, scroll='1m', size=10000)
File "/home/suporte/pysus-env/lib/python3.9/site-packages/elasticsearch/client/utils.py", line 347, in _wrapped
return func(*args, params=params, headers=headers, **kwargs)
File "/home/suporte/pysus-env/lib/python3.9/site-packages/elasticsearch/client/init.py", line 1830, in search
return self.transport.perform_request(
File "/home/suporte/pysus-env/lib/python3.9/site-packages/elasticsearch/transport.py", line 421, in perform_request
_ProductChecker.raise_error(self._verified_elasticsearch)
File "/home/suporte/pysus-env/lib/python3.9/site-packages/elasticsearch/transport.py", line 638, in raise_error
raise UnsupportedProductError(message)
elasticsearch.exceptions.UnsupportedProductError: The client noticed that the server is not Elasticsearch and we do not support this unknown product

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
File "/home/suporte/baixaDengue.py", line 125, in
schedule.run_pending()
File "/home/suporte/pysus-env/lib/python3.9/site-packages/schedule/init.py", line 822, in run_pending
default_scheduler.run_pending()
File "/home/suporte/pysus-env/lib/python3.9/site-packages/schedule/init.py", line 100, in run_pending
self._run_job(job)
File "/home/suporte/pysus-env/lib/python3.9/site-packages/schedule/init.py", line 172, in _run_job
ret = job.run()
File "/home/suporte/pysus-env/lib/python3.9/site-packages/schedule/init.py", line 693, in run
ret = self.job_func()
File "/home/suporte/baixaDengue.py", line 108, in job
time.sleep(10)

======================ESTE É CÓDIGO =======================
#!/usr/bin/env python

from datetime import datetime
from socket import timeout
import pandas as pd
from ast import Global, literal_eval
from dateutil.relativedelta import relativedelta
from elasticsearch import Elasticsearch
from elasticsearch import RequestsHttpConnection
import time
import sqlite3
import schedule

dataExecucao = "Aqui eu coloco o horário que vou executa"
proxies = None

date_format = '%Y-%m-%d'

Configurações para acesso a API

usuario = 'nomeDoUsuario'
senha = 'password'
url = 'https://sinan-es.saude.gov.br/sinan-webservice-dengue-uf/_search?pretty=true'
index = 'sinan-webservice-dengue-uf'

Fim Configurações para acesso a API

class MyConnection(RequestsHttpConnection):
def init(self, *args, **kwargs):
proxies = kwargs.pop('proxies', {})
super(MyConnection, self).init(*args, **kwargs)
self.session.proxies = proxies

def getDadosAPI(query=None, proxies=None):
dados = None
if not query:
query = {
"size": 10000,
"query": {
"match_all": {}
}
}
else:
query = query

es = Elasticsearch(url, connection_class=MyConnection, proxies=proxies, http_auth=(usuario, senha), port=443)
resultado = es.search(index=index, body=query, scroll='1m', size=10000)
sid = resultado['_scroll_id']
scroll_size = len(resultado['hits']['hits'])
dados = [dado['_source'] for dado in resultado['hits']['hits']]

while scroll_size > 0:
    resultado = es.scroll(scroll_id=sid, scroll='1m')
    sid = resultado['_scroll_id']
    scroll_size = len(resultado['hits']['hits'])
    dados += [dado['_source'] for dado in resultado['hits']['hits']]
    print(len(dados))
scroll_size = len(resultado['hits']['hits'])
return dados

def job():
data = datetime.now().strftime('%d-%m-%Y-%H-%M')
date_inicial_str = '2023-01-01'
date_inicial = datetime.strptime(date_inicial_str, date_format)

date_limite_str = datetime.today().strftime(date_format)
date_limite = datetime.strptime(date_limite_str, date_format)

date_final = date_inicial + relativedelta(months=7)
date_final_str = date_final.strftime(date_format)

query = {
    "size": 10000,
    "query": {
        "bool": {
            "filter": [
                {
                    "range": {
                        "_updated_at": {
                            "gte": date_inicial_str,
                            "lt": date_final_str
                        }
                    }
                }
            ]
        }
    }
}

dt_inicio = datetime.now()
notificados = []
print('Procedimento iniciado as {}'.format(dt_inicio))
while date_inicial.date() < date_limite.date():
    # future_date = date_inicial + relativedelta(months=1)
    try:
        query['query']['bool']['filter'][0]['range']['_updated_at']['gte'] = date_inicial.strftime(date_format)
        query['query']['bool']['filter'][0]['range']['_updated_at']['lt'] = date_final.strftime(date_format)
        dados = getDadosAPI(query, proxies)
        print('{} até {} - {}'.format(date_inicial, date_final, len(dados)))
        date_inicial = date_final
        date_final = date_inicial + relativedelta(months=1)
        if len(dados) > 0:
            notificados.append(pd.DataFrame(dados))
    except Exception:
        time.sleep(10)
        print('{} até {} - Ocorreu um Erro na conexao'.format(date_inicial, date_final))

dt_final = datetime.now()
tempo_execucao = dt_final - dt_inicio
print('Procedimento execudado em {} minutos {} segundos'.format(int(tempo_execucao.seconds / 60),
                                                                tempo_execucao.seconds % 60))

df_raw = pd.concat(notificados)
df_raw.to_csv(r'Dengue.csv'.format(data), index=False, sep=';',
              escapechar='"')
print('Aquardando o próximo processamento...')

schedule.every().day.at(dataExecucao).do(job)

schedule.every(7).minutes.do(job)

while 1:
schedule.run_pending()
time.sleep(1)

[fccoelho]

Oi Rosinaldo, o erro indica que a biblioteca cliente do elasticsearch não reconhece o servidor como um servidor elasticsearch. Isto pode ser devido a alguma difereça de versões, ou que a API é implementada por outra ferramenta.

Existe alguma documentação desta API a que possamos ter accesso?

[rosinaldopinto1221]

Solicitei deles e estou aguardando .....

…

On Tue, May 2, 2023 at 7:47 AM Flávio Codeço Coelho < ***@***.***> wrote: Oi Rosinaldo, o erro indica que a biblioteca cliente do elasticsearch não reconhece o servidor como um servidor elasticsearch. Isto pode ser devido a alguma difereça de versões, ou que a API é implementada por outra ferramenta. Existe alguma documentação desta API a que possamos ter accesso? — Reply to this email directly, view it on GitHub <#131 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/ALAVR7ZIYQIZ4KTXQ4XV6QLXED66BANCNFSM6AAAAAAXQG3RN4> . You are receiving this because you authored the thread.Message ID: ***@***.***>