"""Input-Datei-Parser: liest monatliche Sheets und erzeugt EmployeeRecords.""" import re from pathlib import Path from typing import Any, Optional from openpyxl import load_workbook from openpyxl.utils import column_index_from_string from .config import is_month_sheet from .models import EmployeeRecord, FieldSchema def _to_text(value: Any) -> str: if value is None: return "" return str(value).strip() def _to_int_or_none(value: Any) -> Optional[int]: if value is None or str(value).strip() == "": return None try: return int(value) except (ValueError, TypeError): return None def _to_percent_string(value: Any) -> str: """Wandelt einen Prozentwert in einen String ohne %-Zeichen um. Excel speichert Prozentwerte als Dezimalbruch (50 % = 0.5). Falls der Rohwert eine Zahl ≤ 1 ist, wird er als Excel-Prozentwert interpretiert und mit 100 multipliziert. """ if value is None or str(value).strip() == "": return "" # Falls es bereits ein String mit %-Zeichen ist if isinstance(value, str): return value.strip().rstrip("% ") # Numerischer Wert: Excel-Prozentwert (0.5 = 50 %) oder Ganzzahl (50 = 50 %) if isinstance(value, (int, float)): if 0 < value <= 1: return str(int(value * 100)) if value == 0: return "0" # Wert > 1: bereits als Ganzzahl gemeint (z. B. 50) return str(int(value)) return str(value).strip() def _build_column_map(schema: list[FieldSchema]) -> dict[str, int]: """Baut ein Mapping von field_name → 0-basierter Spaltenindex.""" col_map: dict[str, int] = {} for field in schema: col_map[field.field_name] = column_index_from_string(field.column) - 1 return col_map def parse_input_file( filepath: Path, schema: list[FieldSchema], month_sheet_pattern: str = "* 20??", skip_empty_name: bool = True, skip_empty_antragsnummer: bool = True, ) -> list[EmployeeRecord]: """ Öffnet die Input-Datei, iteriert über alle Monats-Sheets und gibt eine Liste von EmployeeRecords zurück. """ wb = load_workbook(filepath, data_only=True) col_map = _build_column_map(schema) records: list[EmployeeRecord] = [] for sheet_name in wb.sheetnames: if not is_month_sheet(sheet_name, month_sheet_pattern): continue ws = wb[sheet_name] # Header ist Zeile 1, Daten ab Zeile 2 # min_col=1 explizit setzen für korrekte Spaltenzuordnung schema_col_count = len(schema) for row in ws.iter_rows(min_row=2, max_row=ws.max_row, min_col=1, max_col=schema_col_count, values_only=True): name = _to_text(row[col_map.get("Name", 0)]) if len(row) > col_map.get("Name", 0) else "" antragsnummer = _to_text(row[col_map.get("Antragsnummer", 0)]) if len(row) > col_map.get("Antragsnummer", 0) else "" if skip_empty_name and not name: continue if skip_empty_antragsnummer and not antragsnummer: continue qualifikation = _to_text(row[col_map.get("Qualifikation", 0)]) if len(row) > col_map.get("Qualifikation", 0) else "" funktion = _to_text(row[col_map.get("Funktion", 0)]) if len(row) > col_map.get("Funktion", 0) else "" stellenanteil_raw = row[col_map.get("Stellenanteil in %", 0)] if len(row) > col_map.get("Stellenanteil in %", 0) else None stellenanteil = _to_percent_string(stellenanteil_raw) monat = _to_text(row[col_map.get("Monat (MM.JJ)", 0)]) if len(row) > col_map.get("Monat (MM.JJ)", 0) else "" taetigkeit = _to_text(row[col_map.get("Tätigkeit", 0)]) if len(row) > col_map.get("Tätigkeit", 0) else "" record_id = _to_int_or_none(row[col_map.get("ID", 0)]) if len(row) > col_map.get("ID", 0) else None kommentar = _to_text(row[col_map.get("Kommentar", 1)]) if "Kommentar" in col_map and len(row) > col_map["Kommentar"] else None records.append(EmployeeRecord( id=record_id, name=name, antragsnummer=antragsnummer, qualifikation=qualifikation, funktion=funktion, stellenanteil=stellenanteil, monat=monat, taetigkeit=taetigkeit, kommentar=kommentar, )) wb.close() return records def group_by_employee(records: list[EmployeeRecord]) -> dict[str, list[EmployeeRecord]]: """Gruppiert Records nach Mitarbeitername.""" groups: dict[str, list[EmployeeRecord]] = {} for record in records: if record.name not in groups: groups[record.name] = [] groups[record.name].append(record) return groups def get_unique_antragsnummern(records: list[EmployeeRecord]) -> list[str]: """Gibt eine sortierte Liste aller eindeutigen Antragsnummern zurück.""" return sorted(set(r.antragsnummer for r in records if r.antragsnummer)) def get_unique_employees(records: list[EmployeeRecord]) -> list[str]: """Gibt eine sortierte Liste aller eindeutigen Mitarbeiternamen zurück.""" return sorted(set(r.name for r in records if r.name))