#118 GeoNode5: parser

Author: etj

rndt/apps.py

@@ -38,15 +38,6 @@ def run_setup_hooks(*args, **kwargs):
         settings.METADATA_PARSERS.extend([RNDT_PARSER_FUNCTION])
         setattr(settings, "METADATA_PARSERS", settings.METADATA_PARSERS)
 
-    RNDT_STORER_FUNCTION = "rndt.metadata.storer.rndt_storer"
-
-    rndt_storers = [RNDT_STORER_FUNCTION]
-    if not getattr(settings, "METADATA_STORERS", None):
-        setattr(settings, "METADATA_STORERS", rndt_storers)
-    elif rndt_storers[0] not in settings.METADATA_STORERS:
-        settings.METADATA_STORERS.extend(rndt_storers)
-        setattr(settings, "METADATA_STORERS", settings.METADATA_STORERS)
-
     urlpatterns += [
         re_path(r"^", include("rndt.api.urls")),
         re_path(r"^catalogue/", include("rndt.catalogue.urls")),

rndt/locale/it/LC_MESSAGES/django.po

@@ -2,26 +2,9 @@
 # Copyright (C) YEAR THE PACKAGE'S COPYRIGHT HOLDER
 # This file is distributed under the same license as the PACKAGE package.
 
-msgid "Data Constraints"
-msgstr "Vincoli sui dati"
-
-msgid "Access constraints"
-msgstr "Vincoli di accesso"
-
-msgid "Use constraints"
-msgstr "Vincoli di fruibilità"
-
-msgid "Free text"
-msgstr "Testo Libero"
-
-msgid "Resolution"
-msgstr "Risoluzione"
-
 msgid "Public Administration"
 msgstr "Pubblica Amministrazione"
 
 msgid "Additional info"
 msgstr "Informazioni aggiuntive"
 
-msgid "Positional Accuracy"
-msgstr "Accuratezza posizionale"

rndt/metadata/parser.py

@@ -10,7 +10,7 @@
 from owslib import util
 from owslib.iso import get_namespaces
 
-ACCESS_CONSTRAINTS_URL = "http://inspire.ec.europa.eu/metadata-codelist/LimitationsOnPublicAccess/noLimitations"
+logger = logging.getLogger(__name__)
 
 
 def rndt_parser(xml, uuid="", vals={}, regions=[], keywords=[], custom={}):
@@ -25,21 +25,96 @@ def rndt_parser(xml, uuid="", vals={}, regions=[], keywords=[], custom={}):
         exml = list(exml)[0]
 
     rndt_parser = RNDTMetadataParser(exml)
-
+    rndt_parser.get_freq(vals)
     keywords, discarded = rndt_parser.resolve_keywords()
     custom["rejected_keywords"] = discarded
 
-    custom["rndt"] = {}
-
-    use_constr = rndt_parser.get_access_costraints(custom)
-    rndt_parser.get_use_costraints(vals, use_constr)
-    rndt_parser.get_resolutions(custom)
-    rndt_parser.get_accuracy(custom)
-    rndt_parser.get_freq(vals)
+    # Next calls parse and store metadata in a jsonschema compliant way (geonode5)
+    jsoninstance = custom.setdefault("jsoninstance", {})
+    resolver = RNDTMetadataResolver(jsoninstance)
+    resolver.resolve_constraints(rndt_parser.parse_constraints())
+    resolver.resolve_resolution(rndt_parser.parse_resolution())
+    resolver.resolve_accuracy(rndt_parser.parse_accuracy())
 
     return uuid, vals, regions, keywords, custom
 
 
+class RNDTMetadataResolver:
+    def __init__(self, jsoninstance:dict):
+        self.jsoninstance = jsoninstance
+
+    def resolve_constraints(self, constraints:list):
+        freetext = ""
+        access = None
+        use = None
+
+        for constr in constraints:
+            logger.debug(f"Resolving constraint: --> {constr}")
+
+            code = constr["code"]
+            if code not in ("otherRestrictions", "limitation not listed"):
+                logger.debug(f"Skipping constraint {constr}")
+                continue
+
+            href = constr["href"]
+            text = constr["text"]
+
+            if not href:
+                logger.debug(f"Collecting text from {constr}")
+                freetext = f"{freetext}\n{text} "
+                continue
+
+            # rndt_LimitationsOnPublicAccess -> url
+            # rndt_ConditionsApplyingToAccessAndUse -> text or url
+
+            t = ThesaurusKeyword.objects.filter(about=href).filter(
+                thesaurus__identifier="LimitationsOnPublicAccess"
+            ).first()
+            if t:
+                if access:
+                    logger.warning(f"Duplicate LimitationsOnPublicAccess overridden {access}")
+                access = {"id":href, "label":t.alt_label}
+                continue
+
+            t = ThesaurusKeyword.objects.filter(about=href).filter(
+                thesaurus__identifier="ConditionsApplyingToAccessAndUse"
+            ).first()
+            if t:
+                if use:
+                    logger.warning(f"Duplicate ConditionsApplyingToAccessAndUse overridden {use}")
+                use = {"inspire_url": True, "url": href}
+                continue
+
+            logger.warning(f"Skipping unknown URL {constr}")
+            # we may try and parse license URLs: that's beyond RNDT requirements, but it would be nice
+        #endfor
+
+        if access:
+            self.jsoninstance["rndt_LimitationsOnPublicAccess"] = access
+        else:
+            logger.info("LimitationsOnPublicAccess not found")
+
+        if use:
+            self.jsoninstance["rndt_ConditionsApplyingToAccessAndUse"] = use
+            if freetext:
+                logger.warning(f"Ignoring freetext constraint [{freetext}]")
+        else:
+            if freetext:
+                self.jsoninstance["rndt_ConditionsApplyingToAccessAndUse"] = {
+                    "inspire_url": False, "freetext": freetext
+                }
+            else:
+                logger.info("ConditionsApplyingToAccessAndUse not found")
+
+    def resolve_resolution(self, val):
+        if val is not None:
+            self.jsoninstance["rndt_resolution"] = val
+
+    def resolve_accuracy(self, val):
+        if val is not None:
+            self.jsoninstance["rndt_accuracy"] = val
+
+
 class RNDTMetadataParser:
     """
     A metadata parser compliant with the RNDT specification
@@ -55,136 +130,112 @@ def __init__(self, exml):
             )
         )
 
-    def get_freq(self, vals):
-        freq_elem = self.exml.find(
-            util.nspath_eval(
-                "gmd:identificationInfo/gmd:MD_DataIdentification/gmd:resourceMaintenance/gmd:MD_MaintenanceInformation/gmd:maintenanceAndUpdateFrequency/gmd:MD_MaintenanceFrequencyCode",
-                self.namespaces,
-            )
-        )
+    def parse_codelist(self, xpath):
+        """ This should be moved into the base parser """
+        elem = self.exml.find(util.nspath_eval(xpath, self.namespaces,))
+        return (elem.attrib.get("codeListValue", None), elem.text) if elem is not None else None
 
-        freq = freq_elem.attrib.get("codeListValue", None) if freq_elem is not None else None
-        vals["maintenance_frequency"] = freq or "unknown"
-
-    def get_access_costraints(self, custom):
+    def parse_constraints(self) -> list:
         """
-        Function responsible to get the access constraints compliant with RNDT
-        - will take all the instances of LegalConstraints
-          - if the restriction MD_RestrictionCode under accessConstraints has a codeListValue = otherRestrictions
-            - If is an anchor item,
-                - will put in the vals dictionary under constraints_other the thesaurus label if exists
-                - otherwise will put in contraints_other the URL parsed
-            - if is a charstring:
-                - will save the value extracted in a variable since is required for get the use_constrains
+        Function responsible to parse the access constraints elements
+        - returns a list of dict:
+           - code: restriction codeListValue
+           - href: if gmx:Anchor in gmd:otherConstraints, its href
+           - text: text content of gmd:otherConstraints, either if CharacterString or Anchor
         """
-        use_constrs = ""
         access_constraints = self.exml.findall(
             util.nspath_eval(
                 "gmd:identificationInfo/gmd:MD_DataIdentification/gmd:resourceConstraints/gmd:MD_LegalConstraints",
                 self.namespaces,
             )
         )
-        for item in access_constraints:
-            md_restriction_code = item.find(
+        ret = []
+        for node in access_constraints:
+            constr = {}
+            logger.debug(f"Parsing constraint: --> {node}")
+
+            md_restriction_code = node.find(
                 util.nspath_eval("gmd:accessConstraints/gmd:MD_RestrictionCode", self.namespaces)
             )
-            if (
-                md_restriction_code is not None
-                and md_restriction_code.attrib.get("codeListValue", "") == "otherRestrictions"
-            ):
-                acc_constr = item.find(util.nspath_eval("gmd:otherConstraints/gmx:Anchor", self.namespaces))
-                if acc_constr is not None:
-                    url = acc_constr.attrib.get("{http://www.w3.org/1999/xlink}href")
-                    t = ThesaurusKeyword.objects.filter(about=url).filter(
-                        thesaurus__identifier="LimitationsOnPublicAccess"
-                    )
-                    if t.exists():
-                        custom["rndt"] = {"constraints_other": url}
-                    else:
-                        custom["rndt"] = {"constraints_other": ACCESS_CONSTRAINTS_URL}
+            if md_restriction_code is not None:
+                constr["type"] = "accessConstraints"
+            else:
+                md_restriction_code = node.find(
+                    util.nspath_eval("gmd:useConstraints/gmd:MD_RestrictionCode", self.namespaces)
+                )
+                if md_restriction_code is not None:
+                    constr["type"] = "useConstraints"
                 else:
-                    use_constrs = item.find(
+                    logger.warning("Missing known restrictioncode")
+                    continue
+
+            constr["code"] = md_restriction_code.attrib.get("codeListValue", "")
+
+            anchor = node.find(util.nspath_eval("gmd:otherConstraints/gmx:Anchor", self.namespaces))
+            if anchor is not None:
+                constr["href"] = anchor.attrib.get("{http://www.w3.org/1999/xlink}href")
+                constr["text"] = anchor.text
+            else:
+                charstring = node.find(
                         util.nspath_eval("gmd:otherConstraints/gco:CharacterString", self.namespaces)
                     ).text
-        return use_constrs
+                constr["href"] = None
+                constr["text"] = charstring
 
-    def get_use_costraints(self, vals, acc_constr):
-        """
-        Function responsible to get the use constraints compliant with RNDT
-        - will take all the instances of LegalConstraints
-          - if the restriction MD_RestrictionCode under useConstraints has a codeListValue = otherRestrictions
-            - If is an anchor item,
-                - will put in the custom dictionary under rndt the thesaurus label if exists
-                - otherwise will put in custom[rndt] the text and the information extracted in the previous step
-            - if is a charstring:
-                - will put in custom[rndt] the text and the information extracted in the previous step
-        """
-        use_constraints = self.exml.findall(
-            util.nspath_eval(
-                "gmd:identificationInfo/gmd:MD_DataIdentification/gmd:resourceConstraints/gmd:MD_LegalConstraints",
-                self.namespaces,
-            )
-        )
-        for item in use_constraints:
-            md_restriction_code = item.find(
-                util.nspath_eval("gmd:useConstraints/gmd:MD_RestrictionCode", self.namespaces)
-            )
-            if (
-                md_restriction_code is not None
-                and md_restriction_code.attrib.get("codeListValue", "") == "otherRestrictions"
-            ):
-                use_constr = item.find(util.nspath_eval("gmd:otherConstraints/gmx:Anchor", self.namespaces))
-                if use_constr is not None:
-                    url = use_constr.attrib.get("{http://www.w3.org/1999/xlink}href")
-                    t = ThesaurusKeyword.objects.filter(about=url).filter(
-                        thesaurus__identifier="ConditionsApplyingToAccessAndUse"
-                    )
-                    if t.exists():
-                        vals["constraints_other"] = url
-                    else:
-                        vals["constraints_other"] = f"{use_constr.text} {acc_constr}"
-                else:
-                    use_constr = item.find(
-                        util.nspath_eval("gmd:otherConstraints/gco:CharacterString", self.namespaces)
-                    )
-                    if use_constr is not None:
-                        vals["constraints_other"] = f"{use_constr.text} {acc_constr}"
-                    else:
-                        vals["constraints_other"] = acc_constr
-        return vals
+            ret.append(constr)
 
-    def get_resolutions(self, custom):
+        return ret
+
+    def parse_resolution(self, default=None):
         resolution = self.exml.find(
             util.nspath_eval(
                 "gmd:identificationInfo/gmd:MD_DataIdentification/gmd:spatialResolution/gmd:MD_Resolution/gmd:distance/gco:Distance",
                 self.namespaces,
             )
         )
 
-        if resolution is not None:
-            custom["rndt"]["resolution"] = (
-                resolution if isinstance(resolution, float) else ast.literal_eval(resolution.text)
-            ) or 0
-        else:
-            logging.error("Resolution cannot be None, using default value 0")
-            custom["rndt"]["resolution"] = 0
-        return custom
+        if resolution is None:
+            logger.info(f"Resolution not found")
+            return default
+        if isinstance(resolution, (float, int)):
+            return resolution
 
-    def get_accuracy(self, custom):
-        accuracy = self.exml.find(
+        try:
+            res = ast.literal_eval(resolution.text)
+            if isinstance(res, (float, int)):
+                return res
+        except ValueError as e:
+            logger.warning(f"Error parsing resolution '{resolution.text}': {e}")
+            return default
+
+        logger.warning(f"Resolution cannot be parsed: [{resolution}]")
+        return default
+
+    def parse_accuracy(self, default=None):
+        acc = self.exml.find(
             util.nspath_eval(
                 "gmd:dataQualityInfo/gmd:DQ_DataQuality/gmd:report/gmd:DQ_AbsoluteExternalPositionalAccuracy/gmd:result/gmd:DQ_QuantitativeResult/gmd:value/gco:Record/gco:Real",
                 self.namespaces,
             )
         )
-        if accuracy is not None:
-            custom["rndt"]["accuracy"] = (
-                accuracy if isinstance(accuracy, float) else ast.literal_eval(accuracy.text)
-            ) or 0
-        else:
-            logging.error("accuracy cannot be None, using default value 0")
-            custom["rndt"]["accuracy"] = 0
-        return custom
+
+        if acc is None:
+            logger.info(f"Accuracy not found")
+            return default
+        if isinstance(acc, (float, int)):
+            return float(acc)
+
+        try:
+            eval_acc = ast.literal_eval(acc.text)
+            if isinstance(eval_acc, (float, int)):
+                return float(eval_acc)
+        except ValueError as e:
+            logger.warning(f"Error parsing resolution '{acc.text}': {e}")
+            return default
+
+        logger.warning(f"Accuracy cannot be parsed: [{acc}]")
+        return default
+
 
     def resolve_keywords(self):
         """
@@ -282,3 +333,14 @@ def _get_keywords(keywords, thesaurus_info):
             else:
                 not_tkey.append(text)
         return available, not_tkey, discarded
+
+    def parse_frequency(self):
+        """ This should be moved into the base parser """
+        return self.parse_codelist("gmd:identificationInfo/gmd:MD_DataIdentification/gmd:resourceMaintenance/gmd:MD_MaintenanceInformation/gmd:maintenanceAndUpdateFrequency/gmd:MD_MaintenanceFrequencyCode")
+
+    def get_freq(self, vals):
+        freq = self.parse_frequency()
+        code = freq[0] if freq else None
+        if freq is None:
+            logger.info(f"Frequency not found")
+        vals["maintenance_frequency"] = code or "unknown"