import re, json, csv from generic.gn_api import gn_api # from generic.gn_request import Request # from generic.JsonByPath import JsonByPath as jxp from xml.etree import ElementTree as ET from html import unescape from xml.etree.ElementTree import fromstring from xmljson import BadgerFish class gn_soap(gn_api): def __init__(self, argv:list): #self.db = False self.setup(argv) super(gn_soap, self, ).__init__() def setup(self, argv): self.apiName = argv[1] # Nome da api self.method = argv[2] # metodo chamado (GET POST PUT DELETE) self.rote = self.data_in = argv[3] # Rota chamada, se ALL, fazer todos self.nmbanco = argv[4] # Banco de Dados self.nmlkgrupo = argv[5] # posicionamento de grupo self.cdempresa = argv[6] # posicionamento de empresa self.cdfilial = argv[7] # posicionamento de filial self.saida = argv[8] # Arquivo de saida api self.saida_cobol = argv[9] # Arquivo de saida cobol self.debug = True if argv[-1] == 'debug' or 'debug' in argv[10:] else False self.data = None def load_input(self) : from pathlib import Path data = '' suffix = Path(self.data_in).suffix if suffix: data = self.read_file(path=self.data_in) self.data_in_suffix = suffix[1:] else: data = self.data_in self.data_in_suffix = 'command_line' # Carrega arquivo apartir de sua extensao # Podendo ser .xml .json .csv .txt ou linha de argumento return self.call(f"load_data_{self.data_in_suffix}", data, is_required=True) """ #!REMOVER ISSO DEPOIS NAO DEIXAR FIXO """ #self.environment = "homologacao" def load_data_command_line(self, data) : # cpfcnpj=20042802000171&inscricaomunicipal=&ambiente=2&controle=4559&protocolo=637298143366160922 _list = data.split("&") self.data = { # KEY : VALUE self.get_index(value.split("="),0):(self.get_index(value.split("="),1) or "").strip() for value in _list } self.action = self.data.get('action',"") if self.action != self.atualWs.get('action'): return False return True def load_data_txt(self, data): #data_file = self.read_file(self.data_in) splitlines = data.split("\n") self.tipo,version,self.action, filler = splitlines[0].split("|") if self.action != self.atualWs.get('action'): return False lines = splitlines[1:] soap_map = {} definitions = json.loads(self.read_file(f"{self.root_dir}/{self.apiName}/definitions.json")) mapa = definitions.get(version,{}).get('mapa',{}).get(self.tipo,{}).get(self.action) for line in lines: dados_linha = line.split("|") chave = dados_linha[0] line_definitions = mapa.get(chave) item_data = dict(zip(line_definitions,dados_linha[1:-1])) if chave in soap_map: temp = soap_map[chave] if not isinstance(soap_map[chave], list): soap_map[chave] = [temp] soap_map[chave].append(item_data) else: soap_map[chave] = item_data self.data = soap_map; return True def soap_handler(self): data = self.data#{"codticket":"75565"} # data response = self.atualWs.get('response',{}) xml = self.atualWs.get('xml',{}) root = xml.get('root',{}) template = xml.get("template",{}) self.xml_data = self.handle_data(template, data, root) self.process_xml() self.mount_request() if self.send_data(): self.xml_response = self.response.content.decode(self.response.encoding or self.response.apparent_encoding) tag = response.get('tag','') result_tag = self.get_tag( tag , self.xml_response) # Remover atributos desnecessarios xml = re.sub(r"<([\w:]+).*?(\/)?>",r"<\1\2>",result_tag) self.response_json = self.xml2json( xml , xml_fromstring=False) self.call(f"out_{self.action}", is_required = True) def xml2json(self, data, **kwargs) : bf_str = BadgerFish(**kwargs) # Keep XML values as strings return bf_str.data(fromstring(data)) def mount_request(self): self.call(self.action, is_required = True) def process_xml(self) -> str: self.xml_data_to_string = ET.tostring(self.xml_data, short_empty_elements=False).decode('utf-8') def send_data(self) -> bool: self.response = self.mysuite_request.doRequest( rote = self.rote , data = self.xml_data_to_string ) return self.response.ok def handle_data(self, mapItens, json, root={}) : """ Manipulacao de dados do map Faz o de para dos dados de entrada para os dados de saida, podendo ser a saida como um XML ou um JSON """ tag = root.get('tag') attrs = root.get('attrs') or {} element = ET.Element(tag) if tag else {} for attr in attrs: element.set(attr, attrs[attr]) for item in mapItens : value = self.handle_item(item, json) if 'to' in item: isRequired = "required" not in item or item.get('required') == True if not isRequired and not value: continue elif isRequired and not value : exit(1) #_utils.eprint(item) cpath = item.get('custom_path_to') item_to = self.call(item['custom_path_to'], item['to'], value) if cpath else item['to'] if item_to: if not tag : element[item_to] = value else : self.build_xpath(element, item_to, value, item.get("attrs", {})) return element def handle_item(self, item, json) : """ Manipulacao e tratamento de cada item """ value = False if "from" in item : cpath = item.get('custom_path_from') _from = self.call(item['custom_path_from'], item['from'], value) if cpath else item['from'] value = self.jsonxpath(_from, json, item.get('trim', False)) if "default" in item and not value: value = item['default'] if "custom" in item : value = self.call(item['custom'], value) if "custom_value" in item : value = self.call(item['custom_value'], value, item) if "by_type" in item: value = self.call(item['by_type'], value, item) if 'list' in item : value = self.handle_list(item, value) #_utils.eprint(value) if 'child' in item : value = self.handle_data(item['child'], value) if "globalThis" in item and item['globalThis']: varName = item['globalThis'] if isinstance(item['globalThis'], str) else item['to'] setattr(self,varName, value ) #if isinstance(value, str): # value = value.strip() return value def handle_list(self, item, value): data = [] root = item.get("root", {}) if isinstance(value, dict) : data = [self.handle_data(item['list'], value, root)] elif isinstance(value, list) : data = [self.handle_data(item['list'], sub_item, root) for sub_item in value ] return data def jsonxpath(self, path ,json, trim) : result = jxp(path=path, json=json, trim=trim) if (isinstance(result.value, str)) : result.value.strip() return result.value def build_xpath(self, node, path, value = False, attrs = {}) : """ Monta o XML baseado em Xpath """ components = re.split(r'''/(?=(?:[^'"]|'[^']*'|"[^"]*")*$)''',path) def append_value(node, value): try: node.append((value)) except : node.text = value if components[0] == node.tag: components.pop(0) while components: # take in account positional indexes in the form /path/para[3] or /path/para[location()=3] if "[" in components[0]: component, trail = components[0].split("[",1) target_index = int(trail.split("=")[-1].strip("]")) else: component = components[0] target_index = 0 if '@' in component : attrs = component.split('@') component = attrs[0] attrs = dict([ attr.replace("'","").split('=') for attr in attrs[1:] ]) components.pop(0) last = not len(components) found_index = -1 for child in list(node): if child.tag == component: found_index += 1 if found_index == target_index: node = child break else: for i in range(target_index - found_index): new_node = ET.Element(component) node.append(new_node) node = new_node for attr in attrs : node.set(attr, attrs[attr] ) attrs = [] if value and last: if isinstance(value, list): for val in value : append_value(node, val) else : append_value(node, value) def get_tag(self, tag, xml, within = True) : #xml = xml.replace("\n","") tagStart = "<{}>".format(tag) tagEnd = "".format(tag) start = xml.find(tagStart) end = xml.find(tagEnd) return xml[start+len(tagStart):end] if within else xml[start:end+len(tagEnd)] #return re.findall(r"<{tag}>(.*)".format(tag = tag), xml)