Add master data parser with German decimal conversion

This commit is contained in:
Rob committed 2026-04-21 15:28:13 +02:00
1 parent 10a27ad4b6
commit f2da062c49
2 files changed
+97 -6

No files matched your search

+92 -1
View File
@@ -10,7 +10,7 @@ from openpyxl import load_workbook
from openpyxl.utils.cell import column_index_from_string
from .errors import ValidationError
from .models import AllocationRecord
from .models import AllocationRecord, MasterDataRecord
def _to_text(value: Any) -> str:
@@ -30,6 +30,27 @@ def _to_float(value: Any, field_name: str) -> float:
raise ValidationError(f"Invalid numeric value for {field_name}: {value}") from exc
def _parse_german_decimal(
value: Any,
field_name: str,
decimal_separator: str,
thousands_separator: str,
) -> float:
if value is None or str(value).strip() == "":
raise ValidationError(f"Missing numeric value for {field_name}.")
text = str(value).strip()
if thousands_separator:
text = text.replace(thousands_separator, "")
if decimal_separator:
text = text.replace(decimal_separator, ".")
try:
return float(text)
except ValueError as exc:
raise ValidationError(f"Invalid numeric value for {field_name}: {value}") from exc
def _parse_month_marker(value: Any, prefix: str, date_format: str) -> datetime | None:
marker = _to_text(value)
if not marker:
@@ -182,3 +203,73 @@ def parse_allocation_records(
workbook.close()
return normalized_records
def parse_masterdata_records(
config: dict[str, Any],
project_root: Path,
) -> list[MasterDataRecord]:
files_cfg = config["files"]
sheets_cfg = config["sheets"]
paths_cfg = config["paths"]
master_cfg = config["masterdata"]
columns_cfg = master_cfg["columns"]
decimal_cfg = master_cfg["decimal"]
workbook_path = (
project_root
/ paths_cfg["input_dir"]
/ files_cfg["masterdata_workbook"]
)
if not workbook_path.exists():
raise ValidationError(f"Master data workbook not found: {workbook_path}")
workbook = load_workbook(workbook_path, data_only=True)
sheet_name = sheets_cfg["masterdata"]
if sheet_name not in workbook.sheetnames:
raise ValidationError(f"Sheet '{sheet_name}' not found in {workbook_path.name}")
worksheet = workbook[sheet_name]
id_col = column_index_from_string(columns_cfg["id"])
antragsnummer_col = column_index_from_string(columns_cfg["antragsnummer"])
name_col = column_index_from_string(columns_cfg["name"])
qualifikation_col = column_index_from_string(columns_cfg["qualifikation"])
stellenanteil_col = column_index_from_string(columns_cfg["stellenanteil"])
funktion_col = column_index_from_string(columns_cfg["funktion"])
taetigkeit_col = column_index_from_string(columns_cfg["taetigkeit"])
records: list[MasterDataRecord] = []
for row_index in range(2, worksheet.max_row + 1):
name = _to_text(worksheet.cell(row=row_index, column=name_col).value)
if not name:
continue
stellenanteil = _parse_german_decimal(
worksheet.cell(row=row_index, column=stellenanteil_col).value,
field_name=f"stellenanteil row {row_index}",
decimal_separator=decimal_cfg.get("decimal_separator", ","),
thousands_separator=decimal_cfg.get("thousands_separator", "."),
)
records.append(
MasterDataRecord(
record_id=_to_text(worksheet.cell(row=row_index, column=id_col).value),
antragsnummer=_to_text(
worksheet.cell(row=row_index, column=antragsnummer_col).value
),
name=name,
qualifikation=_to_text(
worksheet.cell(row=row_index, column=qualifikation_col).value
),
stellenanteil_percent=stellenanteil,
funktion=_to_text(worksheet.cell(row=row_index, column=funktion_col).value),
taetigkeit=_to_text(
worksheet.cell(row=row_index, column=taetigkeit_col).value
),
)
)
workbook.close()
return records