major refactoring

src/wce_crm/custom_widget_registry.py

@@ -0,0 +1,9 @@
+from __future__ import annotations
+
+from typing import Final
+
+CUSTOM_WIDGET_NAMES: Final[frozenset] = frozenset(
+    [
+        "grunderfassung_suche",
+    ]
+)

src/wce_crm/data_models.py

@@ -0,0 +1,297 @@
+from __future__ import annotations
+
+import datetime
+import json
+from typing import Annotated, Any, Final
+
+from pydantic import (
+    AwareDatetime,
+    BaseModel,
+    ConfigDict,
+    EmailStr,
+    Field,
+    field_validator,
+    model_validator,
+)
+
+ValidAge = Annotated[int, Field(ge=0, le=99)]
+
+
+def _parse_json(value: Any) -> str:
+    if isinstance(value, datetime.date):
+        return value.isoformat()
+    elif isinstance(value, datetime.datetime):
+        return value.isoformat()
+    else:
+        raise TypeError
+
+
+COLUMN_SEP: Final[str] = "__"
+
+
+class FlatBaseModel(BaseModel):
+    """
+    Optimised Pydantic base class, which parses JSON strings and column
+    separators recursively and correctly
+    """
+
+    @classmethod
+    def _recursive_parse_json(
+        cls,
+        data: Any,
+    ) -> Any:
+        """look for JSON list strings and parse them"""
+        if isinstance(data, str) and data.startswith("[") and data.endswith("]"):
+            try:
+                parsed = json.loads(data)
+                # Falls die Liste selbst wieder konvertiert werden muss (z.B. Sub-Dicts)
+                return cls._recursive_parse_json(parsed)
+            except json.JSONDecodeError:
+                return data
+        elif isinstance(data, dict):
+            return {k: cls._recursive_parse_json(v) for k, v in data.items()}
+        elif isinstance(data, list):
+            return [cls._recursive_parse_json(item) for item in data]
+        return data
+
+    @classmethod
+    def _recursive_unflatten(
+        cls,
+        data: Any,
+    ) -> Any:
+        """building nested structure using column spearator sequence"""
+        if isinstance(data, dict):
+            unflattened_level = {}
+            for key, value in data.items():
+                if COLUMN_SEP in key:
+                    parts = key.split(COLUMN_SEP)
+                    aktuell = unflattened_level
+                    for part in parts[:-1]:
+                        if part not in aktuell or not isinstance(aktuell[part], dict):
+                            aktuell[part] = {}
+                        aktuell = aktuell[part]
+                    aktuell[parts[-1]] = value
+                else:
+                    unflattened_level[key] = value
+
+            return {k: cls._recursive_unflatten(v) for k, v in unflattened_level.items()}
+
+        elif isinstance(data, list):
+            return [cls._recursive_unflatten(item) for item in data]
+
+        return data
+
+    @model_validator(mode="before")
+    @classmethod
+    def __unflatten_input(
+        cls,
+        data: Any,
+    ) -> Any:  # type: ignore
+        """entry control: prepare flat DB/GUI data for Pydantic"""
+        if not isinstance(data, dict):
+            return data
+
+        # setp 1: convert all JSON-Strings to lists
+        json_parsed_data = cls._recursive_parse_json(data)
+        # step 2: build nested structure based on defined separator sequence
+        final_nested_data = cls._recursive_unflatten(json_parsed_data)
+
+        return final_nested_data
+
+    def to_db(self, *args, **kwargs) -> dict[str, Any]:
+        """output for DB: flat, lists as JSON-Strings"""
+        nested = super().model_dump(*args, **kwargs)
+        return self.__flatten_dict(nested, serialize_lists=True)
+
+    def to_gui(self, *args, **kwargs) -> dict[str, Any]:
+        """output for GUI: flat, but lists remain Python lists"""
+        nested = super().model_dump(*args, **kwargs)
+        return self.__flatten_dict(nested, serialize_lists=False)
+
+    @classmethod
+    def __flatten_dict(
+        cls,
+        nested_dict: dict,
+        parent_key: str = "",
+        serialize_lists: bool = True,
+    ) -> dict[str, Any]:
+        """recursive function to flatten the structure (for outputs)"""
+        items = []
+        for k, v in nested_dict.items():
+            new_key = f"{parent_key}{COLUMN_SEP}{k}" if parent_key else k
+
+            if isinstance(v, dict):
+                items.extend(cls.__flatten_dict(v, new_key, serialize_lists).items())
+            elif isinstance(v, list):
+                processed_list = []
+                for item in v:
+                    if isinstance(item, dict):
+                        processed_list.append(
+                            cls.__flatten_dict(item, serialize_lists=serialize_lists)
+                        )
+                    else:
+                        processed_list.append(item)
+
+                if serialize_lists:
+                    items.append((new_key, json.dumps(processed_list, default=_parse_json)))
+                else:
+                    items.append((new_key, processed_list))
+            else:
+                items.append((new_key, v))
+        return dict(items)
+
+
+class Grunderfassung(FlatBaseModel):
+    # default in SQLAlchemy with lambda and timezone-aware datetime
+    Metadaten_erstellung: AwareDatetime | None = None
+    Metadaten_aktualisierung: AwareDatetime | None = None  # see above
+    Metadaten_nutzer: str | None
+    Metadaten_wiedereintrittsdatum: datetime.date | None = None
+    Grunderfassung_fallnummer: str
+    Grunderfassung_notiz: str | None
+
+    Partnersuche: Grunderfassung_PartnerSuche | None = None
+    Projektrelevanz: Grunderfassung_Projektrelevanz
+    Kontaktperson: Grunderfassung_Kontaktperson
+    Stammdaten: Grunderfassung_Stammdaten
+    WeitereInfos: Grunderfassung_WeitereInfos
+    Schulbildung: list[Grunderfassung_Schulbildung]
+    HoehereBildung: list[Grunderfassung_HoehereBildung]
+    Arbeitserfahrung: list[Grunderfassung_Arbeitserfahrung]
+    Sprachkenntnisse: list[Grunderfassung_Sprachen]
+
+
+class Grunderfassung_PartnerSuche(BaseModel):
+    model_config = ConfigDict(str_strip_whitespace=True)
+
+    un_suche: int | None
+    person_suche: int | None
+    kanal_aufmerksamkeit: str | None
+
+
+class Grunderfassung_Projektrelevanz(BaseModel):
+    model_config = ConfigDict(str_strip_whitespace=True)
+
+    relevanz: str
+    foerderperiode: str | None = None
+
+
+class Grunderfassung_Kontaktperson(BaseModel):
+    model_config = ConfigDict(str_strip_whitespace=True)
+
+    KP_name_partner: str | None
+    KP_titel: str | None
+    KP_anrede_anschrift: str | None
+    KP_name: str | None
+    KP_vorname: str | None
+    KP_festnetznummer: str | None
+    KP_mobilfunknummer: str | None
+    KP_email: EmailStr | None
+    KP_funktion_beziehung: str | None
+    KP_adresse: str | None
+
+
+class Grunderfassung_Stammdaten(BaseModel):
+    model_config = ConfigDict(str_strip_whitespace=True)
+
+    titel: str | None
+    anrede_anschrift: str
+    name: str
+    vorname: str | None
+    geburtsdatum: datetime.date | None
+    herkunftsland: str
+    staatsangehoerigkeit: str | None
+    rueckkehrer: bool | None
+    aufenthaltsort: str | None
+    strasse: str | None
+    hausnummer: str | None
+    PLZ: str | None
+    ort: str | None
+    bundesland: str | None
+    land: str | None
+    festnetznummer: str | None
+    mobilfunknummer: str | None
+    email: EmailStr | None
+    familienstand: str | None
+    anzahl_kinder: Grunderfassung_Stammdaten_AnzahlKinder
+
+    @field_validator("rueckkehrer", mode="before")
+    @classmethod
+    def str_to_bool(cls, value: Any) -> Any:
+        if isinstance(value, str):
+            value = value.strip().lower()
+
+            if value == "ja":
+                return True
+            elif value == "nein":
+                return False
+
+            raise ValueError("Wert muss 'ja', 'nein', True oder False sein.")
+
+        return value
+
+
+class Grunderfassung_Stammdaten_AnzahlKinder(BaseModel):
+    model_config = ConfigDict(str_strip_whitespace=True)
+
+    anzahl: int | None
+    alter: list[ValidAge | None] | None = None
+
+
+class Grunderfassung_WeitereInfos(BaseModel):
+    model_config = ConfigDict(str_strip_whitespace=True)
+
+    WI_deutsch_sprache: str | None
+    WI_aufenthaltstitel: str | None
+    WI_gueltigkeit_aufenthaltstitel: datetime.date | None
+    WI_arbeitsstatus: str | None
+    WI_meldung_institution: str | None
+
+
+class Grunderfassung_Schulbildung(BaseModel):
+    model_config = ConfigDict(str_strip_whitespace=True)
+
+    SB_abschluss: str | None
+    SB_abschlussgrad: str | None
+    SB_schule: str | None
+    SB_ort: str | None
+    SB_land: str | None
+    SB_abschlussjahr: str | None
+    SB_bemerkungsfeld: str | None
+
+
+class Grunderfassung_HoehereBildung(BaseModel):
+    model_config = ConfigDict(str_strip_whitespace=True)
+
+    HB_anerkennung: str | None
+    HB_abschlussgrad: str | None
+    HB_abschlussgrad_dokument: str | None
+    HB_organisation: str | None
+    HB_beruf: str | None
+    HB_land: str | None
+    HB_ort: str | None
+    HB_abschlussjahr: str | None
+    HB_bemerkungsfeld: str | None
+
+
+class Grunderfassung_Arbeitserfahrung(BaseModel):
+    model_config = ConfigDict(str_strip_whitespace=True)
+
+    AE_branche: str | None
+    AE_bezeichnung: str | None
+    AE_funktion: str | None
+    AE_unternehmen: str | None
+    AE_land: str | None
+    AE_zeitspanne: str | None
+    AE_beschaeftigungsart: str | None
+    AE_bemerkungsfeld: str | None
+
+
+class Grunderfassung_Sprachen(BaseModel):
+    model_config = ConfigDict(str_strip_whitespace=True)
+
+    SP_sprache: str | None
+    SP_niveau: str | None
+    SP_nachweis: str | None
+    SP_art_nachweis: str | None = None
+    SP_datum_nachweis: datetime.date | None = None