-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathmarc2finc.py
186 lines (152 loc) · 7.91 KB
/
marc2finc.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
import click
from pymarc import MARCReader
import json
from pathlib import Path
import sys
# Lokale Importe
from help.marc_utils import MarcUtils
from help.slublogging import getSlubLogger
from help.linkml_generator import generate_models_from_schema
def process_marc_files(sourcefile, targetfile=None, models=None):
"""
Verarbeite Marc21-Dateien und erstelle Pydantic- und Dataclass-Objekte
Args:
sourcefile: Pfad zur MARC21-Quelldatei
targetfile: Optional. Pfad zur JSON-Zieldatei
models: Optional. Dictionary mit den zu verwendenden Modellklassen
Returns:
Tuple aus (PydanticFinc-Liste, DataclassFinc-Liste)
"""
log = getSlubLogger('process_marc_files')
log.info(f"Verarbeite Datei: {sourcefile}")
# Wenn keine Modelle übergeben wurden, verwende die Standardmodelle
if models is None:
log.info("Keine Modelle übergeben, generiere Modelle aus Schema")
models = generate_models_from_schema("schema/finc.yaml")
PydanticFinc = models["PydanticFinc"]
DataclassFinc = models["DataclassFinc"]
pydantics = []
dataclasses = []
# Marc21 Datei einlesen
with open(sourcefile, 'rb') as f:
reader = MARCReader(f)
# PPN und Titel ausgeben zur Kontrolle
for record in reader:
log.info(f"PPN: {record['001'].data} - Titel: {record['245']['a']}")
# PPN als ID verwenden
id = f"0-{record['001'].data}"
record_id = record['001'].data
# title = 245ab, clean, join(": "), first
titles = MarcUtils.extract_marc_subfields(record, "245ab", join=": ")
# Nur ein Titel sollte verwendet werden, wenn mehrere vorhanden sind (ungewöhnlich)
title = titles[0] if titles else ""
# topic = 600abcdefghjklmnopqrstuvxyz:610abcdefghklmnoprstuvxyz:611acdefghjklnpqstuvxyz:630adefghklmnoprstvxyz:650abcdevxyz:689agxz:655abvxyz:651avxyz:648avxyz:970de:937abc:653a
complex_topic_spec = "600abcdefghjklmnopqrstuvxyz:610abcdefghklmnoprstuvxyz:611acdefghjklnpqstuvxyz:630adefghklmnoprstvxyz:650abcdevxyz:689agxz:655abvxyz:651avxyz:648avxyz:970de:937abc:653a"
# Verwende die parse_complex_field_spec Methode, um die komplexe Spezifikation zu zerlegen
topic_specs = MarcUtils.parse_complex_field_spec(complex_topic_spec)
log.debug(f"Verarbeite {len(topic_specs)} Themen-Feldspezifikationen")
topics = MarcUtils.extract_marc_subfields(record, *topic_specs)
log.debug(f"Extrahierte {len(topics)} Themen aus dem Record")
# DEMO: recordtype ist Pflichtfeld und soll String sein!
recordtype = "marc"
# isbn = 020a:772z:773z
complex_isbn_spec = "020a:772z:773z"
isbn_specs = MarcUtils.parse_complex_field_spec(complex_isbn_spec)
isbn = MarcUtils.extract_marc_subfields(record, *isbn_specs)
# isbn liefert eine Liste, ist aber nicht Multi-Valued
if isbn and isinstance(isbn, list) and len(isbn) == 1:
isbn = isbn[0]
else:
isbn = None
# DEMO: ISBN die nicht auf den RegEx passt
# isbn = "DIESDAS112"
try:
pydantic_record = PydanticFinc(
id=id,
record_id=record_id,
title=title,
topic=topics,
recordtype=recordtype,
isbn=isbn
)
pydantics.append(pydantic_record)
except Exception as e:
log.error(f"Pydantic: Fehler beim Erstellen des PydanticFinc Objekts: {e}")
try:
dataclass_record = DataclassFinc(
id=id,
record_id=record_id,
title=title,
topic=topics,
recordtype=recordtype,
isbn=isbn
)
dataclasses.append(dataclass_record)
except Exception as e:
log.error(f"Dataclass: Fehler beim Erstellen des Dataclass Finc Objekts: {e}")
# Anschließende Ausgabe oder Verarbeitung der erstellten Objekte, z.B. als JSON speichern
if targetfile:
output_path = Path(targetfile)
# Stelle sicher, dass der Zielordner existiert
output_path.parent.mkdir(parents=True, exist_ok=True)
# Extrahiere den Basisnamen ohne Erweiterung
base_name = output_path.stem
base_dir = output_path.parent
# Erzeuge die Dateinamen für die Ausgabedateien
pydantic_file = base_dir / f"{base_name}.pydantic.jsonl"
dataclass_file = base_dir / f"{base_name}.dataclass.jsonl"
log.info(f"Speichere Pydantic-Modelle in {pydantic_file}")
log.info(f"Speichere Dataclass-Modelle in {dataclass_file}")
try:
# Speichere Pydantic-Modelle im JsonL-Format (ein JSON pro Zeile)
with open(pydantic_file, 'w', encoding='utf-8') as f:
for model in pydantics:
# exclude_none=True entfernt alle None-Werte und exclude_unset=True entfernt ungesetzte Felder
model_dict = model.model_dump(exclude_none=True, exclude_defaults=True)
# Zusätzlich leere Listen entfernen
cleaned_dict = {k: v for k, v in model_dict.items() if not (isinstance(v, list) and len(v) == 0)}
f.write(json.dumps(cleaned_dict, ensure_ascii=False) + '\n')
# Speichere Dataclass-Modelle im JsonL-Format (ein JSON pro Zeile)
with open(dataclass_file, 'w', encoding='utf-8') as f:
for model in dataclasses:
# Konvertiere Dataclass in Dict und entferne None-Werte und leere Listen
model_dict = model.__dict__.copy()
cleaned_dict = {k: v for k, v in model_dict.items() if v is not None and not (isinstance(v, list) and len(v) == 0)}
f.write(json.dumps(cleaned_dict, ensure_ascii=False) + '\n')
log.info(f"Ergebnisse erfolgreich gespeichert: {len(pydantics)} Pydantic-Modelle, {len(dataclasses)} Dataclass-Modelle")
except Exception as e:
log.error(f"Fehler beim Speichern der Ergebnisse: {e}")
return pydantics, dataclasses
@click.command()
@click.option('-s', '--source', required=True, help='Pfad zur MARC21 Quelldatei')
@click.option('-t', '--target', required=True, help='Pfad zur Ausgabedatei (ohne Erweiterung)')
@click.option('--schema', default='schema/finc.yaml', help='Pfad zum LinkML-Schema (default: schema/finc.yaml)')
def main(source, target, schema):
"""Konvertiere MARC21 zu FINC JSON."""
# Optional: Pfad zur Logging-Konfigurationsdatei angeben, falls gewünscht
log = getSlubLogger('marc2finc')
sourcefile = source
targetfile = target
schema_file = schema
log.info(f"Quelle: {sourcefile}")
log.info(f"Ziel-Basis: {targetfile}")
log.info(f"Schema: {schema_file}")
# Generiere die Modelle aus dem Schema
try:
models = generate_models_from_schema(schema_file)
process_marc_files(sourcefile, targetfile, models)
# Erstelle Dateinamen für die Ausgabe
output_path = Path(targetfile)
base_name = output_path.stem
base_dir = output_path.parent
pydantic_file = base_dir / f"{base_name}.pydantic.jsonl"
dataclass_file = base_dir / f"{base_name}.dataclass.jsonl"
click.echo("Verarbeitung abgeschlossen!")
click.echo(f"Pydantic-Modelle wurden in {pydantic_file} gespeichert.")
click.echo(f"Dataclass-Modelle wurden in {dataclass_file} gespeichert.")
except Exception as e:
log.error(f"Fehler bei der Verarbeitung: {e}")
click.echo(f"Fehler: {e}", err=True)
sys.exit(1)
if __name__ == "__main__":
main()