import time from generic.gn_request import Request # from generic.gn_api import gn_api # from generic.gn_json import gn_json # from datetime import datetime import json import urllib3 import unidecode import requests from urllib.parse import urlparse import os urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) class tiino(gn_api, gn_json): def __init__(self, argv:list): self.response = None self.db = False self.output_count = 0 self.suffix = "_app_pedidos3" self.unparsed_data = {} self.setup(argv) self.rote_handler() environment = 'homolog' if self.environment_config.get('ambiente') == 'H' else '' url = self.get_url_environment(environment=environment) token = self.get_token_request( clientId = self.get_token_data('ClientID'), clientSecret = self.get_token_data('ClientSecret'), environment = environment ) self.tiino_request = Request( base_url = url, verify= False, headers={ 'Content-Type': 'application/json' ,'Authorization': f"Bearer {token}" } ) super(tiino, self, ).__init__() print('Fim') def get_token_request(self, clientId, clientSecret, environment='homolog') -> str: url = "" if environment == 'homolog': url = "https://tuneauth.com.br/auth/realms/tiino-dev/protocol/openid-connect/token" # tiino-dev else: url = "https://tuneauth.com.br/auth/realms/tiino/protocol/openid-connect/token" # tiino response = requests.post( url=url, data={ 'client_id': clientId, 'client_secret': clientSecret, 'grant_type': 'client_credentials' }, headers={ 'Content-Type': 'application/x-www-form-urlencoded' } ) if response.ok: return response.json().get('access_token', '') def get_url_environment(self, environment='homolog') -> str: if environment == 'homolog': return 'https://api.sta.tiino.com.br/api/' return 'https://api.tiino.com.br/api/' """ ### ### #### ## # # ## # ### ### #### # # # # # # # ## # # # # # # # ### ### ### # # # ## # # # # ## ### # # # # #### # # #### # # # # # # # #### # # # # # # #### ### ### #### https://api.intelipost.com.br/api/pre-analises POST /api/pre-analises Cria uma nova pré-análise CLR5x6 Retorno COBOL: FLAG-STATUS|DESC-STATUS|DATA-HORA|TICKET """ def post_route_handler_pre_analises(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) def beforeSend_pre_analises(self, data): clienteDesde = data.get('clienteDesde', '') if clienteDesde: data['clienteDesde'] = data['clienteDesde'][:10] return data """ retorno: { "data": "2019-08-24T14:15:22Z", "ticket": "80a7bb07-bd5f-4501-918f-8730e5b79799" } """ def beforeSend_titulos_codigo_titulo_pagamentos(self, data): return data def handle_output_pre_analises(self, data): if self.api_request.req.ok: retorno_cobol = "" response_data = self.api_request.req.json() if self.api_request.req.text else {} retorno_cobol = "OK||{}|{}|".format( response_data.get('data', ''), response_data.get('ticket', '') ) self.relatement({ 'NMTABELA': 'TIINO_API', 'IDSOFTDIB': self.unparsed_data.get('IDSOFTDIB', {}).get('valor', ''), 'NMNOMEAPI': 'tiino', 'IDAPI': response_data.get('ticket', ''), "DTULTIMAALTERACAO" : "now()" }) # Escreve o retorno no arquivo de saída flag = "w+" if self.output_count == 0 else "a+" self.write_file(retorno_cobol, self.saida_cobol, flag=flag, encoding='iso-8859-1') else: retorno_cobol = "ER|{}||||".format(self.api_request.req.text) flag = "w+" if self.output_count == 0 else "a+" self.write_file(retorno_cobol, self.saida_cobol, flag=flag, encoding='iso-8859-1') """ ### #### ### ### ### ## ### # # # # # # # # # # # ### ### # # # # # # # ## # # # # # # # # # # # #### ### ### ### ## ### https://api.intelipost.com.br/api/pedidos POST /api/pedidos Cria um novo pedido Retorno COBOL: FLAG-STATUS|DESC-STATUS|DATA-HORA|TICKET """ def post_route_handler_pedidos(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) def beforeSend_pedidos(self, data): clienteDesde = data.get('clienteDesde', '') if clienteDesde: data['clienteDesde'] = data['clienteDesde'].replace(" ", "T") + "Z" data['realizarVerificacaoComAntifraude'] = data['realizarVerificacaoComAntifraude'] == 'S' return data """ retorno: { "data": "2019-08-24T14:15:22Z", "ticket": "80a7bb07-bd5f-4501-918f-8730e5b79799" } """ def handle_output_pedidos(self, data): if self.api_request.req.ok: retorno_cobol = "" response_data = self.api_request.req.json() if self.api_request.req.text else {} retorno_cobol = "OK||{}|{}|".format( response_data.get('data', ''), response_data.get('ticket', '') ) self.relatement({ 'NMTABELA': 'TIINO_API', 'IDSOFTDIB': self.unparsed_data.get('IDSOFTDIB', {}).get('valor', ''), 'NMNOMEAPI': 'tiino', 'IDAPI': response_data.get('ticket', ''), "DTULTIMAALTERACAO" : "now()" }) # Escreve o retorno no arquivo de saída flag = "w+" if self.output_count == 0 else "a+" self.write_file(retorno_cobol, self.saida_cobol, flag=flag, encoding='iso-8859-1') elif self.api_request.req.status_code == 400: self.response_code_400(self.api_request.req) else: retorno_cobol = "ER|{}||||".format(self.api_request.req.text) flag = "w+" if self.output_count == 0 else "a+" self.write_file(retorno_cobol, self.saida_cobol, flag=flag, encoding='iso-8859-1') """ ### ## # ### ##### ### ### ## ### ### ### #### ### ### ##### ## # # # # # # # # # # # # # # # # # # # # # # ### # # # # # # # # # ## # ### ### # # # # # # # # # # # # # # #### # # # # # # # # # # # # ## #### ### # ### ### # # ### ### # # #### ### ### # ## https://api.intelipost.com.br/api/politicas-de-credito GET /api/politicas-de-credito Lista todas as políticas de crédito RETORNO COBOL: FLAG-STATUS|DESC-STATUS|COD-ERP|NOME|DESCRICAO|ATIVO|PADRAO|DATA-CRIACAO|HORA-CRIACAO|ARQUIVADO| RETORNO EXEMPLO: { "registros": [ { "codigoErp": "string", "nome": "string", "descricao": "string", "ativo": true, "padrao": true, "criadoEm": "2019-08-24T14:15:22Z", "arquivado": true } ], "paginacao": { "paginaAtual": 0, "quantidadeDeRegistros": 0, "possuiAvancarPagina": true, "possuiVoltarPagina": true, "quantidadeDeRegistrosPorPagina": 0, "quantidadeDePaginas": 0 } } """ def politicas_de_credito_handler(self, data): if self.api_request.req.ok: response_data = self.api_request.req.json() if self.api_request.req.text else {} if response_data.get('paginacao', {}): response_data = self.handle_pagination(response_data, self.tiino_request, self.atualWs.get('route')) for record in response_data.get('registros', []): retorno_cobol = "OK||{}|{}|{}|{}|{}|{}|{}|{}|{}|".format( record.get('codigoErp', ''), record.get('nome', ''), record.get('descricao', ''), 'S' if record.get('ativo', False) else 'N', 'S' if record.get('padrao', False) else 'N', record.get('criadoEm', '')[:10].replace('-', ''), record.get('criadoEm', '')[11:19].replace(':', ''), 'S' if record.get('arquivado', False) else 'N' ) # Escreve o retorno no arquivo de saída flag = "w+" if self.output_count == 0 else "a+" self.write_file(retorno_cobol, self.saida_cobol, flag=flag, encoding='iso-8859-1') self.output_count += 1 def beforeSend_clientes(self, data): if self.current_method == "POST": existeTiino = self.tiino_request.doRequest(rote=f"clientes?documento={data.get('documento')}") if existeTiino.ok: dadosTiino = existeTiino.json() if dadosTiino['totalPages'] > 0 and 'registros' in dadosTiino and len(dadosTiino['registros']) > 0: self.ignore_request = True tiinoId = dadosTiino['registros'][0].get('id', '') self.relatement({ 'NMTABELA': 'TBCLIENTE', 'IDSOFTDIB': self.current_data.get('softdib_id', ''), 'NMNOMEAPI': 'tiino', 'IDAPI': tiinoId, "DTULTIMAALTERACAO" : "now()", "HASHMD5": "INTEGRADO" }) self.createLog(201, f"Cliente já existe na Tiino com ID {tiinoId}. Ignorando requisição para cliente: {self.current_data.get('softdib_id', '')} encontrado na base Tiino, ID: {tiinoId}") return data if self.current_data.get('endereco_estado') == "": self.ignore_request = True self.createLog(400, f"Estado do endereço não pode ser vazio. Ignorando requisição para cliente: {self.current_data.get('cliente_codigoExterno', '')} - {self.current_data.get('softdib_id', '')}") return data if self.current_data.get('emails_email') == "": self.ignore_request = True self.createLog(400, f"E-mail não pode ser vazio. Ignorando requisição para cliente: {self.current_data.get('cliente_codigoExterno', '')} - {self.current_data.get('softdib_id', '')}") return data if self.current_method == 'PUT': self.current_method = 'PATCH' """ Converte um objeto (dicionário) em um array de operações patch """ data_array = [] data.pop('telefones', None) # Removendo campo que não pode ser alterado for propriedade, valor in data.items(): data_array.append({ "op": "replace", "path": f"/{propriedade}", "value": valor }) return data_array return data def beforeSend_titulos(self, data): #if self.environment_config.get('ambiente') == 'H': #data['titulo'] = f"TESTE" if self.current_method == 'POST': if not data['dataQuitacao']: del data['dataQuitacao'] # Removendo data de quitacao se nao tiver data de pagamento if self.current_method == 'PUT': self.current_method = 'PATCH' """ Converte um objeto (dicionário) em um array de operações patch """ data_array = [] ### '{"type":"https://httpstatuses.io/422","title":"One or more validation errors occurred.","status":422,"errors":{"operations[1].Numero":["O Numero nao pode ser alterado"],"operations[2].Parcela":["A Parcela nao pode ser alterado"],"operations[10]./cliente":["O caminho \'/cliente\' não existe no objeto \'Titulo\'"]},"traceId":"00-80d7424091acb556404630e4594b2c75-4edbfe95bce45a6a-00"}' del data['numero'] # Removendo campo que não pode ser alterado del data['parcela'] # Removendo campo que não pode ser alterado del data['cliente'] # Removendo campo que não pode ser alterado if not data['dataQuitacao']: del data['dataQuitacao'] # Removendo data de quitação se não tiver data de pagamento if data['status'] == 'pago': del data['status'] # Removendo status pago se não tiver data de quitação for propriedade, valor in data.items(): data_array.append({ "op": "replace", "path": f"/{propriedade}", "value": valor }) return data_array return data def send_handler(self, data) : """ Metodo para manipular o envio """ # o data pode ser alterado aqui dentro self.prepare_send(data) # Faz um tratamento no dados antes de enviar func_rote = self.fixed_rote() func = f"beforeSend_{func_rote}"#.format(func_rote) # Tratamento de dados antes de enviar beforeSend = self.call(func, data, is_required= self.atualWs.get('need_beforeSend',False)) #if self.hasAttr(func, self) : # data = getattr(self, func)( data ) #print("stop") if hasattr(self, 'ignore_request') and self.ignore_request == True: self.ignore_request = False return if beforeSend is not None: data = beforeSend self.send( data ) # Executa apoas enviar - cada rota tem a sua especifica - deve-se criar dentro da arquivo da api if not self.atualWs['send'] or self.send_response.status_code in range(200,300) : func = f"afterSend_{func_rote}"#format(func_rote) # Tratamento de dados depois de enviar #if self.hasAttr(func, self) : # getattr(self, func)( data ) self.call(func, data ) self.current_data = {} def handle_pagination(self, data, request, route): pagina_atual = data.get('paginacao', {}).get('paginaAtual', 0) total_paginas = data.get('paginacao', {}).get('quantidadeDePaginas', 0) registros = data.get('registros', []) while pagina_atual < total_paginas - 1: pagina_atual += 1 response = request.get(route, params={'pagina': pagina_atual}) if response.ok: response_data = response.json() if response.text else {} registros.extend(response_data.get('registros', [])) else: break data['registros'] = registros return data """ .########..########.##.......########.########.######## .##.....##.##.......##.......##..........##....##...... .##.....##.##.......##.......##..........##....##...... .##.....##.######...##.......######......##....######.. .##.....##.##.......##.......##..........##....##...... .##.....##.##.......##.......##..........##....##...... .########..########.########.########....##....######## """ def delete_handler(self) : if not self.hasAttr("deleteFromApi",self) : return self.deleteFromApi() def deleteFromApi(self): rote = self.atualWs.get('rote') def handle_send(self, data) : """ Metodo para manipular o envio """ self.current_data = data self.prepare_send(data) # Faz um tratamento no dados antes de enviar func_rote = self.fixed_rote() func = f"beforeSend_{func_rote}"#.format(func_rote) # Tratamento de dados antes de enviar beforeSend = self.call(func, data, is_required= self.atualWs.get('need_beforeSend',False)) if beforeSend is not None: data = beforeSend response = self.send(data) self.current_data = {} return self.handle_response(response) """ '{"errors":{},"type":"about:blank","status":0,"title":"One or more validation errors occurred.","status":400,"errors":{"pedido":["O pedido de número \'02841300\' já está \'AprovadoAutomatico\' em nosso sistema, para inserir-lo novamente, faça a reprovação pela interface"]}}' """ def response_code_200(self, response): return super().response_code_201(response) def response_code_400(self, req): dados = req.json() # Extrai as mensagens de erro errors = dados.get('errors', {}) or dados.get('exceptionDetails', {}) if not errors: return "Erro desconhecido" # Formata as mensagens mensagens = [] for campo, lista_erros in errors.items(): for erro in lista_erros: mensagens.append(erro) # Tratamento mensagens = unidecode.unidecode(", ".join(mensagens)).replace("\n", " ").replace("\r", " ").replace("|", " ").strip() retorno_cobol = "ER|{}|".format(mensagens) # Retorna todas as mensagens separadas por quebra de linha flag = "w+" if self.output_count == 0 else "a+" self.write_file(retorno_cobol, self.saida_cobol, flag=flag, encoding='iso-8859-1') self.output_count += 1 return False def response_code_500(self, response): dados = response.json() # Extrai as mensagens de erro errors = dados.get('detail', {}) if not errors: return "Erro desconhecido" # Formata as mensagens # Tratamento mensagens = unidecode.unidecode("".join(errors)).replace("\n", " ").replace("\r", " ").replace("|", " ").strip() retorno_cobol = "ER|{}|\n".format(mensagens) # Retorna todas as mensagens separadas por quebra de linha flag = "w+" if self.output_count == 0 else "a+" self.write_file(retorno_cobol, self.saida_cobol, flag=flag, encoding='iso-8859-1') self.output_count += 1 return False def custom_control_id(self, response) : dados = response.json() return dados['id'] def convert_to_patch_operations(self, obj, base_path, operations): """ Converte um objeto (incluindo objetos aninhados) em operações JSON Patch Args: obj: Objeto a ser convertido base_path: Caminho base atual (ex: "/endereco") operations: Lista de operações patch a ser preenchida """ if isinstance(obj, dict): for propriedade, valor in obj.items(): current_path = f"{base_path}/{propriedade}" if isinstance(valor, (dict, list)): # Se for objeto ou array, processa recursivamente self.convert_to_patch_operations(valor, current_path, operations) else: # Se for valor primitivo, adiciona operação operations.append({ "op": "replace", "path": current_path, "value": valor }) elif isinstance(obj, list): for index, item in enumerate(obj): current_path = f"{base_path}/{index}" if isinstance(item, (dict, list)): # Se for objeto ou array, processa recursivamente self.convert_to_patch_operations(item, current_path, operations) else: # Se for valor primitivo, adiciona operação operations.append({ "op": "replace", "path": current_path, "value": item }) def txt_to_dict(self): files = self.rote_data.get('files') entrada = files.get('entrada' ) txt = self.readFile(entrada) linhas_arquivo = txt.split("\n") cabecalho = linhas_arquivo[0] corpo = linhas_arquivo[1:] estrutura = {} mapa_operacoes = {} propiedades_coluna = cabecalho.split(";") for propiedade in propiedades_coluna: estrutura_nome, estrutura_tipo, estrutura_tamanho, estrutura_variacao, estrutura_is_not_null = propiedade.split("|") if estrutura_nome == 'FGCUD': continue estrutura[estrutura_nome] = ({ "tipo":estrutura_tipo, "tamanho":estrutura_tamanho, "variacao":estrutura_variacao, "not_nulo":estrutura_is_not_null == 'S', }) def_ = "" for linha in corpo: linha_dados = linha.split(";") mapa_estrutura = {} for index, valor in enumerate(linha_dados): hed_ = propiedades_coluna[index] if(index == 0 ): def_ = valor if(def_ not in mapa_operacoes): mapa_operacoes[def_] = [] continue estrutura_nome = hed_.split("|")[0] estrutura_atual = estrutura[estrutura_nome] mapa_estrutura[estrutura_nome] = ({ **estrutura_atual, "valor": self.__tratar_valor(valor, **estrutura_atual) }) #linha_dados_json.append(mapa_estrutura) mapa_operacoes[def_].append(mapa_estrutura) return mapa_operacoes def __tratar_valor(self, valor, **kwargs): tipo = kwargs.get('tipo') variacao = kwargs.get('variacao') try: if len(valor) > 0 : if "INT" in tipo : valor = int(valor) elif "DOUBLE" in tipo : valor = float(valor) elif "DECIMAL" in tipo : #valor = valor.replace(".","").replace(",",".") valor = float(valor) elif "DATETIME" in tipo : valor = "{0}-{1}-{2} {3}:{4}:{5}".format(valor[0:4],valor[4:6],valor[6:8],valor[8:10],valor[10:12],valor[12:14]) if int(valor) else None elif "DATE" in tipo : valor = "{0}-{1}-{2}".format(valor[0:4],valor[4:6],valor[6:]) if int(valor) else None elif "I" in variacao : valor = self.__img_encode64(valor) else : valor = valor.strip() valor = valor.replace("§","\n") except: valor = "" if not valor and tipo in ['INT', 'DOUBLE', 'DATETIME', 'DATE'] : valor = None return valor def __img_encode64(self, value) : import base64 import os.path filename, file_extension = os.path.splitext(value) sucesso, image = self.readFile(path=value, flag='rb', encoding=None) extension = file_extension.replace('.','') base64 = (base64.b64encode(image)).decode('utf8') return f"data:image/{extension};base64,{base64}" def handle_response(self, response): reponse_data = self.tryJson(response) response_map = self.atualWs.get('response',{}) response_template = response_map.get('template',{}) data = self.generate_values(response_template, reponse_data) return self.handle_output(data) def handle_output(self, data): rote = self.fixed_rote() return self.call(f"handle_output_{rote}", data) def response_code_401(self, response): # Refazer o token environment = 'homolog' if self.environment_config.get('ambiente') == 'H' else '' url = self.get_url_environment(environment=environment) token = self.get_token_request( clientId = self.get_token_data('ClientID'), clientSecret = self.get_token_data('ClientSecret'), environment = environment ) # Não pode ser o tiino_request, pois ele é usado apenas para inicializar self.api_request = Request( base_url = url, verify= False, headers={ 'Content-Type': 'application/json' ,'Authorization': f"Bearer {token}" } ) self.createLog(401, "Token expirado. Novo token obtido e requisição reenviada.")