Skip to content

Commit

Permalink
lint: apply python black linter
Browse files Browse the repository at this point in the history
Signed-off-by: Vinicius Ruoso <[email protected]>
  • Loading branch information
vkruoso committed Apr 17, 2023
1 parent dc830da commit 2a2077f
Show file tree
Hide file tree
Showing 10 changed files with 178 additions and 192 deletions.
3 changes: 1 addition & 2 deletions receita/__init__.py
Original file line number Diff line number Diff line change
@@ -1,2 +1 @@

__version__ = '2.2.0'
__version__ = "2.2.0"
55 changes: 25 additions & 30 deletions receita/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,45 +18,43 @@


class Cli(object):

def __init__(self):
parser = argparse.ArgumentParser(
usage=BASE_USAGE,
epilog=(
'You can get help for each command using the --help option '
'for each of them.'
)
"You can get help for each command using the --help option "
"for each of them."
),
)
parser.add_argument('command', help='command to run')
parser.add_argument("command", help="command to run")
args = parser.parse_args(sys.argv[1:2])
if not hasattr(self, args.command):
print('unrecognized command\n')
print("unrecognized command\n")
parser.print_help()
exit(1)
getattr(self, args.command)()

def get(self):
parser = argparse.ArgumentParser(
prog='receita get',
description='Download information about a list of companies.',
prog="receita get",
description="Download information about a list of companies.",
epilog=(
'The data is retrieved from ReceitaWS webservice on company '
'information. Please, make sure you read its terms of usage '
'and its documentation at https://www.receitaws.com.br on '
'how the webservice works. You MAY be charged when using '
'some features of this service.'
)
"The data is retrieved from ReceitaWS webservice on company "
"information. Please, make sure you read its terms of usage "
"and its documentation at https://www.receitaws.com.br on "
"how the webservice works. You MAY be charged when using "
"some features of this service."
),
)
parser.add_argument('list', help='CSV file with companies CNPJ')
parser.add_argument("list", help="CSV file with companies CNPJ")
parser.add_argument(
'-d',
"-d",
metavar="DAYS",
dest='days',
help='maximum data deprecation allowed in days')
dest="days",
help="maximum data deprecation allowed in days",
)
parser.add_argument(
'--output',
help='directory to save the output',
default='data'
"--output", help="directory to save the output", default="data"
)
args = parser.parse_args(sys.argv[2:])

Expand All @@ -65,17 +63,14 @@ def get(self):

def build(self):
parser = argparse.ArgumentParser(
prog='receita build',
description='Create CSV files from the retrieved information.')
prog="receita build",
description="Create CSV files from the retrieved information.",
)
parser.add_argument(
'--output',
help='directory to save the generated CSV files',
default='.'
"--output", help="directory to save the generated CSV files", default="."
)
parser.add_argument(
'--input',
help='directory to get input from',
default='data'
"--input", help="directory to get input from", default="data"
)
args = parser.parse_args(sys.argv[2:])

Expand All @@ -87,5 +82,5 @@ def main():
Cli()


if __name__ == '__main__':
if __name__ == "__main__":
main()
164 changes: 82 additions & 82 deletions receita/tools/build.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,88 +6,88 @@


class BaseCSV(object):

ERROR = 'ERROR'
ERROR = "ERROR"

def __init__(self, output):
path = os.path.join(output, self._filename + '.csv')
self._f = open(path, 'w')
path = os.path.join(output, self._filename + ".csv")
self._f = open(path, "w")

self.writer = csv.DictWriter(
self._f, fieldnames=self._fields,
extrasaction='ignore')
self._f, fieldnames=self._fields, extrasaction="ignore"
)
self.writer.writeheader()


class _CompaniesCSV(BaseCSV):

_filename = 'companies'
_filename = "companies"
_fields = [
'cnpj',
'tipo',
'abertura',
'nome',
'fantasia',
'natureza_juridica',
'logradouro',
'numero',
'complemento',
'cep',
'bairro',
'municipio',
'uf',
'email',
'telefone',
'efr',
'situacao',
'data_situacao',
'motivo_situacao',
'situacao_especial',
'data_situacao_especial',
'capital_social'
"cnpj",
"tipo",
"abertura",
"nome",
"fantasia",
"natureza_juridica",
"logradouro",
"numero",
"complemento",
"cep",
"bairro",
"municipio",
"uf",
"email",
"telefone",
"efr",
"situacao",
"data_situacao",
"motivo_situacao",
"situacao_especial",
"data_situacao_especial",
"capital_social",
]

def visit(self, data):
self.writer.writerow(data)


class _ActivitiesCSV(BaseCSV):

_filename = 'activities'
_filename = "activities"
_fields = [
'cnpj',
'tipo',
'codigo',
'descricao',
"cnpj",
"tipo",
"codigo",
"descricao",
]

def visit(self, data):
if data['status'] == self.ERROR:
if data["status"] == self.ERROR:
return

for activity in data['atividade_principal']:
self.writer.writerow({
'cnpj': data['cnpj'],
'tipo': 'principal',
'codigo': activity['code'],
'descricao': activity['text']
})

for activity in data['atividades_secundarias']:
self.writer.writerow({
'cnpj': data['cnpj'],
'tipo': 'secundaria',
'codigo': activity['code'],
'descricao': activity['text']
})
for activity in data["atividade_principal"]:
self.writer.writerow(
{
"cnpj": data["cnpj"],
"tipo": "principal",
"codigo": activity["code"],
"descricao": activity["text"],
}
)

for activity in data["atividades_secundarias"]:
self.writer.writerow(
{
"cnpj": data["cnpj"],
"tipo": "secundaria",
"codigo": activity["code"],
"descricao": activity["text"],
}
)


class _ActivitiesSeenCSV(BaseCSV):

_filename = 'activities_seen'
_filename = "activities_seen"
_fields = [
'codigo',
'descricao',
"codigo",
"descricao",
]

def __init__(self, output):
Expand All @@ -96,47 +96,47 @@ def __init__(self, output):

def _process(self, activities):
for activity in activities:
if activity['code'] == '00.00-0-00':
if activity["code"] == "00.00-0-00":
continue
key = (activity['code'], activity['text'],)
key = (
activity["code"],
activity["text"],
)
if key in self._activities:
continue
self._activities[key] = activity['text']
self.writer.writerow({
'codigo': activity['code'],
'descricao': activity['text']
})
self._activities[key] = activity["text"]
self.writer.writerow(
{"codigo": activity["code"], "descricao": activity["text"]}
)

def visit(self, data):
if data['status'] == self.ERROR:
if data["status"] == self.ERROR:
return
self._process(data['atividade_principal'])
self._process(data['atividades_secundarias'])
self._process(data["atividade_principal"])
self._process(data["atividades_secundarias"])


class _QSACSV(BaseCSV):

_filename = 'qsa'
_filename = "qsa"
_fields = [
'cnpj',
'nome',
'qual',
'pais_origem',
'nome_rep_legal',
'qual_rep_legal',
"cnpj",
"nome",
"qual",
"pais_origem",
"nome_rep_legal",
"qual_rep_legal",
]

def visit(self, data):
if data['status'] == self.ERROR:
if data["status"] == self.ERROR:
return

for qsa in data['qsa']:
qsa.update({'cnpj': data['cnpj']})
for qsa in data["qsa"]:
qsa.update({"cnpj": data["cnpj"]})
self.writer.writerow(qsa)


class Build(object):

def __init__(self, input_, output):
self.input = os.path.abspath(input_)
self.output = os.path.abspath(output)
Expand All @@ -148,11 +148,11 @@ def run(self):
try:
os.mkdir(self.output)
except:
print(('failed to create output directory %s' % self.output))
print("failed to create output directory %s" % self.output)

# Be sure it is a directory
if not os.path.isdir(self.output):
print(('invalid output directory %s' % self.output))
print("invalid output directory %s" % self.output)
sys.exit(1)

# Create the CSV handlers
Expand All @@ -164,8 +164,8 @@ def run(self):
]

# Run by each company populating the CSV files
for path in glob.glob(os.path.join(self.input, '*.json')):
with open(path, 'r') as f:
for path in glob.glob(os.path.join(self.input, "*.json")):
with open(path, "r") as f:
try:
data = json.load(f)
except ValueError:
Expand Down
7 changes: 3 additions & 4 deletions receita/tools/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,19 +3,18 @@


class Client(object):

def __init__(self, cnpj, days=None, token=None):
self.cnpj = cnpj
self.days = days
self.token = token

def get(self):
url = 'https://www.receitaws.com.br/v1/cnpj/%s' % self.cnpj
url = "https://www.receitaws.com.br/v1/cnpj/%s" % self.cnpj
headers = {}

if self.days and self.token:
url = '%s/days/%s' % (url, self.days)
headers.update({'Authorization': 'Bearer %s' % self.token})
url = "%s/days/%s" % (url, self.days)
headers.update({"Authorization": "Bearer %s" % self.token})

try:
response = requests.get(url, headers=headers, timeout=70)
Expand Down
Loading

0 comments on commit 2a2077f

Please sign in to comment.