Source code for swh.coarnotify.indexer
# Copyright (C) 2024-2026 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information
"""Metadata mappings for swh-indexer."""
import json
import logging
from typing import Any
from pyld import jsonld
from rdflib import URIRef
from swh.coarnotify.namespaces import CM
from swh.coarnotify.relationships import MENTIONS_RELATIONSHIPS
from swh.indexer.codemeta import CODEMETA_TERMS
from swh.indexer.metadata_mapping.base import BaseExtrinsicMapping
logger = logging.getLogger(__name__)
[docs]
class CoarNotifyValidationError(Exception):
pass
[docs]
def load_and_compact_notification(content: bytes | str) -> dict[str, Any] | None:
"""Load and compact a notification from the REMS.
Errors logs will be written if something went wrong in the process.
Args:
content: the expanded COAR Notification
Returns:
The compacted form of the COAR Notification or None if we weren't able to
read it
"""
try:
raw_json = json.loads(content)
notification = jsonld.compact(
raw_json,
{
"@context": [
"https://www.w3.org/ns/activitystreams",
"https://coar-notify.net",
]
},
)
except json.JSONDecodeError:
logger.error("Failed to parse JSON document: %s", content)
return None
except jsonld.JsonLdError:
logger.error("Failed to compact JSON-LD document: %s", content)
return None
return notification
[docs]
def validate_mention(notification: dict[str, Any]) -> URIRef:
"""Validate minimal notification's requirements before indexation.
FIXME: CN specs (1.0.1) are a bit unclear about what should context_data contains,
especially the id. It would be more logical to find the paper URI in the id and
then some metadata about it, but instead we might find the software URI in the id
and then metadata about the paper. We are trying to make some changes on the
specs but meanwhile we'll skip verifying that context.id == object.as:subject
Args:
notification: a compact form of a COAR Notification
Raises:
CoarNotifyValidationError: something is wrong with the notification payload
Returns:
The relationship type
"""
object_ = notification.get("object", {}).get("object")
if object_ is None:
raise CoarNotifyValidationError(
f"Missing object[as:object] key in {notification}"
)
if not isinstance(object_, str):
raise CoarNotifyValidationError(
f"object[as:object] value is not a string in {notification}"
)
paper = notification.get("context", {}).get("id")
if not paper:
raise CoarNotifyValidationError(f"Missing context[id] key in {notification}")
if not isinstance(paper, str):
raise CoarNotifyValidationError(
f"context[id] value is not a string in {notification}"
)
relationship = notification.get("object", {}).get("relationship")
if not relationship:
raise CoarNotifyValidationError(
f"Missing object[as:relationship] key in {notification}"
)
if not isinstance(relationship, str):
raise CoarNotifyValidationError(
f"object[as:relationship] value is not a string in {notification}"
)
relationship = URIRef(relationship)
if relationship not in MENTIONS_RELATIONSHIPS:
raise CoarNotifyValidationError(
f"object[as:relationship] value is not valid {notification}"
)
notification_id = notification.get("id")
if not notification_id:
raise CoarNotifyValidationError(f"missing id key in {notification}")
if not isinstance(notification_id, str):
raise CoarNotifyValidationError(f"id value is not a string in {notification}")
return relationship
[docs]
class CoarNotifyMentionMapping(BaseExtrinsicMapping):
"""Map & translate a COAR Notify software mention in a CodeMeta format.
COAR Notify mentions are received by ``swh-coarnotify`` and saved expanded.
Mentions contains metadata on a scientific paper that cites a software.
"""
name = "coarnotify-mention-codemeta"
[docs]
@classmethod
def supported_terms(cls) -> list[str]:
codemeta_terms = [term for term in CODEMETA_TERMS if not term.startswith("@")]
relationships_terms = [
str(term)
for term in (
MENTIONS_RELATIONSHIPS.keys() | MENTIONS_RELATIONSHIPS.values()
)
if not term.startswith(str(CM))
]
return codemeta_terms + relationships_terms
[docs]
def translate(self, content: bytes) -> dict[str, Any] | None:
"""Parse JSON and compact the payload to access the mention.
The whole `context` of the `AnnounceRelationship` notification will be indexed
as it contains metadata about the scientific paper citing the software.
TODO: At some point we might need to fetch metadata from the paper URL as COAR
Notifications are not made to contain **all** the metadata but to indicate
where we should find them.
TODO: We will need to handle cancellations of a mention if it was made by
mistake. Maybe we could use the original notification id and an empty context
to overwrite the previous citation when merging documents ? It is with this in
mind that the notification ID is added to the citation.
Args:
content: the raw expanded COAR Notification
Returns:
A CodeMeta citation if the notification was valid or None
"""
notification = load_and_compact_notification(content)
if not notification:
return None
try:
relationship = validate_mention(notification)
except CoarNotifyValidationError as exc:
logger.error(exc)
return None
inverted_relationship = MENTIONS_RELATIONSHIPS[relationship]
citation = {
"@context": ["http://schema.org/", "https://w3id.org/codemeta/3.0"],
str(inverted_relationship): [
{"id": notification["id"], "ScholarlyArticle": notification["context"]}
],
}
return citation
