import time from generic.gn_request import Request # from generic.gn_serasa import gn_serasa # from datetime import datetime import json, re import base64 import urllib3 from lxml import etree import xml.dom.minidom as minidom import xml.etree.ElementTree as ET from typing import Dict, List, Any, Union, Optional urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) class serasa(gn_serasa): def __init__(self, argv:list): self.response = None self.db = False self.output_count = 0 self.suffix = "_app_credito" self.unparsed_data = {} self.setup(argv) self.rote_handler() clientId = self.get_token_data('clientID') clientSecret = self.get_token_data('clientSecret') environment = 'uat-' if self.environment_config.get('ambiente') == 'H' else '' urls = self.get_url_environment(environment=environment) auth = self.generateOAuth2Token(urls.get('oAuth2'), clientId, clientSecret) self.serasa_request = Request( base_url = urls.get('base_url') , params = {} , data={} , verify= False , headers={ 'Authorization': f"Bearer {auth.get('accessToken')}", 'content-type': 'application/json' } ) super(serasa, self, ).__init__(argv) print('Fim') def get_url_environment(self, environment='uat-'): # PAREI AQUI base = f'{environment}' if environment == 'uat-': base = f"{base}" return { "oAuth2" :f"https://{base}api.serasaexperian.com.br/security/iam/v1/client-identities/login", "base_url" :f"https://{base}api.serasaexperian.com.br/credit-services", } def generateOAuth2Token(self, url, clientId, clientSecret): # Formate como client_id:client_secret token_string = f"{clientId}:{clientSecret}" # Codifique em Base64 token_base64 = base64.b64encode(token_string.encode()).decode() headers = { 'Authorization': f'Basic {token_base64}', } token = Request( method='POST', headers = headers, ).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 pj_handler(self, data): data = self.rote_data.get('path') cnpj, relatorio, features, reportParams, empresa, filial, usuario, periodo_ini, periodo_fin, cliente, participante, tipo_selecao = data['relatorio'].split("|") stringWithReportParameters = "" # features = features[:-1] reportParams = reportParams[:-1] if reportParams != "": mapReportParams = reportParams.split(',') jsonReportParams = { "reportParameters": [] } for keyReport, valueReport in enumerate(mapReportParams): keyItem, valueItem = valueReport.split('#') jsonReportParams["reportParameters"].append({ "name": keyItem, "value": valueItem }) jsonEncoded = json.dumps(jsonReportParams) base64ReportParams = base64.b64encode(jsonEncoded.encode()).decode() stringWithReportParameters = f"&reportParameters={base64ReportParams}" report_req = self.serasa_request.doRequest( rote=f"/business-information-report/v1/reports?reportName={relatorio}&optionalFeatures={features}{stringWithReportParameters}", headers={ "X-Document-Id": cnpj, "Content-Type": "application/json" }, method="GET" ) if report_req.ok: # Salva o JSON em um arquivo separado para depuração with open(f"{self.saida_cobol}.json", "w", encoding="utf-8") as file: file.write(report_req.text) xml_string = self.convert_json_string_to_xml_pj(report_req.text, empresa=empresa, filial=filial, usuario=usuario, periodo_ini=periodo_ini, periodo_fin=periodo_fin, cliente=cliente, participante=participante, tipo_selecao=tipo_selecao) # Salva o XML convertido em um arquivo with open(self.saida_cobol, "w", encoding="utf-8") as file: file.write(xml_string) else: try: data = report_req.json() first_error = data[0] code = first_error['code'] message = self.juglet(f"{first_error['message']}".replace("["," ").replace("]"," ")) print(f"ER|{code}-{message}") except: message = self.juglet(report_req.text.replace("["," ").replace("]"," ")) print(f"ER|{report_req.status_code}-{message}") return ##### ###### ##### #### ##### ###### ##### ###### ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ##### #### ##### ## ## ##### ## ##### #### #### ## ## ## ## #### ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ###### ## #### ## ## ## ## ## def pf_handler(self, data): data = self.rote_data.get('path') cpf, cnpj, relatorio, features, reportParams, empresa, filial, usuario, periodo_ini, periodo_fin, cliente, participante, tipo_selecao = data['relatorio'].split("|") stringWithReportParameters = "" # features = features[:-1] reportParams = reportParams[:-1] if reportParams != "": mapReportParams = reportParams.split(',') jsonReportParams = { "reportParameters": [] } for keyReport, valueReport in enumerate(mapReportParams): keyItem, valueItem = valueReport.split('#') jsonReportParams["reportParameters"].append({ "name": keyItem, "value": valueItem }) jsonEncoded = json.dumps(jsonReportParams) base64ReportParams = base64.b64encode(jsonEncoded.encode()).decode() stringWithReportParameters = f"&reportParameters={base64ReportParams}" report_req = self.serasa_request.doRequest( rote=f"/person-information-report/v1/creditreport?reportName={relatorio}&optionalFeatures={features}{stringWithReportParameters}", headers={ "X-Document-Id": cpf, "X-Retailer-Document-Id": cnpj, "Content-Type": "application/json" }, method="GET" ) if report_req.ok: with open(f"{self.saida_cobol}.json", "w", encoding="utf-8") as file: file.write(report_req.text) xml_string = self.convert_json_string_to_xml_pf(report_req.text, empresa=empresa, filial=filial, usuario=usuario, periodo_ini=periodo_ini, periodo_fin=periodo_fin, cliente=cliente, participante=participante, tipo_selecao=tipo_selecao) with open(self.saida_cobol, "w", encoding="utf-8") as file: file.write(xml_string) else: try: data = report_req.json() first_error = data[0] code = first_error['code'] message = self.juglet(f"{first_error['message']}".replace("["," ").replace("]"," ")) print(f"ER|{code}-{message}") except: message = self.juglet(report_req.text.replace("["," ").replace("]"," ")) print(f"ER|{report_req.status_code}-{message}") return data ###### ## ## ###### ##### ## #### ## ## ## ## ## ## #### ## ## ## ### ## ## ## ## ## ## #### ## ## ##### ###### #### ## #### ## #### ## ## ## ## ## ## ## ## ## ## ## ## ## ###### ## ## ## ## ## ## ## #### 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 convert_json_string_to_xml(self, json_string, empresa, filial, usuario, periodo_ini, periodo_fin, cliente, participante, tipo_selecao): """Converte uma string JSON para uma string XML.""" # Carregar a string JSON como um dicionário json_data = json.loads(json_string) # Converter JSON para XML xml_data = self.json_to_xml(json_data) data = datetime.now().strftime("%d/%m/%Y - %H:%M") xml_data = f"
\n{xml_data}\n" xml_data = self.reordenar(xml_data) # Adicionar cabeçalho XML xml_data = f'\n{xml_data}' return xml_data def convert_json_string_to_xml_pj(self, json_string, empresa, filial, usuario, periodo_ini, periodo_fin, cliente, participante, tipo_selecao): """Converte uma string JSON para uma string XML.""" # Carregar a string JSON como um dicionário json_data = json.loads(json_string) with open('/var/www/html/webservice/gn_integracao/serasa/mapa_pj.json', 'r', encoding='utf-8') as f: mapping_json = json.load(f) data = datetime.now().strftime("%d/%m/%Y - %H:%M") # Converter JSON para XML xml_data = map_json_to_xml(mapping_json, json_data, { "participante": participante, "cliente": cliente, "periodo_fin": periodo_ini, "periodo_ini": periodo_fin, "usuario": usuario, "filial": filial, "empresa": empresa, "datahora": data }) xml_data = format_xml(xml_data) return xml_data def convert_json_string_to_xml_pf(self, json_string, empresa, filial, usuario, periodo_ini, periodo_fin, cliente, participante, tipo_selecao): """Converte uma string JSON para uma string XML.""" # Carregar a string JSON como um dicionário json_data = json.loads(json_string) with open('/var/www/html/webservice/gn_integracao/serasa/mapa_pf.json', 'r', encoding='utf-8') as f: mapping_json = json.load(f) data = datetime.now().strftime("%d/%m/%Y - %H:%M") # Converter JSON para XML xml_data = map_json_to_xml(mapping_json, json_data, { "participante": participante, "cliente": cliente, "periodo_fin": periodo_ini, "periodo_ini": periodo_fin, "usuario": usuario, "filial": filial, "empresa": empresa, "datahora": data }) xml_data = format_xml(xml_data) return xml_data # Desativando pois não é uma rota de pedidos def define_sd_request(self,rota='/integracao/pedido/authentication'): return ##### ## #### #### ## ## ## ###### #### #### ## ## ##### ## ## ## ###### #### ## ## #### ## ## ## ### ## #### ## ## ## ## ### ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ###### ## ## ## ## ## ## ###### ## ## ## ## ## ## ## ##### ###### ## ### ## ###### ###### ## ## ## ## ###### ##### ## ## ## #### #### ## ## ## ## ## ## ## ### ## ## ## ## ## ## ## ### #### ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## #### #### ## ## ## ## ## #### #### ## ## ## ## #### ###### ###### #### 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 juglet(self, message): import unicodedata return ''.join(ch for ch in unicodedata.normalize('NFKD', message) if not unicodedata.combining(ch)) #### ##### ## ## ## ## ####### ###### ##### ##### ###### ## ## ## ## ### ## ## ## ## # ## ## ## ## ## ## ## ## ## ## ## #### ## ## ## ## # ## ## # ## ## ## ## ## ## ## ## #### ## ## #### ##### ##### ## ## ##### ## ## ## ## ### ### ## # ## ## ## ## ## ## ## ## ## ## ## ## ## ### ## # ## ## ## ## ## ## ## ## #### ##### ## ## # ####### #### ## ##### ##### #### ## def map_json_to_xml(mapping_json: List[Dict], source_json: Dict, variables: Dict = None) -> ET.Element: """ Mapeia os dados do JSON de origem para uma estrutura XML de acordo com o mapeamento fornecido. Args: mapping_json: Lista de mapeamentos que definem como os campos devem ser convertidos source_json: JSON de origem contendo os dados a serem mapeados variables: Dicionário de variáveis externas para uso com prefixo @ Returns: Elemento XML raiz contendo a estrutura mapeada """ # Inicializar variáveis externas se não fornecidas if variables is None: variables = {} # Criar o elemento raiz root = ET.Element("root") # Processar cada regra de mapeamento for mapping in mapping_json: from_path = mapping["from"] to_path = mapping["to"] # Verificar se é uma referência a variável externa (com prefixo @) if from_path.startswith('@'): # Extrair o nome da variável sem o @ var_name = from_path[1:] if var_name in variables: # Usar o valor da variável externa source_value = variables[var_name] set_value_in_xml(root, to_path, source_value) continue # Obter o valor no JSON de origem source_value = get_value_from_path(source_json, from_path) # Se o caminho aponta para uma lista e tem 'repeat' definido, processar como repetição if "repeat" in mapping and mapping.get("repeat") and isinstance(source_value, list): # Extrair as partes do caminho parts = to_path.split('/') parent_path = '/'.join(parts[:-1]) if len(parts) > 1 else '' element_name = parts[-1] # Obter o nó pai onde serão adicionados os elementos repetidos parent_node = ensure_path_exists(root, parent_path) if parent_path else root if "flat" in mapping and mapping.get("flat"): # Para flat=true, criar um único elemento contendo todos os itens element_node = ensure_path_exists(root, to_path) for item in source_value: process_mapping_to_xml(element_node, None, item, mapping.get("childrens", []), variables) else: # Para repeat normal (sem flat), repetir o elemento para cada item da lista for item in source_value: # Criar um novo elemento para cada item element_node = ET.SubElement(parent_node, element_name) process_mapping_to_xml(element_node, None, item, mapping.get("childrens", []), variables) else: # Processar como um único item if source_value is not None: if "childrens" in mapping: # Tem filhos, então criar/obter o nó pai e processar os filhos parent_node = ensure_path_exists(root, to_path) process_mapping_to_xml(parent_node, None, source_value, mapping["childrens"], variables) else: # Sem filhos, é um valor simples set_value_in_xml(root, to_path, source_value) return root def process_mapping_to_xml(parent_node: ET.Element, base_path: Optional[str], source_data: Union[Dict, Any], children_mappings: List[Dict], variables: Dict = None) -> None: """ Processa um nível de mapeamento, adicionando valores ao XML. Args: parent_node: Nó XML pai onde os valores serão adicionados base_path: Caminho base para os elementos (pode ser None para nível atual) source_data: Dados de origem para este nível children_mappings: Lista de mapeamentos para os filhos variables: Dicionário de variáveis externas para uso com prefixo @ """ for child_mapping in children_mappings: from_path = child_mapping["from"] to_path = child_mapping["to"] # Verificar se é uma referência a variável externa (com prefixo @) if from_path.startswith('@'): # Extrair o nome da variável sem o @ var_name = from_path[1:] if variables and var_name in variables: # Usar o valor da variável externa full_path = f"{base_path}/{to_path}" if base_path else to_path set_value_in_xml(parent_node, full_path, variables[var_name]) continue # Obter o valor do campo - verificar se source_data é um dicionário if isinstance(source_data, dict): child_value = source_data.get(from_path) else: # Se não for um dicionário, não podemos obter o valor child_value = None # Construir o caminho completo se houver base_path full_path = f"{base_path}/{to_path}" if base_path else to_path # Se o caminho aponta para uma lista e tem 'repeat' definido, processar como repetição if "repeat" in child_mapping and child_mapping.get("repeat") and isinstance(child_value, list): # Extrair as partes do caminho if full_path: parts = full_path.split('/') parent_path = '/'.join(parts[:-1]) if len(parts) > 1 else '' element_name = parts[-1] # Obter o nó pai onde serão adicionados os elementos repetidos list_parent = ensure_path_exists(parent_node, parent_path) if parent_path else parent_node if "flat" in child_mapping and child_mapping.get("flat"): # Para flat=true, criar um único elemento contendo todos os itens list_node = ensure_path_exists(parent_node, full_path) for item in child_value: process_mapping_to_xml(list_node, None, item, child_mapping.get("childrens", []), variables) else: # Para repeat normal (sem flat), repetir o elemento para cada item da lista for item in child_value: # Criar um novo elemento para cada item item_node = ET.SubElement(list_parent, element_name) process_mapping_to_xml(item_node, None, item, child_mapping.get("childrens", []), variables) else: # Se não há full_path, processar diretamente no nó pai if "flat" in child_mapping and child_mapping.get("flat"): # Adicionar cada item diretamente, sem nós intermediários for item in child_value: process_mapping_to_xml(parent_node, None, item, child_mapping.get("childrens", []), variables) else: # Criar elementos filhos nomeados com to_path for item in child_value: item_node = ET.SubElement(parent_node, to_path) process_mapping_to_xml(item_node, None, item, child_mapping.get("childrens", []), variables) else: # Processar como um único item if child_value is not None: if "childrens" in child_mapping: # Tem filhos, então criar/obter o nó pai e processar os filhos child_node = ensure_path_exists(parent_node, full_path) if full_path else parent_node process_mapping_to_xml(child_node, None, child_value, child_mapping["childrens"], variables) else: # Sem filhos, é um valor simples if full_path: set_value_in_xml(parent_node, full_path, child_value) else: # Se não houver caminho, adicionar diretamente como um atributo do nó pai node = ET.SubElement(parent_node, to_path) if child_value is not None: node.text = str(child_value) def get_value_from_path(json_data: Dict, path: str) -> Any: """ Obtém um valor do JSON dado um caminho em formato de string. Args: json_data: Dados JSON path: Caminho do valor (ex: "reports/0/reportName") Returns: O valor encontrado no caminho ou None se não encontrado """ parts = path.split('/') current = json_data for part in parts: if part.isdigit(): # É um índice de array index = int(part) if isinstance(current, list) and 0 <= index < len(current): current = current[index] else: return None else: # É uma chave de objeto if isinstance(current, dict) and part in current: current = current[part] else: return None return current def ensure_path_exists(root: ET.Element, path: str) -> ET.Element: """ Garante que um caminho exista na árvore XML, criando os nós necessários. Args: root: Elemento raiz do XML path: Caminho para garantir (ex: "reports/reportName") Returns: O nó final do caminho """ if not path: return root parts = path.split('/') current = root for part in parts: # Verificar se o elemento já existe existing = current.find(f"./{part}") if existing is None: # Criar o elemento se não existir existing = ET.SubElement(current, part) current = existing return current def set_value_in_xml(root: ET.Element, path: str, value: Any) -> None: """ Define um valor em um nó XML dado um caminho. Args: root: Elemento raiz do XML path: Caminho para o nó (ex: "reports/reportName") value: Valor a ser definido """ node = ensure_path_exists(root, path) # Converter o valor para string, se não for None if value is not None: if isinstance(value, bool): node.text = str(value).lower() else: node.text = str(value) def wrap_with_cdata(xml_str: str) -> str: """ Envolve o conteúdo de tags XML com seções CDATA. Args: xml_str: String XML para processar Returns: String XML com conteúdo envolvido em CDATA """ # Padrão regex para encontrar conteúdo de tags pattern = r'(>)([^<>]*)(<)' # Função para substituir cada correspondência def replacer(match): prefix, content, suffix = match.groups() # Se o conteúdo não estiver vazio e não contiver já uma seção CDATA if content.strip() and not content.strip().startswith("{suffix}" else: return f"{prefix}{content}{suffix}" # Aplicar a substituição result = re.sub(pattern, replacer, xml_str) return result def format_xml(root: ET.Element) -> str: """ Formata o XML com indentação para melhor legibilidade. Args: root: Elemento raiz do XML Returns: String XML formatada """ # Primeiro converter para string usando ElementTree rough_string = ET.tostring(root, encoding='utf-8').decode('utf-8') # Formatar com minidom para obter indentação reparsed = minidom.parseString(rough_string) formatted_xml = reparsed.toprettyxml(indent=" ") # Envolver conteúdo de tags com CDATA cdata_xml = wrap_with_cdata(formatted_xml) return cdata_xml