""" sn_basis/modules/Pruefmanager.py """ from __future__ import annotations from typing import Optional, Any from sn_basis.functions import ask_yes_no, info, warning, error, ask_overwrite_append_cancel_custom from sn_basis.functions.ly_visibility_wrapper import set_layer_visible from sn_basis.modules.pruef_ergebnis import pruef_ergebnis, PruefAktion class Pruefmanager: def __init__(self, ui_modus: str = "qgis", parent: Optional[Any] = None) -> None: self.ui_modus = ui_modus self.parent = parent # ------------------------------------------------------------------ # Meldungen / Zusammenfassungen # ------------------------------------------------------------------ def report_error( self, thema: str, meldung: str, *, aktion: Optional[PruefAktion] = None, kontext: Optional[Any] = None, ) -> None: critical_actions = { "netzwerkfehler", "pruefe_exception", "save_exception", "layer_create_failed", "read_error", "open_error", } warn_actions = { "datei_nicht_gefunden", "pfad_nicht_gefunden", "url_nicht_erreichbar", "falsche_endung", "kein_header", "kein_arbeitsblatt", } if aktion in critical_actions: error(thema, meldung) return if aktion in warn_actions: warning(thema, meldung) return warning(thema, meldung) def report_summary(self, summary: dict) -> None: geladen = summary.get("geladen", []) fehler = summary.get("fehler", {}) ausserhalb = summary.get("ausserhalb", []) relevant = summary.get("relevant", []) message = ( f"Geladene Dienste: {len(geladen)}\n" f"Relevante Dienste: {len(relevant)}\n" f"Dienste ausserhalb: {len(ausserhalb)}\n" f"Fehler: {len(fehler)}" ) info("DataGrabber Zusammenfassung", message) # ------------------------------------------------------------------ # Allgemeine Nutzerinteraktionen # ------------------------------------------------------------------ def zeige_hinweis(self, titel: str, meldung: str) -> None: """Zeigt eine modale Hinweismeldung mit OK-Button.""" from sn_basis.functions.dialog_wrapper import show_info_dialog show_info_dialog(titel, meldung, parent=self.parent) def frage_ja_nein(self, titel: str, meldung: str, default: bool = True) -> bool: """Stellt eine Ja/Nein-Frage. Gibt True zurück, wenn der Nutzer Ja wählt.""" if self.ui_modus != "qgis": return default return ask_yes_no(titel, meldung, default=default, parent=self.parent) def frage_text(self, titel: str, meldung: str, default_text: str = "") -> tuple[str, bool]: """Fragt einen Textwert ab und gibt Text + OK-Status zurück.""" from sn_basis.functions.dialog_wrapper import ask_text if self.ui_modus != "qgis": return default_text, True return ask_text(titel, meldung, default_text=default_text, parent=self.parent) # ------------------------------------------------------------------ # VERFAHRENS-DB-spezifische Entscheidungen # ------------------------------------------------------------------ def _handle_datei_existiert(self, ergebnis: pruef_ergebnis) -> pruef_ergebnis: """Handhabt das Szenario, dass die Ziel-Verfahrens-DB bereits existiert. Zeigt einen einzigen Dialog mit drei Optionen an: - **Überschreiben**: Bestehende Layer ersetzen (entspricht YES) - **Anhängen**: Neue Layer zur Datei hinzufügen (entspricht NO) - **Abbrechen**: Vorgang beenden (entspricht CANCEL) Parameters ---------- ergebnis : pruef_ergebnis Eingabe-Ergebnis mit Dateipfad im ``kontext``-Attribut. Returns ------- pruef_ergebnis Ergebnis mit Aktion: - ``datei_existiert_ueberschreiben`` - ``datei_existiert_anhaengen`` - ``datei_existiert_ueberspringen`` (für Cancel-Fall) """ if self.ui_modus != "qgis": return ergebnis pfad = ergebnis.kontext pfad_str = str(pfad) if pfad else "unbekannt" titel = "Verfahrens-DB existiert bereits" meldung = ( f"Die Datei '{pfad_str}' existiert bereits.\n\n" "Was soll geschehen?\n\n" "• **Überschreiben**: Bestehende Layer ersetzen\n" "• **Anhängen**: Neue Layer hinzufügen\n" "• **Abbrechen**: Vorgang beenden" ) # Einzelner Dialog mit drei Optionen entscheidung = ask_overwrite_append_cancel_custom( parent=self.parent, title=titel, message=meldung ) if entscheidung == "overwrite": return pruef_ergebnis( ok=True, aktion="datei_existiert_ueberschreiben", kontext=ergebnis.kontext, ) elif entscheidung == "append": return pruef_ergebnis( ok=True, aktion="datei_existiert_anhaengen", kontext=ergebnis.kontext, ) else: # cancel return pruef_ergebnis( ok=True, aktion="datei_existiert_ueberspringen", kontext=ergebnis.kontext, ) # ------------------------------------------------------------------ # Basis-Entscheidungen (KORREKT: → pruef_ergebnis) # ------------------------------------------------------------------ def _handle_basic_decision(self, ergebnis: pruef_ergebnis) -> pruef_ergebnis: """Basis-Entscheidung für einfache Ja/Nein-Fragen.""" if self.ui_modus != "qgis": return ergebnis title_map = { "leereingabe_erlaubt": "Ohne Eingabe fortfahren", "standarddatei_vorschlagen": "Standarddatei verwenden", "temporaer_erlaubt": "Temporäre Layer erzeugen", "layer_unsichtbar": "Layer einblenden", } title = title_map.get(ergebnis.aktion, "Entscheidung erforderlich") meldung = ergebnis.meldung or "" try: yes = ask_yes_no(title, meldung, default=False, parent=self.parent) except Exception: return ergebnis if not yes: return ergebnis # Nutzer sagte Ja if ergebnis.aktion == "temporaer_erlaubt": return pruef_ergebnis( ok=True, aktion="temporaer_erzeugen", kontext=ergebnis.kontext ) if ergebnis.aktion == "layer_unsichtbar": try: set_layer_visible(ergebnis.kontext, True) except Exception: return ergebnis return pruef_ergebnis( ok=True, aktion="ok", kontext=ergebnis.kontext, ) return pruef_ergebnis( ok=True, aktion="ok", kontext=ergebnis.kontext ) # ------------------------------------------------------------------ # Hauptlogik: verarbeite() (KORRIGIERT!) # ------------------------------------------------------------------ def verarbeite(self, ergebnis: pruef_ergebnis) -> pruef_ergebnis: # 1. Erfolg → direkt weiter if ergebnis.ok: return ergebnis # 2. VERFAHRENS-DB: Bestehende Datei if ergebnis.aktion == "datei_existiert": return self._handle_datei_existiert(ergebnis) # 3. Basis interaktive Aktionen interactive_actions = { "leereingabe_erlaubt", "standarddatei_vorschlagen", "temporaer_erlaubt", "layer_unsichtbar", } if ergebnis.aktion in interactive_actions: return self._handle_basic_decision(ergebnis) # 4. Fehler behandeln self.report_error( thema=ergebnis.aktion or "pruefung", meldung=ergebnis.meldung or "", aktion=ergebnis.aktion, kontext=ergebnis.kontext, ) return ergebnis def _ask_use_or_replace_pufferlayer(self) -> str: """ Fragt den Nutzer, ob ein vorhandener Pufferlayer verwendet oder ersetzt werden soll. Returns ------- str "verwenden", "ersetzen" oder "abbrechen" """ ergebnis = pruef_ergebnis( ok=False, aktion="layer_existiert", meldung="Ein Pufferlayer ist bereits vorhanden.", ) ergebnis = self.pruefmanager.verarbeite(ergebnis) if not ergebnis.ok: return "abbrechen" return "verwenden" if ergebnis.aktion == "ok" else "ersetzen"