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.55

/rero_mef/marctojson/do_rero_concepts.py

# RERO MEF
# Copyright (C) 2020 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/>.
"""Marctojsons transformer for RERO records."""

import contextlib

from rero_mef.marctojson.helper import build_string_list_from_fields


class Transformation:
    """Transformation MARC21 to JSON for RERO concept."""

    def __init__(self, marc, logger=None, verbose=False, transform=True):
        """Constructor."""
        self.marc = marc
        self.logger = logger
        self.verbose = verbose
        self.json_dict = {}
        if transform:
            self._transform()

    def _transform(self):
        """Call the transformation functions."""
        if self.marc.get_fields("150") or self.marc.get_fields("155"):
            for func in dir(self):
                if func.startswith("trans"):
                    func = getattr(self, func)
                    func()
        else:
            msg = "No 150 or 155"
            if self.logger and self.verbose:
                self.logger.warning(f"NO TRANSFORMATION: {msg}")
            self.json_dict = {"NO TRANSFORMATION": msg}
            self.trans_rero_identifier()

    @property
    def json(self):
        """Json data."""
        return self.json_dict or None

    def trans_rero_identifier(self):
        """Transformation identifier from field 035."""
        if self.logger and self.verbose:
            self.logger.info("Call Function", "trans_rero_identifier")
        identifiers = []
        field_035 = self.marc.get_fields("035")
        if field_035 and field_035[0].get("a"):
            self.json_dict["pid"] = field_035[0]["a"]
            self.json_dict["type"] = "bf:Topic"
            identifiers.append(
                {
                    "type": "bf:Local",
                    "source": "RERO",
                    "value": field_035[0]["a"].strip(),
                }
            )
        for field_016 in self.marc.get_fields("016"):
            identifiers.extend(
                {"type": "bf:Local", "source": "BNF", "value": subfield_a.strip()}
                for subfield_a in field_016.get_subfields("a")
            )
        for field_679 in self.marc.get_fields("679"):
            identifiers.extend(
                {"type": "uri", "value": subfield_u.strip()}
                for subfield_u in field_679.get_subfields("u")
            )
        if identifiers:
            self.json_dict["identifiedBy"] = identifiers

    def trans_rero_bnf_type(self):
        """Transformation bnf type from field 035."""
        if self.logger and self.verbose:
            self.logger.info("Call Function", "trans_rero_bnf_type")
        with contextlib.suppress(Exception):
            if data_075_a := self.marc["075"]["a"].strip():
                self.json_dict["bnf_type"] = data_075_a

    def trans_rero_authorized_access_point(self):
        """Transformation authorized_access_point from field 150 155."""
        if self.logger and self.verbose:
            self.logger.info("Call Function", "trans_rero_authorized_access_point")
        tag = "155" if self.marc.get_fields("155") else "150"
        self.json_dict["authorized_access_point"] = self.marc[tag]["a"].strip()

    def trans_rero_variant_access_point(self):
        """Transformation variant_access_point from field 450 455."""
        if self.logger and self.verbose:
            self.logger.info("Call Function", "trans_rero_variant_access_point")
        tag = "455" if self.marc.get_fields("455") else "450"
        subfields = {"a": ", ", "x": " - "}
        if variant_access_points := build_string_list_from_fields(
            self.marc, tag, subfields
        ):
            self.json_dict["variant_access_point"] = variant_access_points

    def trans_rero_relation(self):
        """Transformation broader related narrower 550 555."""
        if self.logger and self.verbose:
            self.logger.info("Call Function", "trans_rero_relation")
        tag = "555" if self.marc.get_fields("555") else "550"
        relations = {}
        for field in self.marc.get_fields(tag):
            relation_type = "related"
            if field.get("w"):
                if field["w"] == "g":
                    relation_type = "broader"
                elif field["w"] == "h":
                    relation_type = "narrower"
            relations.setdefault(relation_type, [])
            if field.get("0"):
                relations[relation_type].append({"$ref": field["0"]})
            elif field.get("a"):
                relations[relation_type].append(
                    {"authorized_access_point": field["a"].strip()}
                )

        for relation, value in relations.items():
            if value:
                self.json_dict[relation] = value

    def trans_rero_classification(self):
        """Transformation classification from field 072."""
        if self.logger and self.verbose:
            self.logger.info("Call Function", "trans_rero_classification")
        classifications = []
        for field_072 in self.marc.get_fields("072"):
            if field_072.get("a"):
                classification = field_072["a"].split("-")
                if len(classification) == 2:
                    classifications.append(
                        {
                            "type": "bf:ClassificationDdc",
                            "classificationPortion": classification[0].strip(),
                            "name": classification[1].strip(),
                        }
                    )

        if classifications:
            self.json_dict["classification"] = classifications

    def trans_rero_close_match(self):
        """Transformation closeMatch from field 682."""
        if self.logger and self.verbose:
            self.logger.info("Call Function", "trans_rero_close_match")
        close_matchs = []
        for field_682 in self.marc.get_fields("682"):
            with contextlib.suppress(Exception):
                if field_682.get("a") and field_682.get("v"):
                    close_matchs.append(
                        {
                            "authorized_access_point": field_682["a"].strip(),
                            "source": field_682["v"].strip(),
                        }
                    )
        if close_matchs:
            self.json_dict["closeMatch"] = close_matchs

    def trans_rero_note(self):
        """Transformation notes from field.

        670 $a: dataSource
        675 $a: dataNotFound
        680 __ $a: general
        667 $a: nonPublic
        260 _9 $a: seeReference
        260 __ $a: seeReference
        360 __ $a: seeAlsoReference
        016 $9: REROtreatment
        """
        if self.logger and self.verbose:
            self.logger.info("Call Function", "trans_rero_note")
        notes = {
            "dataSource": [],
            "dataNotFound": [],
            "general": [],
            "nonPublic": [],
            "seeReference": [],
            "seeAlsoReference": [],
            "REROtreatment": [],
        }
        for field in self.marc.get_fields("670"):
            if field.get("a"):
                notes["dataSource"].append(field["a"].strip())
        for field in self.marc.get_fields("675"):
            if field.get("a"):
                notes["dataNotFound"].append(field["a"].strip())
        for field in self.marc.get_fields("680"):
            if field.get("a") and field.indicators == [" ", " "]:
                notes["general"].append(field["a"].strip())
        for field in self.marc.get_fields("667"):
            if field.get("a"):
                notes["nonPublic"].append(field["a"].strip())
        for field in self.marc.get_fields("260"):
            if field.get("a") and field.indicators in [[" ", " "], [" ", "9"]]:
                notes["seeReference"].append(field["a"].strip())
        for field in self.marc.get_fields("360"):
            if field.get("a") and field.indicators == [" ", " "]:
                notes["seeAlsoReference"].append(field["a"].strip())
        for field in self.marc.get_fields("016"):
            if field.get("9"):
                notes["REROtreatment"].append(field["9"].strip())
        for note, value in notes.items():
            if value:
                self.json_dict.setdefault("note", [])
                self.json_dict["note"].append({"noteType": note, "label": value})

1	# RERO MEF
2	# Copyright (C) 2020 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	"""Marctojsons transformer for RERO records."""
16
17	import contextlib	1✔
18
19	from rero_mef.marctojson.helper import build_string_list_from_fields	1✔
20
21
22	class Transformation:	1✔
23	"""Transformation MARC21 to JSON for RERO concept."""
24
25	def __init__(self, marc, logger=None, verbose=False, transform=True):	1✔
26	"""Constructor."""
27	self.marc = marc	1✔
28	self.logger = logger	1✔
29	self.verbose = verbose	1✔
30	self.json_dict = {}	1✔
31	if transform:	1✔
32	self._transform()	×
33
34	def _transform(self):	1✔
35	"""Call the transformation functions."""
36	if self.marc.get_fields("150") or self.marc.get_fields("155"):	×
37	for func in dir(self):	×
38	if func.startswith("trans"):	×
39	func = getattr(self, func)	×
40	func()	×
41	else:
42	msg = "No 150 or 155"	×
43	if self.logger and self.verbose:	×
44	self.logger.warning(f"NO TRANSFORMATION: {msg}")	×
45	self.json_dict = {"NO TRANSFORMATION": msg}	×
46	self.trans_rero_identifier()	×
47
48	@property	1✔
49	def json(self):	1✔
50	"""Json data."""
51	return self.json_dict or None	1✔
52
53	def trans_rero_identifier(self):	1✔
54	"""Transformation identifier from field 035."""
55	if self.logger and self.verbose:	1✔
56	self.logger.info("Call Function", "trans_rero_identifier")	1✔
57	identifiers = []	1✔
58	field_035 = self.marc.get_fields("035")	1✔
59	if field_035 and field_035[0].get("a"):	1✔
60	self.json_dict["pid"] = field_035[0]["a"]	1✔
61	self.json_dict["type"] = "bf:Topic"	1✔
62	identifiers.append(	1✔
63	{
64	"type": "bf:Local",
65	"source": "RERO",
66	"value": field_035[0]["a"].strip(),
67	}
68	)
69	for field_016 in self.marc.get_fields("016"):	1✔
70	identifiers.extend(	1✔
71	{"type": "bf:Local", "source": "BNF", "value": subfield_a.strip()}
72	for subfield_a in field_016.get_subfields("a")
73	)
74	for field_679 in self.marc.get_fields("679"):	1✔
75	identifiers.extend(	1✔
76	{"type": "uri", "value": subfield_u.strip()}
77	for subfield_u in field_679.get_subfields("u")
78	)
79	if identifiers:	1✔
80	self.json_dict["identifiedBy"] = identifiers	1✔
81
82	def trans_rero_bnf_type(self):	1✔
83	"""Transformation bnf type from field 035."""
84	if self.logger and self.verbose:	1✔
85	self.logger.info("Call Function", "trans_rero_bnf_type")	1✔
86	with contextlib.suppress(Exception):	1✔
87	if data_075_a := self.marc["075"]["a"].strip():	1✔
88	self.json_dict["bnf_type"] = data_075_a	1✔
89
90	def trans_rero_authorized_access_point(self):	1✔
91	"""Transformation authorized_access_point from field 150 155."""
92	if self.logger and self.verbose:	1✔
93	self.logger.info("Call Function", "trans_rero_authorized_access_point")	1✔
94	tag = "155" if self.marc.get_fields("155") else "150"	1✔
95	self.json_dict["authorized_access_point"] = self.marc[tag]["a"].strip()	1✔
96
97	def trans_rero_variant_access_point(self):	1✔
98	"""Transformation variant_access_point from field 450 455."""
99	if self.logger and self.verbose:	1✔
100	self.logger.info("Call Function", "trans_rero_variant_access_point")	1✔
101	tag = "455" if self.marc.get_fields("455") else "450"	1✔
102	subfields = {"a": ", ", "x": " - "}	1✔
103	if variant_access_points := build_string_list_from_fields(	1✔
104	self.marc, tag, subfields
105	):
106	self.json_dict["variant_access_point"] = variant_access_points	1✔
107
108	def trans_rero_relation(self):	1✔
109	"""Transformation broader related narrower 550 555."""
110	if self.logger and self.verbose:	1✔
111	self.logger.info("Call Function", "trans_rero_relation")	1✔
112	tag = "555" if self.marc.get_fields("555") else "550"	1✔
113	relations = {}	1✔
114	for field in self.marc.get_fields(tag):	1✔
115	relation_type = "related"	1✔
116	if field.get("w"):	1✔
117	if field["w"] == "g":	1✔
118	relation_type = "broader"	1✔
119	elif field["w"] == "h":	1✔
120	relation_type = "narrower"	1✔
121	relations.setdefault(relation_type, [])	1✔
122	if field.get("0"):	1✔
123	relations[relation_type].append({"$ref": field["0"]})	×
124	elif field.get("a"):	1✔
125	relations[relation_type].append(	1✔
126	{"authorized_access_point": field["a"].strip()}
127	)
128
129	for relation, value in relations.items():	1✔
130	if value:	1✔
131	self.json_dict[relation] = value	1✔
132
133	def trans_rero_classification(self):	1✔
134	"""Transformation classification from field 072."""
135	if self.logger and self.verbose:	1✔
136	self.logger.info("Call Function", "trans_rero_classification")	1✔
137	classifications = []	1✔
138	for field_072 in self.marc.get_fields("072"):	1✔
139	if field_072.get("a"):	1✔
140	classification = field_072["a"].split("-")	1✔
141	if len(classification) == 2:	1✔
142	classifications.append(	1✔
143	{
144	"type": "bf:ClassificationDdc",
145	"classificationPortion": classification[0].strip(),
146	"name": classification[1].strip(),
147	}
148	)
149
150	if classifications:	1✔
151	self.json_dict["classification"] = classifications	1✔
152
153	def trans_rero_close_match(self):	1✔
154	"""Transformation closeMatch from field 682."""
155	if self.logger and self.verbose:	1✔
156	self.logger.info("Call Function", "trans_rero_close_match")	1✔
157	close_matchs = []	1✔
158	for field_682 in self.marc.get_fields("682"):	1✔
159	with contextlib.suppress(Exception):	1✔
160	if field_682.get("a") and field_682.get("v"):	1✔
161	close_matchs.append(	1✔
162	{
163	"authorized_access_point": field_682["a"].strip(),
164	"source": field_682["v"].strip(),
165	}
166	)
167	if close_matchs:	1✔
168	self.json_dict["closeMatch"] = close_matchs	1✔
169
170	def trans_rero_note(self):	1✔
171	"""Transformation notes from field.
172
173	670 $a: dataSource
174	675 $a: dataNotFound
175	680 __ $a: general
176	667 $a: nonPublic
177	260 _9 $a: seeReference
178	260 __ $a: seeReference
179	360 __ $a: seeAlsoReference
180	016 $9: REROtreatment
181	"""
182	if self.logger and self.verbose:	1✔
183	self.logger.info("Call Function", "trans_rero_note")	1✔
184	notes = {	1✔
185	"dataSource": [],
186	"dataNotFound": [],
187	"general": [],
188	"nonPublic": [],
189	"seeReference": [],
190	"seeAlsoReference": [],
191	"REROtreatment": [],
192	}
193	for field in self.marc.get_fields("670"):	1✔
194	if field.get("a"):	1✔
195	notes["dataSource"].append(field["a"].strip())	1✔
196	for field in self.marc.get_fields("675"):	1✔
197	if field.get("a"):	1✔
198	notes["dataNotFound"].append(field["a"].strip())	1✔
199	for field in self.marc.get_fields("680"):	1✔
200	if field.get("a") and field.indicators == [" ", " "]:	1✔
201	notes["general"].append(field["a"].strip())	1✔
202	for field in self.marc.get_fields("667"):	1✔
203	if field.get("a"):	1✔
204	notes["nonPublic"].append(field["a"].strip())	1✔
205	for field in self.marc.get_fields("260"):	1✔
206	if field.get("a") and field.indicators in [[" ", " "], [" ", "9"]]:	1✔
207	notes["seeReference"].append(field["a"].strip())	1✔
208	for field in self.marc.get_fields("360"):	1✔
209	if field.get("a") and field.indicators == [" ", " "]:	1✔
210	notes["seeAlsoReference"].append(field["a"].strip())	1✔
211	for field in self.marc.get_fields("016"):	1✔
212	if field.get("9"):	1✔
213	notes["REROtreatment"].append(field["9"].strip())	1✔
214	for note, value in notes.items():	1✔
215	if value:	1✔
216	self.json_dict.setdefault("note", [])	1✔
217	self.json_dict["note"].append({"noteType": note, "label": value})	1✔

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