feat: draft - ipt dwca function

Author: nicokant

pyproject.toml

@@ -16,14 +16,24 @@ dependencies = [
   "structlog>=25.5.0",
   "rich>=14.2.0",
   "typer>=0.20.0",
-  "pygeometa>=0.19.0",
+  "pygeometa @ git+https://github.com/nicokant/pygeometa@gbif-eml#egg=pygeometa",
   "orjson>=3.11.4",
   "duckdb>=1.4.2",
   "lxml>=6.0.2",
   "openpyxl>=3.1.5",
   "python-calamine>=0.6.1",
   "requests>=2.32.5",
-  "sling>=1.5.1"
+  "sling>=1.5.1",
+  "beautifulsoup4>=4.13.3",
+  "fsspec>=2025.3.2",
+  "jinja2>=3.1.6",
+  "s3fs>=2025.3.2",
+  "shapely>=2.1.0",
+  "xmltodict>=0.14.2",
+  "pyarrow>=20.0.0",
+  "httpx>=0.28.1",
+  "setuptools>=65.5.0",
+  "backoff>=2.1.0"
 ]
 description = ""
 license = "GPL-3.0+"
@@ -39,7 +49,10 @@ nva_sync = "datasync.nva:app"
 ubw_sync = "datasync.ubw:app"
 
 [tool.deptry.per_rule_ignores]
-DEP002 = ["rich", "numpy"]
+DEP002 = ["rich", "numpy", "setuptools"]
+
+[tool.hatch.metadata]
+allow-direct-references = true
 
 [tool.ruff]
 fix = true

src/datasync/ipt/__init__.py

@@ -0,0 +1,121 @@
+import json
+
+import pyarrow as pa
+from lxml import etree
+from pygeometa.schemas.gbif_eml import GBIF_EMLOutputSchema
+from pygeometa.schemas.iso19139 import ISO19139OutputSchema
+from shapely.geometry import box
+
+from ..settings import env as logger
+from .settings import (
+    AWS_ENDPOINT_URL,
+    CSW_PATH,
+    GEOAPI_PUBLISH_URL,
+    IPT_URL,
+    RESOURCES_PREFIX,
+    S3_BUCKET,
+    conn,
+)
+
+PARSER = etree.XMLParser(resolve_entities=False)
+eml = GBIF_EMLOutputSchema()
+iso = ISO19139OutputSchema()
+
+
+def get_anytext(bag):
+    """
+    generate bag of text for free text searches
+    accepts list of words, string of XML, or etree.Element
+    """
+
+    if isinstance(bag, list):  # list of words
+        return " ".join([_f for _f in bag if _f]).strip()
+    else:  # xml
+        if isinstance(bag, bytes) or isinstance(bag, str):
+            # serialize to lxml
+            bag = etree.fromstring(bag, PARSER)  # noqa: S320
+        # get all XML element content
+        return " ".join([value.strip() for value in bag.xpath("//text()")])
+
+
+def eml_to_record(ds, text):
+    metadata = eml.import_(text)
+
+    metadata["metadata"]["identifier"] = f"ipt__{ds['id']}"
+
+    xml = iso.write(metadata)
+    fts = get_anytext(xml)
+    idf = metadata["identification"]
+    bbox = idf["extents"]["spatial"][0]["bbox"]
+
+    contribs = []
+    for role, contact in metadata["contact"].items():
+        role = role.split("_")[0]
+        contribs.append(contact["individualname"])
+
+    keywords = []
+    for _k, v in idf["keywords"].items():
+        keywords += v["keywords"]
+
+    if ds.get("ipt_dwca"):
+        links = [
+            {
+                "name": "Parquet",
+                "description": "The resource as (geo)parquet file",
+                "protocol": "FILE:GEO",
+                "url": f"{AWS_ENDPOINT_URL}/{S3_BUCKET}{RESOURCES_PREFIX}{ds['id']}.parquet",  # noqa: E501
+            },
+            {
+                "name": "DWCA",
+                "description": "The resource as Darwin Core Archive",
+                "protocol": "file",
+                "url": f"{IPT_URL}/archive.do?r={ds['id']}",  # noqa: E501
+            },
+        ]
+
+        if GEOAPI_PUBLISH_URL:
+            links.append(
+                {
+                    "name": "OGC API Feature",
+                    "description": "OGC REST API to the resource",
+                    "protocol": "OGCFeat",
+                    "url": f"{GEOAPI_PUBLISH_URL}/collections/ipt__{ds['id']}/items?f=json",  # noqa: E501
+                },
+            )
+    else:
+        links = []
+
+    return {
+        "identifier": metadata["metadata"]["identifier"],
+        "typename": "gmd:MD_Metadata",
+        "schema": "http://www.isotc211.org/2005/gmd",
+        "mdsource": "local",
+        "insert_date": idf["dates"]["publication"],
+        "title": ds["title"],
+        "date_modified": idf["dates"]["publication"],
+        "type": "dataset",
+        "format": None,
+        "wkt_geometry": box(*bbox).wkt,
+        "metadata": xml,
+        "xml": xml,
+        "keywords": ", ".join(set(keywords)),
+        "metadata_type": "application/xml",
+        "anytext": fts,
+        "abstract": metadata["identification"]["abstract"],
+        "date": idf["dates"]["publication"],
+        "creator": "Norsk institutt for naturforskning (NINA)",
+        "publisher": "Norsk institutt for naturforskning (NINA)",
+        "contributor": "; ".join(set(contribs)),
+        "links": json.dumps(links),
+    }
+
+
+def write_eml_record(rows):
+    logger.info("converting to arrow")
+    records = pa.Table.from_pylist(rows)  # noqa: F841
+    logger.info("write to S3")
+    conn.sql("from records").write_parquet(
+        f"s3://{S3_BUCKET}{CSW_PATH}",
+        compression="zstd",
+        overwrite=True,
+    )

src/datasync/ipt/csw.py

@@ -0,0 +1,61 @@
+from ..libs.dms import upsert_dms_element
+from .settings import DMS_PROJECT_ID
+
+
+def create_dms_dataset(ds, parquet_url):
+    ds_id = "IPT__" + ds["id"]
+
+    # Upsert dataset
+    dataset_data = {
+        "id": ds_id,
+        "title": ds["title"],
+        "metadata": {},
+        "version": "latest",
+        "project_id": DMS_PROJECT_ID,
+    }
+    dataset_update_data = {
+        "title": ds["title"],
+        "version": "latest",
+    }
+
+    upsert_dms_element("datasets", ds_id, dataset_data, dataset_update_data)
+
+    if parquet_url:
+        # Upsert tabular resource
+        tabular_resource_data = {
+            "id": ds_id + "_parquet",
+            "dataset_id": ds_id,
+            "title": ds["title"] + " Parquet",
+            "uri": parquet_url,
+            "access_type": "public",
+            "role": "data",
+        }
+        tabular_resource_update_data = {
+            "title": ds["title"] + " Parquet",
+            "uri": parquet_url,
+        }
+
+        upsert_dms_element(
+            "tabularresources",
+            ds_id + "_parquet",
+            tabular_resource_data,
+            tabular_resource_update_data,
+        )
+
+        # Upsert resource
+        resource_data = {
+            "id": ds_id + "_dwca",
+            "dataset_id": ds_id,
+            "title": ds["title"] + " DWCA",
+            "uri": ds["ipt_dwca"],
+            "access_type": "public",
+            "role": "data",
+        }
+        resource_update_data = {
+            "title": ds["title"] + " DWCA",
+            "uri": ds["ipt_dwca"],
+        }
+
+        upsert_dms_element(
+            "resources", ds_id + "_dwca", resource_data, resource_update_data
+        )

src/datasync/ipt/dms.py

@@ -0,0 +1,44 @@
+import pathlib
+import re
+
+import fsspec
+from bs4 import BeautifulSoup
+
+
+class SourceLayer:
+    def __init__(self, node, base_path, extension=False) -> None:
+        self.type = pathlib.Path(node.find("location").text).stem
+        self.headers = []
+        self.path = f"zip://{base_path}/{node.find('location').text}"
+
+        with fsspec.open(
+            f"zip://{node.find('location').text}::{base_path}",
+            encoding=node["encoding"],
+            mode="r",
+        ) as f:
+            sep = re.compile(node["fieldsTerminatedBy"])
+            self.headers = re.split(sep, f.readline().rstrip())
+
+            # extensions nodes have a "coreid" fields that contains the id
+            # of "core" row, this is needed for the join
+            id_field_lookup = "coreid" if extension else "id"
+            self.id = self.headers[int(node.find(id_field_lookup)["index"])]
+
+    def __str__(self) -> str:
+        return self.type
+
+
+def get_context_from_metafile(resource_path: str):
+    meta_file = fsspec.open(f"zip://meta.xml::{resource_path}")
+    with meta_file as meta:
+        soup = BeautifulSoup(meta, features="lxml-xml")
+        extensions = []
+        core = SourceLayer(soup.find("core"), resource_path)
+        all_columns = set(core.headers)
+
+        for extension in soup.find_all("extension"):
+            ext = SourceLayer(extension, resource_path, extension=True)
+            extensions.append(ext)
+            all_columns.union(ext.headers)
+
+        return {"core": core, "extensions": extensions, "columns": all_columns}

src/datasync/ipt/dwca.py

@@ -0,0 +1,63 @@
+import pyarrow as pa
+from lxml import etree
+from pygeometa.schemas.gbif_eml import GBIF_EMLOutputSchema
+
+from ..settings import env as logger
+from .settings import AWS_ENDPOINT_URL, GEOAPI_PATH, RESOURCES_PREFIX, S3_BUCKET, conn
+
+PARSER = etree.XMLParser(resolve_entities=False)
+eml = GBIF_EMLOutputSchema()
+
+
+def to_pygeoapi_resource(ds, eml_text):
+    metadata = eml.import_(eml_text)
+
+    idf = metadata["identification"]
+    spatial = idf["extents"]["spatial"][0]
+
+    contribs = []
+    for role, contact in metadata["contact"].items():
+        role = role.split("_")[0]
+        contribs.append(contact["individualname"])
+
+    keywords = []
+    for _k, v in idf["keywords"].items():
+        keywords += v["keywords"]
+
+    return {
+        "id": f"ipt__{ds['id']}",
+        "type": "collection",
+        "visibility": "default",
+        "title": ds["title"],
+        "extents": {"spatial": spatial},
+        "keywords": list(set(keywords)),
+        "description": metadata["identification"]["abstract"],
+        "providers": [
+            {
+                "type": "feature",
+                "name": "OGR",
+                "default": True,
+                "id_field": "fid",
+                "editable": False,
+                "storage_crs": "http://www.opengis.net/def/crs/OGC/1.3/CRS84",
+                "crs": [
+                    "http://www.opengis.net/def/crs/OGC/1.3/CRS84",
+                    "http://www.opengis.net/def/crs/EPSG/0/4326",
+                ],
+                "data": {
+                    "source_type": "Parquet",
+                    "source": f"/vsicurl/{AWS_ENDPOINT_URL}/{S3_BUCKET}{RESOURCES_PREFIX}{ds['id']}.parquet",  # noqa: E501
+                },
+                "layer": ds["id"],
+            }
+        ],
+    }
+
+
+def write_pygeoapi_resources(rows):
+    logger.info("converting to arrow")
+    records = pa.Table.from_pylist(rows)  # noqa: F841
+    logger.info("write to S3")
+    conn.sql(f"""
+        COPY records to 's3://{S3_BUCKET}{GEOAPI_PATH}' (FORMAT json, ARRAY true)
+    """)  # noqa: E501

src/datasync/ipt/geoapi.py

@@ -0,0 +1,27 @@
+import httpx
+import xmltodict
+from bs4 import BeautifulSoup
+
+from .settings import IPT_URL
+
+
+def get_datasets():
+    res = httpx.get(f"{IPT_URL}/rss")
+    soup = BeautifulSoup(res.text, features="lxml-xml")
+    for item in soup.find_all("item"):
+        content = {
+            k.replace(":", "_"): v
+            for k, v in xmltodict.parse(item.prettify())["item"].items()
+        }
+        resource_id = content["link"].split("=")[1]
+        yield {
+            **content,
+            "id": resource_id,
+            "version": content["guid"]["#text"].split("/")[1].replace("v", ""),
+        }
+
+
+def get_dataset_metadata(resource_id: str):
+    url = IPT_URL + "/eml.do?r=" + resource_id
+    res = httpx.get(url)
+    return res.text