[ Web Proxy ]
URL:
Viewing: https://raw.githubusercontent.com/memgraph/mage/main/python/xml_module.py [Back]  [Original]

import mgp
import defusedxml.ElementTree as ET
import urllib.request

TYPE = "_type"
TEXT = "_text"
CHILDREN = "_children"


def parse_element(element, simple):
    result = {TYPE: element.tag}

    attributes = element.attrib
    if attributes:
        result.update(attributes)

    text_content = element.text
    if text_content and text_content.strip():
        result[TEXT] = text_content

    children = list(element)
    if children:
        children_name = CHILDREN
        if simple:
            children_name = "_" + str(element.tag)
        result[children_name] = [parse_element(child, simple) for child in children]

    return result


def xml_file_to_string(xml_file):
    xml_string = ""
    try:
        with open(xml_file, "r") as xml_file:
            xml_string = xml_file.read().replace("\n", "").replace("  ", "")
    except PermissionError:
        raise PermissionError(
            "You don't have permissions to write into that file."
            "Make sure to give the necessary permissions to user memgraph."
        )
    except Exception:
        raise OSError("Could not open or write to file.")
    return xml_string


@mgp.function
def parse(xml_input: str, simple: bool = False, path: str = "") -> mgp.Map:
    """
    Function to parse xml string (or file) into a map.

    Parameters
    ----------
    xml_input : str
        XML string which is to be parsed.
    simple: bool = false
        Boolean which specifies how should the children list be named,
        when it is false, all children lists are named _children
        if true, all children lists are named based on their parent.
    path: str = ""
        Path to XML file which is to be parsed, if it is not "",
        XML string is ignored and only file is parsed, if it is left as
        default, it is ignored.

    Returns:
        mgp.Map -> XML file parsed as map
    """

    root = None
    parser = ET.DefusedXMLParser()
    if path:
        root = ET.fromstring(xml_file_to_string(path), parser)
    else:
        root = ET.fromstring(xml_input, parser)
    output_map = parse_element(root, simple)
    return output_map


def check_url(url):
    if not url.endswith(".xml"):
        raise ValueError("File must be xml!")


def xpath_search(root, xpath_expression):
    try:
        result = root.findall(xpath_expression)
        return result
    except Exception as e:
        raise ValueError(f"XPath search error: {e}")


"""
Note: our implementation of xpath is different from neo4j,
because python offers different support for xpath than java
We cant have absolute paths
And our xpath search starts from the root node,
so ./something is equivalent to /root/something
For example,
python input : .//ZONE is equivalent to java input //ZONE
https://docs.python.org/3/library/xml.etree.elementtree.html#xpath-support
here are our avaliable xpath options
"""


@mgp.read_proc
def load(
    xml_url: str,
    simple: bool = False,
    path: str = "",
    xpath: str = "",
    headers: mgp.Map = {},
) -> mgp.Record(output_map=mgp.Map):
    """
    Procedure to load XML from url or from file to a map.

    Parameters
    ----------
    xml_url : str
        Url of the xml file to be parsed.
    simple: bool = false
        Boolean which specifies how should the children list be named,
        when it is false, all children lists are named _children
        if true, all children lists are named based on their parent.
    path: str = ""
        Path to XML file which is to be parsed, if it is not "",
        XML url is ignored and only file is parsed. If it is left as default,
        file path is ignored.
    xpath: str = ""
        Xpath expression which specifies which elements shall be returned.
        If left as "", it will be ignored, otherwise,
        only elements in XML file which satisfy
        expression are returned.
    headers: mgp.Map ={}
        Additional HTTP headers used in url request.

    Returns:
        mgp.Map -> XML file or URL parsed as map.
        In case XPATH is active, a map of for each element will be returned.
    """
    root = None
    parser = ET.DefusedXMLParser()
    if path:
        root = ET.fromstring(xml_file_to_string(path), parser)
    else:
        check_url(xml_url)
        try:
            request = urllib.request.Request(xml_url, headers=headers)
            response = urllib.request.urlopen(request).read()
            root = ET.fromstring(response)
        except Exception as e:
            raise ValueError(f"Error while fetching or parsing XML: {e}")

    if xpath:
        record_list = list()
        xpath_list = xpath_search(root, xpath)
        for element in xpath_list:
            record_list.append(mgp.Record(output_map=parse_element(element, simple)))
        return record_list
    output_map = parse_element(root, simple)
    return mgp.Record(output_map=output_map)

Web Proxy Viewer  |  New URL  |  Original Page