Merge pull request #55 from ImpulsoGov/correcoes_etl_cnes

Correções ETL de Profissionais

implementar_fluxos.py

@@ -77,11 +77,11 @@
 )
 
 logger.info("Lendo módulos disponíveis...")
-#geral = import_module(".scripts.geral", "impulsoetl")
+geral = import_module(".scripts.geral", "impulsoetl")
 impulso_previne = import_module(".scripts.impulso_previne", "impulsoetl")
 saude_mental = import_module(".scripts.saude_mental", "impulsoetl")
 
-modulos = [impulso_previne,saude_mental] #geral #saude_mental
+modulos = [impulso_previne,geral,saude_mental] 
 
 
 if __name__ == "__main__":

src/impulsoetl/scnes/estabelecimentos_equipes/tratamento.py

@@ -109,8 +109,8 @@ def ordenar_colunas(df_extraido: pd.DataFrame, COLUNAS_TIPOS: dict):
         + "a partir da página do CNES."
     ),
     tags=["cnes", "equipes", "tratamento"],
-    retries=2,
-    retry_delay_seconds=120,
+    retries=0,
+    retry_delay_seconds=None,
 )
 def tratamento_dados(
     df_extraido: pd.DataFrame, periodo_id: str, unidade_geografica_id: str

src/impulsoetl/scnes/estabelecimentos_horarios/extracao.py

@@ -14,7 +14,16 @@
 from impulsoetl.scnes.extracao_lista_cnes import extrair_lista_cnes
 from impulsoetl.loggers import logger, habilitar_suporte_loguru
 
-
+@task(
+    name="Extrair Horários de Funcionamento dos Estabelecimentos de Saúde",
+    description=(
+        "Extrai os dados dos horários de funcionamento dos estabelecimentos de saúde"
+        + "a partir da página do CNES."
+    ),
+    tags=["cnes", "equipes", "extracao"],
+    retries=2,
+    retry_delay_seconds=120,
+)
 def extrair_horarios_estabelecimentos(
     codigo_municipio: str, lista_cnes: list, periodo_data_inicio: date
 ) -> pd.DataFrame:
@@ -46,7 +55,6 @@ def extrair_horarios_estabelecimentos(
                 + cnes
                 + "?competencia={:%Y%m}".format(periodo_data_inicio)
             )
-            # url = "http://cnes.datasus.gov.br/services/estabelecimentos/atendimento/"+codigo_municipio+cnes
             payload = {}
             headers = {
                 "Accept": "application/json, text/plain, */*",
@@ -67,6 +75,35 @@ def extrair_horarios_estabelecimentos(
             df["estabelecimento_cnes_id"] = cnes
             df_extraido = df_extraido.append(df)
 
+            if  df.empty:
+                try:
+                    url = 'https://cnes.datasus.gov.br/services/estabelecimentos/'+codigo_municipio+cnes+"?competencia={:%Y%m}".format(periodo_data_inicio)
+                    payload = {}
+                    headers = {
+                        "Accept": "application/json, text/plain, */*",
+                        "Accept-Language": "pt-BR,pt;q=0.9,en-US;q=0.8,en;q=0.7",
+                        "Connection": "keep-alive",
+                        "Referer": "http://cnes.datasus.gov.br/pages/estabelecimentos/",
+                        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/105.0.0.0 Safari/537.36",
+                        }
+                    response = requests.request("GET", url, headers=headers, data=payload)
+                    res = response.text
+                    parsed = json.loads(res)
+
+                    if "tpSempreAberto"  in parsed:
+                        horario_sempre_aberto = parsed.get("tpSempreAberto")     
+                    if horario_sempre_aberto == "S":
+                        df_parcial = pd.DataFrame()
+                        df_parcial['municipio_id_sus'] = [codigo_municipio]
+                        df_parcial['estabelecimento_cnes_id'] = [cnes]
+                        df_parcial['diaSemana'] = 'Sempre Aberto'
+                        df_parcial['hrInicioAtendimento'] = ""
+                        df_parcial['hrFimAtendimento'] = ""
+                        df_extraido = df_extraido.append(df_parcial)
+
+                except json.JSONDecodeError:
+                    pass  
+
         except json.JSONDecodeError:
             logger.info(
                 "Não foi possível extrair os horarios para o estabelecimento: "

src/impulsoetl/scnes/estabelecimentos_horarios/principal.py

@@ -15,8 +15,6 @@
     tratamento_dados,
 )
 from impulsoetl.scnes.extracao_lista_cnes import extrair_lista_cnes
-
-# from impulsoetl.scnes.verificacao_etl_scnes import verificar_dados
 from impulsoetl.utilitarios.bd import carregar_dataframe
 
 
@@ -63,10 +61,6 @@ def obter_horarios_estabelecimentos(
         unidade_geografica_id=unidade_geografica_id,
     )
 
-    # verificar_dados(
-    #    df_extraido=df_extraido, df_tratado=df_tratado
-    # )
-
     carregar_dataframe(
         sessao=sessao, df=df_tratado, tabela_destino=tabela_destino
     )

src/impulsoetl/scnes/estabelecimentos_horarios/tratamento.py

@@ -69,7 +69,16 @@ def transformar_horarios_json(df_extraido: pd.DataFrame) -> pd.DataFrame:
 
     return df_consolidado
 
-
+@task(
+    name="Transforma dados extraídos dos Horários de Funcionamento dos Estabelecimentos",
+    description=(
+        "Transforma os dados dos horários de funcionamento dos estabelecimentos de saúde"
+        + "a partir da página do CNES."
+    ),
+    tags=["cnes", "equipes", "tratamento"],
+    retries=0,
+    retry_delay_seconds=None,
+)
 def tratamento_dados(
     df_extraido: pd.DataFrame, periodo_id: str, unidade_geografica_id: str
 ) -> pd.DataFrame:

src/impulsoetl/scnes/estabelecimentos_profissionais_com_ine/extracao.py

@@ -0,0 +1,84 @@
+import warnings
+
+warnings.filterwarnings("ignore")
+import json
+from datetime import date
+
+import pandas as pd
+import requests
+from prefect import task
+
+from impulsoetl.loggers import habilitar_suporte_loguru, logger
+from impulsoetl.scnes.estabelecimentos_equipes.extracao import extrair_equipes
+from impulsoetl.scnes.extracao_lista_cnes import extrair_lista_cnes
+
+@task(
+    name="Extrair dados dos profissionais de saúde com INE por estabelecimento",
+    description=(
+        "Realiza a extração dos dados dos profissionais vinculados à alguma equipe de saúde"
+        + "a partir da página do CNES"
+    ),
+    tags=["cnes", "profissionais", "extracao"],
+    retries=0,
+    retry_delay_seconds=None,
+)
+def extrair_profissionais_com_ine (
+    codigo_municipio:str,
+    lista_codigos:list,
+    periodo_data_inicio:date
+)->pd.DataFrame:
+    """
+    Extrai informaçãoes dos profissionais de saúde que fazem parte de alguma equipe a partir da página do CNES
+     Argumentos:
+        codigo_municipio: Id sus do municipio.
+        lista_cnes: Lista contento os códigos CNES dos estabelecimentos presentes no município
+                    (conforme retornado pela função [`extrair_lista_cnes()`][]).
+        periodo_data_inicio: Data da competência 
+     Retorna:
+        Objeto [`pandas.DataFrame`] com os dados extraídos.
+    """
+
+    # Extrai as equipes do municipio
+    #lista_cnes = extrair_lista_cnes(codigo_municipio)
+    equipes = extrair_equipes(codigo_municipio,lista_codigos,periodo_data_inicio).reset_index()
+    linhas = equipes.shape[0]
+
+    df_extraido = pd.DataFrame()
+
+    # Para cada equipe seleciona a seqEquipe (número INE sem os zeros no inicio) e o coArea, e extrai os profissionais vinculados àquela equipe:
+    cont = 0
+    while cont <= (linhas-1):
+        cnes = equipes.loc[cont,'estabelecimento_cnes_id']
+        seqEquipe = equipes.loc[cont, 'seqEquipe']
+        coArea = equipes.loc[cont, 'coArea']
+        coEquipe = equipes.loc[cont, 'coEquipe']
+
+        try:
+            url = "http://cnes.datasus.gov.br/services/estabelecimentos-equipes/profissionais/"+codigo_municipio+cnes+"?coMun="+codigo_municipio+"&coArea="+coArea+"&coEquipe="+seqEquipe+"&competencia={:%Y%m}".format(periodo_data_inicio)
+            payload={}
+            headers = {
+                'Accept': 'application/json, text/plain, */*',
+                'Accept-Language': 'pt-BR,pt;q=0.9,en-US;q=0.8,en;q=0.7',
+                'Connection': 'keep-alive',
+                'Referer': 'http://cnes.datasus.gov.br/pages/estabelecimentos/ficha/equipes/',
+                'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/105.0.0.0 Safari/537.36'
+                }
+
+            response = requests.request("GET", url, headers=headers, data=payload)
+            res = response.text
+            parsed = json.loads(res)
+            df = pd.DataFrame(parsed)
+            df['INE'] = coEquipe
+            df['coArea'] = coArea
+            df['estabelecimento_cnes_id'] = cnes
+            df['municipio_id_sus'] = codigo_municipio
+            df_extraido = df_extraido.append(df)
+            cont += 1
+
+        except Exception as e:
+            logger.error(e)
+            logger.error("Erro ao extrair os profissionais com INE")
+            pass
+
+    return df_extraido    
+