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

90.91

/rero_mef/concepts/idref/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 IdRef agent."""

from flask import current_app
from invenio_search.api import RecordsSearch

from ..api import ConceptIndexer, ConceptRecord
from .fetchers import idref_id_fetcher
from .minters import idref_id_minter
from .models import ConceptIdrefMetadata
from .providers import ConceptIdrefProvider


class ConceptIdrefSearch(RecordsSearch):
    """RecordsSearch."""

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

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

        default_filter = None


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

    minter = idref_id_minter
    fetcher = idref_id_fetcher
    provider = ConceptIdrefProvider
    name = "idref"
    viaf_source_code = "RAMEAU"
    pid_type = "concept_idref_pid"
    model_cls = ConceptIdrefMetadata
    search = ConceptIdrefSearch

    @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 idref_get_record

        return idref_get_record(id_=id_, debug=debug)

    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
            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:
                # Extra code, because the data from GND is not completely correct
                # and we need to find out whether there is one exact match with possible close matches elsewhere.
                # find exact matches in GND:
                exact_pids = []
                for hit in query.source(["pid", "exactMatch"]).scan():
                    data = hit.to_dict()
                    for exact_match in data.get("exactMatch", []):
                        for identified_by in exact_match.get("identifiedBy", []):
                            if (
                                identified_by.get("source") == "BNF"
                                and identified_by.get("type") == "bf:Nbn"
                                and identified_by.get("value", "").startswith("FRBNF")
                            ):
                                exact_pids.append(hit.pid)
                # we have one associated record with an exact match
                if len(exact_pids) == 1:
                    return association_cls.get_record_by_pid(exact_pids[0])
                # we have multiple associated records
                current_app.logger.error(
                    f"MULTIPLE ASSOCIATIONS IDENTIFIERS FOUND FOR: {self.name} {self.pid} "
                    f"| {association_identifier}"
                )
            # we have one associated record
            elif query.count() == 1:
                hit = next(query.source("pid").scan())
                return association_cls.get_record_by_pid(hit.pid)
        return None

    @property
    def association_identifier(self):
        """Get associated identifier from identifiedBy."""
        if pids := [
            identified_by.get("value")
            for identified_by in self.get("identifiedBy", [])
            if identified_by.get("source") == "BNF"
            and identified_by.get("value", "").startswith("FRBNF")
        ]:
            if len(pids) > 1:
                current_app.logger.error(
                    f"MULTIPLE ASSOCIATIONS FOUND FOR: {self.name} {self.pid} | {', '.join(pids)}"
                )
            if pids:
                return pids[-1]
        return None

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

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


class ConceptIdrefIndexer(ConceptIndexer):
    """Concept IDREF indexer."""

    record_cls = ConceptIdrefRecord

    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=ConceptIdrefSearch.Meta.index, doc_type="cidref"
        )

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 IdRef 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 idref_id_fetcher	1✔
23	from .minters import idref_id_minter	1✔
24	from .models import ConceptIdrefMetadata	1✔
25	from .providers import ConceptIdrefProvider	1✔
26
27
28	class ConceptIdrefSearch(RecordsSearch):	1✔
29	"""RecordsSearch."""
30
31	class Meta:	1✔
32	"""Search only on index."""
33
34	index = "concepts_idref"	1✔
35	doc_types = None	1✔
36	fields = ("*",)	1✔
37	facets = {}	1✔
38
39	default_filter = None	1✔
40
41
42	class ConceptIdrefRecord(ConceptRecord):	1✔
43	"""Concepts Authority class."""
44
45	minter = idref_id_minter	1✔
46	fetcher = idref_id_fetcher	1✔
47	provider = ConceptIdrefProvider	1✔
48	name = "idref"	1✔
49	viaf_source_code = "RAMEAU"	1✔
50	pid_type = "concept_idref_pid"	1✔
51	model_cls = ConceptIdrefMetadata	1✔
52	search = ConceptIdrefSearch	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 idref_get_record	×
63
64	return idref_get_record(id_=id_, debug=debug)	×
65
66	def get_association_record(self, association_cls, association_search):	1✔
67	"""Get associated record.
68
69	:params association_cls: Association class
70	:params association_search: Association search class.
71	:returns: Associated record.
72	"""
73	if association_identifier := self.association_identifier:	1✔
74	# Test if my identifier is unique
75	count = (	1✔
76	self.search()
77	.filter("term", _association_identifier=association_identifier)
78	.count()
79	)
80	if count > 1:	1✔
81	current_app.logger.error(	×
82	f"MULTIPLE IDENTIFIERS FOUND FOR: {self.name} {self.pid} "
83	f"\| {association_identifier}"
84	)
85	return None	×
86	# Get associated record
87	query = association_search().filter(	1✔
88	"term", _association_identifier=association_identifier
89	)
90	if query.count() > 1:	1✔
91	# Extra code, because the data from GND is not completely correct
92	# and we need to find out whether there is one exact match with possible close matches elsewhere.
93	# find exact matches in GND:
94	exact_pids = []	1✔
95	for hit in query.source(["pid", "exactMatch"]).scan():	1✔
96	data = hit.to_dict()	1✔
97	for exact_match in data.get("exactMatch", []):	1✔
98	for identified_by in exact_match.get("identifiedBy", []):	1✔
99	if (	1✔
100	identified_by.get("source") == "BNF"
101	and identified_by.get("type") == "bf:Nbn"
102	and identified_by.get("value", "").startswith("FRBNF")
103	):
104	exact_pids.append(hit.pid)	1✔
105	# we have one associated record with an exact match
106	if len(exact_pids) == 1:	1✔
107	return association_cls.get_record_by_pid(exact_pids[0])	1✔
108	# we have multiple associated records
109	current_app.logger.error(	1✔
110	f"MULTIPLE ASSOCIATIONS IDENTIFIERS FOUND FOR: {self.name} {self.pid} "
111	f"\| {association_identifier}"
112	)
113	# we have one associated record
114	elif query.count() == 1:	1✔
115	hit = next(query.source("pid").scan())	1✔
116	return association_cls.get_record_by_pid(hit.pid)	1✔
117	return None	1✔
118
119	@property	1✔
120	def association_identifier(self):	1✔
121	"""Get associated identifier from identifiedBy."""
122	if pids := [	1✔
123	identified_by.get("value")
124	for identified_by in self.get("identifiedBy", [])
125	if identified_by.get("source") == "BNF"
126	and identified_by.get("value", "").startswith("FRBNF")
127	]:
128	if len(pids) > 1:	1✔
129	current_app.logger.error(	×
130	f"MULTIPLE ASSOCIATIONS FOUND FOR: {self.name} {self.pid} \| {', '.join(pids)}"
131	)
132	if pids:	1✔
133	return pids[-1]	1✔
134	return None	1✔
135
136	@property	1✔
137	def association_info(self):	1✔
138	"""Get associated record."""
139	from rero_mef.concepts import (	1✔
140	ConceptGndRecord,
141	ConceptGndSearch,
142	ConceptMefRecord,
143	)
144
145	ConceptGndRecord.flush_indexes()	1✔
146	return {	1✔
147	"identifier": self.association_identifier,
148	"record": self.get_association_record(
149	association_cls=ConceptGndRecord, association_search=ConceptGndSearch
150	),
151	"record_cls": ConceptGndRecord,
152	"search_cls": ConceptGndSearch,
153	"mef_cls": ConceptMefRecord,
154	}
155
156
157	class ConceptIdrefIndexer(ConceptIndexer):	1✔
158	"""Concept IDREF indexer."""
159
160	record_cls = ConceptIdrefRecord	1✔
161
162	def bulk_index(self, record_id_iterator):	1✔
163	"""Bulk index records.
164
165	:param record_id_iterator: Iterator yielding record UUIDs.
166	"""
167	super().bulk_index(	×
168	record_id_iterator, index=ConceptIdrefSearch.Meta.index, doc_type="cidref"
169	)

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