import time from generic.gn_request import Request # from generic.gn_unicred import gn_unicred # from datetime import datetime import json import urllib3 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) class cobranca_unicred(gn_unicred): def __init__(self, argv:list): self.response = None self.db = False self.output_count = 0 self.suffix = "app_generic" self.unparsed_data = {} self.setup(argv) self.rote_handler() nomeUsuario = self.get_token('nomeUsuario') senha = self.get_token('senha') apiKey = self.get_token('apiKey') cooperativa = self.get_token('cooperativa') IdBeneficiario = self.get_token('IdBeneficiario') # ClientID = self.get_token('ClientID') # ClientSecret = self.get_token('ClientSecret') environment = 'homolog' if self.environment_config.get('ambiente') == 'H' else '' urls = self.get_url_environment(environment=environment) auth = self.generateOAuth2TokenSample(urls.get('oAuth2'), apiKey, nomeUsuario, senha) # auth = self.generateOAuth2Token(urls.get('oAuth2'), ClientID, ClientSecret ) self.cobranca_unicred_request = Request( base_url = urls.get('base_url') , params = {}, verify= False, headers={ **apiKey, **cooperativa, 'Authorization': f"Bearer {auth.get('accessToken')}", 'content-type': 'application/json' } ) super(cobranca_unicred, self, ).__init__(argv) print('Fim') def get_url_environment(self, environment='homolog'): # PAREI AQUI base = f'{environment}' if environment == 'homolog': base = f"/{base}" return { "oAuth2" :f"https://api.e-unicred.com.br{base}/oauth2/v2/grant-token", "base_url" :f"https://api.e-unicred.com.br{base}/cobranca/v2/" } def generateOAuth2TokenSample(self, url, apiKey, nomeUsuario, senha): headers = { 'apiKey': apiKey['apiKey'], "Content-Type" : "application/json", } data = { 'nomeUsuario': nomeUsuario['nomeUsuario'], 'senha': senha['senha'] } token = Request( #auth = (client_id, client_secret) method='POST', # verify=False, headers = headers, data = json.dumps(data) ).doRequest(url=url); if not token.ok: self.createLog('APIERROR',f"ERRO ao criar token - Necessario criacao de novo token {token.text}") exit(1) return token.json() def generateOAuth2Token(self, url, client_id, client_secret): import base64 Authorization = "Basic "+base64.b64encode( bytes('{client_id}:{client_secret}'.format( **client_id, **client_secret ), encoding='utf8' ) ).decode("utf-8") OAuth2 = Request( #auth = (client_id, client_secret) method='POST', verify=False, headers = { "Authorization": Authorization, "Content-Type":"application/x-www-form-urlencoded", }, data= { "grant_type": "client_credentials", "scope": "cobrancas.boletos-requisicao cobrancas.boletos-info" } ).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 get_route_handler_beneficiarios_id_beneficiario_titulos_titulo_status(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) self.dados_requisicao = request #data_parsed = self.parser(request, request_structure) self.handle_send(data_parsed) # ------------------------------------------------------------------# # # # ROTA DE CRIAÇÃO DE BOLETOS # # # # ------------------------------------------------------------------# def post_route_handler_beneficiarios_id_beneficiario_titulos(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) self.dados_requisicao = request #data_parsed = self.parser(request, request_structure) self.handle_send(data_parsed) def beforeSend_beneficiarios_id_beneficiario_titulos(self, data): # INCLUSAO DO ID BENEFICIARIO NA ROTA token = self.get_token('IdBeneficiario') rota_atual = self.atualWs['rote'] self.current_rote = rota_atual.replace('id_beneficiario', token['IdBeneficiario']) def buscar_dados_extras(self, id_beneficiario, id_titulo, parametros): url = f"beneficiarios/{id_beneficiario}/titulos/{id_titulo}/status" boleto = self.cobranca_unicred_request.doRequest( rote=url, method="GET" ) if boleto.ok and boleto.status_code == 200: boleto = json.loads(boleto.text) # print(json.dumps(boleto['headers'])) parametros = parametros.replace("LINHA_DIGITAVEL", boleto['linhaDigitavel']) parametros = parametros.replace("COD_BARRA_NUM", boleto['codBarras']) parametros = parametros.replace("QR_EMV", boleto['qrCodePix']) return parametros else: return False def handle_output_beneficiarios_id_beneficiario_titulos(self, data): #"""Tratamento de retorno dos metodos PATCH e POST de boletodos""" idBenef = self.get_token('IdBeneficiario') id_beneficiario = idBenef['IdBeneficiario'] status = "OK" if self.api_request.req.status_code == 200 else "ER" desc_status = "Enviado com sucesso" extra_err_status = "" id_titulo = self.api_request.req.text if self.api_request.req.status_code == 200 else "" if self.api_request.req.status_code != 200: retorno = json.loads(self.api_request.req.text) desc_status = retorno['message'] try: extra_err_status = " - {}".format(retorno["body"][0]['message']) except: extra_err_status = "" nossoNumero = self.dados_requisicao.get('nosso_nro')['valor'] beneficiarioVariacaoCarteira = self.dados_requisicao.get('benef_var')['valor'] tituloBenef = self.dados_requisicao.get('tit_benef')['valor'] structure = f"{status}|{desc_status}{extra_err_status}|{nossoNumero}|||{tituloBenef}|{beneficiarioVariacaoCarteira}|LINHA_DIGITAVEL|COD_BARRA_NUM|||||QR_EMV|{id_titulo}|\n" if status == "OK": buscar_dados = self.buscar_dados_extras(id_beneficiario, id_titulo, structure) tentativas = 0 while buscar_dados == False and tentativas < 6: time.sleep(8) buscar_dados = self.buscar_dados_extras(id_beneficiario, id_titulo, structure) tentativas += 1 print(f"|> Tentativas de busca: {tentativas}") if buscar_dados != False: structure = buscar_dados else: structure = structure.replace("LINHA_DIGITAVEL", "") structure = structure.replace("COD_BARRA_NUM", "") structure = structure.replace("QR_EMV", "") else: structure = structure.replace("LINHA_DIGITAVEL", "") structure = structure.replace("COD_BARRA_NUM", "") structure = structure.replace("QR_EMV", "") flag = "w+" if self.output_count == 0 else "a+" self.write_file(structure, self.saida_cobol ,flag= flag, encoding='iso-8859-1' ) self.output_count += 1 def custom_flag_status(self, value, key): return "OK" if not value else "ER" def convert_to_boolean(self, value): return True if value == "S" else False def pagination_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"boletos?{params}" return None return None def beforeSend_beneficiarios_id_beneficiario_titulos_id_titulo_status(self, data): # INCLUSAO DO ID BENEFICIARIO NA ROTA token = self.get_token('IdBeneficiario') rota_atual = self.atualWs['rote'] dados = self.unparsed_data id_titulo = dados.get('id_titulo')['valor'] rota_atual = rota_atual.replace('id_beneficiario', token['IdBeneficiario']) rota_atual = rota_atual.replace('id_titulo', id_titulo) self.current_rote = rota_atual #TRATAMENTO DO GET def beneficiarios_id_beneficiario_titulos_id_titulo_status_handler(self, boletos): data = self.txt_to_dict() for request in data.get('C'): self.unparsed_data = request self.handle_send({}) def handle_output_beneficiarios_id_beneficiario_titulos_id_titulo_status(self, data): #"""Tratamento de retorno dos metodos PATCH e POST de boletodos""" # '{"codBarras":"13699958100000150004371000034847300000400335","linhaDigitavel":"13694371060003484730100004003356995810000015000","nossoNumero":"00000400335","dataDeVencimento":"2023-12-31","valor":150.00,"status":"ABERTO","motivo":null,"qrCodePix":"00020101021226940014br.gov.bcb.pix2572qrcode-h.unicred.com.br/pix/v1/cobv/4411730a-8753-41f5-aee0-21facba764ad5204000053039865802BR5917Cooperado Joao5786007CHAPECO61088981046062070503***630467EE","pagamentos":[]}' status = "OK" if self.api_request.req.status_code == 200 else "ER" desc_status = "Consultado com sucesso" if self.api_request.req.status_code != 200: retorno = json.loads(self.api_request.req.text) desc_status = f"{self.api_request.req.status_code} - {retorno['message']}" estrutura = f"{status}|{desc_status}||||||||||\n" if(self.api_request.req.ok): dados = json.loads(self.api_request.req.text) if len(dados['pagamentos']) > 0: pag_data = dados['pagamentos'][0]['data'] pag_valor = dados['pagamentos'][0]['valor'] dados['pag_data'] = pag_data.replace('-', "") dados['pag_valor'] = pag_valor else: dados['pag_data'] = "" dados['pag_valor'] = "" dados['motivo'] = dados['motivo'] if dados['motivo'] else "" dados['dataDeVencimento'] = dados['dataDeVencimento'].replace('-', "") estrutura = f"{status}|{desc_status}|{dados['codBarras']}|{dados['linhaDigitavel']}|{dados['nossoNumero']}|{dados['dataDeVencimento']}|{dados['valor']}|{dados['status']}|{dados['motivo']}|{dados['qrCodePix']}|{dados['pag_data']}|{dados['pag_valor']}\n" flag = "w+" if self.output_count == 0 else "a+" self.write_file(estrutura, self.saida_cobol ,flag= flag, encoding='iso-8859-1' ) self.output_count += 1