16621609190

Committed 30 Jul 2025 11:43AM UTC coverage: 84.491% (+0.008%) from 84.483%

Build # 16621609190

Build Type

push

github

Committed by

rerowep

Commit Message

chore: update dependencies

Co-Authored-by: Peter Weber <peter.weber@rero.ch>

Run Details

4560 of 5397 relevant lines covered (84.49%)

0.84 hits per line

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

89.23

/rero_mef/concepts/gnd/api.py

# RERO MEF
# Copyright (C) 2024 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/>.

"""API for manipulating Gnd agent."""

from flask import current_app
from invenio_search.api import RecordsSearch

from ..api import ConceptIndexer, ConceptRecord
from .fetchers import gnd_id_fetcher
from .minters import gnd_id_minter
from .models import ConceptGndMetadata
from .providers import ConceptGndProvider


class ConceptGndSearch(RecordsSearch):
    """RecordsSearch."""

    class Meta:
        """Search only on index."""

        index = "concepts_gnd"
        doc_types = None
        fields = ("*",)
        facets = {}

        default_filter = None


class ConceptGndRecord(ConceptRecord):
    """Concepts Authority class."""

    minter = gnd_id_minter
    fetcher = gnd_id_fetcher
    provider = ConceptGndProvider
    name = "gnd"
    viaf_source_code = "RAMEAU"
    pid_type = "concept_gnd_pid"
    model_cls = ConceptGndMetadata
    search = ConceptGndSearch

    @classmethod
    def get_online_record(cls, id_, debug=False):
        """Get online Record.

        :param id_: Id of online record.
        :param debug: Debug print.
        :returns: record or None
        """
        from .tasks import gnd_get_record

        return gnd_get_record(id_=id_, debug=debug)

    @property
    def association_identifier(self):
        """Get associated identifier."""
        for match_type, max_count in current_app.config.get(
            "RERO_MEF_CONCEPTS_GND_MATCHES", {}
        ).items():
            matches = self.get(match_type, [])
            match_count = 0
            match_value = ""
            for match in matches:
                for identified_by in match.get("identifiedBy", []):
                    if (
                        identified_by.get("source") == "BNF"
                        and identified_by.get("type") == "bf:Nbn"
                        and identified_by.get("value", "").startswith("FRBNF")
                    ):
                        match_count += 1
                        match_value = identified_by.get("value")
            if match_value and match_count <= max_count:
                return match_value[:13]
        return None

    def get_association_record(self, association_cls, association_search):
        """Get associated record.

        :params association_cls: Association class
        :params association_search: Association search class.
        :returns: Associated record.
        """
        if association_identifier := self.association_identifier:
            # Test if my identifier is unique
            exact_count = (
                self.search()
                .filter("term", exactMatch__identifiedBy__source="BNF")
                .filter("term", exactMatch__identifiedBy__type="bf:Nbn")
                .filter("term", exactMatch__identifiedBy__value=association_identifier)
                .count()
            )
            if exact_count != 1:
                # we have 0 or multiple exact matches
                count = (
                    self.search()
                    .filter("term", _association_identifier=association_identifier)
                    .count()
                )
                if count > 1:
                    current_app.logger.error(
                        f"MULTIPLE IDENTIFIERS FOUND FOR: {self.name} {self.pid} "
                        f"| {association_identifier}"
                    )
                    return None
            # Get associated record
            query = association_search().filter(
                "term", _association_identifier=association_identifier
            )
            if query.count() > 1:
                current_app.logger.error(
                    f"MULTIPLE ASSOCIATIONS IDENTIFIERS FOUND FOR: {self.name} {self.pid} "
                    f"| {association_identifier}"
                )
            elif query.count() == 1:
                hit = next(query.source("pid").scan())
                return association_cls.get_record_by_pid(hit.pid)
        return None

    @property
    def association_info(self):
        """Get associated record."""
        from rero_mef.concepts import (
            ConceptIdrefRecord,
            ConceptIdrefSearch,
            ConceptMefRecord,
        )

        ConceptIdrefRecord.flush_indexes()
        return {
            "identifier": self.association_identifier,
            "record": self.get_association_record(
                association_cls=ConceptIdrefRecord,
                association_search=ConceptIdrefSearch,
            ),
            "record_cls": ConceptIdrefRecord,
            "search_cls": ConceptIdrefSearch,
            "mef_cls": ConceptMefRecord,
        }


class ConceptGndIndexer(ConceptIndexer):
    """Concept GND indexer."""

    record_cls = ConceptGndRecord

    def bulk_index(self, record_id_iterator):
        """Bulk index records.

        :param record_id_iterator: Iterator yielding record UUIDs.
        """
        super().bulk_index(
            record_id_iterator, index=ConceptGndSearch.Meta.index, doc_type="cognd"
        )

1	# RERO MEF
2	# Copyright (C) 2024 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	"""API for manipulating Gnd agent."""
17
18	from flask import current_app	1✔
19	from invenio_search.api import RecordsSearch	1✔
20
21	from ..api import ConceptIndexer, ConceptRecord	1✔
22	from .fetchers import gnd_id_fetcher	1✔
23	from .minters import gnd_id_minter	1✔
24	from .models import ConceptGndMetadata	1✔
25	from .providers import ConceptGndProvider	1✔
26
27
28	class ConceptGndSearch(RecordsSearch):	1✔
29	"""RecordsSearch."""
30
31	class Meta:	1✔
32	"""Search only on index."""
33
34	index = "concepts_gnd"	1✔
35	doc_types = None	1✔
36	fields = ("*",)	1✔
37	facets = {}	1✔
38
39	default_filter = None	1✔
40
41
42	class ConceptGndRecord(ConceptRecord):	1✔
43	"""Concepts Authority class."""
44
45	minter = gnd_id_minter	1✔
46	fetcher = gnd_id_fetcher	1✔
47	provider = ConceptGndProvider	1✔
48	name = "gnd"	1✔
49	viaf_source_code = "RAMEAU"	1✔
50	pid_type = "concept_gnd_pid"	1✔
51	model_cls = ConceptGndMetadata	1✔
52	search = ConceptGndSearch	1✔
53
54	@classmethod	1✔
55	def get_online_record(cls, id_, debug=False):	1✔
56	"""Get online Record.
57
58	:param id_: Id of online record.
59	:param debug: Debug print.
60	:returns: record or None
61	"""
62	from .tasks import gnd_get_record	×
63
64	return gnd_get_record(id_=id_, debug=debug)	×
65
66	@property	1✔
67	def association_identifier(self):	1✔
68	"""Get associated identifier."""
69	for match_type, max_count in current_app.config.get(	1✔
70	"RERO_MEF_CONCEPTS_GND_MATCHES", {}
71	).items():
72	matches = self.get(match_type, [])	1✔
73	match_count = 0	1✔
74	match_value = ""	1✔
75	for match in matches:	1✔
76	for identified_by in match.get("identifiedBy", []):	1✔
77	if (	1✔
78	identified_by.get("source") == "BNF"
79	and identified_by.get("type") == "bf:Nbn"
80	and identified_by.get("value", "").startswith("FRBNF")
81	):
82	match_count += 1	1✔
83	match_value = identified_by.get("value")	1✔
84	if match_value and match_count <= max_count:	1✔
85	return match_value[:13]	1✔
86	return None	1✔
87
88	def get_association_record(self, association_cls, association_search):	1✔
89	"""Get associated record.
90
91	:params association_cls: Association class
92	:params association_search: Association search class.
93	:returns: Associated record.
94	"""
95	if association_identifier := self.association_identifier:	1✔
96	# Test if my identifier is unique
97	exact_count = (	1✔
98	self.search()
99	.filter("term", exactMatch__identifiedBy__source="BNF")
100	.filter("term", exactMatch__identifiedBy__type="bf:Nbn")
101	.filter("term", exactMatch__identifiedBy__value=association_identifier)
102	.count()
103	)
104	if exact_count != 1:	1✔
105	# we have 0 or multiple exact matches
106	count = (	1✔
107	self.search()
108	.filter("term", _association_identifier=association_identifier)
109	.count()
110	)
111	if count > 1:	1✔
112	current_app.logger.error(	×
113	f"MULTIPLE IDENTIFIERS FOUND FOR: {self.name} {self.pid} "
114	f"\| {association_identifier}"
115	)
116	return None	×
117	# Get associated record
118	query = association_search().filter(	1✔
119	"term", _association_identifier=association_identifier
120	)
121	if query.count() > 1:	1✔
122	current_app.logger.error(	×
123	f"MULTIPLE ASSOCIATIONS IDENTIFIERS FOUND FOR: {self.name} {self.pid} "
124	f"\| {association_identifier}"
125	)
126	elif query.count() == 1:	1✔
127	hit = next(query.source("pid").scan())	1✔
128	return association_cls.get_record_by_pid(hit.pid)	1✔
129	return None	×
130
131	@property	1✔
132	def association_info(self):	1✔
133	"""Get associated record."""
134	from rero_mef.concepts import (	1✔
135	ConceptIdrefRecord,
136	ConceptIdrefSearch,
137	ConceptMefRecord,
138	)
139
140	ConceptIdrefRecord.flush_indexes()	1✔
141	return {	1✔
142	"identifier": self.association_identifier,
143	"record": self.get_association_record(
144	association_cls=ConceptIdrefRecord,
145	association_search=ConceptIdrefSearch,
146	),
147	"record_cls": ConceptIdrefRecord,
148	"search_cls": ConceptIdrefSearch,
149	"mef_cls": ConceptMefRecord,
150	}
151
152
153	class ConceptGndIndexer(ConceptIndexer):	1✔
154	"""Concept GND indexer."""
155
156	record_cls = ConceptGndRecord	1✔
157
158	def bulk_index(self, record_id_iterator):	1✔
159	"""Bulk index records.
160
161	:param record_id_iterator: Iterator yielding record UUIDs.
162	"""
163	super().bulk_index(	×
164	record_id_iterator, index=ConceptGndSearch.Meta.index, doc_type="cognd"
165	)

rero / rero-mef / 16621609190

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