17425918180

Committed 03 Sep 2025 07:11AM UTC coverage: 95.796% (-0.6%) from 96.378%

Build # 17425918180

Build Type

push

github

Committed by

PascalRepond

Commit Message

translations: extract messages

Co-Authored-by: Pascal Repond <pascal.repond@rero.ch>

Run Details

7816 of 8159 relevant lines covered (95.8%)

0.96 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

99.33

/sonar/modules/documents/serializers/schemas/dc.py

# Swiss Open Access Repository
# Copyright (C) 2021 RERO
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, version 3 of the License.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

"""Dublin Core marshmallow schema."""

import re

from flask import current_app, request
from marshmallow import fields

from sonar.modules.documents.api import DocumentRecord
from sonar.modules.documents.views import part_of_format

from .base_schema import BaseSchema


class DublinCoreSchema(BaseSchema):
    """Schema for records v1 in JSON."""

    contributors = fields.Method("get_contributors")
    creators = fields.Method("get_creators")
    dates = fields.Method("get_dates")
    descriptions = fields.Method("get_descriptions")
    formats = fields.Method("get_formats")
    identifiers = fields.Method("get_identifiers")
    languages = fields.Method("get_languages")
    publishers = fields.Method("get_publishers")
    relations = fields.Method("get_relations")
    rights = fields.Method("get_rights")
    sources = fields.Method("get_sources")
    subjects = fields.Method("get_subjects")
    titles = fields.Method("get_titles")
    types = fields.Method("get_types")

    def translate_language(self, language):
        """Translate language code ISO-639-3 to ISO-639-2 if possible.

        :param language: language with ISO-639-3 format.
        :returns: language code ISO-639-2 if possible or ISO-639-3.
        """
        langs = current_app.config["SONAR_APP_LANGUAGES_MAP"]
        if langs.get(language):
            return langs[language]
        return language

    def get_contributors(self, obj):
        """Get contributors."""
        return [
            self.format_contributor(contributor)
            for contributor in obj["metadata"].get("contribution", [])
            if contributor["role"][0] != "cre" and contributor["agent"].get("preferred_name")
        ]

    def get_creators(self, obj):
        """Get creators."""
        return [
            self.format_contributor(contributor)
            for contributor in obj["metadata"].get("contribution", [])
            if contributor["role"][0] == "cre" and contributor["agent"].get("preferred_name")
        ]

    def get_dates(self, obj):
        """Get dates."""
        items = [
            provision_activity["startDate"]
            for provision_activity in obj["metadata"].get("provisionActivity", [])
            if provision_activity["type"] == "bf:Publication" and provision_activity.get("startDate")
        ]

        if obj["metadata"].get("mainFile") and obj["metadata"]["mainFile"]["restriction"]["date"]:
            items.append(f"info:eu-repo/date/embargoEnd/{obj['metadata']['mainFile']['embargo_date']}")

        return items

    def get_descriptions(self, obj):
        """Get descriptions."""
        items = []
        for abstract in obj["metadata"].get("abstracts", []):
            if "language" in abstract:
                items.append(
                    {
                        "@attrs": [
                            {
                                "prefix": "xml",
                                "name": "lang",
                                "value": self.translate_language(abstract["language"]),
                            }
                        ],
                        "value": abstract["value"],
                    }
                )
            else:
                items.append(abstract["value"])

        return items

    def get_formats(self, obj):
        """Get formats."""
        main_file = obj["metadata"].get("mainFile")

        if main_file and main_file.get("mimetype"):
            return [main_file["mimetype"]]

        return []

    def get_identifiers(self, obj):
        """Get identifiers."""
        pid = obj["metadata"]["pid"]
        items = list(
            {
                DocumentRecord.get_permanent_link(request.host_url, pid, ignore_ark=True),
                DocumentRecord.get_permanent_link(request.host_url, pid),
            }
        )
        # If files on the document
        if "_files" in obj["metadata"]:
            # Extraction of files only with a type file
            files = filter(
                lambda f: ("type" in f and f["type"] == "file"),
                obj["metadata"]["_files"],
            )
            # Files sorting
            files = sorted(files, key=lambda file: file.get("order", 100))
            # Remove / at the end of host_url
            host = request.host_url[:-1]
            # Add file only the the link is defined in download
            for file in files:
                links = file.get("links", {})
                if "download" in links and links.get("download"):
                    items.append(f"{host}{links['download']}")
                # if the file is restricted it does not appears on the download link
                elif file["restriction"]["restricted"]:
                    items.append(f"{host}/documents/{pid}/files/{file['key']}")
        return items

    def get_languages(self, obj):
        """Get languages."""
        return [language["value"] for language in obj["metadata"].get("language", [])]

    def get_publishers(self, obj):
        """Get publishers."""
        if not obj["metadata"].get("provisionActivity"):
            return []

        items = []

        for provision_activity in obj["metadata"]["provisionActivity"]:
            if provision_activity["type"] == "bf:Publication" and provision_activity.get("statement"):
                for statement in provision_activity["statement"]:
                    if statement["type"] == "bf:Agent":
                        items.append(statement["label"][0]["value"])  # noqa: PERF401

        return items

    def get_relations(self, obj):
        """Get relations."""
        items = [
            other_edition["document"]["electronicLocator"] for other_edition in obj["metadata"].get("otherEdition", [])
        ] + [other_edition["document"]["electronicLocator"] for other_edition in obj["metadata"].get("relatedTo", [])]

        result = "info:eu-repo/semantics/altIdentifier/{schema}/{identifier}"

        # Identifiers
        for identifier in obj["metadata"].get("identifiedBy", []):
            # ARK
            matches = re.match(r"^ark\:\/(.*)$", identifier["value"])
            if matches:
                items.append(result.format(schema="ark", identifier=matches.group(1)))

            # DOI
            matches = re.match(r"^(10\..*)$", identifier["value"])
            if identifier["type"] == "bf:Doi" and matches:
                items.append(result.format(schema="doi", identifier=matches.group(1)))

            # ISBN
            if identifier["type"] == "bf:Isbn":
                items.append(result.format(schema="isbn", identifier=identifier["value"]))

            # ISSN
            if identifier["type"] == "bf:Issn":
                items.append(result.format(schema="issn", identifier=identifier["value"]))

            # PMID
            if (
                identifier["type"] == "bf:Local"
                and identifier.get("source")
                and identifier["source"].lower().find("pmid") != -1
            ):
                items.append(result.format(schema="pmid", identifier=identifier["value"]))

            # URN
            if identifier["type"] == "bf:Urn":
                items.append(result.format(schema="urn", identifier=identifier["value"]))

        return items

    def get_rights(self, obj):
        """Get rights."""
        items = []

        # Main file
        result = "info:eu-repo/semantics/{access}"

        main_file = obj["metadata"].get("mainFile")
        if main_file:
            if main_file["restriction"]["restricted"]:
                # Embargo
                if main_file["restriction"]["date"]:
                    items.append(result.format(access="embargoedAccess"))
                # Restricted
                else:
                    items.append(result.format(access="restrictedAccess"))
            # No restriction
            else:
                items.append(result.format(access="openAccess"))

        # Usage en access policy
        if obj["metadata"].get("usageAndAccessPolicy"):
            result = [obj["metadata"]["usageAndAccessPolicy"]["license"]]

            if obj["metadata"]["usageAndAccessPolicy"].get("label"):
                result.append(obj["metadata"]["usageAndAccessPolicy"]["label"])

            items.append(", ".join(result))

        return items

    def get_sources(self, obj):
        """Get sources."""
        return [part_of_format(part_of) for part_of in obj["metadata"].get("partOf", [])]

    def get_subjects(self, obj):
        """Get subjects."""
        items = []

        # Subjects
        for subjects in obj["metadata"].get("subjects", []):
            if "language" in subjects["label"]:
                for value in subjects["label"]["value"]:
                    items.append(  # noqa: PERF401
                        {
                            "@attrs": [
                                {
                                    "prefix": "xml",
                                    "name": "lang",
                                    "value": self.translate_language(subjects["label"]["language"]),
                                }
                            ],
                            "value": value,
                        }
                    )
            else:
                items = items + subjects["label"]["value"]

        # Classification
        for classification in obj["metadata"].get("classification", []):
            classification_type = "udc"

            if classification["type"] == "bf:ClassificationDdc":
                classification_type = "ddc"

            items.append(f"info:eu-repo/classification/{classification_type}/{classification['classificationPortion']}")

        return items

    def get_titles(self, obj):
        """Get titles."""
        title = {
            "@attrs": [
                {
                    "prefix": "xml",
                    "name": "lang",
                    "value": self.translate_language(obj["metadata"]["title"][0]["mainTitle"][0]["language"]),
                }
            ],
            "value": obj["metadata"]["title"][0]["mainTitle"][0]["value"].strip(),
        }
        if obj["metadata"]["title"][0].get("subtitle"):
            subtitle = obj["metadata"]["title"][0]["subtitle"][0]["value"].strip()
            title["value"] = f"{title['value']} : {subtitle}"

        return [title]

    def get_types(self, obj):
        """Get types."""
        if obj["metadata"].get("documentType", ""):
            types = obj["metadata"].get("documentType", "").split(":")
            if len(types) == 1:
                return [f"{types[0]}"]
            if len(types) == 2:
                return [f"http://purl.org/coar/resource_type/{types[1]}"]

        return []

    def format_contributor(self, contributor):
        """Format contributor item.

        :param contributor: Contributor dict.
        :returns: Formatted string representing the contributor.
        """
        data = contributor["agent"]["preferred_name"]

        info = []
        if contributor["agent"].get("number"):
            info.append(contributor["agent"]["number"])

        if contributor["agent"].get("date"):
            info.append(contributor["agent"]["date"])

        if contributor["agent"].get("place"):
            info.append(contributor["agent"]["place"])

        if info:
            data += f" ({' : '.join(info)})"

        return data

1	# Swiss Open Access Repository
2	# Copyright (C) 2021 RERO
3	#
4	# This program is free software: you can redistribute it and/or modify
5	# it under the terms of the GNU Affero General Public License as published by
6	# the Free Software Foundation, version 3 of the License.
7	#
8	# This program is distributed in the hope that it will be useful,
9	# but WITHOUT ANY WARRANTY; without even the implied warranty of
10	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11	# GNU Affero General Public License for more details.
12	#
13	# You should have received a copy of the GNU Affero General Public License
14	# along with this program. If not, see <http://www.gnu.org/licenses/>.
15
16	"""Dublin Core marshmallow schema."""
17
18	import re	1✔
19
20	from flask import current_app, request	1✔
21	from marshmallow import fields	1✔
22
23	from sonar.modules.documents.api import DocumentRecord	1✔
24	from sonar.modules.documents.views import part_of_format	1✔
25
26	from .base_schema import BaseSchema	1✔
27
28
29	class DublinCoreSchema(BaseSchema):	1✔
30	"""Schema for records v1 in JSON."""
31
32	contributors = fields.Method("get_contributors")	1✔
33	creators = fields.Method("get_creators")	1✔
34	dates = fields.Method("get_dates")	1✔
35	descriptions = fields.Method("get_descriptions")	1✔
36	formats = fields.Method("get_formats")	1✔
37	identifiers = fields.Method("get_identifiers")	1✔
38	languages = fields.Method("get_languages")	1✔
39	publishers = fields.Method("get_publishers")	1✔
40	relations = fields.Method("get_relations")	1✔
41	rights = fields.Method("get_rights")	1✔
42	sources = fields.Method("get_sources")	1✔
43	subjects = fields.Method("get_subjects")	1✔
44	titles = fields.Method("get_titles")	1✔
45	types = fields.Method("get_types")	1✔
46
47	def translate_language(self, language):	1✔
48	"""Translate language code ISO-639-3 to ISO-639-2 if possible.
49
50	:param language: language with ISO-639-3 format.
51	:returns: language code ISO-639-2 if possible or ISO-639-3.
52	"""
53	langs = current_app.config["SONAR_APP_LANGUAGES_MAP"]	1✔
54	if langs.get(language):	1✔
55	return langs[language]	1✔
56	return language	1✔
57
58	def get_contributors(self, obj):	1✔
59	"""Get contributors."""
60	return [	1✔
61	self.format_contributor(contributor)
62	for contributor in obj["metadata"].get("contribution", [])
63	if contributor["role"][0] != "cre" and contributor["agent"].get("preferred_name")
64	]
65
66	def get_creators(self, obj):	1✔
67	"""Get creators."""
68	return [	1✔
69	self.format_contributor(contributor)
70	for contributor in obj["metadata"].get("contribution", [])
71	if contributor["role"][0] == "cre" and contributor["agent"].get("preferred_name")
72	]
73
74	def get_dates(self, obj):	1✔
75	"""Get dates."""
76	items = [	1✔
77	provision_activity["startDate"]
78	for provision_activity in obj["metadata"].get("provisionActivity", [])
79	if provision_activity["type"] == "bf:Publication" and provision_activity.get("startDate")
80	]
81
82	if obj["metadata"].get("mainFile") and obj["metadata"]["mainFile"]["restriction"]["date"]:	1✔
83	items.append(f"info:eu-repo/date/embargoEnd/{obj['metadata']['mainFile']['embargo_date']}")	1✔
84
85	return items	1✔
86
87	def get_descriptions(self, obj):	1✔
88	"""Get descriptions."""
89	items = []	1✔
90	for abstract in obj["metadata"].get("abstracts", []):	1✔
91	if "language" in abstract:	1✔
92	items.append(	1✔
93	{
94	"@attrs": [
95	{
96	"prefix": "xml",
97	"name": "lang",
98	"value": self.translate_language(abstract["language"]),
99	}
100	],
101	"value": abstract["value"],
102	}
103	)
104	else:
105	items.append(abstract["value"])	1✔
106
107	return items	1✔
108
109	def get_formats(self, obj):	1✔
110	"""Get formats."""
111	main_file = obj["metadata"].get("mainFile")	1✔
112
113	if main_file and main_file.get("mimetype"):	1✔
114	return [main_file["mimetype"]]	1✔
115
116	return []	1✔
117
118	def get_identifiers(self, obj):	1✔
119	"""Get identifiers."""
120	pid = obj["metadata"]["pid"]	1✔
121	items = list(	1✔
122	{
123	DocumentRecord.get_permanent_link(request.host_url, pid, ignore_ark=True),
124	DocumentRecord.get_permanent_link(request.host_url, pid),
125	}
126	)
127	# If files on the document
128	if "_files" in obj["metadata"]:	1✔
129	# Extraction of files only with a type file
130	files = filter(	1✔
131	lambda f: ("type" in f and f["type"] == "file"),
132	obj["metadata"]["_files"],
133	)
134	# Files sorting
135	files = sorted(files, key=lambda file: file.get("order", 100))	1✔
136	# Remove / at the end of host_url
137	host = request.host_url[:-1]	1✔
138	# Add file only the the link is defined in download
139	for file in files:	1✔
140	links = file.get("links", {})	1✔
141	if "download" in links and links.get("download"):	1✔
142	items.append(f"{host}{links['download']}")	1✔
143	# if the file is restricted it does not appears on the download link
144	elif file["restriction"]["restricted"]:	1✔
145	items.append(f"{host}/documents/{pid}/files/{file['key']}")	1✔
146	return items	1✔
147
148	def get_languages(self, obj):	1✔
149	"""Get languages."""
150	return [language["value"] for language in obj["metadata"].get("language", [])]	1✔
151
152	def get_publishers(self, obj):	1✔
153	"""Get publishers."""
154	if not obj["metadata"].get("provisionActivity"):	1✔
155	return []	1✔
156
157	items = []	1✔
158
159	for provision_activity in obj["metadata"]["provisionActivity"]:	1✔
160	if provision_activity["type"] == "bf:Publication" and provision_activity.get("statement"):	1✔
161	for statement in provision_activity["statement"]:	1✔
162	if statement["type"] == "bf:Agent":	1✔
163	items.append(statement["label"][0]["value"]) # noqa: PERF401	1✔
164
165	return items	1✔
166
167	def get_relations(self, obj):	1✔
168	"""Get relations."""
169	items = [	1✔
170	other_edition["document"]["electronicLocator"] for other_edition in obj["metadata"].get("otherEdition", [])
171	] + [other_edition["document"]["electronicLocator"] for other_edition in obj["metadata"].get("relatedTo", [])]
172
173	result = "info:eu-repo/semantics/altIdentifier/{schema}/{identifier}"	1✔
174
175	# Identifiers
176	for identifier in obj["metadata"].get("identifiedBy", []):	1✔
177	# ARK
178	matches = re.match(r"^ark\:\/(.*)$", identifier["value"])	1✔
179	if matches:	1✔
180	items.append(result.format(schema="ark", identifier=matches.group(1)))	1✔
181
182	# DOI
183	matches = re.match(r"^(10\..*)$", identifier["value"])	1✔
184	if identifier["type"] == "bf:Doi" and matches:	1✔
185	items.append(result.format(schema="doi", identifier=matches.group(1)))	1✔
186
187	# ISBN
188	if identifier["type"] == "bf:Isbn":	1✔
189	items.append(result.format(schema="isbn", identifier=identifier["value"]))	1✔
190
191	# ISSN
192	if identifier["type"] == "bf:Issn":	1✔
193	items.append(result.format(schema="issn", identifier=identifier["value"]))	1✔
194
195	# PMID
196	if (	1✔
197	identifier["type"] == "bf:Local"
198	and identifier.get("source")
199	and identifier["source"].lower().find("pmid") != -1
200	):
201	items.append(result.format(schema="pmid", identifier=identifier["value"]))	1✔
202
203	# URN
204	if identifier["type"] == "bf:Urn":	1✔
205	items.append(result.format(schema="urn", identifier=identifier["value"]))	1✔
206
207	return items	1✔
208
209	def get_rights(self, obj):	1✔
210	"""Get rights."""
211	items = []	1✔
212
213	# Main file
214	result = "info:eu-repo/semantics/{access}"	1✔
215
216	main_file = obj["metadata"].get("mainFile")	1✔
217	if main_file:	1✔
218	if main_file["restriction"]["restricted"]:	1✔
219	# Embargo
220	if main_file["restriction"]["date"]:	1✔
221	items.append(result.format(access="embargoedAccess"))	1✔
222	# Restricted
223	else:
224	items.append(result.format(access="restrictedAccess"))	1✔
225	# No restriction
226	else:
227	items.append(result.format(access="openAccess"))	1✔
228
229	# Usage en access policy
230	if obj["metadata"].get("usageAndAccessPolicy"):	1✔
231	result = [obj["metadata"]["usageAndAccessPolicy"]["license"]]	1✔
232
233	if obj["metadata"]["usageAndAccessPolicy"].get("label"):	1✔
234	result.append(obj["metadata"]["usageAndAccessPolicy"]["label"])	1✔
235
236	items.append(", ".join(result))	1✔
237
238	return items	1✔
239
240	def get_sources(self, obj):	1✔
241	"""Get sources."""
242	return [part_of_format(part_of) for part_of in obj["metadata"].get("partOf", [])]	1✔
243
244	def get_subjects(self, obj):	1✔
245	"""Get subjects."""
246	items = []	1✔
247
248	# Subjects
249	for subjects in obj["metadata"].get("subjects", []):	1✔
250	if "language" in subjects["label"]:	1✔
251	for value in subjects["label"]["value"]:	1✔
252	items.append( # noqa: PERF401	1✔
253	{
254	"@attrs": [
255	{
256	"prefix": "xml",
257	"name": "lang",
258	"value": self.translate_language(subjects["label"]["language"]),
259	}
260	],
261	"value": value,
262	}
263	)
264	else:
265	items = items + subjects["label"]["value"]	×
266
267	# Classification
268	for classification in obj["metadata"].get("classification", []):	1✔
269	classification_type = "udc"	1✔
270
271	if classification["type"] == "bf:ClassificationDdc":	1✔
272	classification_type = "ddc"	1✔
273
274	items.append(f"info:eu-repo/classification/{classification_type}/{classification['classificationPortion']}")	1✔
275
276	return items	1✔
277
278	def get_titles(self, obj):	1✔
279	"""Get titles."""
280	title = {	1✔
281	"@attrs": [
282	{
283	"prefix": "xml",
284	"name": "lang",
285	"value": self.translate_language(obj["metadata"]["title"][0]["mainTitle"][0]["language"]),
286	}
287	],
288	"value": obj["metadata"]["title"][0]["mainTitle"][0]["value"].strip(),
289	}
290	if obj["metadata"]["title"][0].get("subtitle"):	1✔
291	subtitle = obj["metadata"]["title"][0]["subtitle"][0]["value"].strip()	1✔
292	title["value"] = f"{title['value']} : {subtitle}"	1✔
293
294	return [title]	1✔
295
296	def get_types(self, obj):	1✔
297	"""Get types."""
298	if obj["metadata"].get("documentType", ""):	1✔
299	types = obj["metadata"].get("documentType", "").split(":")	1✔
300	if len(types) == 1:	1✔
301	return [f"{types[0]}"]	1✔
302	if len(types) == 2:	1✔
303	return [f"http://purl.org/coar/resource_type/{types[1]}"]	1✔
304
305	return []	1✔
306
307	def format_contributor(self, contributor):	1✔
308	"""Format contributor item.
309
310	:param contributor: Contributor dict.
311	:returns: Formatted string representing the contributor.
312	"""
313	data = contributor["agent"]["preferred_name"]	1✔
314
315	info = []	1✔
316	if contributor["agent"].get("number"):	1✔
317	info.append(contributor["agent"]["number"])	1✔
318
319	if contributor["agent"].get("date"):	1✔
320	info.append(contributor["agent"]["date"])	1✔
321
322	if contributor["agent"].get("place"):	1✔
323	info.append(contributor["agent"]["place"])	1✔
324
325	if info:	1✔
326	data += f" ({' : '.join(info)})"	1✔
327
328	return data	1✔

rero / sonar / 17425918180

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous