from generic.gn_request import Request # from generic.gn_argo import gn_argo # from datetime import datetime import re import urllib3 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) #https://api-hml.useargo.com/oauth/access-token/ - Url para Autenticacao Homologacao #https://api.useargo.com/oauth/access-token/ - Url para Autenticacao Producao #https://api-hml.useargo.com/argo/v2/ - Servidor de Homologacao #https://api.useargo.com/argo/v2/ - Servidor de Producao #client_id = "934a8cd3-af2a-468e-af8e-62821a83a06c" #client_secret = "edf9ded9-e252-4165-afde-3443a10d7eaa" class argo(gn_argo): def __init__(self, argv:list): self.date_now_formatted = datetime.today().strftime('%Y-%m-%d') self.response = None self.db = False self.output_count = 0 self.suffix = "generic_" self.unparsed_data = {} self.setup(argv) self.rote_handler() client_id = self.get_token('ClientID') client_secret = self.get_token('ClientSecret') client_username = self.get_token('username') client_password = self.get_token('password') client_reseller = self.get_token('reseller') client_corp = self.get_token('corp') client_grant_type = self.get_token('grant_type') ambiente_api = self.get_token_data('ambiente_api') environment = '-hml' if ambiente_api == 'H' else '' urls = self.get_url_environment(environment=environment) auth = self.generateOAuth2Token(urls.get('oAuth2'), client_id, client_secret,client_username,client_password,client_reseller,client_corp,client_grant_type ) # temp_token = '060beff6-e8d9-47ce-9a31-4a7dd29f54b7' self.argo_request = Request( base_url = urls.get('base_url') , params = { **self.get_token('access_token') }, verify= False, headers={ 'Content-Type': 'application/json', 'access_token': f"{auth.get('access_token')}", # 'access_token': f"{temp_token}", 'client_id': f"{self.environment_config.get('tokens',{0}).get('ClientID')}" } ) super(argo, self, ).__init__(argv) print('Fim') def get_url_environment(self, environment='homolog'): # PAREI AQUI base = f'{environment}' return { "oAuth2" :f"https://api{base}.useargo.com/oauth/access-token", "base_url" :f"https://api{base}.useargo.com/argo/v2/" } def generateOAuth2Token(self, url, client_id, client_secret,client_username,client_password,client_reseller,client_corp,client_grant_type): # """Api do banco do brasil requeer autenticacao OAuth2 - # Args: # url (str): endpoit de authenticacao # client_id (str): cliente id fornecido dentro do do portal BB # client_secret (str): cliente secret fornecido dentro do portal BB # """ import base64 import json as jsonDump Authorization = "Basic "+base64.b64encode( bytes('{ClientID}:{ClientSecret}'.format( **client_id, **client_secret ), encoding='utf8' ) ).decode("utf-8") #payload = jsonDump.dumps('"username":"{username}","password":"{password}","reseller":"{reseller}","corp":"{corp}","grant_type":"{grant_type}"'.format( # **client_username,**client_password,**client_reseller,**client_corp,**client_grant_type)) payload = jsonDump.dumps({ "username" : "{username}".format(**client_username), "password" : "{password}".format(**client_password), "reseller" : "{reseller}".format(**client_reseller), "corp" : "{corp}".format(**client_corp), "grant_type": "{grant_type}".format(**client_grant_type) }) OAuth2 = Request( #auth = (client_id, client_secret) method='POST', verify=False, headers = { "Content-Type":"application/json" ,"Authorization": Authorization }, data=payload ).doRequest(url=url+"/"); if not OAuth2.ok: self.createLog('APIERROR',f"ERRO ao criar token - Necessario criacao de novo token {OAuth2.text}") exit(1) return OAuth2.json() def post_route_handler_boletos(self): #""" Rota de inclusao e alteracao de boleto tratamento de envio""" data = self.txt_to_dict() for request in data.get('C'): self.unparsed_data = request request_structure = self.atualWs.get('request') data_parsed = self.parser(request, request_structure) #data_parsed = self.parser(request, request_structure) self.handle_send(data_parsed) def saida_requests_requestId_flights_quotations(self, dados): text = '' flag = "w+" if not dados and not self.response.ok: status = self.response.status_code erros = (self.tryJson(self.response) or {}).get('erros', []) #erros = (self.get_index(erros,0) or {}).get('textoMensagem', 'houve um erro durante a execucao') erros = (self.get_index(erros,0) or {}).get('textoMensagem', self.response.content.decode("utf-8")) text = f"ER|{status}-{erros}|" else: try: #dados = dados[-10:] texts = [] for index, dado in enumerate(dados): #is_new_line = "\n" if (index + 1) != total else "" flag = "w+" if not index else "a+" linha = [] linha.append(dado.get('webServiceId')) linha.append(dado.get('locator')) linha.append(dado.get('ticket')) linha.append(dado.get('totalValue')) linha.append(dado.get('approvedTotalValue')) voos = dado.get('airQuotedsLegs') linha.append(voos[0].get('departure')) linha.append(voos[0].get('departureDate')) linha.append(voos[1].get('departure')) linha.append(voos[1].get('departureDate')) linha.append(dado.get('observation').replace("\n", "@@")) texts.append("OK||" + ("|".join(str(campos) for campos in linha))) text = "\n".join(texts) except Exception as er: text = f"ER|{str(er)}|" self.write_file(text , self.saida_cobol ,flag= flag, encoding='iso-8859-1') def saida_requests_requestId_rentals_quotations(self, dados): text = '' flag = "w+" if not dados and not self.response.ok: status = self.response.status_code erros = (self.tryJson(self.response) or {}).get('erros', []) #erros = (self.get_index(erros,0) or {}).get('textoMensagem', 'houve um erro durante a execucao') erros = (self.get_index(erros,0) or {}).get('textoMensagem', self.response.content.decode("utf-8")) text = f"ER|{status}-{erros}|" else: try: #dados = dados[-10:] texts = [] for index, dado in enumerate(dados): #is_new_line = "\n" if (index + 1) != total else "" flag = "w+" if not index else "a+" linha = [] linha.append(dado.get('quotationId')) linha.append(dado.get('carRental')) linha.append(dado.get('carRentalIata')) linha.append(dado.get('vehicle')) linha.append(dado.get('pickUpAdress')) linha.append(dado.get('totalValue')) linha.append(dado.get('approvedTotalValue')) texts.append("OK||" + ("|".join(str(campos) for campos in linha))) text = "\n".join(texts) except Exception as er: text = f"ER|{str(er)}|" self.write_file(text , self.saida_cobol ,flag= flag, encoding='iso-8859-1') def saida_requests_requestId_hotels_quotations(self, dados): text = '' flag = "w+" if not dados and not self.response.ok: status = self.response.status_code erros = (self.tryJson(self.response) or {}).get('erros', []) #erros = (self.get_index(erros,0) or {}).get('textoMensagem', 'houve um erro durante a execucao') erros = (self.get_index(erros,0) or {}).get('textoMensagem', self.response.content.decode("utf-8")) text = f"ER|{status}-{erros}|" else: try: #dados = dados[-10:] texts = [] for index, dado in enumerate(dados): #is_new_line = "\n" if (index + 1) != total else "" flag = "w+" if not index else "a+" linha = [] linha.append(dado.get('quotationId')) linha.append(dado.get('hotel')) linha.append(dado.get('city')) linha.append(dado.get('state')) dadosHotel = dado.get('hotelData') if not dadosHotel : linha.append(dado.get('localization')) else : linha.append(dadosHotel.get('address')) linha.append(dado.get('voucherDate')) linha.append(dado.get('totalValue')) linha.append(dado.get('approvedTotalValue')) texts.append("OK||" + ("|".join(str(campos) for campos in linha))) text = "\n".join(texts) except Exception as er: text = f"ER|{str(er)}|" self.write_file(text , self.saida_cobol ,flag= flag, encoding='iso-8859-1') """ :'######:::::'###::::'####:'########:::::'###:::::::'########::'########::'#######::'##::::'##:'########::'######::'########::'######:: '##... ##:::'## ##:::. ##:: ##.... ##:::'## ##:::::: ##.... ##: ##.....::'##.... ##: ##:::: ##: ##.....::'##... ##:... ##..::'##... ##: ;##:::..:::'##:. ##::: ##:: ##:::: ##::'##:. ##::::: ##:::: ##: ##::::::: ##:::: ##: ##:::: ##: ##::::::: ##:::..::::: ##:::: ##:::..:: . ######::'##:::. ##:: ##:: ##:::: ##:'##:::. ##:::: ########:: ######::: ##:::: ##: ##:::: ##: ######:::. ######::::: ##::::. ######:: :..... ##: #########:: ##:: ##:::: ##: #########:::: ##.. ##::: ##...:::: ##:'## ##: ##:::: ##: ##...:::::..... ##:::: ##:::::..... ##: '##::: ##: ##.... ##:: ##:: ##:::: ##: ##.... ##:::: ##::. ##:: ##::::::: ##:.. ##:: ##:::: ##: ##:::::::'##::: ##:::: ##::::'##::: ##: . ######:: ##:::: ##:'####: ########:: ##:::: ##:::: ##:::. ##: ########:: ##### ##:. #######:: ########:. ######::::: ##::::. ######:: :......:::..:::::..::....::........:::..:::::..:::::..:::::..::........:::.....:..:::.......:::........:::......::::::..::::::......::: """ def saveRequestIdInHistory(self, request_id): with open(f"/tmp/requests_ids_history_{self.date_now_formatted}.tmp", "a+") as f: f.write(f"{request_id}\n") f.close() def searchRequestIdInHistory(self, request_id): try: with open(f"/tmp/requests_ids_history_{self.date_now_formatted}.tmp", "r+") as f: data = f.read().split("\n") if request_id in data: return True else: return False except: return False def saida_requests(self, dados): text = '' flag = "w+" fixed_tarefa = self.rote_data['path']['os_vinculada']; requisicoes = 1 requisicoes_restantes = '9999' try: if 'x-ratelimit-remaning' in self.response.headers: requisicoes_restantes = self.response.headers["x-ratelimit-remaning"] except: pass print("Requisicoes restantes: " + requisicoes_restantes) if not dados and not self.response.ok: status = self.response.status_code erros = (self.tryJson(self.response) or {}).get('erros', []) #erros = (self.get_index(erros,0) or {}).get('textoMensagem', 'houve um erro durante a execucao') erros = (self.get_index(erros,0) or {}).get('textoMensagem', self.response.content.decode("utf-8")) text = f"ER|{status}-{erros}|" else: try: #dados = dados[-10:] texts = [] for index, dado in enumerate(dados): #is_new_line = "\n" if (index + 1) != total else "" # flag = "w+" if not index else "a+" amountFlights = dado.get('amountFlights') amountHotels = dado.get('amountHotels') amountRentals = dado.get('amountRentals') requestId = dado.get('requestId') requestNumber = dado.get('requestNumber') if self.searchRequestIdInHistory(requestId): continue if( (amountFlights and amountFlights == 0) and (amountHotels and amountHotels == 0) and (amountRentals and amountRentals == 0) ): continue managementfieldsGet = self.argo_request.doRequest( rote = f"requests/{requestId}/managementfields" , method = "GET" ) requisicoes += 1 if(managementfieldsGet.ok): managementsFields = managementfieldsGet.json() for index, mf_dados in enumerate(managementsFields): if(mf_dados['managementFieldId'] == "ODS_PROJETO"): ods_project_data = re.findall(r"[0-9]+", mf_dados['content']) ods_project_data_number = list(map(int, ods_project_data)) if(len(ods_project_data_number) > 0 and (ods_project_data_number[0] == int(fixed_tarefa)) ): #print(managementsFields) self.saveRequestIdInHistory(requestId) if(amountFlights and amountFlights > 0): flightsGet = self.argo_request.doRequest( rote = f"requests/{requestId}/flights/quotations" , method = "GET" ) requisicoes += 1 if(flightsGet.ok): flightsFields = flightsGet.json() tmp_txts = '' try: #dados = dados[-10:] texts = [] for index, flightsDado in enumerate(flightsFields): #is_new_line = "\n" if (index + 1) != total else "" linha = [] linha.append(requestNumber) linha.append(requisicoes_restantes) linha.append(flightsDado.get('webServiceId')) linha.append(flightsDado.get('locator')) linha.append(flightsDado.get('ticket')) linha.append(flightsDado.get('totalValue')) linha.append(flightsDado.get('approvedTotalValue')) voos = flightsDado.get('airQuotedsLegs') linha.append(voos[0].get('departure')) linha.append(voos[0].get('departureDate')) if len(voos) > 1: linha.append(voos[1].get('departure')) linha.append(voos[1].get('departureDate')) linha.append(flightsDado.get('observation').replace("\n", "@@")) texts.append("OK||F|" + ("|".join(str(campos) for campos in linha)) + "\n") tmp_txts = "".join(texts) except Exception as er: tmp_txts = f"ER|{str(er)}|" self.write_file(tmp_txts , self.saida_cobol ,flag= 'a+', encoding='iso-8859-1') #print(flightsFields) if(amountHotels and amountHotels > 0): hotelsGet = self.argo_request.doRequest( rote = f"requests/{requestId}/hotels/quotations" , method = "GET" ) requisicoes += 1 if(hotelsGet.ok): hotelsFields = hotelsGet.json() tmp_txts = '' try: #dados = dados[-10:] texts = [] for index, hotelsDado in enumerate(hotelsFields): #is_new_line = "\n" if (index + 1) != total else "" flag = "a+" linha = [] linha.append(requestNumber) linha.append(requisicoes_restantes) linha.append(hotelsDado.get('quotationId')) linha.append(hotelsDado.get('hotel')) linha.append(hotelsDado.get('city')) linha.append(hotelsDado.get('state')) dadosHotel = hotelsDado.get('hotelData') if not dadosHotel : linha.append(hotelsDado.get('localization')) else : linha.append(dadosHotel.get('address')) linha.append(hotelsDado.get('voucherDate')) linha.append(hotelsDado.get('totalValue')) linha.append(hotelsDado.get('approvedTotalValue')) texts.append("OK||H|" + ("|".join(str(campos) for campos in linha))) tmp_txts = "|".join(texts) + "\n" except Exception as er: tmp_txts = f"ER|{str(er)}|" self.write_file(tmp_txts , self.saida_cobol ,flag= 'a+', encoding='iso-8859-1') #print(hotelsFields) if(amountRentals and amountRentals > 0): rentalsGet = self.argo_request.doRequest( rote = f"requests/{requestId}/rentals/quotations" , method = "GET" ) requisicoes += 1 if(rentalsGet.ok): rentalsFields = rentalsGet.json() tmp_txts = '' try: texts = [] for index, rentalsDado in enumerate(rentalsFields): linha = [] linha.append(requestNumber) linha.append(requisicoes_restantes) linha.append(rentalsDado.get('quotationId')) linha.append(rentalsDado.get('carRental')) linha.append(rentalsDado.get('carRentalIata')) linha.append(rentalsDado.get('vehicle')) linha.append(rentalsDado.get('pickUpAdress')) linha.append(rentalsDado.get('totalValue')) linha.append(rentalsDado.get('approvedTotalValue')) texts.append("OK||R|" + ("|".join(str(campos) for campos in linha))) tmp_txts = "|".join(texts) + "\n" except Exception as er: tmp_txts = f"ER|{str(er)}|" self.write_file(tmp_txts , self.saida_cobol ,flag= 'a+', encoding='iso-8859-1') #print(rentalsFields) else: continue else: continue # response_first = self.handle_send({ # rote = 'requests/{requestId}/managementfields?;path->requestId=' + requestId # }) # if(amountFlights > 0): # linha = [] # linha.append(dado.get('amountFlights')) # linha.append(dado.get('amountHotels')) # linha.append(dado.get('amountRentals')) # linha.append(dado.get('requestId')) # linha.append(dado.get('requestNumber')) # linha.append(dado.get('tripStatus')) # linha.append(dado.get('expenseStatus')) # texts.append("OK||" + ("|".join(str(campos) for campos in linha))) # text = "\n".join(texts) except Exception as er: text = f"ER|{str(er)}|" if text != '': self.write_file(text , self.saida_cobol ,flag= flag, encoding='iso-8859-1') else: if self.read_file(self.saida_cobol, 'r', 'iso-8859-1') == None or self.read_file(self.saida_cobol, 'r', 'iso-8859-1') == "": self.write_file("OK||X||{}|".format(requisicoes_restantes), self.saida_cobol, flag=flag, encoding='iso-8859-1') print (f"Nro de Requisicoes executadas: {requisicoes}") def saida_requests_requestId_managementfields(self, dados): text = '' flag = "w+" if not dados and not self.response.ok: status = self.response.status_code erros = (self.tryJson(self.response) or {}).get('erros', []) #erros = (self.get_index(erros,0) or {}).get('textoMensagem', 'houve um erro durante a execucao') erros = (self.get_index(erros,0) or {}).get('textoMensagem', self.response.content.decode("utf-8")) text = f"ER|{status}-{erros}|" else: try: #dados = dados[-10:] texts = [] for index, dado in enumerate(dados): #is_new_line = "\n" if (index + 1) != total else "" flag = "w+" if not index else "a+" linha = [] linha.append(dado.get('managementFieldId')) linha.append(dado.get('content')) texts.append("OK||" + ("|".join(str(campos) for campos in linha))) text = "\n".join(texts) except Exception as er: text = f"ER|{str(er)}|" self.write_file(text , self.saida_cobol ,flag= flag, encoding='iso-8859-1') def saida_managementfields(self, dados): text = '' flag = "w+" if not dados and not self.response.ok: status = self.response.status_code erros = (self.tryJson(self.response) or {}).get('erros', []) #erros = (self.get_index(erros,0) or {}).get('textoMensagem', 'houve um erro durante a execucao') erros = (self.get_index(erros,0) or {}).get('textoMensagem', self.response.content.decode("utf-8")) text = f"ER|{status}-{erros}|" else: try: #dados = dados[-10:] texts = [] for index, dado in enumerate(dados): #is_new_line = "\n" if (index + 1) != total else "" flag = "w+" if not index else "a+" linha = [] linha.append(dado.get('fieldId')) linha.append(dado.get('clientId')) linha.append(dado.get('name')) texts.append("OK||" + ("|".join(str(campos) for campos in linha))) text = "\n".join(texts) except Exception as er: text = f"ER|{str(er)}|" self.write_file(text , self.saida_cobol ,flag= flag, encoding='iso-8859-1') # """ Rota para alteracao de boletodos tratamento de envio""" def patch_route_handler_boletos_id(self): self.post_route_handler_boletos() def handle_output_boletos_id(self, data): self.handle_output_boletos(data) # """ Rota para get dos dados da requisicao """ def beforeGet_requests_requestId_managementfields(self): self.unparsed_data = self.rote_data['path'] self.handle_send(self.rote_data['path']) def beforeGet_requests_requestId_flights_quotations(self): self.unparsed_data = self.rote_data['path'] self.handle_send(self.rote_data['path']) def beforeGet_requests_requestId_rentals_quotations(self): self.unparsed_data = self.rote_data['path'] self.handle_send(self.rote_data['path']) def beforeGet_requests_requestId_hotels_quotations(self): self.unparsed_data = self.rote_data['path'] self.handle_send(self.rote_data['path']) def XXXhandle_output_requests_requestId_managementfields(self, data): self.users_handler(data) """ def db_data_boletos(self, data): rote_filter = self.rote_data.get('filter') if(rote_filter): return list(filter(lambda item: int(item.get('nro_remessa')) == int(rote_filter.get('remessa')), data)) return None def db_data_boletos_id(self, data): return self.db_data_boletos(data) """ def format_date_bb(self, value): """ Metodo para formatacao de dadas para o BB""" if value: date = datetime.strptime(value,"%Y-%m-%d") return datetime.strftime(date,"%d.%m.%Y") return "" def custom_flag_status(self, value, key): return "OK" if not value else "ER" #"Metodos com valores padrao" def custom_default_value(self, data, item): default_values = { "numero" : 'numeroTituloCliente' , "numeroCarteira" : 'numeroCarteira' , "numeroVariacaoCarteira" : 'numeroVariacaoCarteira' , "numeroTituloBeneficiario" : 'numeroTituloBeneficiario' } return data or self.current_data.get(default_values.get(item.get('from',''),''),'') def custom_default_value_patch_boletos(self, data, item): default_values = { "nosso_nro" : 'nosso_nro/valor', "nro_titulo_benef" : 'nro_titulo_benef/valor', } return data or self.jsonValueByPath(default_values.get(item.get('from','')), self.unparsed_data) #self.file_data.get(default_values.get(key.get('from',''),''),'') def custom_default_value_post_boletos_baixar(self, data, item): default_values = { "numeroConveio" : 'numeroConveio', } return data or self.jsonValueByPath(default_values.get(item.get('from','')), self.unparsed_data) #self.file_data.get(default_values.get(key.get('from',''),''),'') def XXpagination_rules(self, response, data): #'pedidos/?status=2&alterado_apos=2021-07-12 15:41:29' if response.ok: body = self.tryJson(response) if body.get('indicadorContinuidade') == 'S': query = self.rote_data.get('query') query['indice'] = body.get('proximoIndice') params = self.mountRouteParams(query) return f"usuarios?{params}" return None return None def XXafterGet_users(self, data): return data.get('users',[]) def XafterGet_users(self, data): return data.get('result',{}).get('entityList',[]) #return data.get('result',[]) #TRATAMENTO DO GET def users_handler(self, dados): if 'requests_requestId_flights_quotations' in self.fixed_rote(): self.saida_requests_requestId_flights_quotations(dados) return if 'requests_requestId_rentals_quotations' in self.fixed_rote(): self.saida_requests_requestId_rentals_quotations(dados) return if 'requests_requestId_hotels_quotations' in self.fixed_rote(): self.saida_requests_requestId_hotels_quotations(dados) return if 'requests_requestId_managementfields' in self.fixed_rote(): self.saida_requests_requestId_managementfields(dados) return if 'requests' in self.fixed_rote(): self.saida_requests(dados) return if 'managementsfields' in self.fixed_rote(): self.saida_managementfields(dados) return self.get_users_handler(dados) def get_users_handler(self, dados): text = '' flag = "w+" if not dados and not self.response.ok: status = self.response.status_code erros = (self.tryJson(self.response) or {}).get('erros', []) #erros = (self.get_index(erros,0) or {}).get('textoMensagem', 'houve um erro durante a execucao') erros = (self.get_index(erros,0) or {}).get('textoMensagem', self.response.content.decode("utf-8")) text = f"ER|{status}-{erros}|" else: try: #dados = dados[-10:] texts = [] for index, dado in enumerate(dados): #is_new_line = "\n" if (index + 1) != total else "" flag = "w+" if not index else "a+" texts.append("OK||" + ("|".join(str(dado) for dado in dados ))) text = "\n".join(texts) except Exception as er: text = f"ER|{str(er)}|" self.write_file(text , self.saida_cobol ,flag= flag, encoding='iso-8859-1') """ .########.####.##.......########.########...#######.....########.....###.......########...#######..########....###... .##........##..##..........##....##.....##.##.....##....##.....##...##.##......##.....##.##.....##....##......##.##.. .##........##..##..........##....##.....##.##.....##....##.....##..##...##.....##.....##.##.....##....##.....##...##. .######....##..##..........##....########..##.....##....##.....##.##.....##....########..##.....##....##....##.....## .##........##..##..........##....##...##...##.....##....##.....##.#########....##...##...##.....##....##....######### .##........##..##..........##....##....##..##.....##....##.....##.##.....##....##....##..##.....##....##....##.....## .##.......####.########....##....##.....##..#######.....########..##.....##....##.....##..#######.....##....##.....## """ def rote_filter_users(self) : # somente usuarios ativos queryParams = self.mountRouteParams( { "actives" : "true" #, "paramFilter" : paramFilter.replace('\n','').replace(" ","") }) return queryParams def rote_filter_requests(self) : query = self.rote_data.get('query') if query: return self.mountRouteParams(query) return None def rote_filter_boletos(self,): query = self.rote_data.get('query') if query: return self.mountRouteParams(query) return None def rote_filter_boletos_id(self,): return self.rote_filter_boletos() #"Tratamento de retorno do GET" def usuarios_id_handler(self, usuario): text = "OK||" + ("|".join(str(dado) for dado in usuario.values())) + "\n" self.write_file(text , self.saida_cobol ,flag= "w+", encoding='iso-8859-1' ) def format_rote_users_id(self, rote): path_params = self.rote_data.get('path') if path_params: return rote.format(**path_params) return None def format_rote_requests_requestId_managementfields(self, rote): path_params = self.rote_data.get('path') if path_params: return rote.format(**path_params) return None def format_rote_requests_requestId_flights_quotations(self, rote): path_params = self.rote_data.get('path') if path_params: return rote.format(**path_params) return None def format_rote_requests_requestId_rentals_quotations(self, rote): path_params = self.rote_data.get('path') if path_params: return rote.format(**path_params) return None def format_rote_requests_requestId_hotels_quotations(self, rote): path_params = self.rote_data.get('path') if path_params: return rote.format(**path_params) return None #! NÃO SERA PRECISO GERAR O QRCODE POR AQUI """ def gerarCrc16(self, payload): crc16 = crcmod.mkCrcFun(poly=0x11021, initCrc=0xFFFF, rev=False, xorOut=0x0000) self.crc16Code = hex(crc16(str(payload).encode('utf-8'))) self.crc16Code_formatado = str(self.crc16Code).replace('0x', '').upper() self.payload_completa = f'{payload}{self.crc16Code_formatado}' self.gerarQrCode(self.payload_completa) def gerarQrCode(self, payload): import qrcode self.qrcode = qrcode.make(payload) self.qrcode.save('logs/pixqrcode.png') return print(payload) """