import requests, re #,json from datetime import date, datetime def getIndex(arr:list, index:int, default = False) : try: return arr[index] except: return default != False and default or None # https://www.receitaws.com.br/v1/cnpj/17055652000126 # python3 /var/www/html/webservice/receita/receitaws.py 17055652000126 <- Executará no modo debug MAP = [ {"get":"status" } , {"get":"message" } , {"get":"situacao" } , {"get":"nome" } , {"get":"fantasia" } , {"get":"data_situacao" ,"custom":"data_situacao" } , {"get":"tipo" } , {"get":"porte" } , {"get":"abertura" ,"custom":"abertura" } , {"get":"natureza_juridica" } , {"get":"ultima_atualizacao" ,"custom":"ultima_atualizacao" } , {"get":"logradouro" } , {"get":"numero" ,"custom":"apenas_numeros" } , {"get":"cep" ,"custom":"cep" } , {"get":"municipio" } , {"get":"bairro" } , {"get":"uf" } , {"get":"complemento" } , {"get":"telefone" ,"custom":"telefone" } , {"get":"email" } , {"get":"atividade_principal/0" , "custom": "atividade_principal" } , {"get":"atividades_secundarias" , "custom": "atividades_secundarias" } ] class Receita : """Class para capturar dados de empresas vindo da receita""" def __init__(self,cnpj:str,arquivo_saida:str): """Inicia classe Args: cnpj (str): CNPJ do cliente a ser consultado """ self.cnpj = cnpj self.arquivo_saida = arquivo_saida or "lixo.txt" self.endpoint = "https://www.receitaws.com.br/v1/cnpj/{}".format(cnpj) self.consultarCnpj() def consultarCnpj(self): if not self.cnpj : response = {"status":"ERROR","message":"É necessário informar o CNPJ para consultar-lo"} else: try: req = requests.get(self.endpoint) response = req.json() except Exception as e: response = {"status":"ERROR","message":"Retorno com erros: " + str(e)} #print(str(e)) #import json #response = '{"atividade_principal": [{"text": "Fabrica\\u00e7\\u00e3o de aparelhos e equipamentos de medida, teste e controle", "code": "26.51-5-00"}], "data_situacao": "08/10/2012", "complemento": "BRCAO 04 COND LINHAO CT EMPRESARIA", "tipo": "MATRIZ", "nome": "ACX AUTOMACAO E CONTROLE LTDA", "uf": "PR", "telefone": "(41) 8884-8895", "atividades_secundarias": [{"text": "Manuten\\u00e7\\u00e3o e repara\\u00e7\\u00e3o de aparelhos e instrumentos de medida, teste e controle", "code": "33.12-1-02"}, {"text": "Instala\\u00e7\\u00e3o de m\\u00e1quinas e equipamentos industriais", "code": "33.21-0-00"}, {"text": "Instala\\u00e7\\u00e3o e manuten\\u00e7\\u00e3o el\\u00e9trica", "code": "43.21-5-00"}, {"text": "Com\\u00e9rcio atacadista de M\\u00e1quinas e equipamentos para uso industrial; partes e pe\\u00e7as", "code": "46.63-0-00"}, {"text": "Desenvolvimento de programas de computador sob encomenda", "code": "62.01-5-01"}, {"text": "Desenvolvimento e licenciamento de programas de computador customiz\\u00e1veis", "code": "62.02-3-00"}, {"text": "Desenvolvimento e licenciamento de programas de computador n\\u00e3o-customiz\\u00e1veis", "code": "62.03-1-00"}, {"text": "Suporte t\\u00e9cnico, manuten\\u00e7\\u00e3o e outros servi\\u00e7os em tecnologia da informa\\u00e7\\u00e3o", "code": "62.09-1-00"}], "qsa": [{"qual": "49-S\\u00f3cio-Administrador", "nome": "BRUNO CESAR XAVIER"}, {"qual": "49-S\\u00f3cio-Administrador", "nome": "RAUL BERTONCELLO NETO"}, {"qual": "22-S\\u00f3cio", "nome": "MATHEUS HENRIQUE DE OLIVEIRA LELES"}], "situacao": "ATIVA", "bairro": "FAZENDINHA", "logradouro": "R EDVINO ANTONIO DEBONI", "numero": "225", "cep": "81.330-600", "municipio": "CURITIBA", "porte": "MICRO EMPRESA", "abertura": "08/10/2012", "natureza_juridica": "206-2 - Sociedade Empres\\u00e1ria Limitada", "fantasia": "ACX INDUSTRIAL", "cnpj": "17.055.652/0001-26", "ultima_atualizacao": "2020-12-01T16:28:51.964Z", "status": "OK", "email": "", "efr": "", "motivo_situacao": "", "situacao_especial": "", "data_situacao_especial": "", "capital_social": "210000.00", "extra": {}, "billing": {"free": true, "database": true}}' #response = json.loads(response) self.tranlate(response) #print(response.json()) def tranlate(self, data): copy_book = [] for item in MAP: #_to = item.get("to", "item_{}".format(key)) value = self.handle_item(item, data) copy_book.append(value) self.write_file( data = '|'.join(copy_book) + "|" , encoding = 'iso-8859-1' , path = self.arquivo_saida ) def handle_item(self, item, json) : """ Manipulacao e tratamento de cada item """ value = False if "get" in item : cpath = item.get('custom_path_get') _get = self.call(item['custom_path_get'], item['get'], value) if cpath else item['get'] value = self.jsonxpath(_get, json, item.get('trim', False)) if "default" in item and not value: value = item['default'] if "custom" in item : value = self.call(item['custom'], value, item) if "by_type" in item: value = self.call(item['by_type'], value, item) return value def hasAttr(self, attribute, _object) : return attribute in dir(_object) def call(self, *args, **kwargs) : """ Parameters ---------- `` args[0]`` str The name's funtion to be called. ``args[1:]`` parameter's function ``kwargs`` object params. ``is_required`` bool To determine if this function is required or not ``default`` None. Raises ------ ``Exception`` If args[0] not defined will raise Exception """ required = kwargs.get("is_required") if args : func = args[0] else: raise Exception("call() missing 1 required positional argument") if self.hasAttr(func, self ) : if required : del kwargs['is_required'] try: return getattr(self, func)(*args[1:], **kwargs) except : raise else : if required : raise Exception("name '{}' is not defined, this function is required".format(func)) else : return False def jsonxpath(self, path ,json, trim) : result = gn_jsonpath(path=path, json=json, trim=trim) if (isinstance(result.value, str)) : result.value.strip() return result.value def read_file(self, path ="",flag = "r", encoding='utf8') : return self.handle_file(handle="read", path =path,flag = flag, encoding=encoding) def write_file(self, data="", path="", flag = "w+", encoding='utf8') : return self.handle_file(handle = "write",data=data, path =path,flag = flag, encoding=encoding) def handle_file(self, **kwargs) : try: with open(kwargs['path'] , kwargs['flag'], encoding=kwargs['encoding']) as file: if kwargs['handle'] == 'read': return file.read() elif kwargs['handle'] == 'write': return file.write(kwargs['data']) else: pass except IOError as error: print(error) exit() ############################################################################ #### ### #### CUSTOMIZACAO ### #### ### ############################################################################ def atividade_principal(self, data, props) : if isinstance(data, dict): return self.remover_mascara_code(data.get('code',"")) #data = self.remover_mascara_code(data) #return ";".join(data.values() or []) + ";" return "" def atividades_secundarias(self, data, props) : if isinstance(data, list): return ";".join([self.remover_mascara_code(atividade.get('code',"")) for atividade in data]) + ";" return "" def remover_mascara_code(self, code) : return re.sub(r'[^\d]','', code) def telefone(self, data, props) : if "/" in data: data = getIndex(data.split("/"), 0, "") telefone = re.sub(r'(^([0]|[(])+|[^\d])', '', data) #telefone_numero = re.sub(r'(^([0]|[(])+|[^\d])', '', data) return "{ddd}|{numero}".format( ddd = telefone[0:2], numero = telefone[2:] ) def data_situacao(self, data, props) : return self.convert_date(data, '%d/%m/%Y') #date = datetime.strptime(data,'%d/%M/%Y') #return date.strftime("%Y%M%d") def ultima_atualizacao(self, data, props) : return self.convert_date(data, '%Y-%m-%dT%H:%M:%S.%fz') def abertura(self, data, props): return self.convert_date(data, '%d/%m/%Y') def convert_date(self, str, dt_to, dt_from="%Y%m%d"): #data2 = datetime.strptime(str,dt_to) #data3 = datetime.date(data2) #return data3.strftime(dt_from) while True: try: data2 = datetime.strptime(str,dt_to) data3 = datetime.date(data2) return data3.strftime(dt_from) except ValueError: return "00000000" def cep(self, data, props) : return data.replace(".","").replace("-","") def apenas_numeros(self, data, props): return re.sub(r'[^\d]','',data) class gn_jsonpath(): def __init__(self, **data): self.trim = data.get('trim', False) self.path = re.sub( r'[\s\t\r\n](?=(\`[^\`]*\`|[^\`])*$)', "", data['path'] ) self.jsonIn = data['json'] self.value = list() self.error_path = list() self.process() self.current_path = '' # print(self.value) def __repr__(self): return self.value def process(self): #self.all = self.processPath(self.path, "||") #_filter = self.onlyValidValues(self.all) self.all , _filter = self.processByLogic(self.path, "||") # To return only first value valid in || self.value = _filter[0] if _filter else '' """ if _filter: self.value = else: self.value = '' """ def processByLogic(self,path, operator) : paths_processed = self.processPath(path, operator) filter_done = self.onlyValidValues(paths_processed) return paths_processed, filter_done def processPath(self, path, separator): paths = path.split(separator) return list(map(self.execute, paths)) def onlyValidValues(self, values): """ Metodo para retornar valores verdadeiros """ #flt = list(filter(lambda value: value is not None and value != 0, values)) #print(values) return [value for value in values if value or value == 0 or type(value) == str and re.match(r'^\s+$', value)] def execute(self, path): # save path to show log. self.current_path = path value = '' try: if "&&" in path: #plus = self.processPath(path, "&&") #_filter = self.onlyValidValues(plus) plus, _filter = self.processByLogic(path, "&&") _filter = self.onlyValidValues(plus) value = "".join( map(lambda value: value if not self.trim else value.strip(), _filter) ) if len(_filter) == len(plus) else '' #value = "".join(map(str.strip, _filter)) else: value = self.goThroughPath(path) except ValueError as error: print("error: " + error) return value def goThroughPath(self, path): has_quote = re.findall(r'\`(.*?)\`', path) if not "/" in path and len(has_quote): return has_quote[0] path_splited = path.split("/") current_value = self.tryPath(self.jsonIn, path_splited[0]) # if not isinstance(current_value, dict) and not isinstance(current_value, list): # current_value = current_value # else: return self.getValue(path_splited, current_value) # return current_value def isnumber(self, value): try: float(value) except ValueError: return False return True def getValue(self, path_splited, initial_value): value = initial_value for index, key in enumerate(path_splited[1:]): try: key = int(key) except: pass if ( key != '' and not isinstance(key, int) and ("*" in key[0] or ("[" in key and "]" in key))) and index != len(path_splited): _charToJoin = re.findall(r"\`(.*?)\`", key) _charToJoin = _charToJoin[0] if _charToJoin else " " key1 = None key2 = None keyS = re.findall(r'\[(.*?)\]', key) if keyS: keyS = keyS[0].split(":") key1, key2 = self.tryInt(keyS[0]), self.tryInt(keyS[1]) value = self.onlyValidValues( [self.getValue(path_splited[index + 1:], val) for val in value]) return _charToJoin.join(map(str.strip, value[key1: key2])) else: value = self.tryPath(value, key) return value def tryInt(self, value): try: return int(value) except: return None def tryPath(self, arr, key): try: return arr[key] except: self.error_path.append(f"PATH:{self.current_path} - KEY: {key or None}") """print( ###### path pode estar errado ####### PATH : {path} KEY : {key} .format(path=self.current_path, key=key))""" return '' if __name__ == "__main__" : import sys cnpj = getIndex(sys.argv,1,"") arquivo_saida = getIndex(sys.argv,2,False) Receita(cnpj = cnpj, arquivo_saida=arquivo_saida)