feat: get taxonomy from GBIF

Author: frafra

pyproject.toml

@@ -25,7 +25,7 @@ dependencies = [
   "typer>=0.20.0",
   "pygeometa @ git+https://github.com/nicokant/pygeometa@gbif-eml#egg=pygeometa",
   "orjson>=3.11.4",
-  "duckdb>=1.4.3",
+  "duckdb>=1.4.2",
   "lxml>=6.0.2",
   "openpyxl>=3.1.5",
   "python-calamine>=0.6.1",
@@ -41,7 +41,7 @@ dependencies = [
   "httpx>=0.28.1",
   "setuptools>=65.5.0",
   "backoff>=2.1.0",
-  "pyyaml>=6.0.3",
+  "pyyaml>=6.0.3"
 ]
 description = ""
 license = "GPL-3.0+"

src/datasync/gbif_backbone.py

@@ -0,0 +1,51 @@
+from importlib.resources import files
+
+import duckdb
+import fsspec
+import typer
+
+from .libs.helpers import pa_read_tsv
+from .settings import (
+    env,
+    log,
+)
+
+resources = files(__package__).joinpath("gbif_backbone")
+taxon_sql = resources.joinpath("taxon.sql").read_text()
+vernacularname_sql = resources.joinpath("vernacularname.sql").read_text()
+
+
+log.debug("Importing GBIF Backbone settings")
+
+
+GBIF_BACKBONE_DUCKDB_NAME = env.path(
+    "GBIF_BACKBONE_DUCKDB_FILE_NAME", default="gbif_backbone.duckdb"
+).root
+GBIF_BACKBONE_URL = env.url(
+    "GBIF_BACKBONE_URL",
+    default="https://hosted-datasets.gbif.org/datasets/backbone/current/backbone.zip",
+)
+
+app = typer.Typer(help="export GBIF Backbone data to DuckDB database")
+
+
+def import_taxon(conn, archive):
+    log.debug("Importing Taxon.tsv")
+    pa_taxon = pa_read_tsv(archive, "Taxon.tsv")  # noqa: F841
+    conn.execute(taxon_sql)
+
+
+def import_vernacular_name(conn, archive):
+    log.debug("Importing VernacularName.tsv")
+    pa_vernacular_names = pa_read_tsv(archive, "VernacularName.tsv")  # noqa: F841
+    conn.execute(vernacularname_sql)
+
+
+@app.command()
+def import_all():
+    """Import GBIF Backbone data into a DuckDB database."""
+    archive = fsspec.filesystem("zip", fo=GBIF_BACKBONE_URL.geturl(), mode="r")
+    with duckdb.connect(GBIF_BACKBONE_DUCKDB_NAME) as conn:
+        import_taxon(conn, archive)
+        import_vernacular_name(conn, archive)
+    log.info("GBIF Backbone data imported successfully")

src/datasync/gbif_backbone/taxon.sql

@@ -0,0 +1,6 @@
+CREATE OR REPLACE TABLE taxon AS FROM pa_taxon;
+CREATE INDEX taxon_taxonid ON taxon (taxonid);
+PRAGMA create_fts_index(
+    'taxon', 'taxonID', 'canonicalName',
+    overwrite=TRUE
+);

src/datasync/gbif_backbone/vernacularname.sql

@@ -0,0 +1,11 @@
+CREATE OR REPLACE TABLE vernacular_name AS
+SELECT DISTINCT
+    CONCAT_WS('|', taxonid, language, vernacularname) AS vernacularid,
+    *
+FROM pa_vernacular_names;
+CREATE INDEX vernacular_names_taxonid ON vernacular_name (taxonid);
+CREATE INDEX vernacular_names_language ON vernacular_name (language);
+PRAGMA create_fts_index(
+    'vernacular_name', 'vernacularID', 'vernacularName',
+    overwrite = TRUE
+);

src/datasync/libs/helpers.py

@@ -2,6 +2,7 @@
 
 from lxml import etree
 from lxml.etree import _Element
+from pyarrow import csv
 
 PARSER: etree.XMLParser = etree.XMLParser(resolve_entities=False)
 
@@ -29,3 +30,15 @@ def get_anytext(bag: str | _Element | list[str]) -> str:
         # NOTE: this should never happen as the xpath evaluation always returns a list
         # but the type annotation is generic as xpath might return any type
         raise TypeError("xpath result was not a list of strings")
+
+
+TSV_PARSE_OPTIONS = csv.ParseOptions(
+    delimiter="\t", quote_char=False, double_quote=False
+)
+
+
+def pa_read_tsv(archive, filename):
+    return csv.read_csv(
+        archive.open(filename),
+        parse_options=TSV_PARSE_OPTIONS,
+    )

src/datasync/main.py

@@ -4,7 +4,16 @@
 
 import typer
 
-from . import dms, grass, ninagen, nva, pit_registering_salmon, services, ubw
+from . import (
+    dms,
+    gbif_backbone,
+    grass,
+    ninagen,
+    nva,
+    pit_registering_salmon,
+    services,
+    ubw,
+)
 
 app = typer.Typer(
     help="Provide subcommands for synchronizing different resources, see subcommands"
@@ -16,6 +25,7 @@
 app.add_typer(pit_registering_salmon.app, name="pit-registering-salmon")
 app.add_typer(grass.app, name="grass-gis")
 app.add_typer(services.app, name="services")
+app.add_typer(gbif_backbone.app, name="gbif-backbone")
 
 if __name__ == "__main__":
     app()