deepseek den plan umsetzen lassen

src/parser.py

@@ -0,0 +1,118 @@
+"""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_float(value: Any, field_name: str) -> float:
+    if value is None or str(value).strip() == "":
+        return 0.0
+    text = str(value).strip().replace(",", ".")
+    try:
+        return float(text)
+    except ValueError:
+        return 0.0
+
+
+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 _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
+        for row in ws.iter_rows(min_row=2, max_row=ws.max_row, 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 = _to_float(row[col_map.get("Stellenanteil in %", 0)], "Stellenanteil in %") if len(row) > col_map.get("Stellenanteil in %", 0) else 0.0
+            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))