Harvest dataservices

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## Current (in progress)
 
+- Harvest dataservices [#3029](https://github.com/opendatateam/udata/pull/3029)
 - Refactor catalog exports [#3052](https://github.com/opendatateam/udata/pull/3052)
 - Add a filter to filter dataservices by dataset [#3056](https://github.com/opendatateam/udata/pull/3056)
 - Fix reuses' datasets references [#3057](https://github.com/opendatateam/udata/pull/3057)

js/components/harvest/item.vue

@@ -34,6 +34,19 @@
                     :dataset="item.dataset">
                 </dataset-card>
             </dd>
+            <dt v-if="item.dataservice">{{ _('Dataservice') }}</dt>
+            <dd v-if="item.dataservice">
+                <div class="card">
+                    <div class="card-body">
+                        <h4>
+                            <a :title="item.dataservice.title" :href="item.dataservice.self_web_url">
+                                {{ item.dataservice.title | truncate 80 }}
+                            </a>
+                            <div class="clamp-3">{{{ item.dataservice.description | markdown 180 }}}</div>
+                        </h4>
+                    </div>
+                </div>
+            </dd>
             <dt v-if="item.errors.length">{{ _('Errors') }}</dt>
             <dd v-if="item.errors.length">
                 <div v-for="error in item.errors">

udata/api_fields.py

@@ -70,10 +70,14 @@ def convert_db_to_field(key, field, info = {}):
         constructor_write = restx_fields.String
     elif isinstance(field, mongo_fields.EmbeddedDocumentField):
         nested_fields = info.get('nested_fields')
-        if nested_fields is None:
-            raise ValueError(f"EmbeddedDocumentField `{key}` requires a `nested_fields` param to serialize/deserialize.")
+        if nested_fields is not None:
+            constructor = lambda **kwargs: restx_fields.Nested(nested_fields, **kwargs)
+        elif hasattr(field.document_type_obj, '__read_fields__'):
+            constructor_read = lambda **kwargs: restx_fields.Nested(field.document_type_obj.__read_fields__, **kwargs)
+            constructor_write = lambda **kwargs: restx_fields.Nested(field.document_type_obj.__write_fields__, **kwargs)
+        else:
+            raise ValueError(f"EmbeddedDocumentField `{key}` requires a `nested_fields` param to serialize/deserialize or a `@generate_fields()` definition.")
 
-        constructor = lambda **kwargs: restx_fields.Nested(nested_fields, **kwargs)
     else:
         raise ValueError(f"Unsupported MongoEngine field type {field.__class__.__name__}")
 

udata/core/dataservices/models.py

@@ -31,6 +31,35 @@ def hidden(self):
                     db.Q(deleted_at__ne=None) |
                     db.Q(archived_at__ne=None))
 
+@generate_fields()
+class HarvestMetadata(db.EmbeddedDocument):
+    backend = field(db.StringField())
+    domain = field(db.StringField())
+
+    source_id = field(db.StringField())
+    source_url = field(db.URLField())
+
+    remote_id = field(db.StringField())
+    remote_url = field(db.URLField())
+
+    # If the node ID is a `URIRef` it means it links to something external, if it's not an `URIRef` it's often a
+    # auto-generated ID just to link multiple RDF node togethers. When exporting as RDF to other catalogs, we 
+    # want to re-use this node ID (only if it's not auto-generated) to improve compatibility.
+    uri = field(
+        db.URLField(),
+        description="RDF node ID if it's an `URIRef`. `None` if it's not present or if it's a random auto-generated ID inside the graph.",
+    )
+
+    created_at = field(
+        db.DateTimeField(),
+        description="Date of the creation as provided by the harvested catalog"
+    )
+    last_update = field(
+        db.DateTimeField(),
+        description="Date of the last harvesting"
+    )
+    archived_at = field(db.DateTimeField())
+
 @generate_fields()
 class Dataservice(WithMetrics, Owned, db.Document):
     meta = {
@@ -119,12 +148,18 @@ class Dataservice(WithMetrics, Owned, db.Document):
         },
     )
 
+    harvest = field(
+        db.EmbeddedDocumentField(HarvestMetadata),
+        readonly=True,
+    )
+
     @function_field(description="Link to the API endpoint for this dataservice")
     def self_api_url(self):
         return endpoint_for('api.dataservice', dataservice=self, _external=True)
 
-    def self_web_url():
-        pass
+    @function_field(description="Link to the udata web page for this dataservice")
+    def self_web_url(self):
+        return endpoint_for('dataservices.show', dataservice=self, _external=True)
 
     # TODO
     # frequency = db.StringField(choices=list(UPDATE_FREQUENCIES.keys()))

udata/core/dataservices/rdf.py

@@ -0,0 +1,58 @@
+
+from datetime import datetime
+from typing import List, Optional
+from rdflib import RDF, Graph, URIRef
+
+from udata.core.dataservices.models import Dataservice, HarvestMetadata as HarvestDataserviceMetadata
+from udata.core.dataset.models import Dataset, License
+from udata.core.dataset.rdf import sanitize_html
+from udata.harvest.models import HarvestSource
+from udata.rdf import DCAT, DCT, contact_point_from_rdf, rdf_value, remote_url_from_rdf, theme_labels_from_rdf, themes_from_rdf, url_from_rdf
+
+def dataservice_from_rdf(graph: Graph, dataservice: Dataservice, node, all_datasets: List[Dataset]) -> Dataservice :
+    '''
+    Create or update a dataset from a RDF/DCAT graph
+    '''
+    if node is None:  # Assume first match is the only match
+        node = graph.value(predicate=RDF.type, object=DCAT.DataService)
+
+    d = graph.resource(node)
+
+    dataservice.title = rdf_value(d, DCT.title)
+    dataservice.description = sanitize_html(d.value(DCT.description) or d.value(DCT.abstract))
+
+    dataservice.base_api_url = url_from_rdf(d, DCAT.endpointURL)
+    dataservice.endpoint_description_url = url_from_rdf(d, DCAT.endpointDescription)
+
+    dataservice.contact_point = contact_point_from_rdf(d, dataservice) or dataservice.contact_point
+
+    datasets = []
+    for dataset_node in d.objects(DCAT.servesDataset):
+        id = dataset_node.value(DCT.identifier)
+        dataset = next((d for d in all_datasets if d is not None and d.harvest.remote_id == id), None)
+
+        if dataset is None:
+            # We try with `endswith` because Europe XSLT have problems with IDs. Sometimes they are prefixed with the domain of the catalog, sometimes not.
+            dataset = next((d for d in all_datasets if d is not None and d.harvest.remote_id.endswith(id)), None)
+
+        if dataset is not None:
+            datasets.append(dataset.id)
+
+    if datasets:
+        dataservice.datasets = datasets
+
+    license = rdf_value(d, DCT.license)
+    if license is not None:
+        dataservice.license = License.guess(license)
+
+    if not dataservice.harvest:
+        dataservice.harvest = HarvestDataserviceMetadata()
+
+    dataservice.harvest.uri = d.identifier.toPython() if isinstance(d.identifier, URIRef) else None
+    dataservice.harvest.remote_url = remote_url_from_rdf(d)
+    dataservice.harvest.created_at = rdf_value(d, DCT.issued)
+    dataservice.metadata_modified_at = rdf_value(d, DCT.modified)
+
+    dataservice.tags = themes_from_rdf(d)
+
+    return dataservice

udata/core/dataset/rdf.py

@@ -6,7 +6,6 @@
 import logging
 
 from datetime import date
-from html.parser import HTMLParser
 from typing import Optional
 from dateutil.parser import parse as parse_dt
 from flask import current_app
@@ -18,14 +17,14 @@
 
 from udata import i18n, uris
 from udata.core.spatial.models import SpatialCoverage
-from udata.frontend.markdown import parse_html
 from udata.core.dataset.models import HarvestDatasetMetadata, HarvestResourceMetadata
-from udata.models import db, ContactPoint
+from udata.harvest.exceptions import HarvestSkipException
+from udata.models import db
 from udata.rdf import (
-    DCAT, DCATAP, DCT, FREQ, SCV, SKOS, SPDX, SCHEMA, EUFREQ, EUFORMAT, IANAFORMAT, VCARD, RDFS,
-    HVD_LEGISLATION, namespace_manager, schema_from_rdf, url_from_rdf
+    DCAT, DCATAP, DCT, FREQ, SCV, SKOS, SPDX, SCHEMA, EUFREQ, EUFORMAT, IANAFORMAT, TAG_TO_EU_HVD_CATEGORIES, RDFS, 
+    namespace_manager, rdf_value, remote_url_from_rdf, sanitize_html, schema_from_rdf, themes_from_rdf, url_from_rdf, HVD_LEGISLATION,
+    contact_point_from_rdf,
 )
-from udata.tags import slug as slugify_tag
 from udata.utils import get_by, safe_unicode
 from udata.uris import endpoint_for
 
@@ -77,44 +76,6 @@
     EUFREQ.NEVER: 'punctual',
 }
 
-# Map High Value Datasets URIs to keyword categories
-EU_HVD_CATEGORIES = {
-    "http://data.europa.eu/bna/c_164e0bf5": "Météorologiques",
-    "http://data.europa.eu/bna/c_a9135398": "Entreprises et propriété d'entreprises",
-    "http://data.europa.eu/bna/c_ac64a52d": "Géospatiales",
-    "http://data.europa.eu/bna/c_b79e35eb": "Mobilité",
-    "http://data.europa.eu/bna/c_dd313021": "Observation de la terre et environnement",
-    "http://data.europa.eu/bna/c_e1da4e07": "Statistiques"
-}
-TAG_TO_EU_HVD_CATEGORIES = {slugify_tag(EU_HVD_CATEGORIES[uri]): uri for uri in EU_HVD_CATEGORIES}
-
-
-class HTMLDetector(HTMLParser):
-    def __init__(self, *args, **kwargs):
-        HTMLParser.__init__(self, *args, **kwargs)
-        self.elements = set()
-
-    def handle_starttag(self, tag, attrs):
-        self.elements.add(tag)
-
-    def handle_endtag(self, tag):
-        self.elements.add(tag)
-
-
-def is_html(text):
-    parser = HTMLDetector()
-    parser.feed(text)
-    return bool(parser.elements)
-
-
-def sanitize_html(text):
-    text = text.toPython() if isinstance(text, Literal) else ''
-    if is_html(text):
-        return parse_html(text)
-    else:
-        return text.strip()
-
-
 def temporal_to_rdf(daterange, graph=None):
     if not daterange:
         return
@@ -255,18 +216,6 @@ def dataset_to_rdf(dataset, graph=None):
 }
 
 
-def serialize_value(value):
-    if isinstance(value, (URIRef, Literal)):
-        return value.toPython()
-    elif isinstance(value, RdfResource):
-        return value.identifier.toPython()
-
-
-def rdf_value(obj, predicate, default=None):
-    value = obj.value(predicate)
-    return serialize_value(value) if value else default
-
-
 def temporal_from_literal(text):
     '''
     Parse a temporal coverage from a literal ie. either:
@@ -341,29 +290,6 @@ def temporal_from_rdf(period_of_time):
         # so we log the error for future investigation and improvement
         log.warning('Unable to parse temporal coverage', exc_info=True)
 
-
-def contact_point_from_rdf(rdf, dataset):
-    contact_point = rdf.value(DCAT.contactPoint)
-    if contact_point:
-        name = rdf_value(contact_point, VCARD.fn) or ''
-        email = (rdf_value(contact_point, VCARD.hasEmail)
-                 or rdf_value(contact_point, VCARD.email)
-                 or rdf_value(contact_point, DCAT.email))
-        if not email:
-            return
-        email = email.replace('mailto:', '').strip()
-        if dataset.organization:
-            contact_point = ContactPoint.objects(
-                name=name, email=email, organization=dataset.organization).first()
-            return (contact_point or
-                    ContactPoint(name=name, email=email, organization=dataset.organization).save())
-        elif dataset.owner:
-            contact_point = ContactPoint.objects(
-                name=name, email=email, owner=dataset.owner).first()
-            return (contact_point or
-                    ContactPoint(name=name, email=email, owner=dataset.owner).save())
-
-
 def spatial_from_rdf(graph):
     geojsons = []
     for term in graph.objects(DCT.spatial):
@@ -503,43 +429,6 @@ def title_from_rdf(rdf, url):
         else:
             return i18n._('Nameless resource')
 
-
-def remote_url_from_rdf(rdf):
-    '''
-    Return DCAT.landingPage if found and uri validation succeeds.
-    Use RDF identifier as fallback if uri validation succeeds.
-    '''
-    landing_page = url_from_rdf(rdf, DCAT.landingPage)
-    uri = rdf.identifier.toPython()
-    for candidate in [landing_page, uri]:
-        if candidate:
-            try:
-                uris.validate(candidate)
-                return candidate
-            except uris.ValidationError:
-                pass
-
-
-def theme_labels_from_rdf(rdf):
-    '''
-    Get theme labels to use as keywords.
-    Map HVD keywords from known URIs resources if HVD support is activated.
-    '''
-    for theme in rdf.objects(DCAT.theme):
-        if isinstance(theme, RdfResource):
-            uri = theme.identifier.toPython()
-            if current_app.config['HVD_SUPPORT'] and uri in EU_HVD_CATEGORIES:
-                label = EU_HVD_CATEGORIES[uri]
-                # Additionnally yield hvd keyword
-                yield 'hvd'
-            else:
-                label = rdf_value(theme, SKOS.prefLabel)
-        else:
-            label = theme.toPython()
-        if label:
-            yield label
-
-
 def resource_from_rdf(graph_or_distrib, dataset=None, is_additionnal=False):
     '''
     Map a Resource domain model to a DCAT/RDF graph
@@ -617,6 +506,9 @@ def dataset_from_rdf(graph: Graph, dataset=None, node=None):
     d = graph.resource(node)
 
     dataset.title = rdf_value(d, DCT.title)
+    if not dataset.title:
+        raise HarvestSkipException("missing title on dataset")
+
     # Support dct:abstract if dct:description is missing (sometimes used instead)
     description = d.value(DCT.description) or d.value(DCT.abstract)
     dataset.description = sanitize_html(description)
@@ -634,9 +526,7 @@ def dataset_from_rdf(graph: Graph, dataset=None, node=None):
     if acronym:
         dataset.acronym = acronym
 
-    tags = [tag.toPython() for tag in d.objects(DCAT.keyword)]
-    tags += theme_labels_from_rdf(d)
-    dataset.tags = list(set(tags))
+    dataset.tags = themes_from_rdf(d)
 
     temporal_coverage = temporal_from_rdf(d.value(DCT.temporal))
     if temporal_coverage:

udata/harvest/api.py

@@ -5,6 +5,7 @@
 from udata.api import api, API, fields
 from udata.auth import admin_permission
 
+from udata.core.dataservices.models import Dataservice
 from udata.core.dataset.api_fields import dataset_ref_fields, dataset_fields
 from udata.core.organization.api_fields import org_ref_fields
 from udata.core.organization.permissions import EditOrganizationPermission
@@ -45,6 +46,9 @@ def backends_ids():
     'dataset': fields.Nested(dataset_ref_fields,
                              description='The processed dataset',
                              allow_null=True),
+    'dataservice': fields.Nested(Dataservice.__read_fields__,
+                             description='The processed dataservice',
+                             allow_null=True),
     'status': fields.String(description='The item status',
                             required=True,
                             enum=list(HARVEST_ITEM_STATUS)),