minor bug fixes.

Author: rhysrevans3

extraction_methods/plugins/geometry.py

@@ -40,7 +40,7 @@ class GeometryInput(Input):
     ] = Field(
         description="Type of geometry to be produced.",
     )
-    coordinates: list[str | float] = Field(
+    coordinates: list[Any] = Field(
         description="list of coordinates to convert to geometry. Ordering is respected.",
     )
     output_key: str = Field(

extraction_methods/plugins/lambda.py

@@ -12,7 +12,6 @@
 __contact__ = "[email protected]"
 
 
-import ast
 import logging
 from typing import Any
 
@@ -86,7 +85,7 @@ def run(self, body: dict[str, Any]) -> dict[str, Any]:
 
         output_body = body.copy()
 
-        function = ast.literal_eval(self.input.function)
+        function = eval(self.input.function)  # nosec B307
 
         result = function(*self.input.args, **self.input.kwargs)
 

extraction_methods/plugins/regex_rename.py

@@ -15,6 +15,7 @@
 # Python imports
 import logging
 import re
+from collections.abc import KeysView
 from typing import Any
 
 from pydantic import Field
@@ -46,6 +47,10 @@ class RegexRenameInput(Input):
     regex_swaps: list[RegexOutputKey] = Field(
         description="Regex and output key combinations.",
     )
+    nest_delimiter: str = Field(
+        default="",
+        description="delimiter for nested term.",
+    )
 
 
 class RegexRenameExtract(ExtractionMethod):
@@ -76,13 +81,59 @@ class RegexRenameExtract(ExtractionMethod):
 
     input_class = RegexRenameInput
 
+    def matching_keys(self, keys: KeysView[str], key_regex: str) -> list[str]:
+        """
+        Find all keys that match regex
+
+        :param keys: dictionary keys to test
+        :type keys: KeysView
+        :param key_regex: regex to test against
+        :type key_regex: str
+
+        :return: matching keys
+        :rtype: list
+        """
+
+        regex = re.compile(key_regex)
+
+        return list(filter(regex.match, keys))
+
+    def rename(
+        self, body: dict[str, Any], key_parts: list[str], output_key: str
+    ) -> dict[str, Any]:
+        """
+        Rename terms
+
+        :param body: current body
+        :type body: dict
+        :param key_parts: key parts seperated by delimiter
+        :type key_parts: list
+
+        :return: dict
+        :rtype: update body
+        """
+
+        for key in self.matching_keys(body.keys(), key_parts[0]):
+
+            if len(key_parts) > 1:
+                body[key] = self.rename(body[key], key_parts[1:], output_key)
+
+            else:
+                body[output_key] = body[key]
+                del body[key]
+
+        return body
+
     @update_input
     def run(self, body: dict[str, Any]) -> dict[str, Any]:
 
-        output = body.copy()
-        for key in body.keys():
-            for swap in self.input.regex_swaps:
-                if re.fullmatch(rf"{swap.regex}", key):
-                    output[swap.output_key] = body[key]
+        for swap in self.input.regex_swaps:
+            nest = (
+                swap.regex.split(self.input.delimiter)
+                if self.input.delimiter
+                else [swap.regex]
+            )
+
+            body = self.rename(body, nest, swap.output_key)
 
-        return output
+        return body

extraction_methods/plugins/xml.py

@@ -12,14 +12,15 @@
 __contact__ = "[email protected]"
 
 import logging
+import os.path
 
 # Python imports
 from collections import defaultdict
 
 # Package imports
 from typing import Any
 
-from lxml.etree import ElementTree  # nosec B410
+from lxml import etree  # nosec B410
 from pydantic import Field
 
 from extraction_methods.core.extraction_method import ExtractionMethod
@@ -114,13 +115,14 @@ def run(self, body: dict[str, Any]) -> dict[str, Any]:
 
         # Extract the keys
         try:
-            if isinstance(self.input.input_term, str):
-                xml_file = ElementTree.parse(self.input.input_term)
+
+            if os.path.isfile(self.input.input_term):
+                xml_file = etree.parse(self.input.input_term)
 
             else:
-                xml_file = ElementTree.XML(self.input.input_term)
+                xml_file = etree.XML(self.input.input_term.encode("ascii", "ignore"))
 
-        except (ElementTree.ParseError, FileNotFoundError, TypeError):
+        except (etree.ParseError, FileNotFoundError, TypeError):
             return body
 
         output: dict[str, list[str]] = defaultdict(list)