470 lines
15 KiB
Python
470 lines
15 KiB
Python
from argparse import ArgumentParser
|
|
from collections import UserDict
|
|
import json
|
|
from pathlib import Path
|
|
import re
|
|
import shutil
|
|
from urllib.parse import urljoin
|
|
|
|
from datalad.api import catalog_add, catalog_create, catalog_remove, catalog_set
|
|
from datalad_catalog.schema_utils import get_metadata_sources
|
|
from rdflib import Graph, URIRef
|
|
from rdflib.namespace import RDF, SKOS
|
|
from requests_cache import CachedSession
|
|
import shortuuid
|
|
|
|
|
|
class IgnorantDict(UserDict):
|
|
"""A dict which ignores None values"""
|
|
|
|
def __setitem__(self, key, value):
|
|
if value is not None:
|
|
self.data[key] = value
|
|
|
|
|
|
def query_record(session, pid, collection="public", fmt="json"):
|
|
endpoint_url = urljoin(BASEURL, f"{collection}/record")
|
|
r = session.get(endpoint_url, params={"pid": pid, "format": fmt})
|
|
return r.json() if r.ok else None
|
|
|
|
|
|
def resolve_display_label(session, uriref, include_pid=True, collection="public"):
|
|
"""Resolve a URI to its display label"""
|
|
d = query_record(session, uriref, collection=collection, fmt="json")
|
|
if d is not None:
|
|
label = d.get("display_label", "unknown")
|
|
pid = d.get("pid")
|
|
return f"{label} ({pid})" if include_pid else label
|
|
else:
|
|
return "UNKNOWN"
|
|
|
|
|
|
def fetch_class_records(session, klass, g, collection="public"):
|
|
"""Fetch all records for a class
|
|
|
|
Updates a given graph in-place. Until we have server-side queries,
|
|
we need all records of certain classes to do client-side
|
|
filtering.
|
|
|
|
"""
|
|
endpoint = urljoin(BASEURL, f"{collection}/records/{klass}")
|
|
r = session.get(endpoint, params={"format": "ttl"})
|
|
if r.ok:
|
|
records = r.json()
|
|
for record in records:
|
|
g.parse(data=record, format="turtle")
|
|
elif r.status_code == 413:
|
|
# Request Entity Too Large
|
|
fetch_class_records_paginated(session, klass, g, collection)
|
|
|
|
|
|
def fetch_class_records_paginated(session, klass, g, collection):
|
|
"""Fetch all records for a class using paginated API
|
|
|
|
This is risky because it can take a long, long time. With limited
|
|
records and careful choice of classes, this should not be a
|
|
problem.
|
|
|
|
"""
|
|
endpoint = urljoin(BASEURL, f"{collection}/records/p/{klass}")
|
|
page = 1
|
|
last_page = None
|
|
while last_page is None or page <= last_page:
|
|
r = session.get(endpoint, params={"format": "ttl", "page": page})
|
|
if r.ok:
|
|
res = r.json()
|
|
# parse items from the page
|
|
for record in res.get("items"):
|
|
g.parse(data=record, format="turtle")
|
|
# update last page if not yet known
|
|
if last_page is None:
|
|
last_page = res.get("pages", 0)
|
|
elif last_page is None:
|
|
# if the first request errors out, break to avoid infinite loop
|
|
# for subsequent requests just skip the page
|
|
break
|
|
page += 1
|
|
|
|
|
|
def get_contributor_name(d):
|
|
# start with the preferred labels for person or organization
|
|
for k in ("formatted_name", "name", "short_name"):
|
|
if k in d:
|
|
return d[k]
|
|
# otherwise, try several alternatives
|
|
if "family_name" in d or "given_name" in d:
|
|
# default to Western name order
|
|
name = " ".join([d.get("given_name"), d.get("family_name")]).strip()
|
|
if prefix := d.get("honorific_name_prefix", False):
|
|
name = " ".join(prefix, name)
|
|
if suffix := d.get("honorific_name_suffix", False):
|
|
name = " ".join(prefix, name)
|
|
elif "display_label" in d:
|
|
# a display label may do
|
|
name = d["display_label"]
|
|
else:
|
|
# use pid (required) as last resort
|
|
name = d["pid"]
|
|
return name
|
|
|
|
|
|
def person_to_author(d):
|
|
author = IgnorantDict()
|
|
author["givenName"] = d.get("given_name")
|
|
author["familyName"] = d.get("family_name")
|
|
author["name"] = d.get("formatted_name")
|
|
if (orcid := d.get("orcid")) is not None:
|
|
author["identifiers"] = [{"type": "ORCID", "identifier": orcid}]
|
|
return author.data
|
|
|
|
|
|
def org_to_author(d):
|
|
author = IgnorantDict()
|
|
author["name"] = d.get("name")
|
|
identifiers = []
|
|
# process identifiers associated with the organization
|
|
for identifier in d.get("identifiers", []):
|
|
if "schema_agency" in identifier and "notation" in identifier:
|
|
identifiers.append(
|
|
{
|
|
"type": identifier.get("schema_agency"),
|
|
"identifier": identifier.get("notation"),
|
|
}
|
|
)
|
|
# ROR is likely to be used as the pid, but not included in identifiers
|
|
if "ror" not in {x["type"].lower() for x in identifiers}:
|
|
m = re.search("ror.org/(0[a-z|0-9]{6}[0-9]{2})$", d["pid"])
|
|
if m is not None:
|
|
identifiers.append({"type": "ROR", "identifier": m.group(0)})
|
|
if len(identifiers) > 0:
|
|
author["identifiers"] = identifiers
|
|
return author.data
|
|
|
|
|
|
def resolve_attribution(session, attr):
|
|
attr_object = query_record(session, attr["object"])
|
|
attr_roles = [query_record(session, role) for role in attr["roles"]]
|
|
return {"object": attr_object, "roles": attr_roles}
|
|
|
|
|
|
def process_attributions(ds):
|
|
authors = []
|
|
contributors = {}
|
|
funders = []
|
|
|
|
for x in ds.get("attributed_to", []):
|
|
|
|
attr = resolve_attribution(session, x)
|
|
if attr["object"]["schema_type"] == "trr379ra:TRR379Person":
|
|
author = person_to_author(attr["object"])
|
|
elif attr["object"]["schema_type"] == "trr379ra:TRR379Organization":
|
|
author = org_to_author(attr["object"])
|
|
else:
|
|
continue
|
|
|
|
# go through role labels, single out funder
|
|
role_labels = []
|
|
is_funder = False
|
|
for role in attr["roles"]:
|
|
label = role.get("display_label")
|
|
if label == "Funder":
|
|
# special treatment for marcrel:fnd
|
|
is_funder = True
|
|
else:
|
|
role_labels.append(label)
|
|
|
|
if len(role_labels) > 0:
|
|
# at least one non-funder role required for author & contributor lists
|
|
authors.append(author)
|
|
contributors[get_contributor_name(attr["object"])] = role_labels
|
|
if is_funder:
|
|
# funding goes to a separate funding tab
|
|
funders.append({"name": get_contributor_name(attr["object"])})
|
|
|
|
res = {
|
|
"authors": authors if len(authors) > 0 else None,
|
|
"contributors": contributors if len(contributors) > 0 else None,
|
|
"funders": funders if len(funders) > 0 else None,
|
|
}
|
|
|
|
return res
|
|
|
|
|
|
def process_rules(session, ds):
|
|
spdx_rules = []
|
|
# resolve PIDs to (known) definitions
|
|
for rule_pid in ds.get("rules", []):
|
|
rule = query_record(session, rule_pid)
|
|
if rule is not None and rule["pid"].startswith("spdxlic"):
|
|
spdx_rules.append(rule)
|
|
|
|
# convert single spdx rule to catalog's license
|
|
if len(spdx_rules) == 1:
|
|
rule = spdx_rules[0]
|
|
license = {
|
|
"name": rule.get("title", "unknown"),
|
|
"url": rule.get("pid").replace("spdxlic:", "https://spdx.org/licenses/"),
|
|
}
|
|
else:
|
|
# ignore multiple licenses for now
|
|
license = None
|
|
|
|
return license
|
|
|
|
|
|
def process_data_items(g, ds_pid):
|
|
"""Get Data item - part of - Dataset
|
|
|
|
Finds data items corresponding to the dataset, and counts by kind.
|
|
|
|
"""
|
|
data_item_kind_query = """
|
|
SELECT ?diKind (COUNT(?di) as ?diCount)
|
|
WHERE {
|
|
?di a trr379ra:TRR379DataItem .
|
|
?di dlrelationsmx:part_of ?thisDataset .
|
|
?di dlrelationsmx:kind ?diKind .
|
|
}
|
|
"""
|
|
qres = g.query(data_item_kind_query, initBindings={"thisDataset": URIRef(ds_pid)})
|
|
|
|
# resolve the data item kinds to their names and format them for display in the catalog
|
|
content = {}
|
|
for row in qres:
|
|
kind, count = row
|
|
if (kind is not None) and (count.toPython() != 0):
|
|
|
|
content[resolve_display_label(session, kind)] = count.value
|
|
|
|
return content if len(content) > 0 else None
|
|
|
|
|
|
def process_distributions(g, ds_pid):
|
|
"""Find matching distributions
|
|
|
|
In general these do not have to be DataLad dataset versions
|
|
|
|
"""
|
|
distribution_query = """
|
|
SELECT DISTINCT ?dist ?kind ?convention ?bm
|
|
WHERE {
|
|
?dist a trr379ra:TRR379Distribution .
|
|
?dist dlfilesmx:distribution_of ?thisDataset .
|
|
OPTIONAL { ?dist dlrelationsmx:kind ?kind . }
|
|
OPTIONAL { ?dist dlrelationsmx:conforms_to ?convention . }
|
|
}
|
|
"""
|
|
qres = g.query(distribution_query, initBindings={"thisDataset": URIRef(ds_pid)})
|
|
content = {}
|
|
for row in qres:
|
|
dist, kind, convention, broad = row
|
|
# TRR379Distribution has optional slot: kind
|
|
if kind is not None:
|
|
content[dist.toPython()] = {"kind": resolve_display_label(session, kind)}
|
|
else:
|
|
# in practice we've seen kind declared in broad mappings
|
|
known_kinds = {"dlvocab:datalad-dataset-version"} # could be more
|
|
broad_matches = g.objects(
|
|
subject=dist, predicate=SKOS.broadMatch, unique=True
|
|
)
|
|
for bm in broad_matches:
|
|
if bm.toPython() in known_kinds:
|
|
content[dist.toPython()] = {
|
|
"kind": resolve_display_label(session, bm.toPython())
|
|
}
|
|
break # stop at 1st known
|
|
else:
|
|
# did not find a known one
|
|
content[dist.toPython()] = {"kind": "UNKNOWN"}
|
|
# TRR379Distribution has optional slot: kind
|
|
if convention is not None:
|
|
content[dist.toPython()]["conforms_to"] = resolve_display_label(
|
|
session, convention
|
|
)
|
|
return content if len(content) > 0 else None
|
|
|
|
|
|
def determine_id_ver(ds_pid):
|
|
"""Determine dataset ID and version"""
|
|
m = re.match(
|
|
r"https://pid.datalad.org/datasets/([0-9a-f]{8}-([0-9a-f]{4}-){3}[0-9a-f]{12})",
|
|
ds_pid,
|
|
)
|
|
# regex above is neat but not precise (e.g. 3rd group must start with 4, etc.)
|
|
if m is not None:
|
|
ds_id = m.group(1)
|
|
else:
|
|
# mint a short uuid (equivalent to UUIDv5 with ns:URL if pid starts with https)
|
|
ds_id = shortuuid.uuid(name=ds_pid)
|
|
|
|
ds_ver = "any" # temporary workaround
|
|
|
|
return ds_id, ds_ver
|
|
|
|
|
|
def process_dataset(session, ds_pid, g):
|
|
endpoint = urljoin(BASEURL, "public/record")
|
|
r = session.get(endpoint, params={"pid": ds_pid, "format": "json"})
|
|
if r.ok:
|
|
# todo: graceful exit if not
|
|
d = r.json()
|
|
|
|
# create an empty dataset record
|
|
# required fields: [ "type", "dataset_id", "dataset_version", "metadata_sources"]
|
|
cat_ds = IgnorantDict(
|
|
type="dataset",
|
|
metadata_sources=get_metadata_sources(name="pool-to-catalog", version="0.1.0"),
|
|
)
|
|
# id and version
|
|
cat_ds["dataset_id"], cat_ds["dataset_version"] = determine_id_ver(ds_pid)
|
|
|
|
# additional display - simplifies adding, can pop if not needed
|
|
cat_ds["additional_display"] = []
|
|
|
|
# name, short name, description
|
|
cat_ds["name"] = d.get("title")
|
|
cat_ds["short_name"] = d.get("short_name")
|
|
cat_ds["description"] = d.get("description")
|
|
|
|
# Attributions (authors, funding, contributors)
|
|
attributions = process_attributions(d)
|
|
cat_ds["authors"] = attributions["authors"]
|
|
cat_ds["funding"] = attributions["funders"]
|
|
if attributions["contributors"] is not None:
|
|
cat_ds["additional_display"].append(
|
|
{
|
|
"name": "Contributors",
|
|
"content": attributions["contributors"],
|
|
"icon": "fa-solid fa-users",
|
|
}
|
|
)
|
|
|
|
# Rights (license)
|
|
cat_ds["license"] = process_rules(session, d)
|
|
|
|
# Data items
|
|
data_items = process_data_items(g, ds_pid)
|
|
if data_items is not None:
|
|
cat_ds["additional_display"].append(
|
|
{
|
|
"name": "Data items",
|
|
"content": data_items,
|
|
"icon": "fa-solid fa-file",
|
|
}
|
|
)
|
|
|
|
# Distributions
|
|
distributions = process_distributions(g, ds_pid)
|
|
if distributions is not None:
|
|
cat_ds["additional_display"].append(
|
|
{
|
|
"name": "Distributions",
|
|
"content": distributions,
|
|
"icon": "fa-solid fa-file-export",
|
|
}
|
|
)
|
|
|
|
# clean up
|
|
if len(cat_ds["additional_display"]) == 0:
|
|
cat_ds.pop("additional_display")
|
|
|
|
return cat_ds.data
|
|
|
|
|
|
def change_catalog_favicon(cat_path):
|
|
favicon_dir = cat_path / "assets" / "favicon"
|
|
shutil.rmtree(favicon_dir)
|
|
favicon_dir.mkdir()
|
|
shutil.copy(Path("assets/favicon.ico"), favicon_dir)
|
|
|
|
|
|
def add_to_catalog(cat_path, obj, create=True, delete=True):
|
|
if create and not cat_path.joinpath("config.json").exists():
|
|
catalog_create(cat_path, config_file=Path("assets/config.json"))
|
|
# automagic copies config-specified logo, but favicon needs doing
|
|
change_catalog_favicon(cat_path)
|
|
if delete:
|
|
catalog_remove(
|
|
cat_path,
|
|
dataset_id=obj["dataset_id"],
|
|
dataset_version=obj["dataset_version"],
|
|
reckless=True,
|
|
on_failure="ignore",
|
|
)
|
|
catalog_add(cat_path, metadata=json.dumps(obj))
|
|
|
|
|
|
def set_catalog_homepage(cat_path, datasets):
|
|
# create an empty dataset record
|
|
# required fields: [ "type", "dataset_id", "dataset_version", "metadata_sources"]
|
|
super_id = "main"
|
|
super_version = "latest"
|
|
superds = {
|
|
"type": "dataset",
|
|
"dataset_id": super_id,
|
|
"dataset_version": super_version,
|
|
"name": "TRR379 datasets",
|
|
"metadata_sources": get_metadata_sources(
|
|
name="pool-to-catalog", version="0.1.0"
|
|
),
|
|
"subdatasets": [],
|
|
}
|
|
for subds in datasets:
|
|
superds["subdatasets"].append(
|
|
{
|
|
"dataset_id": subds["dataset_id"],
|
|
"dataset_version": subds["dataset_version"],
|
|
"dataset_path": (
|
|
subds["short_name"]
|
|
if "short_name" in subds
|
|
else subds["dataset_id"]
|
|
),
|
|
}
|
|
)
|
|
add_to_catalog(cat_path, superds)
|
|
catalog_set(
|
|
cat_path,
|
|
property="home",
|
|
dataset_id=super_id,
|
|
dataset_version=super_version,
|
|
reckless="overwrite",
|
|
)
|
|
|
|
|
|
BASEURL = "https://pool.v0.trr379.de/api/"
|
|
|
|
parser = ArgumentParser()
|
|
parser.add_argument("catalog_path", type=Path)
|
|
args = parser.parse_args()
|
|
|
|
# cache requests for 2 hours; enough for re-runs, without having to clear manually
|
|
session = CachedSession(
|
|
".cache/requests-cache/http_cache", backend="sqlite", expire_after=7200
|
|
)
|
|
|
|
# create shared graph for graph-based operations
|
|
# todo: verify if it is useful
|
|
g = Graph()
|
|
|
|
# load up the graph with data items and distributions
|
|
fetch_class_records(session, "TRR379Dataset", g)
|
|
fetch_class_records(session, "TRR379DataItem", g)
|
|
fetch_class_records(session, "TRR379Distribution", g)
|
|
|
|
dataset_ids = (
|
|
uri.toPython()
|
|
for uri in g.subjects(
|
|
predicate=RDF.type,
|
|
object=g.namespace_manager.expand_curie("trr379ra:TRR379Dataset"),
|
|
)
|
|
)
|
|
|
|
# process and add datasets
|
|
cataloged_datasets = []
|
|
for ds_pid in dataset_ids:
|
|
ds = process_dataset(session, ds_pid, g)
|
|
add_to_catalog(args.catalog_path, ds)
|
|
cataloged_datasets.append(ds)
|
|
|
|
# set catalog homepage
|
|
set_catalog_homepage(args.catalog_path, cataloged_datasets)
|