research/src/fetch_cnpj_info.py
from concurrent import futures
import json
import argparse
import time
import random
import itertools
import numpy as np
import os.path
import sys
import pandas as pd
import shutil
import requests
import requests.exceptions
import re
import logging
import json
from datetime import datetime, timedelta
LOGGER_NAME = 'fetch_cnpj'
TEMP_DATASET_PATH = os.path.join('data', 'companies-partial.xz')
INFO_DATASET_PATH = os.path.join('data', '{0}-{1}-{2}-companies.xz')
global logger, cnpj_list, num_threads, proxies_list
# source files mapped for extract cnpj
with open(os.path.join(os.path.dirname(os.path.realpath(__file__)),
'table_config.json')) as json_file:
json_config = json.load(json_file)
datasets_cols = json_config['cnpj_cpf']
def configure_logger(verbosity):
logger = logging.getLogger(LOGGER_NAME)
logger.setLevel(verbosity)
ch = logging.StreamHandler(sys.stdout)
ch.setLevel(verbosity)
logger.addHandler(ch)
return logger
def transform_and_translate_data(json_data):
"""
Transform main activity, secondary activity and partners list in
multi columns and translate column names.
"""
try:
data = pd.DataFrame(columns=['atividade_principal',
'data_situacao',
'tipo',
'nome',
'telefone',
'atividades_secundarias',
'situacao',
'bairro',
'logradouro',
'numero',
'cep',
'municipio',
'uf',
'abertura',
'natureza_juridica',
'fantasia',
'cnpj',
'ultima_atualizacao',
'status',
'complemento',
'email',
'efr',
'motivo_situacao',
'situacao_especial',
'data_situacao_especial',
'qsa'])
data = data.append(json_data, ignore_index=True)
except Exception as e:
logger.error("Error trying to transform and translate data:")
logger.error(json_data)
raise e
def decompose_main_activity(value):
struct = value
if struct:
return pd.Series(struct[0]). \
rename_axis({'code': 'main_activity_code',
'text': 'main_activity'})
else:
return pd.Series({}, index=['main_activity_code', 'main_activity'])
def decompose_secondary_activities(value):
struct = value
if struct and struct[0].get('text') != 'Não informada':
new_attributes = [pd.Series(activity).
rename_axis({'code': 'secondary_activity_%i_code' % (index + 1),
'text': 'secondary_activity_%i' % (index + 1)})
for index, activity in enumerate(struct)]
return pd.concat(new_attributes)
else:
return pd.Series()
def decompose_partners_list(value):
struct = value
if struct and len(struct) > 0:
new_attributes = [pd.Series(partner).
rename_axis({
'nome_rep_legal': 'partner_%i_legal_representative_name' % (index + 1),
'qual_rep_legal': 'partner_%i_legal_representative_qualification' % (index + 1),
'pais_origem': 'partner_%i_contry_origin' % (index + 1),
'nome': 'partner_%i_name' % (index + 1),
'qual': 'partner_%i_qualification' % (index + 1)})
for index, partner in enumerate(struct)]
return pd.concat(new_attributes)
else:
return pd.Series()
data = data.rename(columns={
'abertura': 'opening',
'atividade_principal': 'main_activity',
'atividades_secundarias': 'secondary_activities',
'bairro': 'neighborhood',
'cep': 'zip_code',
'complemento': 'additional_address_details',
'data_situacao_especial': 'special_situation_date',
'data_situacao': 'situation_date',
'efr': 'responsible_federative_entity',
'fantasia': 'trade_name',
'logradouro': 'address',
'motivo_situacao': 'situation_reason',
'municipio': 'city',
'natureza_juridica': 'legal_entity',
'nome': 'name',
'numero': 'number',
'situacao_especial': 'special_situation',
'situacao': 'situation',
'telefone': 'phone',
'tipo': 'type',
'uf': 'state',
'ultima_atualizacao': 'last_updated',
})
data['main_activity'] = data['main_activity'].fillna('{}')
data['secondary_activities'] = data['secondary_activities'].fillna('{}')
data['qsa'] = data['qsa'].fillna('{}')
data = pd.concat([
data.drop(['main_activity', 'secondary_activities', 'qsa'], axis=1),
data['main_activity'].apply(decompose_main_activity),
data['secondary_activities'].apply(decompose_secondary_activities),
data['qsa'].apply(decompose_partners_list)],
axis=1)
return data
def load_temp_dataset():
if os.path.exists(TEMP_DATASET_PATH):
return pd.read_csv(TEMP_DATASET_PATH, low_memory=False)
else:
return pd.DataFrame(columns=['cnpj'])
def read_cnpj_source_files(cnpj_source_files):
"""
Read the files passed as arguments and extract CNPJs from them.
The file needs to be mapped in datasets_cols.
"""
def extract_file_name_from_args(filepath):
date = re.compile('\d+-\d+-\d+-').findall(os.path.basename(filepath))
if date:
filename_without_date = os.path.basename(
filepath).replace(date[0], '')
else:
filename_without_date = os.path.basename(filepath)
return filename_without_date[:filename_without_date.rfind('.')]
def read_cnpj_list_to_import(filename, column):
cnpj_list = pd.read_csv(filename,
usecols=([column]),
dtype={column: np.str}
)[column]
cnpj_list = cnpj_list.map(lambda cnpj:
str(cnpj).replace(r'[./-]', '')).where(cnpj_list.str.len() == 14).unique()
return list(cnpj_list)
filesNotFound = list(filter(lambda file: not os.path.exists(file) or
datasets_cols.get(
extract_file_name_from_args(file.lower())) is None,
cnpj_source_files))
filesFound = list(filter(lambda file: os.path.exists(file) and
datasets_cols.get(
extract_file_name_from_args(file.lower())),
cnpj_source_files))
cnpj_list_to_import = list(itertools.chain.from_iterable(
map(lambda file: read_cnpj_list_to_import(file, datasets_cols.get(extract_file_name_from_args(file.lower()))),
filesFound)))
return cnpj_list_to_import, filesFound, filesNotFound
def remaining_cnpjs(cnpj_list_to_import, temp_dataset):
cnpj_list = set(cnpj_list_to_import)
already_fetched = set(temp_dataset['cnpj'].str.replace(r'[./-]', ''))
return list(cnpj_list - already_fetched)
def fetch_cnpj_info(cnpj, timeout=60):
url = 'http://receitaws.com.br/v1/cnpj/%s' % cnpj
try:
result = requests.get(url,
timeout=timeout,
proxies={'http': random.choice(proxies_list + [None])})
if result.status_code == 200:
cnpj_list.remove(cnpj)
json_return = json.loads(result.text.replace(
'\n', '').replace('\r', '').replace(';', ''))
return json_return
elif result.status_code == 429:
logger.debug('Sleeping 60 seconds to try again.')
logger.debug(result.text)
time.sleep(60)
logger.debug('Thread starting fetch again. {} CNPJs remaining.'.format(
len(cnpj_list)))
else:
logger.debug(result.text)
except requests.exceptions.Timeout as e:
logger.debug(e)
except requests.exceptions.ConnectionError as e:
logger.debug(e)
parser = argparse.ArgumentParser(epilog="ex: python fetch_cnpj_info.py \
./data/2016-12-10-reimbursements.xz 2016-12-14-amendments.xz \
-p 177.67.84.135:8080 177.67.82.80:8080 -t 10")
parser.add_argument('-v', '--verbosity', default='INFO',
help='level of logging messages.')
parser.add_argument('-t', '--threads', type=int, default=10,
help='number of threads to use in fetch process')
parser.add_argument('-p', '--proxies', nargs='*',
help='proxies list to distribuite requests in fetch process')
parser.add_argument('args', nargs='+',
help='source files from where to get CNPJs to fetch')
args = parser.parse_args()
if args.args:
logger = configure_logger(args.verbosity)
if args.proxies:
proxies_list = ['http://' + proxy for proxy in args.proxies]
else:
proxies_list = []
num_threads = args.threads
cnpj_list_to_import, filesFound, filesNotFound = read_cnpj_source_files(
args.args)
temp_dataset = load_temp_dataset()
cnpj_list = remaining_cnpjs(cnpj_list_to_import, temp_dataset)
print('%i CNPJ\'s to be fetched' % len(cnpj_list))
print('Starting fetch. {0} worker threads and {1} http proxies'.format(
num_threads, len(proxies_list)))
# Try again in case of error during fetch_cnpj_info
while len(cnpj_list) > 0:
with futures.ThreadPoolExecutor(max_workers=num_threads) as executor:
future_to_cnpj_info = dict((executor.submit(fetch_cnpj_info, cnpj), cnpj)
for cnpj in cnpj_list)
last_saving_point = 0
for future in futures.as_completed(future_to_cnpj_info):
cnpj = future_to_cnpj_info[future]
if future.exception() is None and future.result() is not None and future.result()['status'] == 'OK':
result_translated = transform_and_translate_data(
future.result())
temp_dataset = pd.concat([temp_dataset, result_translated])
if last_saving_point < divmod(len(temp_dataset.index), 100)[0]:
last_saving_point = divmod(len(temp_dataset.index), 100)[0]
print('###################################')
print('Saving information already fetched. {0} records'.format(
len(temp_dataset.index)))
temp_dataset.to_csv(TEMP_DATASET_PATH,
compression='xz',
encoding='utf-8',
index=False)
temp_dataset.to_csv(TEMP_DATASET_PATH,
compression='xz',
encoding='utf-8',
index=False)
os.rename(TEMP_DATASET_PATH, INFO_DATASET_PATH.format(
datetime.today().strftime("%Y"),
datetime.today().strftime("%m"),
datetime.today().strftime("%d")))
if len(filesNotFound) > 0:
print('The following files were not found:')
for file in filesNotFound:
print(file)
print('Maybe they were misspelled or the CNPJ\'s columns are not mapped:')
for file in datasets_cols:
print('File: %s | Column: %s' % (file, datasets_cols[file]))
print('%i CNPJ\'s listed in file(s)' % len(set(cnpj_list_to_import)))
cnpj_list = remaining_cnpjs(cnpj_list_to_import, temp_dataset)
print('%i CNPJ\'s remaining' % len(cnpj_list))
else:
print('no files to fetch CNPJ\'s')
print('type python fetch_cnpj_info.py -h for help')