#from generic.xmljson import parker from generic.JsonByPath import JsonByPath as jxp from xml.etree import ElementTree as ET import re, json from xml.etree.ElementTree import fromstring class gn_json(): def __init__(self) -> None: pass #self = injection def parser(self, data, structure): root = structure.get('root',{}) template = structure.get("template",{}) #xml = self.handle_data(template, data, root) paths = self.handle_values(template, data) transformer = DataTransform() return transformer.tojson(paths) #return self.xml2json(xml,xml_fromstring=False) def handle_values(self, template, data): data_to_transform = [] for item in template: _path = item.get('to').replace("/",".") value = self.handle_item(item, data) #self.jsonxpath(_from, data) isRequired = "required" in item and item.get('required') == True accept_falsy = "accept_falsy" in item and item.get('accept_falsy') == True if not isRequired and not accept_falsy and not value: continue elif isRequired and not value : raise Exception(f"Campo {item} obrigatorio") data_to_transform.append({"path": _path, "value": value }) return data_to_transform def xml2json(self, xml, **kwargs) : def _fromstring(value): '''Convert XML string value to None, boolean, int or float''' # NOTE: Is this even possible ? if value is None: return None # FIXME: In XML, booleans are either 0/false or 1/true (lower-case !) if value.lower() == 'true': return True elif value.lower() == 'false': return False # FIXME: Using int() or float() is eating whitespaces unintendedly here try: return json.loads(value) except ValueError: return value from xmljson import Parker parker = Parker(xml_fromstring=_fromstring) xml_to_string = self.process_xml(xml) return json.loads(json.dumps(parker.data(fromstring(xml_to_string)))) #bf_str = BadgerFish(**kwargs) # Keep XML values as strings #return bf_str.data(data) def process_xml(self, xml) -> str: return ET.tostring(xml, encoding='unicode') 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 item['from'] =='juros_tipo': # print('sdf') if 'to' in item: isRequired = "required" in item and item.get('required') == True accept_falsy = "accept_falsy" in item and item.get('accept_falsy') == True if not isRequired and not accept_falsy and not value: continue elif isRequired and not value : raise Exception(f"Campo {item} obrigatorio") #_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", {})) #self.xml_data_to_string = ET.tostring(element, short_empty_elements=False).decode('utf-8') #self.json.handle_data(json.get("template"), data ,json.get("root")) return element def handle_item(self, item, json) : """ Manipulacao e tratamento de cada item """ value = False if item.get("from") : 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 item.get("default") and not value: value = item['default'] if item.get("custom") : value = self.call(item['custom'], value) if item.get("custom_value") : value = self.call(item['custom_value'], value, item) if item.get("by_type"): value = self.call(item['by_type'], value, item) if item.get("list") : value = self.handle_list(item, value) #_utils.eprint(value) if item.get("child") : value = self.handle_data(item['child'], value) if item.get("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=False) : 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 = str(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 is not None and last: if isinstance(value, list): for val in value : append_value(node, val) else : append_value(node, value) class DataTransform: def __init__(self, **opts) -> None: self.customs = {} def custom(self, name) : def decorator(f): self.__add_custom_rules(name, f) return f return decorator def __add_custom_rules(self, key, f): self.customs[key] = f def toxml(self, data) -> str: return self.jsontoxml( data ) def tojson(self, paths) -> dict : return self.__create_json(paths) def __create_json(self, paths): from pydash import set_ json = {} for item in paths: is_required = item.get('required', True) value = item.get('value') current_path = item.get('path') value = value if '@' in current_path else self.__handle_json_value(value, item ) if is_required and value is not None : set_(json, current_path, value) value = None return json def __handle_json_value(self,value, item): custom = self.customs.get(item.get('path')) tolist = item.get('tolist') if callable(custom): value = custom(value, item) if tolist: data = [] for vals in value: nlist = [] paths = item.get(tolist,[]) for i, val in enumerate(vals): nlist.append({**paths[i], "value": val}) data.append(self.__create_json(nlist)) return data if isinstance(value, list) and len(value) and isinstance(value[0], dict) and value[0].get('path'): return self.__create_json(value) return value def jsontoxml(self, paths, root='data'): import xmltodict json_data = self.__create_json(paths) return xmltodict.unparse(json_data, pretty=True)