Extração dos dados
ReceitaCNPJApi
Descrição:
A ReceitaCNPJApi é uma interface especializada para interagir com a API de Dados Abertos da Receita Federal. Esta classe automatiza várias tarefas relacionadas à obtenção de informações sobre CNPJs, como empresas, sócios e outros
detalhes. Ela foi projetada para simplificar a aquisição, descompactação e preparação de dados para análise, removendo a necessidade de interação manual com a API.
A ReceitaCNPJApi é uma interface especializada para interagir com a API de Dados Abertos da Receita Federal. Esta classe automatiza várias tarefas relacionadas à obtenção de informações sobre CNPJs, como empresas, sócios e outros
detalhes. Ela foi projetada para simplificar a aquisição, descompactação e preparação de dados para análise, removendo a necessidade de interação manual com a API.
A Classe foi desenvolvida para ser utilizada localmente
Para visualizar a implementação da extração em um nuvem vá para aba: Disponibilização dos dados em Nuvem
Foi implemendo boas práticas para não sobrecarregar servidor entre uma requisição e outra
Métodos:
__init__()
Inicializa a instância da classe e estabelece um sistema de logging para rastrear informações e erros durante as operações.
get_last_month() -> datetime.date
Descrição: Retorna a data atual do sistema.
Retorno: Um objeto datetime.date representando a data atual.
get_output_path_for_prefix(prefix: str) -> str
Descrição: Constrói um caminho de saída baseado no prefixo fornecido para armazenar arquivos.
Retorno: Uma string representando o caminho completo de saída para o prefixo especificado.
get_most_common_date() -> Optional[datetime.date]
Descrição: Consulta a API para identificar a data mais comum de atualização dos arquivos.
Retorno: Um objeto datetime.date representando a data mais comum de atualização, ou None se não puder ser determinado.
lista_urls_receita(*prefixes: str) -> List[str]
Descrição: Gera uma lista de URLs de download com base nos prefixos fornecidos.
Retorno: Uma lista de strings contendo URLs completas para os arquivos correspondentes aos prefixos.
fetch_data(url: str, save_path: str, log_accumulator: Optional[List[str]] = None, max_attempts: int = 15, wait_time: int = 180) -> str
Descrição: Automatiza o processo de download de um arquivo da URL especificada e o salva no caminho especificado.
Retorno: Uma string representando o caminho completo do arquivo baixado.
download_and_unzip(url: str, save_base_path: str = "./temp", output_base_path: str = "./output", headers: Optional[dict] = None, log_accumulator: Optional[List[str]] = None, data_update: bool = True) -> Tuple[str, str]
Descrição: Combina as operações de download e descompactação, facilitando a obtenção e preparação de arquivos.
Retorno: Um tuple contendo a URL processada e uma mensagem ("Success" ou a razão da falha).
unzip_files(zip_file_path: str, output_base_path: str, log_accumulator: Optional[List[str]] = None)
Descrição: Extrai os conteúdos de um arquivo zip fornecido e os salva no caminho de saída especificado.
Retorno: Nenhum. A operação modifica o sistema de arquivos para refletir os dados extraídos.
Atributos:
BASE_URL: A URL base da API de dados abertos da Receita Federal.
FILE_PREFIXES: Uma lista contendo prefixos dos arquivos que podem ser baixados da API.
MAX_ATTEMPTS: Um número representando o máximo de tentativas ao fazer uma solicitação para a API.
WAIT_TIME: Um número que indica o tempo de espera em segundos entre tentativas de solicitação.
Utilizando a classe ReceitaCNPJApi, os desenvolvedores têm uma ferramenta robusta e intuitiva para extrair e processar dados relacionados a CNPJs da Receita Federal sem a necessidade de interação manual ou familiaridade com os detalhes da API subjacente.
Inicializa a instância da classe e estabelece um sistema de logging para rastrear informações e erros durante as operações.
get_last_month() -> datetime.date
Descrição: Retorna a data atual do sistema.
Retorno: Um objeto datetime.date representando a data atual.
get_output_path_for_prefix(prefix: str) -> str
Descrição: Constrói um caminho de saída baseado no prefixo fornecido para armazenar arquivos.
Retorno: Uma string representando o caminho completo de saída para o prefixo especificado.
get_most_common_date() -> Optional[datetime.date]
Descrição: Consulta a API para identificar a data mais comum de atualização dos arquivos.
Retorno: Um objeto datetime.date representando a data mais comum de atualização, ou None se não puder ser determinado.
lista_urls_receita(*prefixes: str) -> List[str]
Descrição: Gera uma lista de URLs de download com base nos prefixos fornecidos.
Retorno: Uma lista de strings contendo URLs completas para os arquivos correspondentes aos prefixos.
fetch_data(url: str, save_path: str, log_accumulator: Optional[List[str]] = None, max_attempts: int = 15, wait_time: int = 180) -> str
Descrição: Automatiza o processo de download de um arquivo da URL especificada e o salva no caminho especificado.
Retorno: Uma string representando o caminho completo do arquivo baixado.
download_and_unzip(url: str, save_base_path: str = "./temp", output_base_path: str = "./output", headers: Optional[dict] = None, log_accumulator: Optional[List[str]] = None, data_update: bool = True) -> Tuple[str, str]
Descrição: Combina as operações de download e descompactação, facilitando a obtenção e preparação de arquivos.
Retorno: Um tuple contendo a URL processada e uma mensagem ("Success" ou a razão da falha).
unzip_files(zip_file_path: str, output_base_path: str, log_accumulator: Optional[List[str]] = None)
Descrição: Extrai os conteúdos de um arquivo zip fornecido e os salva no caminho de saída especificado.
Retorno: Nenhum. A operação modifica o sistema de arquivos para refletir os dados extraídos.
Atributos:
BASE_URL: A URL base da API de dados abertos da Receita Federal.
FILE_PREFIXES: Uma lista contendo prefixos dos arquivos que podem ser baixados da API.
MAX_ATTEMPTS: Um número representando o máximo de tentativas ao fazer uma solicitação para a API.
WAIT_TIME: Um número que indica o tempo de espera em segundos entre tentativas de solicitação.
Utilizando a classe ReceitaCNPJApi, os desenvolvedores têm uma ferramenta robusta e intuitiva para extrair e processar dados relacionados a CNPJs da Receita Federal sem a necessidade de interação manual ou familiaridade com os detalhes da API subjacente.
Para uma análise mais detalhada, consulte a documentação presente na classe.
Script da Classe
ReceitaCNPJApi Teste da Classe
Exemplo do código
from ReceitaCNPJApi import ReceitaCNPJApi
from concurrent.futures import ThreadPoolExecutor
api = ReceitaCNPJApi()
# Pode ser uma tabela ou varias
urls = api.lista_urls_receita('Municipios')
urls = api.lista_urls_receita('Municipios', 'Cnaes',
'Naturezas', 'Qualificacoes', 'Paises','Motivos')
def process_url(url):
return api.download_and_unzip(url)
# Mais do 3 aconteceu de tomar timeout. Sugiro manter
with ThreadPoolExecutor(max_workers=3) as executor:
results = list(executor.map(process_url, urls))
for url, result in zip(urls, results):
print(f"Result for {url}: {result[1]}")
Output
INFO:ReceitaCNPJApi:Trying to unzip ./temp\Naturezas.zip
INFO:ReceitaCNPJApi:File https://dadosabertos.rfb.gov.br/CNPJ/Naturezas.zip has been unzipped successfully to ./output
INFO:ReceitaCNPJApi:Trying to unzip ./temp\Cnaes.zip
INFO:ReceitaCNPJApi:File https://dadosabertos.rfb.gov.br/CNPJ/Cnaes.zip has been unzipped successfully to ./output
INFO:ReceitaCNPJApi:Trying to unzip ./temp\Municipios.zip
INFO:ReceitaCNPJApi:File https://dadosabertos.rfb.gov.br/CNPJ/Municipios.zip has been unzipped successfully to ./output
INFO:ReceitaCNPJApi:Trying to unzip ./temp\Qualificacoes.zip
INFO:ReceitaCNPJApi:File https://dadosabertos.rfb.gov.br/CNPJ/Qualificacoes.zip has been unzipped successfully to ./output
INFO:ReceitaCNPJApi:Trying to unzip ./temp\Paises.zip
INFO:ReceitaCNPJApi:File https://dadosabertos.rfb.gov.br/CNPJ/Paises.zip has been unzipped successfully to ./output
INFO:ReceitaCNPJApi:Trying to unzip ./temp\Motivos.zip
INFO:ReceitaCNPJApi:File https://dadosabertos.rfb.gov.br/CNPJ/Motivos.zip has been unzipped successfully to ./output
Result for https://dadosabertos.rfb.gov.br/CNPJ/Municipios.zip: Success
Result for https://dadosabertos.rfb.gov.br/CNPJ/Cnaes.zip: Success
Result for https://dadosabertos.rfb.gov.br/CNPJ/Naturezas.zip: Success
Result for https://dadosabertos.rfb.gov.br/CNPJ/Qualificacoes.zip: Success
Result for https://dadosabertos.rfb.gov.br/CNPJ/Paises.zip: Success
Result for https://dadosabertos.rfb.gov.br/CNPJ/Motivos.zip: Success