284 lines
9.3 KiB
Python
284 lines
9.3 KiB
Python
"""Parsing and normalization for source workbooks."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from openpyxl import load_workbook
|
|
from openpyxl.utils.cell import column_index_from_string
|
|
|
|
from .errors import ValidationError
|
|
from .models import AllocationRecord, MasterDataRecord
|
|
|
|
|
|
def _to_text(value: Any) -> str:
|
|
if value is None:
|
|
return ""
|
|
return str(value).strip()
|
|
|
|
|
|
def _to_float(value: Any, field_name: str) -> float:
|
|
if value is None or str(value).strip() == "":
|
|
raise ValidationError(f"Missing numeric value for {field_name}.")
|
|
|
|
text = str(value).strip().replace(",", ".")
|
|
try:
|
|
return float(text)
|
|
except ValueError as exc:
|
|
raise ValidationError(f"Invalid numeric value for {field_name}: {value}") from exc
|
|
|
|
|
|
def _to_float_or_none(value: Any) -> float | None:
|
|
if value is None or str(value).strip() == "":
|
|
return None
|
|
|
|
text = str(value).strip().replace(",", ".")
|
|
try:
|
|
return float(text)
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
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:
|
|
return None
|
|
|
|
if not marker.lower().startswith(prefix.lower()):
|
|
return None
|
|
|
|
date_text = marker[len(prefix) :].strip()
|
|
return datetime.strptime(date_text, date_format)
|
|
|
|
|
|
def _resolve_month_column_index(
|
|
worksheet: Any,
|
|
header_row: int,
|
|
first_month_col: str,
|
|
target_month: int,
|
|
target_year: int,
|
|
month_prefix: str,
|
|
month_date_format: str,
|
|
) -> int:
|
|
start_index = column_index_from_string(first_month_col)
|
|
matched_columns: list[int] = []
|
|
|
|
for column_index in range(start_index, worksheet.max_column + 1):
|
|
value = worksheet.cell(row=header_row, column=column_index).value
|
|
parsed = _parse_month_marker(value, month_prefix, month_date_format)
|
|
if parsed is None:
|
|
continue
|
|
if parsed.month == target_month and parsed.year == target_year:
|
|
matched_columns.append(column_index)
|
|
|
|
if len(matched_columns) == 0:
|
|
raise ValidationError(
|
|
f"No month column found for {target_month:02d}/{target_year} in allocation workbook."
|
|
)
|
|
|
|
if len(matched_columns) > 1:
|
|
raise ValidationError(
|
|
f"Multiple month columns found for {target_month:02d}/{target_year} in allocation workbook."
|
|
)
|
|
|
|
return matched_columns[0]
|
|
|
|
|
|
def _is_inactive_value(value: Any, inactive_values: list[Any]) -> bool:
|
|
normalized = _to_text(value)
|
|
if normalized in {"", "-"}:
|
|
return True
|
|
|
|
for inactive in inactive_values:
|
|
if str(inactive).strip() == normalized:
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
def parse_allocation_records(
|
|
config: dict[str, Any],
|
|
report_month: int,
|
|
report_year: int,
|
|
project_root: Path,
|
|
) -> list[AllocationRecord]:
|
|
allocation_cfg = config["allocation"]
|
|
columns_cfg = allocation_cfg["columns"]
|
|
files_cfg = config["files"]
|
|
sheets_cfg = config["sheets"]
|
|
paths_cfg = config["paths"]
|
|
|
|
workbook_path = (
|
|
project_root
|
|
/ paths_cfg["input_dir"]
|
|
/ files_cfg["allocation_workbook"]
|
|
)
|
|
|
|
if not workbook_path.exists():
|
|
raise ValidationError(f"Allocation workbook not found: {workbook_path}")
|
|
|
|
workbook = load_workbook(workbook_path, data_only=True)
|
|
sheet_name = sheets_cfg["allocation"]
|
|
if sheet_name not in workbook.sheetnames:
|
|
raise ValidationError(f"Sheet '{sheet_name}' not found in {workbook_path.name}")
|
|
worksheet = workbook[sheet_name]
|
|
|
|
header_row = allocation_cfg["header_month_marker_row"]
|
|
month_marker_cfg = allocation_cfg["month_marker"]
|
|
month_column = _resolve_month_column_index(
|
|
worksheet=worksheet,
|
|
header_row=header_row,
|
|
first_month_col=columns_cfg["first_month_column"],
|
|
target_month=report_month,
|
|
target_year=report_year,
|
|
month_prefix=month_marker_cfg["prefix"],
|
|
month_date_format=month_marker_cfg["date_format"],
|
|
)
|
|
|
|
employee_col = column_index_from_string(columns_cfg["employee_name"])
|
|
contract_col = column_index_from_string(columns_cfg["contract_hours_weekly"])
|
|
project_col = column_index_from_string(columns_cfg["project_key"])
|
|
remarks_col = column_index_from_string(columns_cfg["remarks"])
|
|
inactive_values = allocation_cfg.get("inactive_values", ["", "-", "0", 0])
|
|
|
|
current_employee = ""
|
|
current_contract_hours: float | None = None
|
|
normalized_records: list[AllocationRecord] = []
|
|
|
|
for row_index in range(header_row + 1, worksheet.max_row + 1):
|
|
row_employee = _to_text(worksheet.cell(row=row_index, column=employee_col).value)
|
|
row_contract = worksheet.cell(row=row_index, column=contract_col).value
|
|
row_project = _to_text(worksheet.cell(row=row_index, column=project_col).value)
|
|
|
|
if row_employee:
|
|
current_employee = row_employee
|
|
current_contract_hours = _to_float_or_none(row_contract)
|
|
|
|
if not row_project:
|
|
continue
|
|
|
|
if not current_employee or current_contract_hours is None:
|
|
raise ValidationError(
|
|
f"Project row {row_index} has no inherited employee context."
|
|
)
|
|
|
|
month_value = worksheet.cell(row=row_index, column=month_column).value
|
|
if _is_inactive_value(month_value, inactive_values):
|
|
continue
|
|
|
|
funded_hours = _to_float(month_value, f"funded_hours row {row_index}")
|
|
if funded_hours <= 0:
|
|
continue
|
|
|
|
if current_contract_hours <= 0:
|
|
raise ValidationError(
|
|
f"Contract hours must be > 0 for employee '{current_employee}'"
|
|
)
|
|
|
|
notes = _to_text(worksheet.cell(row=row_index, column=remarks_col).value)
|
|
normalized_records.append(
|
|
AllocationRecord(
|
|
employee_name=current_employee,
|
|
contract_hours_weekly=current_contract_hours,
|
|
project_key=row_project,
|
|
funded_hours_weekly=funded_hours,
|
|
notes=notes,
|
|
)
|
|
)
|
|
|
|
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
|