Dados de CNPJ's podem ser uma rica fonte de informações para diversos tipos de análises. Entretanto, frequentemente só temos acesso ao número de um CNPJ, não conhecendo informações importantes como sua atividade principal, situação cadastral, sócios da empresa ou capital social, por exemplo. Felizmente, podemos obter essas e outras informações através da API Minha Receita, uma iniciativa da sociedade civil que disponibiliza, de forma gratuita e acessível, dados da Receita Federal.
Neste projeto, utilizaremos Python para fazer requisições à API Minha Receita e armazenar informações sobre +46mil CNPJ's. Veremos:
- Como realizar conexão com PostgreSQL para que o Python possa acessar o conteúdo de um banco de dados;
- Como fazer requisições para a API Minha Receita e obter informações associadas a um CNPJ segundo a Receita Federal;
- Como armazenar a resposta da API em um banco de dados PostgreSQL.
Conexão com PostgreSQL
Nosso objetivo é fazer requisições à API Minha Receita para obter dados de +46mil CNPJ's. Esses CNPJ's estão armazenados em uma tabela PostgreSQL, por isso, primeiro precisamos realizar a conexão com o banco de dados e fazer com que o Python tenha acesso à lista de CNPJ's.
Para fazer essa conexão, estamos usando a biblioteca Psycopg, que permite que o Python conecte-se a um banco de dados PostgreSQL. A função sql_conection() realiza a conexão e será utilizada em todos os momentos em que for necessário interagir com o banco de dados.
python
import psycopg2
def sql_connection():
# Realiza a conexão com o banco de dados PostgreSQL e retorna a variável que será utilizada para interagir com o banco
database = ''
user = ''
password = ''
host = ''
port = ''
con = psycopg2.connect(database=database,
user=user,
password=password,
host=host,
port=port)
return con
Nosso banco de dados contém uma tabela com vários dados, entre os quais está o CNPJ. Vamos fazer uma query para essa tabela e armazenar os CNPJ's únicos em um dataframe.
python
import pandas as pd
import pandas.io.sql as sqlio
def query_to_df(query):
# Dada uma query PostgreSQL, realiza a query no banco de dados e armazena o resultado em um dataframe
con = sql_connection()
data_set = sqlio.read_sql_query(query, con)
return data_set
query = 'select distinct(cnpj) from table_name'
df_cnpj = query_to_df(query)
Fazendo requisições para a API Minha Receita
No meu caso, o código acima retorna um dataframe com 46.911 CNPJ's, mas, por enquanto, não temos nenhuma informação adicional sobre eles. Felizmente, podemos fazer requisições à API Minha Receita e obter informações da Receita Federal sobre cada um desses CNPJ's.
A função get_cnpj_data(cnpj) utiliza a biblioteca Requests para realizar uma requisição à API, procurando por um CNPJ, e retorna os dados encontrados.
import requests
def get_cnpj_data(cnpj):
# Dado um CNPJ, faz uma requisição para a API Minha Receita. Caso a requisição seja bem sucedida, retorna o conteúdo da requisição em formato json
minha_receita_api_url = 'https://minhareceita.org/'
r = requests.post(minha_receita_api_url, data=cnpj, timeout=None)
if r.status_code == 200:
return json.loads(r.content)
Para nos familiarizarmos com a resposta da API, vamos fazer uma requisição de exemplo e utilizar a função jprint() para gerar uma visualização amigável do objeto json retornado pela API.
import json
def jprint(obj):
# Cria visualização amigável de um objeto json
text = json.dumps(obj, sort_keys=True, indent=4, ensure_ascii=False)
print(text)
cnpj_example = {'cnpj': 19131243000197}
response_example = get_cnpj_data(cnpj_example)
jprint(response_example)
O código acima deve retornar algo semelhante a:
{
"bairro": "BELA VISTA",
"capital_social": 0,
"cep": "01311902",
"cnae_fiscal": 9430800,
"cnae_fiscal_descricao": "Atividades de associações de defesa de direitos sociais",
"cnaes_secundarias": [
{
"codigo": 9493600,
"descricao": "Atividades de organizações associativas ligadas à cultura e à arte"
},
{
"codigo": 9499500,
"descricao": "Atividades associativas não especificadas anteriormente"
},
{
"codigo": 8599699,
"descricao": "Outras atividades de ensino não especificadas anteriormente"
},
{
"codigo": 8230001,
"descricao": "Serviços de organização de feiras, congressos, exposições e festas"
},
{
"codigo": 6204000,
"descricao": "Consultoria em tecnologia da informação"
}
],
"cnpj": "19131243000197",
"codigo_municipio": 7107,
"codigo_natureza_juridica": 3999,
"complemento": "ANDAR 4",
"data_exclusao_do_simples": null,
"data_inicio_atividade": "2013-10-03",
"data_opcao_pelo_simples": null,
"data_situacao_cadastral": "2013-10-03",
"data_situacao_especial": null,
"ddd_fax": "",
"ddd_telefone_1": "11 23851939",
"ddd_telefone_2": "",
"descricao_matriz_filial": "Matriz",
"descricao_porte": "Demais",
"descricao_situacao_cadastral": "Ativa",
"descricao_tipo_logradouro": "AVENIDA",
"identificador_matriz_filial": 1,
"logradouro": "PAULISTA 37",
"motivo_situacao_cadastral": 0,
"municipio": "SAO PAULO",
"nome_cidade_exterior": "",
"nome_fantasia": "REDE PELO CONHECIMENTO LIVRE",
"numero": "37",
"opcao_pelo_mei": false,
"opcao_pelo_simples": false,
"porte": 5,
"qsa": [
{
"cnpj": "19131243000197",
"cnpj_cpf_do_socio": "",
"codigo_qualificacao_representante_legal": 0,
"codigo_qualificacao_socio": 16,
"cpf_representante_legal": "",
"data_entrada_sociedade": "2019-10-25",
"identificador_de_socio": 2,
"nome_representante_legal": "",
"nome_socio": "FERNANDA CAMPAGNUCCI PEREIRA",
"percentual_capital_social": 0
}
],
"qualificacao_do_responsavel": 16,
"razao_social": "OPEN KNOWLEDGE BRASIL",
"situacao_cadastral": 2,
"situacao_especial": "",
"uf": "SP"
}
Obtendo dados de +46mil CNPJ's
Agora que fizemos uma requisição de exemplo e estamos familiarizados com a resposta da API, podemos seguir para a parte divertida e finalmente obter os dados da Receita Federal sobre os 46.911 CNPJ's nos quais estamos interessados.
Primeiro, vamos criar uma tabela PostgreSQL para armazenar nossos dados. Entre todas as informações que retornam da API, escolhemos apenas algumas variáveis principais, que são mais interessantes para as análises futuras.
Utilizamos o método connection.cursor(), que permite que o Python execute comandos em um banco de dados PostgreSQL. Repare que, após executar o comando, é necessário utilizar também o método connection.commit() para garantir que as mudanças no banco de dados sejam feitas.
def create_sql_table():
# Cria uma tabela PostgreSQL para armazenar os dados dos CNPJ's
con = sql_connection()
with con.cursor() as cur:
create_table_command = 'create table if not exists cnpj_data('\
'cnpj varchar(100),'\
'razao_social varchar(250),'\
'nome_fantasia varchar(250),'\
'atividade_principal_codigo varchar(50),'\
'atividade_principal_descricao varchar(1000),'\
'situacao_cadastral varchar(50),'\
'capital_social float,'\
'porte varchar(50),'\
'codigo_natureza_juridica int,'\
'data_abertura date,'\
'cep varchar(50),'\
'municipio varchar(100),'\
'uf varchar(2)'\
')'
cur.execute(create_table_command)
con.commit()
create_sql_table()
Uma vez criada a tabela, podemos começar a alimentá-la com os dados dos CNPJ's. Para isso, vamos considerar a orientação da documentação do PostgreSQL de que a melhor forma de alimentar uma tabela é usando o comando COPY FROM. Uma demonstração sobre o melhor desempenho do COPY FROM em relação a outros métodos pode ser encontrada nesta análise, que avalia a performance de diferentes formas de importar dados de uma fonte remota para uma tabela PostgreSQL.
Assim, para alimentar nossa tabela, seguiremos o seguinte fluxo:
Vamos começar definindo as funções que utilizaremos para armazenar os dados da resposta da API em um dataframe. Repare que, na função cnpj_data_to_df(), caso o CNPJ não seja encontrado e a resposta da API volte com valor nulo, iremos adicionar o CNPJ ao dataframe mesmo assim, porém com valores nulos nas colunas de informações. Isso nos ajudará a ter controle sobre quais CNPJ's foram encontrados e quais não foram.
def clean_text(full_text):
# Limpa um texto, retirando quebras de linha e ponto-e-vírgulas
clean_text = full_text.replace("\n", " ")
clean_text = full_text.replace(";", " ")
return clean_text
def cnpj_data_to_df(df_cnpj):
# Recebe um dataframe contendo os CNPJ's e, a partir das requisições à API Minha Receita, estrutura um dataframe contendo os dados do CNPJ
data_set = pd.DataFrame(columns=[
'cnpj',
'razao_social',
'nome_fantasia',
'atividade_principal_codigo',
'atividade_principal_descricao',
'situacao_cadastral',
'capital_social',
'porte',
'codigo_natureza_juridica',
'data_abertura',
'cep',
'municipio',
'uf'
])
for cnpj in df_cnpj['cpf_cnpj']:
cnpj_data = get_cnpj_data({'cnpj': cnpj})
if cnpj_data != None:
new_row = {
'cnpj': cnpj,
'razao_social': cnpj_data['razao_social'],
'nome_fantasia': cnpj_data['nome_fantasia'],
'atividade_principal_codigo': cnpj_data['cnae_fiscal'],
'atividade_principal_descricao': clean_text(cnpj_data['cnae_fiscal_descricao']),
'situacao_cadastral': cnpj_data['descricao_situacao_cadastral'],
'capital_social': float(cnpj_data['capital_social']),
'porte': cnpj_data['descricao_porte'],
'codigo_natureza_juridica': int(cnpj_data['codigo_natureza_juridica']),
'data_abertura': cnpj_data['data_inicio_atividade'],
'cep': cnpj_data['cep'],
'municipio': cnpj_data['municipio'],
'uf': cnpj_data['uf'],
}
else:
new_row = {'cnpj': cnpj}
data_set = data_set.append(new_row, ignore_index=True)
return data_set
Agora podemos definir a função df_to_csv(), que exporta o dataframe para um csv temporário, e a função csv_to_sql_table(), que utiliza o COPY FROM para alimentar a tabela PostgreSQL.
def df_to_csv(csv_path, data_set):
# Exporta um dataframe para um arquivo csv
data_set.to_csv(csv_path, header=False, index=False, sep=';',
encoding='utf-8', quoting=csv.QUOTE_MINIMAL)
def csv_to_sql_table(csv_path):
# Alimenta a tabela PostgreSQL com os dados dos CNPJ's
con = sql_connection()
with con.cursor() as cur:
cur.execute('copy cnpj_data from %s delimiter %s csv', [csv_path, ';'])
con.commit()
Com todas as funções definidas, podemos agora iterar sobre o dataframe df_cnpj, que contém os CNPJ's, e fazer requisições à API Minha Receita. Vamos armazenar as respostas na tabela PostgreSQL em lotes de 1000 CNPJ's para garantir que, caso ocorra um erro enquanto o código roda, não perderemos os dados das requisições já realizadas.
csv_path = r'\temp_cnpj_data.csv'
df_cnpj_data = pd.DataFrame()
for i in range(0, len(df_cnpj), 1000):
new_batch = cnpj_data_to_df(df_cnpj[i:i+1000])
df_to_csv(csv_path, new_batch)
csv_to_sql_table(csv_path)
df_cnpj_data = df_cnpj_data.append(new_batch, ignore_index=True)
print(i)
Para finalizar, vamos checar quantos CNPJ's foram encontrados e quantos não foram.
query = 'select * from cnpj_data'
df_cnpj_data = query_to_df(query)
empty_columns = ['razao_social',
'nome_fantasia',
'atividade_principal_codigo',
'atividade_principal_descricao',
'situacao_cadastral',
'capital_social',
'porte',
'codigo_natureza_juridica',
'data_abertura',
'cep',
'municipio',
'uf'
]
df_found_cnpj = df_cnpj_data.dropna(axis=0, how='all', subset=empty_columns)
percentage_not_found_cnpj = 100*(1 - (len(df_found_cnpj)/len(df_cnpj_data)))
print('De', len(df_cnpj_data), 'CNPJs, foram encontrados', len(df_found_cnpj))
print('Não foram encontrados', round(percentage_not_found_cnpj, 2), '% dos CNPJs buscados.')
De 46911 CNPJs, foram encontrados 46851
Não foram encontrados 0.13 % dos CNPJs buscados.
Conclusões
Neste projeto, utilizamos Python para acessar e manipular bancos de dados PostgreSQL. Conseguimos obter informações de +46mil CNPJ's, com apenas 0,13% dos dados não encontrados. Para isso, a API Minha Receita foi um recurso essencial, já que ela disponibiliza os dados da Receita Federal de forma gratuita e acessível.
Com este tutorial, espero ter contribuído para desvendar possíveis dúvidas sobre como utilizar a biblioteca Psycopg para conectar o Python a um banco de dados PostgreSQL e a biblioteca Requests para realizar requisições a uma API.
Espero também que o projeto contribua para que mais pessoas conheçam e utilizem a API Minha Receita.
O meu repositório no Github com o notebook do projeto está disponível neste link.
Top comments (4)
Olá, estou replicando este código para utilizar numa base de cnpjs, porém quando rodo esse trecho, o jupyter retorna dizendo "NameError: name 'csv' is not defined"
"csv_path = r'\temp_cnpj_data.csv'
df_cnpj_data = pd.DataFrame()
for i in range(0, len(df_cnpj), 1000):
new_batch = cnpj_data_to_df(df_cnpj[i:i+1000])
df_to_csv(csv_path, new_batch)
csv_to_sql_table(csv_path)
df_cnpj_data = df_cnpj_data.append(new_batch, ignore_index=True)
print(i)"
Alguém sabe o que pode ser?
Parabéns pelo post!! Já salvei aqui ❤️
como eu traria as informações de qsa ?