import time, decimal, re from generic.gn_request import Request # from generic.gn_santander import gn_santander # from datetime import datetime import json import urllib3 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) class cobranca_santand(gn_santander): 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() clientID = self.get_token_data('ClientID') environment = 'homolog' if self.environment_config.get('ambiente') == 'H' else '' urls = self.get_url_environment(environment=environment) auth = self.generateOAuth2Token(urls.get('oAuth2')) cert = ( self.get_token_data('Certificado_certKey'), self.get_token_data('Certificado_priKey') ) self.cobranca_santand_request = Request( base_url = urls.get('base_url') , params = {}, verify= False, headers={ 'X-Application-Key': clientID, 'Authorization': f"Bearer {auth.get('access_token')}", 'content-type': 'application/json' }, cert = cert ) super(cobranca_santand, self, ).__init__(argv) print('Fim') def get_url_environment(self, environment='homolog'): WorkspaceID = self.get_token_data("WorkspaceID"); ambiente = "" if environment == 'homolog': ambiente = "sandbox" else: ambiente = "open" base_url = f"https://trust-{ambiente}.api.santander.com.br/collection_bill_management/v2/" if self.rote != "bills": base_url = "{}{}".format(base_url, f"workspaces/{WorkspaceID}/") if environment == 'homolog': return { "oAuth2": f"https://trust-sandbox.api.santander.com.br/auth/oauth/v2/token", "base_url": base_url } return { "oAuth2" :f"https://trust-open.api.santander.com.br/auth/oauth/v2/token", "base_url" : base_url } def generateOAuth2Token(self, url): headers = { "Content-Type" : "application/x-www-form-urlencoded", } data = { 'client_id': self.get_token_data('ClientID'), 'client_secret': self.get_token_data('ClientSecret'), 'grant_type': 'client_credentials' } cert = ( self.get_token_data('Certificado_certKey'), self.get_token_data('Certificado_priKey') ) token = Request( method='POST', headers = headers, data = data, cert = cert ).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() # ------------------------------------------------------------------# # # # ROTA DE CRIAÇÃO DE BOLETOS # # # # ------------------------------------------------------------------# def post_route_handler_bank_slips(self): 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 if 'environment' in data_parsed and data_parsed['environment'] == "TESTE": data_parsed['nsuCode'] = "TST{}".format(data_parsed['nsuCode']) data_parsed['payer']['zipCode'] = self.to_cep(data_parsed['payer']['zipCode']) data_parsed['nominalValue'] = self.two_decimal(data_parsed['nominalValue']) self.handle_send(data_parsed) def handle_output_bank_slips(self, data): #"""Tratamento de retorno dos metodos PATCH e POST de boletodos""" status = "OK" if self.api_request.req.status_code in [200, 201] else "ER" desc_status = "Enviado com sucesso" extra_err_status = "" if self.api_request.req.status_code not in [200, 201]: retorno = json.loads(self.api_request.req.text) desc_status = retorno['_message'] # '{"_errorCode":400,"_message":"Altair error","_timestamp":"2024-04-16T17:09:46.020939-03:00","_traceId":"9a657b08-5a41-4991-9375-2c027d0addb7","_errors":[{"_code":"490","_message":"Cnpj raiz do pagador nao pode ser igual ao do beneficiario final - usar bda"}]}' try: extra_err_status = " - ({}) {}".format(retorno["_errors"][0]['_code'], retorno["_errors"][0]['_message']) except: extra_err_status = "" nossoNumero = self.dados_requisicao.get('nosso_nro')['valor'] tituloBenef = self.dados_requisicao.get('tit_benef')['valor'] retorno = json.loads(self.api_request.req.text) self.write_file(self.api_request.req.text, "/tmp/cobranca_santader.log.txt", flag="w+", encoding='iso-8859-1') # Criando boletos if self.current_method == "POST": structure = f"{status}|{desc_status}{extra_err_status}|{nossoNumero}|||{tituloBenef}||LINHA_DIGITAVEL|COD_BARRA_NUM|||||QR_EMV||\n" if status == "OK": structure = structure.replace("LINHA_DIGITAVEL", retorno['barCode']) structure = structure.replace("COD_BARRA_NUM", retorno['digitableLine']) structure = structure.replace("QR_EMV", retorno['qrCodePix'] if 'qrCodePix' in retorno else "HOMOLOGACAO") 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 # Atualizando boletos elif self.current_method == "PATCH": convenio = self.dados_requisicao.get('convenio')['valor'] structure = f"{status}|{desc_status}{extra_err_status}|{tituloBenef}|{nossoNumero}|{convenio}|\n" 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 # ------------------------------------------------------------------# # # # ROTA DE ATUALIZAR DE BOLETOS # # # # ------------------------------------------------------------------# def patch_route_handler_bank_slips(self): 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 self.handle_send(data_parsed) # ------------------------------------------------------------------# # # # ROTA DE CONSULTA DE BOLETOS # # # # ------------------------------------------------------------------# def beforeSend_bills(self, data): rota_atual = self.atualWs['rote'] dados = self.unparsed_data nosso_nro = dados.get('nosso_nro')['valor'] convenio = dados.get('convenio')['valor'] rota_atual = "{}/{}.{}?tipoConsulta=settlement".format(rota_atual, convenio, nosso_nro) self.current_rote = rota_atual def bills_handler(self, boletos): data = self.txt_to_dict() for request in data.get('C'): self.unparsed_data = request self.handle_send({}) def handle_output_bills(self, data): 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) retorno = dados retorno['dueDate'] = retorno['dueDate'].replace("-", "") retorno['issueDate'] = retorno['issueDate'].replace("-", "") payment_date = "" payment_date_credit = "" payment_value = "" if retorno['status'] == 'BAIXADO': base = retorno['writeOffData'][0] payment_date = base['writeOffDate'] if base['writeOffDate'] != None else "" payment_value = base['writeOffValue'] if {base['writeOffValue']} != None else "" elif retorno['status'] == 'LIQUIDADO': base = retorno['settlementData'][0] payment_date = base['settlementDate'] if base['settlementDate'] != None else "" payment_date_credit = base['settlementCreditDate'] if base['settlementCreditDate'] != None else "" payment_value = base['settlementValue'] if base['settlementValue'] != None else "" payment_date = payment_date.replace("-", "") payment_date_credit = payment_date_credit.replace("-", "") extra_status = "PIX" if "PIX" in retorno['writeOffData'][0]['writeOffDescription'] else retorno['writeOffData'][0]['writeOffDescription'] estrutura = f"{status}|{desc_status}|{retorno['documentNumber']}|{retorno['bankNumber']}|{retorno['clientNumber']}|{retorno['dueDate']}|{retorno['nominalValue']}|{retorno['issueDate']}|{retorno['participantCode']}|{retorno['status']}|{extra_status}|{retorno['settlementData'][0]['interestValue']}|{retorno['settlementData'][0]['discountValue']}|{retorno['settlementData'][0]['deductionValue']}|{payment_value}|{payment_date}|{payment_date_credit}|\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 # ------------------------------------------------------------------# # # # FUNÇÕES EXTRAS # # # # ------------------------------------------------------------------# def two_decimal(self, value): return "{:.2f}".format(value) def to_cep(self, value): cep = value[:5] + "-" + value[5:] return cep def converter_para_lista(self, value, item): return value