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

84.76

/rero_mef/marctojson/do_idref_places.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/>.
"""Marctojsons transformer for IdRef records."""

import contextlib
from datetime import datetime, timezone

from rero_mef.marctojson.helper import (
    build_string_from_field,
    build_string_list_from_fields,
)


class Transformation:
    """Transformation MARC21 to JSON for Idref 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("008"):
            for func in dir(self):
                if func.startswith("trans"):
                    func = getattr(self, func)
                    func()
        else:
            msg = "No 008"
            if self.logger and self.verbose:
                self.logger.warning(f"NO TRANSFORMATION: {msg}")
            self.json_dict = {"NO TRANSFORMATION": msg}
            self.trans_idref_pid()

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

    def trans_idref_pid(self):
        """Transformation identifier from field 001."""
        if self.logger and self.verbose:
            self.logger.info("Call Function", "trans_idref_pid")
        if field_001 := self.marc.get_fields("001"):
            self.json_dict["pid"] = field_001[0].data
            self.json_dict["type"] = "bf:Place"

    def trans_idref_bnf_type(self):
        """Transformation bnf_type from field 008."""
        if self.logger and self.verbose:
            self.logger.info("Call Function", "trans_idref_bnf_type")
        self.json_dict["bnf_type"] = "sujet Rameau"

    def trans_idref_identifier(self):
        """Transformation identifier from field 003 033."""
        if self.logger and self.verbose:
            self.logger.info("Call Function", "trans_idref_identifier")
        identifiers = self.json_dict.get("identifiedBy", [])
        if field_003 := self.marc.get_fields("003"):
            uri = field_003[0].data.strip()
            identifiers.append({"type": "uri", "value": uri, "source": "IDREF"})
        for field_033 in self.marc.get_fields("033"):
            if field_033.get("2") and field_033.get("a"):
                subfield_a = field_033.get("a")
                if isinstance(subfield_a, list):
                    subfield_a = subfield_a[0]
                subfield_2 = field_033.get("2")
                if isinstance(subfield_2, list):
                    subfield_2 = subfield_2[0]
                identifiers.append(
                    {
                        "type": "uri" if subfield_a.startswith("http") else "bf:Nbn",
                        "value": subfield_a.strip(),
                        "source": subfield_2.upper(),
                    }
                )
        if identifiers:
            self.json_dict["identifiedBy"] = identifiers

    def trans_idref_relation_pid(self):
        """Transformation old pids 035 $a $9 = sudoc."""
        if self.logger and self.verbose:
            self.logger.info("Call Function", "trans_idref_relation_pid")
        for field_035 in self.marc.get_fields("035"):
            subfield_a = field_035.get("a")
            if isinstance(subfield_a, list):
                subfield_a = subfield_a[0]
            subfield_2 = field_035.get("2")
            if isinstance(subfield_2, list):
                subfield_2 = subfield_2[0]
            subfield_9 = field_035.get("9")
            if isinstance(subfield_9, list):
                subfield_9 = subfield_9[0]
            if subfield_a and subfield_9 == "sudoc":
                self.json_dict["relation_pid"] = {
                    "value": field_035["a"],
                    "type": "redirect_from",
                }
            elif subfield_2:
                identified_by = self.json_dict.get("identifiedBy", [])
                identified_by.append(
                    {
                        "source": subfield_2.upper(),
                        "type": "uri" if subfield_a.startswith("http") else "bf:Nbn",
                        "value": subfield_a,
                    }
                )
                self.json_dict["identifiedBy"] = identified_by

    def trans_idref_deleted(self):
        """Transformation deleted leader 5 == d."""
        if self.logger and self.verbose:
            self.logger.info("Call Function", "trans_idref_deleted")
        if self.marc.leader[5] == "d":
            self.json_dict["deleted"] = datetime.now(timezone.utc).isoformat()

    def trans_idref_authorized_access_point(self):
        """Transformation authorized_access_point from field 215."""
        if self.logger and self.verbose:
            self.logger.info("Call Function", "trans_idref_authorized_access_point")
        tag = "215"
        subfields = {"a": ", ", "x": " - ", "y": " - ", "z": " - "}
        try:
            if authorized_ap := build_string_from_field(self.marc[tag], subfields):
                self.json_dict["authorized_access_point"] = authorized_ap
        except Exception:
            self.json_dict["authorized_access_point"] = f"TAG: {tag} NOT FOUND"

    def trans_idref_variant_access_point(self):
        """Transformation variant_access_point from field 415."""
        if self.logger and self.verbose:
            self.logger.info("Call Function", "trans_idref_variant_access_point")
        tag = "415"
        subfields = {"a": ", ", "x": " - ", "y": " - ", "z": " - "}
        if variant_access_points := build_string_list_from_fields(
            self.marc, tag, subfields
        ):
            self.json_dict["variant_access_point"] = variant_access_points

    def trans_idref_relation(self):
        """Transformation broader related narrower 515."""
        if self.logger and self.verbose:
            self.logger.info("Call Function", "trans_idref_relation")
        relations = {}
        for tag in ["515"]:
            for field in self.marc.get_fields(tag):
                relation_type = None
                if subfield_5 := field.get("5"):
                    if subfield_5[0] == "g":
                        relation_type = "broader"
                    elif subfield_5[0] == "h":
                        relation_type = "narrower"
                    elif subfield_5[0] == "z":
                        relation_type = "related"
                if relation_type:
                    relations.setdefault(relation_type, [])
                    # TODO: $ref relation
                    # if subfield_3 := field.get('3'):
                    #     relations[relation_type].append({
                    #         '$ref':
                    #         f'https://.../concepts/idref/{subfield_3}'
                    #     })
                    # else:
                    subfields = {"a": ", ", "x": " - ", "y": " - ", "z": " - "}
                    if authorized_ap := build_string_from_field(field, subfields):
                        relations.setdefault(relation_type, [])
                        relations[relation_type].append(
                            {"authorized_access_point": authorized_ap}
                        )
        for relation, value in relations.items():
            if value:
                self.json_dict[relation] = value

    def trans_idref_classification(self):
        """Transformation classification from field 686."""
        if self.logger and self.verbose:
            self.logger.info("Call Function", "trans_idref_classification")
        classifications = []
        for field_686 in self.marc.get_fields("686"):
            if field_686.get("a"):
                classification = {
                    "type": "bf:ClassificationDdc",
                    "classificationPortion": field_686["a"].strip(),
                }
                if field_686.get("c"):
                    classification["name"] = field_686["c"]
                classifications.append(classification)
        if classifications:
            self.json_dict["classification"] = classifications

    def trans_idref_close_match(self):
        """Transformation closeMatch from field 822."""
        if self.logger and self.verbose:
            self.logger.info("Call Function", "trans_idref_close_match")
        close_matchs = []
        for field_822 in self.marc.get_fields("822"):
            with contextlib.suppress(Exception):
                close_match = {}
                if field_822.get("a") and field_822.get("2"):
                    close_match = {
                        "authorized_access_point": field_822["a"].strip(),
                        "source": field_822["2"].strip(),
                    }
                    if field_822.get("u"):
                        close_match["identifiedBy"] = [
                            {
                                "type": "uri",
                                "value": field_822["u"].strip(),
                                "source": field_822["2"].strip(),
                            }
                        ]
                if close_match:
                    close_matchs.append(close_match)
        if close_matchs:
            self.json_dict["closeMatch"] = close_matchs

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

        810 $a: dataSource
        815 $a: dataNotFound
        300 $a: general
        330 $a: general
        356 $a: general
        305 $a: seeReference
        310 $a: seeReference
        320 $a: seeReference
        """
        if self.logger and self.verbose:
            self.logger.info("Call Function", "trans_idref_note")
        notes = {
            "dataSource": [],
            "dataNotFound": [],
            "general": [],
            "seeReference": [],
        }
        for field in self.marc.get_fields("810"):
            if field.get("a"):
                notes["dataSource"].append(field["a"].strip())
        for field in self.marc.get_fields("815"):
            if field.get("a"):
                notes["dataNotFound"].append(field["a"].strip())
        for field in self.marc.get_fields("300", "330", "356"):
            if field.get("a"):
                notes["general"].append(field["a"].strip())
        for field in self.marc.get_fields("305", "310", "320"):
            if field.get("a"):
                notes["seeReference"].append(field["a"].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) 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	"""Marctojsons transformer for IdRef records."""
16
17	import contextlib	1✔
18	from datetime import datetime, timezone	1✔
19
20	from rero_mef.marctojson.helper import (	1✔
21	build_string_from_field,
22	build_string_list_from_fields,
23	)
24
25
26	class Transformation:	1✔
27	"""Transformation MARC21 to JSON for Idref concept."""
28
29	def __init__(self, marc, logger=None, verbose=False, transform=True):	1✔
30	"""Constructor."""
31	self.marc = marc	1✔
32	self.logger = logger	1✔
33	self.verbose = verbose	1✔
34	self.json_dict = {}	1✔
35	if transform:	1✔
36	self._transform()	×
37
38	def _transform(self):	1✔
39	"""Call the transformation functions."""
40	if self.marc.get_fields("008"):	×
41	for func in dir(self):	×
42	if func.startswith("trans"):	×
43	func = getattr(self, func)	×
44	func()	×
45	else:
46	msg = "No 008"	×
47	if self.logger and self.verbose:	×
48	self.logger.warning(f"NO TRANSFORMATION: {msg}")	×
49	self.json_dict = {"NO TRANSFORMATION": msg}	×
50	self.trans_idref_pid()	×
51
52	@property	1✔
53	def json(self):	1✔
54	"""Json data."""
55	return self.json_dict or None	1✔
56
57	def trans_idref_pid(self):	1✔
58	"""Transformation identifier from field 001."""
59	if self.logger and self.verbose:	1✔
60	self.logger.info("Call Function", "trans_idref_pid")	1✔
61	if field_001 := self.marc.get_fields("001"):	1✔
62	self.json_dict["pid"] = field_001[0].data	1✔
63	self.json_dict["type"] = "bf:Place"	1✔
64
65	def trans_idref_bnf_type(self):	1✔
66	"""Transformation bnf_type from field 008."""
67	if self.logger and self.verbose:	×
68	self.logger.info("Call Function", "trans_idref_bnf_type")	×
69	self.json_dict["bnf_type"] = "sujet Rameau"	×
70
71	def trans_idref_identifier(self):	1✔
72	"""Transformation identifier from field 003 033."""
73	if self.logger and self.verbose:	1✔
74	self.logger.info("Call Function", "trans_idref_identifier")	1✔
75	identifiers = self.json_dict.get("identifiedBy", [])	1✔
76	if field_003 := self.marc.get_fields("003"):	1✔
77	uri = field_003[0].data.strip()	1✔
78	identifiers.append({"type": "uri", "value": uri, "source": "IDREF"})	1✔
79	for field_033 in self.marc.get_fields("033"):	1✔
80	if field_033.get("2") and field_033.get("a"):	1✔
81	subfield_a = field_033.get("a")	1✔
82	if isinstance(subfield_a, list):	1✔
83	subfield_a = subfield_a[0]	×
84	subfield_2 = field_033.get("2")	1✔
85	if isinstance(subfield_2, list):	1✔
86	subfield_2 = subfield_2[0]	×
87	identifiers.append(	1✔
88	{
89	"type": "uri" if subfield_a.startswith("http") else "bf:Nbn",
90	"value": subfield_a.strip(),
91	"source": subfield_2.upper(),
92	}
93	)
94	if identifiers:	1✔
95	self.json_dict["identifiedBy"] = identifiers	1✔
96
97	def trans_idref_relation_pid(self):	1✔
98	"""Transformation old pids 035 $a $9 = sudoc."""
99	if self.logger and self.verbose:	1✔
100	self.logger.info("Call Function", "trans_idref_relation_pid")	1✔
101	for field_035 in self.marc.get_fields("035"):	1✔
102	subfield_a = field_035.get("a")	1✔
103	if isinstance(subfield_a, list):	1✔
104	subfield_a = subfield_a[0]	×
105	subfield_2 = field_035.get("2")	1✔
106	if isinstance(subfield_2, list):	1✔
107	subfield_2 = subfield_2[0]	×
108	subfield_9 = field_035.get("9")	1✔
109	if isinstance(subfield_9, list):	1✔
110	subfield_9 = subfield_9[0]	×
111	if subfield_a and subfield_9 == "sudoc":	1✔
112	self.json_dict["relation_pid"] = {	1✔
113	"value": field_035["a"],
114	"type": "redirect_from",
115	}
116	elif subfield_2:	1✔
117	identified_by = self.json_dict.get("identifiedBy", [])	1✔
118	identified_by.append(	1✔
119	{
120	"source": subfield_2.upper(),
121	"type": "uri" if subfield_a.startswith("http") else "bf:Nbn",
122	"value": subfield_a,
123	}
124	)
125	self.json_dict["identifiedBy"] = identified_by	1✔
126
127	def trans_idref_deleted(self):	1✔
128	"""Transformation deleted leader 5 == d."""
129	if self.logger and self.verbose:	1✔
130	self.logger.info("Call Function", "trans_idref_deleted")	1✔
131	if self.marc.leader[5] == "d":	1✔
132	self.json_dict["deleted"] = datetime.now(timezone.utc).isoformat()	1✔
133
134	def trans_idref_authorized_access_point(self):	1✔
135	"""Transformation authorized_access_point from field 215."""
136	if self.logger and self.verbose:	1✔
137	self.logger.info("Call Function", "trans_idref_authorized_access_point")	1✔
138	tag = "215"	1✔
139	subfields = {"a": ", ", "x": " - ", "y": " - ", "z": " - "}	1✔
140	try:	1✔
141	if authorized_ap := build_string_from_field(self.marc[tag], subfields):	1✔
142	self.json_dict["authorized_access_point"] = authorized_ap	1✔
143	except Exception:	×
144	self.json_dict["authorized_access_point"] = f"TAG: {tag} NOT FOUND"	×
145
146	def trans_idref_variant_access_point(self):	1✔
147	"""Transformation variant_access_point from field 415."""
148	if self.logger and self.verbose:	1✔
149	self.logger.info("Call Function", "trans_idref_variant_access_point")	1✔
150	tag = "415"	1✔
151	subfields = {"a": ", ", "x": " - ", "y": " - ", "z": " - "}	1✔
152	if variant_access_points := build_string_list_from_fields(	1✔
153	self.marc, tag, subfields
154	):
155	self.json_dict["variant_access_point"] = variant_access_points	1✔
156
157	def trans_idref_relation(self):	1✔
158	"""Transformation broader related narrower 515."""
159	if self.logger and self.verbose:	1✔
160	self.logger.info("Call Function", "trans_idref_relation")	1✔
161	relations = {}	1✔
162	for tag in ["515"]:	1✔
163	for field in self.marc.get_fields(tag):	1✔
164	relation_type = None	1✔
165	if subfield_5 := field.get("5"):	1✔
166	if subfield_5[0] == "g":	1✔
167	relation_type = "broader"	1✔
168	elif subfield_5[0] == "h":	×
169	relation_type = "narrower"	×
170	elif subfield_5[0] == "z":	×
171	relation_type = "related"	×
172	if relation_type:	1✔
173	relations.setdefault(relation_type, [])	1✔
174	# TODO: $ref relation
175	# if subfield_3 := field.get('3'):
176	# relations[relation_type].append({
177	# '$ref':
178	# f'https://.../concepts/idref/{subfield_3}'
179	# })
180	# else:
181	subfields = {"a": ", ", "x": " - ", "y": " - ", "z": " - "}	1✔
182	if authorized_ap := build_string_from_field(field, subfields):	1✔
183	relations.setdefault(relation_type, [])	1✔
184	relations[relation_type].append(	1✔
185	{"authorized_access_point": authorized_ap}
186	)
187	for relation, value in relations.items():	1✔
188	if value:	1✔
189	self.json_dict[relation] = value	1✔
190
191	def trans_idref_classification(self):	1✔
192	"""Transformation classification from field 686."""
193	if self.logger and self.verbose:	1✔
194	self.logger.info("Call Function", "trans_idref_classification")	1✔
195	classifications = []	1✔
196	for field_686 in self.marc.get_fields("686"):	1✔
197	if field_686.get("a"):	1✔
198	classification = {	1✔
199	"type": "bf:ClassificationDdc",
200	"classificationPortion": field_686["a"].strip(),
201	}
202	if field_686.get("c"):	1✔
203	classification["name"] = field_686["c"]	1✔
204	classifications.append(classification)	1✔
205	if classifications:	1✔
206	self.json_dict["classification"] = classifications	1✔
207
208	def trans_idref_close_match(self):	1✔
209	"""Transformation closeMatch from field 822."""
210	if self.logger and self.verbose:	1✔
211	self.logger.info("Call Function", "trans_idref_close_match")	1✔
212	close_matchs = []	1✔
213	for field_822 in self.marc.get_fields("822"):	1✔
214	with contextlib.suppress(Exception):	1✔
215	close_match = {}	1✔
216	if field_822.get("a") and field_822.get("2"):	1✔
217	close_match = {	1✔
218	"authorized_access_point": field_822["a"].strip(),
219	"source": field_822["2"].strip(),
220	}
221	if field_822.get("u"):	1✔
222	close_match["identifiedBy"] = [	1✔
223	{
224	"type": "uri",
225	"value": field_822["u"].strip(),
226	"source": field_822["2"].strip(),
227	}
228	]
229	if close_match:	1✔
230	close_matchs.append(close_match)	1✔
231	if close_matchs:	1✔
232	self.json_dict["closeMatch"] = close_matchs	1✔
233
234	def trans_idref_note(self):	1✔
235	"""Transformation notes from field.
236
237	810 $a: dataSource
238	815 $a: dataNotFound
239	300 $a: general
240	330 $a: general
241	356 $a: general
242	305 $a: seeReference
243	310 $a: seeReference
244	320 $a: seeReference
245	"""
246	if self.logger and self.verbose:	1✔
247	self.logger.info("Call Function", "trans_idref_note")	1✔
248	notes = {	1✔
249	"dataSource": [],
250	"dataNotFound": [],
251	"general": [],
252	"seeReference": [],
253	}
254	for field in self.marc.get_fields("810"):	1✔
255	if field.get("a"):	1✔
256	notes["dataSource"].append(field["a"].strip())	1✔
257	for field in self.marc.get_fields("815"):	1✔
258	if field.get("a"):	1✔
259	notes["dataNotFound"].append(field["a"].strip())	1✔
260	for field in self.marc.get_fields("300", "330", "356"):	1✔
261	if field.get("a"):	1✔
262	notes["general"].append(field["a"].strip())	1✔
263	for field in self.marc.get_fields("305", "310", "320"):	1✔
264	if field.get("a"):	1✔
265	notes["seeReference"].append(field["a"].strip())	1✔
266	for note, value in notes.items():	1✔
267	if value:	1✔
268	self.json_dict.setdefault("note", [])	1✔
269	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