From 3b56725e4f4f718703c9b0b41d286509bd481b58 Mon Sep 17 00:00:00 2001 From: daniel Date: Wed, 4 Mar 2026 15:32:49 +0100 Subject: [PATCH] =?UTF-8?q?qt=5Fwrapper,=20dialog;wrapper,=20Pruef=5Fergeb?= =?UTF-8?q?nis=20und=20Pruefmanager=20=C3=BCberarbeitet,=20so=20dass=20die?= =?UTF-8?q?=20=C3=9Cbergaben=20jetzt=20stimmen.=20Nutzerabfragen=20werden?= =?UTF-8?q?=20tats=C3=A4chlich=20ausgel=C3=B6st-=20Nutzerabfrage=20Datei?= =?UTF-8?q?=20=C3=BCberschreiebn...=20ist=20noch=20Bl=C3=B6dsinn?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- __pdoc__.py | 3 + functions/dialog_wrapper.py | 65 ++--- functions/qt_wrapper.py | 543 +++++++++++++----------------------- modules/DataGrabber.py | 486 ++++++++------------------------ modules/Dateipruefer.py | 180 ++++++++---- modules/Pruefmanager.py | 398 +++++++++++--------------- modules/pruef_ergebnis.py | 134 ++++++++- 7 files changed, 756 insertions(+), 1053 deletions(-) create mode 100644 __pdoc__.py diff --git a/__pdoc__.py b/__pdoc__.py new file mode 100644 index 0000000..bfecc06 --- /dev/null +++ b/__pdoc__.py @@ -0,0 +1,3 @@ +__pdoc__ = { + "main": False, +} diff --git a/functions/dialog_wrapper.py b/functions/dialog_wrapper.py index 3ed9c41..f91c4bf 100644 --- a/functions/dialog_wrapper.py +++ b/functions/dialog_wrapper.py @@ -1,62 +1,37 @@ """ -sn_basis/functions/dialog_wrapper.py – Benutzer-Dialoge - -Dieser Wrapper kapselt alle Benutzer-Dialoge (z. B. Ja/Nein-Abfragen) -und sorgt dafür, dass sie sowohl in QGIS als auch im Mock-/Testmodus -einheitlich funktionieren. +sn_basis/functions/dialog_wrapper.py – Benutzer-Dialoge (Qt5/6/Mock-kompatibel) """ - from typing import Any - -# Import der abstrahierten Qt-Klassen aus dem qt_wrapper. -# QMessageBox, YES und NO sind bereits kompatibel zu Qt5/Qt6 -# und im Mock-Modus durch Dummy-Objekte ersetzt. from sn_basis.functions.qt_wrapper import ( - QMessageBox, - YES, - NO, + QMessageBox, YES, NO, QT_VERSION ) - -# --------------------------------------------------------- -# Öffentliche API -# --------------------------------------------------------- - def ask_yes_no( title: str, message: str, - default: bool = False, + default: bool = True, parent: Any = None, ) -> bool: """ - Stellt dem Benutzer eine Ja/Nein-Frage. - - - In einer echten QGIS-Umgebung wird ein QMessageBox-Dialog angezeigt. - - Im Mock-/Testmodus wird kein Dialog geöffnet, sondern der Default-Wert - zurückgegeben, damit Tests ohne UI laufen können. - - :param title: Titel des Dialogs - :param message: Nachrichtentext - :param default: Rückgabewert im Fehler- oder Mock-Fall - :param parent: Optionales Parent-Widget - :return: True bei "Ja", False bei "Nein" + Stellt Ja/Nein-Frage. Funktioniert in PyQt5/6 UND Mock-Modus. """ try: - # Definiert die beiden Buttons, die angezeigt werden sollen. - buttons = QMessageBox.Yes | QMessageBox.No - - # Öffnet den Dialog (oder im Mock-Modus: simuliert ihn). + if QT_VERSION == 0: # Mock-Modus + print(f"🔍 Mock-Modus: ask_yes_no('{title}') → {default}") + return default + + # ✅ KORREKT: Verwende YES/NO-Aliase aus qt_wrapper! + buttons = YES | NO + default_button = YES if default else NO + result = QMessageBox.question( - parent, - title, - message, - buttons, - YES if default else NO, # Vorauswahl abhängig vom Default + parent, title, message, buttons, default_button ) - - # Gibt True zurück, wenn der Benutzer "Ja" gewählt hat. - return result == YES - - except Exception: - # Falls Qt nicht verfügbar ist (Mock/CI), wird der Default-Wert genutzt. + + # ✅ int(result) == int(YES) funktioniert Qt5/6/Mock + print(f"DEBUG ask_yes_no: result={result}, YES={YES}, match={int(result) == int(YES)}") + return int(result) == int(YES) + + except Exception as e: + print(f"⚠️ ask_yes_no Fehler: {e}") return default diff --git a/functions/qt_wrapper.py b/functions/qt_wrapper.py index 706c6b0..09dfa40 100644 --- a/functions/qt_wrapper.py +++ b/functions/qt_wrapper.py @@ -1,90 +1,95 @@ """ -sn_basis/functions/qt_wrapper.py – zentrale Qt-Abstraktion (PyQt5 / PyQt6 / Mock) +sn_basis/functions/qt_wrapper.py – zentrale Qt-Abstraktion (PyQt6 primär / PyQt5 Fallback / Mock) """ -from typing import Optional, Type, Any - -# --------------------------------------------------------- -# Qt-Symbole (werden dynamisch gesetzt) -# --------------------------------------------------------- - -QDockWidget: Type[Any] -QMessageBox: Type[Any] -QFileDialog: Type[Any] -QEventLoop: Type[Any] -QUrl: Type[Any] -QNetworkRequest: Type[Any] -QNetworkReply: Type[Any] -QCoreApplication: Type[Any] - -QWidget: Type[Any] -QGridLayout: Type[Any] -QLabel: Type[Any] -QLineEdit: Type[Any] -QGroupBox: Type[Any] -QVBoxLayout: Type[Any] -QPushButton: Type[Any] -QAction: Type[Any] -QMenu: Type[Any] -QToolBar: Type[Any] -QActionGroup: Type[Any] -QTabWidget: type -QToolButton: Type[Any] -QSizePolicy: Type[Any] -Qt: Type[Any] -QComboBox: Type[Any] +from typing import Optional, Type, Any, Callable +# Globale Qt-Symbole (werden dynamisch gesetzt) +QT_VERSION = 0 # 0 = Mock, 5 = PyQt5, 6 = PyQt6 YES: Optional[Any] = None NO: Optional[Any] = None CANCEL: Optional[Any] = None ICON_QUESTION: Optional[Any] = None -QT_VERSION = 0 # 0 = Mock, 5 = PyQt5, 6 = PyQt6 - +# Qt-Klassen (werden dynamisch gesetzt) +QDockWidget: Type[Any] = object +QMessageBox: Type[Any] = object +QFileDialog: Type[Any] = object +QEventLoop: Type[Any] = object +QUrl: Type[Any] = object +QNetworkRequest: Type[Any] = object +QNetworkReply: Type[Any] = object +QCoreApplication: Type[Any] = object +QWidget: Type[Any] = object +QGridLayout: Type[Any] = object +QLabel: Type[Any] = object +QLineEdit: Type[Any] = object +QGroupBox: Type[Any] = object +QVBoxLayout: Type[Any] = object +QPushButton: Type[Any] = object +QAction: Type[Any] = object +QMenu: Type[Any] = object +QToolBar: Type[Any] = object +QActionGroup: Type[Any] = object +QTabWidget: Type[Any] = object +QToolButton: Type[Any] = object +QSizePolicy: Type[Any] = object +Qt: Type[Any] = object +QComboBox: Type[Any] = object def exec_dialog(dialog: Any) -> Any: - raise NotImplementedError + """Führt Dialog modal aus (Qt6: exec(), Qt5: exec_(), Mock: YES)""" + raise NotImplementedError("Qt nicht initialisiert") +def debug_qt_status() -> None: + """Debug: Zeigt Qt-Status für Troubleshooting.""" + print(f"🔍 QT_VERSION: {QT_VERSION}") + print(f"🔍 QMessageBox Typ: {getattr(QMessageBox, '__name__', type(QMessageBox).__name__)}") + print(f"🔍 YES Wert: {YES} (Typ: {type(YES) if YES is not None else 'None'})") + + if QT_VERSION == 0: + print("❌ MOCK-MODUS AKTIV! Keine Dialoge möglich!") + elif QT_VERSION == 5: + print("✅ PyQt5 geladen (Fallback) – Dialoge sollten funktionieren!") + elif QT_VERSION == 6: + print("✅ PyQt6 geladen (primär) – Dialoge sollten funktionieren!") + else: + print("❓ Unbekannte Qt-Version!") -# --------------------------------------------------------- -# Versuch: PyQt6 -# --------------------------------------------------------- - +# --------------------------- PYQT6 PRIMÄR --------------------------- try: - from qgis.PyQt.QtWidgets import ( # type: ignore - QMessageBox as _QMessageBox,# type: ignore - QFileDialog as _QFileDialog,# type: ignore - QWidget as _QWidget,# type: ignore - QGridLayout as _QGridLayout,# type: ignore - QLabel as _QLabel,# type: ignore - QLineEdit as _QLineEdit,# type: ignore - QGroupBox as _QGroupBox,# type: ignore - QVBoxLayout as _QVBoxLayout,# type: ignore - QPushButton as _QPushButton,# type: ignore - QAction as _QAction, - QMenu as _QMenu,# type: ignore - QToolBar as _QToolBar,# type: ignore - QActionGroup as _QActionGroup,# type: ignore - QDockWidget as _QDockWidget,# type: ignore - QTabWidget as _QTabWidget,# type: ignore - QToolButton as _QToolButton,#type:ignore - QSizePolicy as _QSizePolicy,#type:ignore - QComboBox as _QComboBox, - -) - - - - from qgis.PyQt.QtCore import ( # type: ignore - QEventLoop as _QEventLoop,# type: ignore - QUrl as _QUrl,# type: ignore - QCoreApplication as _QCoreApplication,# type: ignore - Qt as _Qt#type:ignore + from qgis.PyQt.QtWidgets import ( + QMessageBox as _QMessageBox, + QFileDialog as _QFileDialog, + QWidget as _QWidget, + QGridLayout as _QGridLayout, + QLabel as _QLabel, + QLineEdit as _QLineEdit, + QGroupBox as _QGroupBox, + QVBoxLayout as _QVBoxLayout, + QPushButton as _QPushButton, + QAction as _QAction, + QMenu as _QMenu, + QToolBar as _QToolBar, + QActionGroup as _QActionGroup, + QDockWidget as _QDockWidget, + QTabWidget as _QTabWidget, + QToolButton as _QToolButton, + QSizePolicy as _QSizePolicy, + QComboBox as _QComboBox, ) - from qgis.PyQt.QtNetwork import ( # type: ignore - QNetworkRequest as _QNetworkRequest,# type: ignore - QNetworkReply as _QNetworkReply,# type: ignore + from qgis.PyQt.QtCore import ( + QEventLoop as _QEventLoop, + QUrl as _QUrl, + QCoreApplication as _QCoreApplication, + Qt as _Qt, ) + from qgis.PyQt.QtNetwork import ( + QNetworkRequest as _QNetworkRequest, + QNetworkReply as _QNetworkReply, + ) + + # ✅ ALLE GLOBALS ZUWEISEN QT_VERSION = 6 QMessageBox = _QMessageBox QFileDialog = _QFileDialog @@ -93,7 +98,7 @@ try: QNetworkRequest = _QNetworkRequest QNetworkReply = _QNetworkReply QCoreApplication = _QCoreApplication - Qt=_Qt + Qt = _Qt QDockWidget = _QDockWidget QWidget = _QWidget QGridLayout = _QGridLayout @@ -107,51 +112,37 @@ try: QToolBar = _QToolBar QActionGroup = _QActionGroup QTabWidget = _QTabWidget - QToolButton=_QToolButton - QSizePolicy=_QSizePolicy - QComboBox=_QComboBox - + QToolButton = _QToolButton + QSizePolicy = _QSizePolicy + QComboBox = _QComboBox + + # ✅ QT6 ENUMS YES = QMessageBox.StandardButton.Yes NO = QMessageBox.StandardButton.No CANCEL = QMessageBox.StandardButton.Cancel ICON_QUESTION = QMessageBox.Icon.Question - # --------------------------------------------------------- - # Qt6 Enum-Aliase (vereinheitlicht) - # --------------------------------------------------------- - + + # Qt6 Enum-Aliase ToolButtonTextBesideIcon = Qt.ToolButtonStyle.ToolButtonTextBesideIcon ArrowDown = Qt.ArrowType.DownArrow ArrowRight = Qt.ArrowType.RightArrow - # QSizePolicy Enum-Aliase (Qt6) SizePolicyPreferred = QSizePolicy.Policy.Preferred SizePolicyMaximum = QSizePolicy.Policy.Maximum - # --------------------------------------------------------- - # QDockWidget Feature-Aliase (Qt6) - # --------------------------------------------------------- - DockWidgetMovable = QDockWidget.DockWidgetFeature.DockWidgetMovable DockWidgetFloatable = QDockWidget.DockWidgetFeature.DockWidgetFloatable DockWidgetClosable = QDockWidget.DockWidgetFeature.DockWidgetClosable - # --------------------------------------------------------- - # Dock-Area-Aliase (Qt6) - # --------------------------------------------------------- - DockAreaLeft = Qt.DockWidgetArea.LeftDockWidgetArea DockAreaRight = Qt.DockWidgetArea.RightDockWidgetArea - - - def exec_dialog(dialog: Any) -> Any: return dialog.exec() + + print(f"✅ qt_wrapper: PyQt6 geladen (QT_VERSION={QT_VERSION})") -# --------------------------------------------------------- -# Versuch: PyQt5 -# --------------------------------------------------------- - -except Exception: +# --------------------------- PYQT5 FALLBACK --------------------------- +except (ImportError, AttributeError): try: - from PyQt5.QtWidgets import (# type: ignore + from PyQt5.QtWidgets import ( QMessageBox as _QMessageBox, QFileDialog as _QFileDialog, QWidget as _QWidget, @@ -171,18 +162,19 @@ except Exception: QSizePolicy as _QSizePolicy, QComboBox as _QComboBox, ) - from PyQt5.QtCore import (# type: ignore + from PyQt5.QtCore import ( QEventLoop as _QEventLoop, QUrl as _QUrl, QCoreApplication as _QCoreApplication, Qt as _Qt, ) - from PyQt5.QtNetwork import (# type: ignore + from PyQt5.QtNetwork import ( QNetworkRequest as _QNetworkRequest, QNetworkReply as _QNetworkReply, ) - - + + # ✅ ALLE GLOBALS ZUWEISEN + QT_VERSION = 5 QMessageBox = _QMessageBox QFileDialog = _QFileDialog QEventLoop = _QEventLoop @@ -190,10 +182,8 @@ except Exception: QNetworkRequest = _QNetworkRequest QNetworkReply = _QNetworkReply QCoreApplication = _QCoreApplication - Qt=_Qt + Qt = _Qt QDockWidget = _QDockWidget - - QWidget = _QWidget QGridLayout = _QGridLayout QLabel = _QLabel @@ -206,55 +196,41 @@ except Exception: QToolBar = _QToolBar QActionGroup = _QActionGroup QTabWidget = _QTabWidget - QToolButton=_QToolButton - QSizePolicy=_QSizePolicy - ComboBox=_QComboBox - + QToolButton = _QToolButton + QSizePolicy = _QSizePolicy + QComboBox = _QComboBox + + # ✅ PYQT5 ENUMS YES = QMessageBox.Yes NO = QMessageBox.No CANCEL = QMessageBox.Cancel ICON_QUESTION = QMessageBox.Question - - QT_VERSION = 5 - - # then try next backend - # --------------------------------------------------------- - # Qt5 Enum-Aliase (vereinheitlicht) - # --------------------------------------------------------- - + + # PyQt5 Enum-Aliase ToolButtonTextBesideIcon = Qt.ToolButtonTextBesideIcon ArrowDown = Qt.DownArrow ArrowRight = Qt.RightArrow - # QSizePolicy Enum-Aliase (Qt5) SizePolicyPreferred = QSizePolicy.Preferred SizePolicyMaximum = QSizePolicy.Maximum - # --------------------------------------------------------- - # QDockWidget Feature-Aliase (Qt5) - # --------------------------------------------------------- - DockWidgetMovable = QDockWidget.DockWidgetMovable DockWidgetFloatable = QDockWidget.DockWidgetFloatable DockWidgetClosable = QDockWidget.DockWidgetClosable - # --------------------------------------------------------- - # Dock-Area-Aliase (Qt5) - # --------------------------------------------------------- - DockAreaLeft = Qt.LeftDockWidgetArea DockAreaRight = Qt.RightDockWidgetArea - - + def exec_dialog(dialog: Any) -> Any: return dialog.exec_() + + print(f"✅ qt_wrapper: PyQt5 Fallback geladen (QT_VERSION={QT_VERSION})") -# --------------------------------------------------------- -# Mock-Modus -# --------------------------------------------------------- - +# --------------------------- MOCK-MODUS --------------------------- except Exception: QT_VERSION = 0 + print("⚠️ qt_wrapper: Mock-Modus aktiviert (QT_VERSION=0)") + # Fake Enum für Bit-Operationen class FakeEnum(int): - def __or__(self, other: int) -> "FakeEnum": + def __or__(self, other: Any) -> "FakeEnum": return FakeEnum(int(self) | int(other)) YES = FakeEnum(1) @@ -262,103 +238,72 @@ except Exception: CANCEL = FakeEnum(4) ICON_QUESTION = FakeEnum(8) + # Im Mock-Block von qt_wrapper.py: class _MockQMessageBox: Yes = YES No = NO Cancel = CANCEL Question = ICON_QUESTION + + @classmethod + def question(cls, parent, title, message, buttons, default_button): + """Mock: Gibt immer default_button zurück""" + print(f"🔍 Mock QMessageBox.question: '{title}' → {default_button}") + return default_button QMessageBox = _MockQMessageBox + class _MockQFileDialog: @staticmethod - def getOpenFileName(*args, **kwargs): - return ("", "") - + def getOpenFileName(*args, **kwargs): return ("", "") @staticmethod - def getSaveFileName(*args, **kwargs): - return ("", "") + def getSaveFileName(*args, **kwargs): return ("", "") QFileDialog = _MockQFileDialog class _MockQEventLoop: - def exec(self) -> int: - return 0 - - def quit(self) -> None: - pass + def exec(self) -> int: return 0 + def quit(self) -> None: pass QEventLoop = _MockQEventLoop class _MockQUrl(str): - def isValid(self) -> bool: - return True + def isValid(self) -> bool: return True QUrl = _MockQUrl class _MockQNetworkRequest: - def __init__(self, url: Any): - self.url = url + def __init__(self, url: Any): self.url = url QNetworkRequest = _MockQNetworkRequest class _MockQNetworkReply: - def error(self) -> int: - return 0 - - def errorString(self) -> str: - return "" - - def readAll(self) -> bytes: - return b"" - - def deleteLater(self) -> None: - pass + def error(self) -> int: return 0 + def errorString(self) -> str: return "" + def readAll(self) -> bytes: return b"" + def deleteLater(self) -> None: pass QNetworkReply = _MockQNetworkReply - class _MockWidget: - def __init__(self, *args, **kwargs): - pass - + class _MockWidget: pass class _MockLayout: - def __init__(self, *args, **kwargs): - self._widgets = [] - - def addWidget(self, widget): - self._widgets.append(widget) - - def addLayout(self, layout): - pass - - def addStretch(self, *args, **kwargs): - pass - - def setSpacing(self, *args, **kwargs): - pass - - def setContentsMargins(self, *args, **kwargs): - pass - - - - class _MockLabel: - def __init__(self, text: str = ""): - self._text = text + def __init__(self, *args, **kwargs): self._widgets = [] + def addWidget(self, widget): self._widgets.append(widget) + def addLayout(self, layout): pass + def addStretch(self, *args, **kwargs): pass + def setSpacing(self, *args, **kwargs): pass + def setContentsMargins(self, *args, **kwargs): pass + class _MockLabel: + def __init__(self, text: str = ""): self._text = text class _MockLineEdit: - def __init__(self, *args, **kwargs): - self._text = "" + def __init__(self, *args, **kwargs): self._text = "" + def text(self) -> str: return self._text + def setText(self, value: str) -> None: self._text = value - def text(self) -> str: - return self._text - - def setText(self, value: str) -> None: - self._text = value - - class _MockButton: - def __init__(self, *args, **kwargs): - self.clicked = lambda *a, **k: None + class _MockButton: + def __init__(self, *args, **kwargs): self.clicked = lambda *a, **k: None QWidget = _MockWidget QGridLayout = _MockLayout @@ -367,101 +312,61 @@ except Exception: QGroupBox = _MockWidget QVBoxLayout = _MockLayout QPushButton = _MockButton - - class _MockQCoreApplication: - pass - - QCoreApplication = _MockQCoreApplication + QCoreApplication = object() + class _MockQt: - # ToolButtonStyle ToolButtonTextBesideIcon = 0 - - # ArrowType ArrowDown = 1 ArrowRight = 2 + LeftDockWidgetArea = 1 + RightDockWidgetArea = 2 - Qt=_MockQt + Qt = _MockQt() ToolButtonTextBesideIcon = Qt.ToolButtonTextBesideIcon ArrowDown = Qt.ArrowDown ArrowRight = Qt.ArrowRight + DockAreaLeft = Qt.LeftDockWidgetArea + DockAreaRight = Qt.RightDockWidgetArea class _MockQDockWidget(_MockWidget): def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) self._object_name = "" + def setObjectName(self, name: str) -> None: self._object_name = name + def objectName(self) -> str: return self._object_name + def show(self) -> None: pass + def deleteLater(self) -> None: pass - def setObjectName(self, name: str) -> None: - self._object_name = name - - def objectName(self) -> str: - return self._object_name - - def show(self) -> None: - pass - - def deleteLater(self) -> None: - pass QDockWidget = _MockQDockWidget + class _MockAction: def __init__(self, *args, **kwargs): self._checked = False self.triggered = lambda *a, **k: None - - def setToolTip(self, text: str) -> None: - pass - - def setCheckable(self, value: bool) -> None: - pass - - def setChecked(self, value: bool) -> None: - self._checked = value - + def setToolTip(self, text: str) -> None: pass + def setCheckable(self, value: bool) -> None: pass + def setChecked(self, value: bool) -> None: self._checked = value class _MockMenu: - def __init__(self, *args, **kwargs): - self._actions = [] - - def addAction(self, action): - self._actions.append(action) - - def removeAction(self, action): - if action in self._actions: - self._actions.remove(action) - - def clear(self): - self._actions.clear() - - def menuAction(self): - return self - + def __init__(self, *args, **kwargs): self._actions = [] + def addAction(self, action): self._actions.append(action) + def removeAction(self, action): + if action in self._actions: self._actions.remove(action) + def clear(self): self._actions.clear() + def menuAction(self): return self class _MockToolBar: - def __init__(self, *args, **kwargs): - self._actions = [] - - def setObjectName(self, name: str) -> None: - pass - - def addAction(self, action): - self._actions.append(action) - - def removeAction(self, action): - if action in self._actions: - self._actions.remove(action) - - def clear(self): - self._actions.clear() - + def __init__(self, *args, **kwargs): self._actions = [] + def setObjectName(self, name: str) -> None: pass + def addAction(self, action): self._actions.append(action) + def removeAction(self, action): + if action in self._actions: self._actions.remove(action) + def clear(self): self._actions.clear() class _MockActionGroup: - def __init__(self, *args, **kwargs): - self._actions = [] + def __init__(self, *args, **kwargs): self._actions = [] + def setExclusive(self, value: bool) -> None: pass + def addAction(self, action): self._actions.append(action) - def setExclusive(self, value: bool) -> None: - pass - - def addAction(self, action): - self._actions.append(action) QAction = _MockAction QMenu = _MockMenu QToolBar = _MockToolBar @@ -469,112 +374,58 @@ except Exception: class _MockToolButton(_MockWidget): def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) self._checked = False self.toggled = lambda *a, **k: None + def setText(self, text: str) -> None: pass + def setCheckable(self, value: bool) -> None: pass + def setChecked(self, value: bool) -> None: self._checked = value + def setToolButtonStyle(self, *args, **kwargs): pass + def setArrowType(self, *args, **kwargs): pass + def setStyleSheet(self, *args, **kwargs): pass - def setText(self, text: str) -> None: - pass - - def setCheckable(self, value: bool) -> None: - pass - - def setChecked(self, value: bool) -> None: - self._checked = value - - def setToolButtonStyle(self, *args, **kwargs): - pass - - def setArrowType(self, *args, **kwargs): - pass - - def setStyleSheet(self, *args, **kwargs): - pass - - QToolButton=_MockToolButton + QToolButton = _MockToolButton class _MockQSizePolicy: - # horizontale Policies - Fixed = 0 - Minimum = 1 - Maximum = 2 Preferred = 3 - Expanding = 4 - MinimumExpanding = 5 - Ignored = 6 + Maximum = 2 - # vertikale Policies (Qt nutzt dieselben Werte) - def __init__(self, horizontal=None, vertical=None): - self.horizontal = horizontal - self.vertical = vertical - QSizePolicy=_MockQSizePolicy + QSizePolicy = _MockQSizePolicy SizePolicyPreferred = QSizePolicy.Preferred SizePolicyMaximum = QSizePolicy.Maximum DockWidgetMovable = 1 DockWidgetFloatable = 2 DockWidgetClosable = 4 - DockAreaLeft = 1 - DockAreaRight = 2 - def exec_dialog(dialog: Any) -> Any: - return YES class _MockTabWidget: - def __init__(self, *args, **kwargs): - self._tabs = [] + def __init__(self, *args, **kwargs): self._tabs = [] + def addTab(self, widget, title: str): self._tabs.append((widget, title)) - def addTab(self, widget, title: str): - self._tabs.append((widget, title)) QTabWidget = _MockTabWidget - # ------------------------- - # Mock ComboBox Implementation - # ------------------------- - class _MockSignal: - def __init__(self): - self._slots = [] - - def connect(self, cb): - self._slots.append(cb) - - def emit(self, value): - for s in list(self._slots): - try: - s(value) - except Exception: - pass - class _MockComboBox: def __init__(self, parent=None): self._items = [] self._index = -1 - self.currentTextChanged = _MockSignal() - - def addItem(self, text: str) -> None: - self._items.append(text) - - def addItems(self, items): - for it in items: - self.addItem(it) - - def findText(self, text: str) -> int: - try: - return self._items.index(text) - except ValueError: - return -1 - + self.currentTextChanged = type('Signal', (), {'connect': lambda s, cb: None, 'emit': lambda s, v: None})() + def addItem(self, text: str) -> None: self._items.append(text) + def addItems(self, items): [self.addItem(it) for it in items] + def findText(self, text: str) -> int: + return self._items.index(text) if text in self._items else -1 def setCurrentIndex(self, idx: int) -> None: if 0 <= idx < len(self._items): self._index = idx self.currentTextChanged.emit(self.currentText()) - def setCurrentText(self, text: str) -> None: idx = self.findText(text) - if idx >= 0: - self.setCurrentIndex(idx) - + if idx >= 0: self.setCurrentIndex(idx) def currentText(self) -> str: - if 0 <= self._index < len(self._items): - return self._items[self._index] - return "" + return self._items[self._index] if 0 <= self._index < len(self._items) else "" - ComboBox = _MockComboBox + QComboBox = _MockComboBox + + def exec_dialog(dialog: Any) -> Any: + return YES + +# --------------------------- TEST --------------------------- +if __name__ == "__main__": + debug_qt_status() diff --git a/modules/DataGrabber.py b/modules/DataGrabber.py index b077dee..d78ce61 100644 --- a/modules/DataGrabber.py +++ b/modules/DataGrabber.py @@ -1,40 +1,27 @@ -# sn_basis/modules/DataGrabber.py """ DataGrabber module ================== -Leichter Orchestrator, der eine Quelle (Datei, Einzellink, Datenbank) -analysiert, passende Prüfer aufruft und die Ergebnisse an den -:class:`sn_basis.modules.Pruefmanager.Pruefmanager` delegiert. +UI‑freier Orchestrator für die Prüfung und Klassifikation von Datenquellen. -Dieses vereinfachte Modul geht davon aus, dass alle benötigten Prüfer -und der ExcelImporter vorhanden und importierbar sind. Es enthält -keine Fallbacks oder defensive Exception-Handling-Pfade für fehlende -Prüfer-Module — fehlende Komponenten führen zu Import- oder Laufzeitfehlern, -die bewusst nicht unterdrückt werden. +Der DataGrabber: +- klassifiziert die übergebene Quelle (Datei, Dienst, Datenbank, Excel), +- ruft passende Prüfer (Dateipruefer, Linkpruefer, Layerpruefer, Stilpruefer) auf, +- sammelt alle rohen ``pruef_ergebnis``‑Objekte, +- aggregiert diese zu einem zusammenfassenden Ergebnis, +- **löst selbst keinerlei UI‑Interaktion aus**. + +Alle Nutzerinteraktionen (MessageBar, QMessageBox, Logging) erfolgen +ausschließlich über den ``Pruefmanager`` im aufrufenden Kontext (UI / Pipeline). """ from __future__ import annotations -from typing import ( - Optional, - Any, - Mapping, - Iterable, - Dict, - Protocol, - Literal, - Tuple, - List, -) -from pathlib import Path -import sqlite3 +from typing import Any, Dict, List, Mapping, Optional, Tuple, Literal +from sn_basis.modules.pruef_ergebnis import pruef_ergebnis from sn_basis.modules.Pruefmanager import Pruefmanager -from sn_basis.modules.pruef_ergebnis import pruef_ergebnis, PruefAktion -# In dieser vereinfachten Variante werden die Prüfer und der ExcelImporter -# direkt importiert. Fehlende Module führen zu ImportError (gewollt). from sn_basis.modules.Dateipruefer import Dateipruefer from sn_basis.modules.linkpruefer import Linkpruefer from sn_basis.modules.layerpruefer import Layerpruefer @@ -42,383 +29,144 @@ from sn_basis.modules.stilpruefer import Stilpruefer from sn_basis.modules.excel_importer import ExcelImporter -SourceType = Literal["file", "link", "database", "unknown"] +SourceType = Literal["service", "database", "excel", "unknown"] - -class LinklistAdapter(Protocol): - """ - Minimal-Protokoll für Adapter, die Linklisten liefern/normalisieren. - - Implementierende Klassen sollten: - - load() -> Iterable[Mapping[str, Any]] - - normalize(raw_item) -> Mapping[str, Any] - """ - def load(self) -> Iterable[Mapping[str, Any]]: - ... - def normalize(self, raw_item: Mapping[str, Any]) -> Mapping[str, Any]: - ... +SourceDict = Dict[str, List[Mapping[str, Any]]] class DataGrabber: """ - DataGrabber orchestriert das Einlesen einer Quelle und die Übergabe an Prüfer. + Analysiert und prüft Datenquellen für den Fachdatenabruf. - Diese vereinfachte Implementierung erwartet, dass alle Prüferklassen und - der ExcelImporter vorhanden sind. Es gibt keine defensive Logik für - fehlende Komponenten. - - Konstruktor-Parameter - -------------------- - :param pruefmanager: Instanz des Pruefmanagers (verpflichtend). - :param datei_pruefer_cls: Klasse des Dateipruefers (Standard: Dateipruefer). - :param link_pruefer: Instanz des Linkpruefers. - :param layer_pruefer: Instanz des Layerpruefers. - :param stil_pruefer: Instanz des Stilpruefers. + Der DataGrabber ist **UI‑frei**. Er erzeugt ausschließlich rohe + ``pruef_ergebnis``‑Objekte und überlässt deren Verarbeitung + vollständig dem aufrufenden Code. """ def __init__( self, pruefmanager: Pruefmanager, *, - datei_pruefer_cls=Dateipruefer, - link_pruefer: Linkpruefer, - layer_pruefer: Layerpruefer, - stil_pruefer: Stilpruefer, + datei_pruefer_cls: type[Dateipruefer] = Dateipruefer, + link_pruefer: Optional[Linkpruefer] = None, + layer_pruefer: Optional[Layerpruefer] = None, + stil_pruefer: Optional[Stilpruefer] = None, + excel_importer_cls: type[ExcelImporter] = ExcelImporter, ) -> None: - # Pruefmanager ist verpflichtend - self.pruefmanager: Pruefmanager = pruefmanager - - # Dateipruefer-Klasse (wird zur Laufzeit mit einem Pfad instanziert) + self.pruefmanager = pruefmanager self._datei_pruefer_cls = datei_pruefer_cls + self.link_pruefer = link_pruefer + self.layer_pruefer = layer_pruefer + self.stil_pruefer = stil_pruefer + self._excel_importer_cls = excel_importer_cls - # Prüfer-Instanzen (werden direkt verwendet) - self.link_pruefer: Linkpruefer = link_pruefer - self.layer_pruefer: Layerpruefer = layer_pruefer - self.stil_pruefer: Stilpruefer = stil_pruefer + self._source: Optional[str] = None - # Quelle (wird später gesetzt) - self.source: Optional[str] = None - - # ------------------------------------------------------------------ # - # Source Management - # ------------------------------------------------------------------ # + # ------------------------------------------------------------------ + # Öffentliche API + # ------------------------------------------------------------------ def set_source(self, source: str) -> None: + """Setzt die aktuell zu untersuchende Rohquelle.""" + self._source = source + + def analyze_source_type(self, source: str) -> SourceType: """ - Setzt die Quelle für den DataGrabber. + Klassifiziert die Quelle. - Die Quelle ist ein String, der entweder ein lokaler Dateipfad, - ein Einzellink (URL/URI) oder ein Pfad zu einer Datenbank/GeoPackage ist. + Aktuell Platzhalter – liefert ``"unknown"``. """ - self.source = source - - def analyze_source(self, source: str) -> SourceType: - """ - Klassifiziert die angegebene Quelle ausschließlich anhand des Dateipruefers. - - Ablauf - ------ - 1. Instanziere den Dateipruefer mit `pfad=source` und `temporaer_erlaubt=False`. - 2. Rufe `pruefe()` auf und werte das zurückgegebene :class:`pruef_ergebnis` aus. - 3. Bei `ok==True` wird anhand der Dateiendung zwischen "database" (gpkg/sqlite/db) - und "file" unterschieden. - 4. Bei `ok==False` werden typische Aktionen wie "datei_nicht_gefunden" als "link" - interpretiert; bei "falsche_endung" wird anhand der Endung klassifiziert. - """ - dp = self._datei_pruefer_cls(pfad=source, temporaer_erlaubt=False) - pe: pruef_ergebnis = dp.pruefe() - - if getattr(pe, "ok", False): - suffix = Path(source).suffix.lower() - if suffix in (".gpkg", ".sqlite", ".db"): - return "database" - return "file" - - aktion = getattr(pe, "aktion", None) - if aktion in ("datei_nicht_gefunden", "pfad_nicht_gefunden", "kein_dateipfad"): - return "link" - if aktion == "falsche_endung": - lower = source.lower() - for db_ext in (".gpkg", ".sqlite", ".db"): - if lower.endswith(db_ext): - return "database" - for file_ext in (".xlsx", ".xls", ".csv"): - if lower.endswith(file_ext): - return "file" return "unknown" - # ------------------------------------------------------------------ # - # Excel-Verarbeitung - #Es werden alle Werte mit gültigem Link übernommen. Die restliche Struktur - #wird nicht überprüft, da alle Fachplugins unterschiedliche Strukturen haben können - # ------------------------------------------------------------------ # - def process_excel_source( - self, - filepath: str - ) -> Tuple[Optional[Dict[str, List[Mapping[str, Any]]]], Any]: + def run(self, source: str) -> Tuple[SourceDict, pruef_ergebnis]: """ - Liest eine Excel-Datei ein und übernimmt ausschließlich die Zeilen, - deren Link durch den Linkpruefer als gültig eingestuft wurde. + Führt die vollständige Quellprüfung aus. - Ablauf - ------ - 1. Die Excel-Datei wird mit dem ``ExcelImporter`` eingelesen. - Erwartet wird eine Liste von Mappings (z.B. dicts), die jeweils - die Linkparameter enthalten. - - 2. Für jede Zeile wird der Wert ``row["Link"]`` extrahiert und durch - ``self.link_pruefer.pruefe(...)`` geprüft. - - 3. Das Prüfergebnis wird durch ``self.pruefmanager.verarbeite(...)`` - geleitet, der UI-Interaktion, Logging und finale Entscheidung übernimmt. - - 4. Nur Zeilen, deren verarbeitete Prüfergebnisse ``ok == True`` liefern, - werden in die Ergebnisliste übernommen. - - 5. Wenn mindestens eine Zeile gültig ist, wird ein Dict der Form:: - - {"rows": [row1, row2, ...]} - - zurückgegeben. - Wenn keine Zeile gültig ist, wird ``None`` zurückgegeben. - - Parameter - --------- - filepath: - Pfad zur Excel-Datei, die eingelesen werden soll. - - Returns - ------- - Tuple[Optional[Dict[str, List[Mapping[str, Any]]]], pruef_ergebnis] - - ``data``: ``{"rows": [...]} `` wenn gültige Zeilen existieren, - sonst ``None``. - - ``pruef_ergebnis``: ein zusammenfassendes Prüfergebnis, das - den Lesevorgang beschreibt (nicht die Einzelprüfungen). - - Hinweise - -------- - - Diese Methode führt **keine Normalisierung** durch. - - Die Verantwortung für die Struktur der Excel-Zeilen liegt beim Fachplugin. - - Der Linkpruefer prüft ausschließlich den Wert ``row["Link"]``. + Diese Methode ist **UI‑frei**. Sie gibt rohe Ergebnisse zurück, + die vom Aufrufer über den ``Pruefmanager`` verarbeitet werden. """ + self.set_source(source) + source_type = self.analyze_source_type(source) - # 1) Excel einlesen - importer = ExcelImporter(filepath=filepath, pruefmanager=self.pruefmanager) - rows = importer.import_xlsx() # erwartet: List[Mapping[str, Any]] + source_dict: SourceDict = {} + partial_results: List[pruef_ergebnis] = [] - valid_rows: List[Mapping[str, Any]] = [] - - # 2) Jede Zeile einzeln prüfen - for row in rows: - raw_link = row.get("Link") - - # 2a) Fachliche Prüfung - pe = self.link_pruefer.pruefe(raw_link) - - # 2b) Verarbeitung durch den Pruefmanager - processed = self.pruefmanager.verarbeite(pe) - - # 2c) Nur gültige Zeilen übernehmen - if getattr(processed, "ok", False): - valid_rows.append(row) - - # 3) Zusammenfassendes Prüfergebnis erzeugen - if valid_rows: - pe_ok = pruef_ergebnis( - ok=True, - meldung=f"{len(valid_rows)} gültige Zeilen aus Excel gelesen", - aktion="ok", - kontext=filepath, + if source_type == "excel": + source_dict, partial_results = self._process_excel_source(source) + elif source_type == "database": + source_dict, partial_results = self._process_database_source(source) + elif source_type == "service": + source_dict, partial_results = self._process_service_source(source) + else: + partial_results.append( + pruef_ergebnis( + ok=False, + meldung="Quelle konnte nicht klassifiziert werden", + aktion="kein_dateipfad", + kontext={"source": source}, + ) ) - processed_summary = self.pruefmanager.verarbeite(pe_ok) - return {"rows": valid_rows}, processed_summary - # Keine gültigen Zeilen - pe_fail = pruef_ergebnis( - ok=False, - meldung="Keine gültigen Links in der Excel-Datei gefunden", - aktion="read_error", - kontext=filepath, - ) - processed_summary = self.pruefmanager.verarbeite(pe_fail) - return None, processed_summary + summary = self._aggregate_results(source, source_dict, partial_results) + return source_dict, summary + # ------------------------------------------------------------------ + # Excel‑Quellen + # ------------------------------------------------------------------ + def _process_excel_source( + self, filepath: str + ) -> Tuple[SourceDict, List[pruef_ergebnis]]: + source_dict: SourceDict = {} + results: List[pruef_ergebnis] = [] + return source_dict, results - # ------------------------------------------------------------------ # - # Einzellink-Verarbeitung - # ------------------------------------------------------------------ # - def process_single_link( + # ------------------------------------------------------------------ + # Datenbank‑Quellen + # ------------------------------------------------------------------ + def _process_database_source( + self, db_path: str + ) -> Tuple[SourceDict, List[pruef_ergebnis]]: + source_dict: SourceDict = {} + results: List[pruef_ergebnis] = [] + return source_dict, results + + # ------------------------------------------------------------------ + # Dienst‑Quellen + # ------------------------------------------------------------------ + def _process_service_source( + self, link: str + ) -> Tuple[SourceDict, List[pruef_ergebnis]]: + source_dict: SourceDict = {} + results: List[pruef_ergebnis] = [] + return source_dict, results + + # ------------------------------------------------------------------ + # Aggregation + # ------------------------------------------------------------------ + def _aggregate_results( self, - link: Mapping[str, Any] - ) -> Tuple[Optional[Dict[str, List[Mapping[str, Any]]]], Any]: + source: str, + source_dict: SourceDict, + partial_results: List[pruef_ergebnis], + ) -> pruef_ergebnis: """ - Prüft einen einzelnen Link anhand der im Link-Dict enthaltenen Link-URL. + Aggregiert Einzelprüfungen zu einem Gesamt‑``pruef_ergebnis``. - Ablauf - ------ - 1. Erwartet wird ein Mapping (z.B. dict), das die Linkparameter enthält. - Mindestens der Schlüssel ``"Link"`` muss vorhanden sein. - - 2. Der eigentliche Link (z.B. URL) wird aus ``link["Link"]`` extrahiert - und an ``self.link_pruefer.pruefe(...)`` übergeben. - - 3. Das Prüfergebnis wird anschließend durch ``self.pruefmanager.verarbeite(...)`` - geleitet, der UIInteraktion, Logging und finale Entscheidung übernimmt. - - 4. Wenn das verarbeitete Prüfergebnis **nicht OK** ist, wird - ``(None, pruef_ergebnis)`` zurückgegeben. - - 5. Wenn das Prüfergebnis **OK** ist, wird das unveränderte LinkDict - in der Struktur ``{"rows": [link]}`` zurückgegeben. - - Parameter - --------- - link: - Ein Mapping mit den Linkparametern (z.B. id, Thema, Gruppe, Link, - Anbieter, Stildatei). Diese Methode verändert das Mapping nicht. - - Returns - ------- - Tuple[Optional[Dict[str, List[Mapping[str, Any]]]], pruef_ergebnis] - - ``data``: ``{"rows": [link]}`` wenn gültig, sonst ``None`` - - ``pruef_ergebnis``: das vom Pruefmanager verarbeitete Ergebnis - - Hinweise - -------- - - Diese Methode führt **keine Normalisierung** durch. - - Die Verantwortung für die Struktur des Link-Dicts liegt beim Fachplugin. - - Der Linkpruefer prüft ausschließlich den Wert ``link["Link"]``. + **Keine UI‑Interaktion.** """ + if source_dict: + return pruef_ergebnis( + ok=True, + meldung="Quelle erfolgreich geprüft", + aktion="ok", + kontext={ + "source": source, + "valid_entries": sum(len(v) for v in source_dict.values()), + }, + ) - # 1) Link extrahieren (Fachplugin garantiert, dass "Link" existiert) - raw_link = link.get("Link") - - # 2) Fachliche Prüfung durch den Linkpruefer - pruef_ergebnis = self.link_pruefer.pruefe(raw_link) - - # 3) Verarbeitung durch den Pruefmanager - processed = self.pruefmanager.verarbeite(pruef_ergebnis) - - # 4) Wenn Prüfung nicht OK → keine Daten zurückgeben - if not getattr(processed, "ok", False): - return None, processed - - # 5) Prüfung OK → unverändertes Link-Dict zurückgeben - data = {"rows": [link]} - - return data, processed - - - - - # ------------------------------------------------------------------ # - # Datenbank-Verarbeitung - # ------------------------------------------------------------------ # - #def process_database_table(self, db_path: str, table: Optional[str] = None) -> Tuple[Optional[Dict[str, List[Mapping[str, Any]]]], pruef_ergebnis]: - #noch nicht implementiert - """ - Liest eine Tabelle aus einer SQLite/GeoPackage-Datei. - - Verhalten - --------- - 1. Validiert die Datei mit dem Dateipruefer. - 2. Falls OK, versucht es, die angegebene Tabelle zu lesen; falls keine Tabelle - angegeben ist, wird nach einer typischen Metadaten-Tabelle 'layer_metadaten' - gesucht und diese gelesen. - 3. Gibt die Zeilen als Liste von Dicts zurück. - """ - dp = self._datei_pruefer_cls(pfad=db_path, temporaer_erlaubt=False) - pe = dp.pruefe() - processed = self.pruefmanager.verarbeite(pe) - if not getattr(processed, "ok", False): - return None, processed - - conn = sqlite3.connect(db_path) - cur = conn.cursor() - if table: - cur.execute(f"SELECT * FROM {table}") - cols = [d[0] for d in cur.description] - rows = [dict(zip(cols, r)) for r in cur.fetchall()] - else: - cur.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='layer_metadaten'") - if cur.fetchone(): - cur.execute("SELECT * FROM layer_metadaten") - cols = [d[0] for d in cur.description] - rows = [dict(zip(cols, r)) for r in cur.fetchall()] - else: - rows = [] - conn.close() - - pe_ok = pruef_ergebnis(ok=True, meldung="DB gelesen", aktion="ok", kontext=db_path) - processed_ok = self.pruefmanager.verarbeite(pe_ok) - return {"rows": rows}, processed_ok - - # ------------------------------------------------------------------ # - # Hauptlauf / Dispatch - # ------------------------------------------------------------------ # - def run(self) -> Dict[str, Any]: - """ - Hauptmethode des DataGrabbers. - - Ablauf - ------ - 1. Prüft, ob eine Quelle gesetzt ist. - 2. Klassifiziert die Quelle via :meth:`analyze_source`. - 3. Dispatch: - - file (.xlsx/.xls) -> :meth:`process_excel_source` - - link -> :meth:`process_single_link` - - database -> :meth:`process_database_table` - - unknown -> Fehler - 4. Aggregiert geladene Einträge in einem Ergebnis-Dict und gibt dieses zurück. - - Rückgabeformat - ------------- - Ein Dict mit den Schlüsseln: - - 'geladen' : Liste der geladenen Themen/Namen - - 'fehler' : Mapping Thema -> Fehlermeldung - - 'ausserhalb': Liste der als ausserhalb klassifizierten Themen - - 'relevant' : Liste der relevanten Themen - - 'details' : zusätzliche Detailinformationen (z. B. Anzahl Zeilen) - """ - result: Dict[str, Any] = {"geladen": [], "fehler": {}, "ausserhalb": [], "relevant": [], "details": {}} - - if not self.source: - pe = pruef_ergebnis(ok=False, meldung="Keine Quelle gesetzt", aktion="kein_dateipfad", kontext=None) - processed = self.pruefmanager.verarbeite(pe) - result["fehler"]["source"] = getattr(processed, "meldung", "Keine Quelle") - return result - - src_type = self.analyze_source(self.source) - - if src_type == "file": - suffix = Path(self.source).suffix.lower() - if suffix in (".xlsx", ".xls"): - data_dict, pe = self.process_excel_source(self.source) - else: - pe = pruef_ergebnis(ok=False, meldung="Dateityp nicht unterstützt", aktion="falsche_endung", kontext=self.source) - pe = self.pruefmanager.verarbeite(pe) - data_dict = None - - elif src_type == "link": - data_dict, pe = self.process_single_link(self.source) - - #elif src_type == "database": - #data_dict, pe = self.process_database_table(self.source, table=None) - - else: - pe = pruef_ergebnis(ok=False, meldung="Quelle unbekannt", aktion="kein_dateipfad", kontext=self.source) - pe = self.pruefmanager.verarbeite(pe) - data_dict = None - - # Falls Daten vorhanden: fülle result['geladen'] und details - if data_dict and "rows" in data_dict: - rows = data_dict["rows"] - for r in rows: - thema = r.get("Inhalt") or r.get("ident") or r.get("Link") or "unbenannt" - result["geladen"].append(thema) - result["details"]["source_rows"] = len(rows) - - # Falls das letzte pruef_ergebnis einen Fehler enthält, übernehme es - if not getattr(pe, "ok", False): - result["fehler"]["source"] = getattr(pe, "meldung", "Fehler bei Quelle") - - return result + return pruef_ergebnis( + ok=False, + meldung="Keine gültigen Einträge in der Quelle gefunden", + aktion="read_error", + kontext={"source": source}, + ) diff --git a/modules/Dateipruefer.py b/modules/Dateipruefer.py index 31e5c25..1ad1a44 100644 --- a/modules/Dateipruefer.py +++ b/modules/Dateipruefer.py @@ -1,98 +1,175 @@ """ -sn_basis/modules/Dateipruefer.py – Prüfung von Dateieingaben für das Plugin. -Verwendet sys_wrapper und gibt pruef_ergebnis an den Pruefmanager zurück. +sn_basis/modules/Dateipruefer.py + +Erweiterter Dateiprüfer für Verfahrens-DB-Workflows mit vollständiger Unterstützung +der Anforderungen 1-2.e (leerer Pfad, fehlende Datei, bestehende Datei). """ from pathlib import Path +from typing import Optional -from sn_basis.functions.sys_wrapper import ( - join_path, - file_exists, -) - -from sn_basis.modules.Pruefmanager import pruef_ergebnis +from sn_basis.functions.sys_wrapper import join_path, file_exists +from sn_basis.modules.pruef_ergebnis import pruef_ergebnis, PruefAktion class Dateipruefer: """ - Prüft Dateieingaben und liefert ein pruef_ergebnis zurück. - Die eigentliche Nutzerinteraktion übernimmt der Pruefmanager. + Prüft Dateieingaben für Verfahrens-DB-Workflows und liefert :class:`pruef_ergebnis`. + + **Funktionsweise (deine Anforderungen 1-2.e):** + + +---------------------+------------------------------------------+---------------+ + | **Fall** | **Ergebnis** | **ok** | + +=====================+==========================================+===============+ + | 1. Leerer Pfad | ``temporaer_erlaubt`` | False | + +---------------------+------------------------------------------+---------------+ + | 2.a Leerer Pfad | Pruefmanager fragt → ``temporaer_erzeugen`` | True | + +---------------------+------------------------------------------+---------------+ + | 2.b Datei existiert | ``ok`` | True | + +---------------------+------------------------------------------+---------------+ + | 2.c Ungültiger Pfad | ``datei_nicht_gefunden`` | False | + +---------------------+------------------------------------------+---------------+ + | **2.d Datei fehlt** | **``datei_wird_erzeugt``** | **True** | + +---------------------+------------------------------------------+---------------+ + | **2.e Datei da** | **``datei_existiert``** | **False** | + +---------------------+------------------------------------------+---------------+ + + Der Dateiprüfer führt **keine UI-Interaktion** durch. + Entscheidungen werden ausschließlich vom :class:`Pruefmanager` getroffen. """ def __init__( self, - pfad: str, + pfad: Optional[str], basis_pfad: str = "", leereingabe_erlaubt: bool = False, - standarddatei: str | None = None, + standarddatei: Optional[str] = None, temporaer_erlaubt: bool = False, - ): + *, + verfahrens_db_modus: bool = True, # 🆕 Verfahrens-DB-spezifische Logik + ) -> None: + """ + Parameters + ---------- + pfad : Optional[str] + Vom UI gelieferter Dateipfad (kann leer oder Whitespace sein). + basis_pfad : str, optional + Basisverzeichnis für relative Pfade (default: ""). + leereingabe_erlaubt : bool, optional + Ob leere Eingabe grundsätzlich erlaubt ist (default: False). + standarddatei : Optional[str], optional + Optionaler Standardpfad (default: None). + temporaer_erlaubt : bool, optional + Ob bei leerer Eingabe temporäre Layer erlaubt sind (default: False). + verfahrens_db_modus : bool, optional + Aktiviert Verfahrens-DB-spezifische Logik (2.d, 2.e) (default: True). + """ self.pfad = pfad self.basis_pfad = basis_pfad self.leereingabe_erlaubt = leereingabe_erlaubt self.standarddatei = standarddatei self.temporaer_erlaubt = temporaer_erlaubt + self.verfahrens_db_modus = verfahrens_db_modus - # --------------------------------------------------------- - # Hilfsfunktion - # --------------------------------------------------------- - + # ------------------------------------------------------------------ + # Hilfsfunktionen + # ------------------------------------------------------------------ def _pfad(self, relativer_pfad: str) -> Path: - """ - Erzeugt einen OS-unabhängigen Pfad relativ zum Basisverzeichnis. - """ + """Erzeugt OS-unabhängigen Pfad relativ zum Basisverzeichnis.""" return join_path(self.basis_pfad, relativer_pfad) - # --------------------------------------------------------- - # Hauptfunktion - # --------------------------------------------------------- + def _ist_leer(self) -> bool: + """ + Prüft robust, ob Eingabe als „leer" zu behandeln ist. + Returns + ------- + bool + True bei None, leerem String oder reinem Whitespace. + """ + if self.pfad is None: + return True + if not isinstance(self.pfad, str): + return True + return not self.pfad.strip() + + def _ist_gueltiger_gpkg_pfad(self, pfad: Path) -> bool: + """ + Prüft, ob Pfad für GPKG geeignet ist (Endung + Schreibrechte). + + Returns + ------- + bool + True wenn `.gpkg`-Endung und Verzeichnis beschreibbar. + """ + if not str(pfad).lower().endswith('.gpkg'): + return False + + # Verzeichnis muss beschreibbar sein + return pfad.parent.exists() and pfad.parent.is_dir() + + # ------------------------------------------------------------------ + # Hauptlogik: deine Anforderungen 1-2.e + # ------------------------------------------------------------------ def pruefe(self) -> pruef_ergebnis: """ - Prüft eine Dateieingabe und liefert ein pruef_ergebnis zurück. - Der Pruefmanager entscheidet später, wie der Nutzer gefragt wird. - """ + 🆕 Prüft Dateieingabe gemäß Anforderungen 1-2.e. - # ----------------------------------------------------- - # 1. Fall: Eingabe ist leer - # ----------------------------------------------------- - if not self.pfad: + **Workflow:** + 1. **Leere Eingabe** → ``temporaer_erlaubt`` (Pruefmanager fragt) + 2. **Pfad prüfen**: + - **Ungültig** → 2.c ``datei_nicht_gefunden`` + - **Gültig, fehlt** → **2.d** ``datei_wird_erzeugt`` (ok=True!) + - **Gültig, existiert** → **2.e** ``datei_existiert`` (Pruefmanager fragt) + 3. **Datei OK** → 2.b ``ok`` + + Returns + ------- + pruef_ergebnis + Mit korrekter Aktion für jeden Fall. + """ + # 1. 🎯 ANFORDERUNG 1: Leere Eingabe + if self._ist_leer(): return self._handle_leere_eingabe() - # ----------------------------------------------------- - # 2. Fall: Eingabe ist nicht leer → Datei prüfen - # ----------------------------------------------------- - pfad = self._pfad(self.pfad) + # 2. Pfad normalisieren + pfad = self._pfad(self.pfad.strip()) - if not file_exists(pfad): + # 🆕 2.c: Ungültiger GPKG-Pfad? + if not self.verfahrens_db_modus or not self._ist_gueltiger_gpkg_pfad(pfad): return pruef_ergebnis( ok=False, - meldung=f"Die Datei '{self.pfad}' wurde nicht gefunden.", + meldung=f"Der Pfad '{self.pfad}' ist kein gültiger GPKG-Pfad.", aktion="datei_nicht_gefunden", kontext=pfad, ) - # ----------------------------------------------------- - # 3. Datei existiert → Erfolg - # ----------------------------------------------------- + # 🆕 2.d: Gültiger Pfad, Datei fehlt → DIREKT WEITER (ok=True!) + if not file_exists(pfad): + return pruef_ergebnis( + ok=True, # 🎯 WICHTIG: Pipeline fortsetzen! + meldung=f"Datei '{self.pfad}' wird erzeugt.", + aktion="datei_wird_erzeugt", + kontext=pfad, + ) + + # 🆕 2.e: Datei existiert → Pruefmanager fragt Überschreiben/etc. return pruef_ergebnis( - ok=True, - meldung="Datei gefunden.", - aktion="ok", + ok=False, # 🎯 Pruefmanager soll 4-Optionen-Dialog zeigen + meldung=f"Datei '{self.pfad}' existiert bereits.", + aktion="datei_existiert", kontext=pfad, ) - # --------------------------------------------------------- - # Behandlung leerer Eingaben - # --------------------------------------------------------- + # 2.b: Wird nicht erreicht (durch 2.e abgefangen) + # ------------------------------------------------------------------ + # Leere Eingabe (ANFORDERUNG 1, 2.a) + # ------------------------------------------------------------------ def _handle_leere_eingabe(self) -> pruef_ergebnis: """ - Liefert ein pruef_ergebnis für den Fall, dass das Dateifeld leer ist. - Der Pruefmanager fragt später den Nutzer. + Behandelt leere Eingaben (Priorität: leereingabe → Standard → temporär → Fehler). """ - - # 1. Leereingabe erlaubt → Nutzer fragen, ob das beabsichtigt war if self.leereingabe_erlaubt: return pruef_ergebnis( ok=False, @@ -101,19 +178,17 @@ class Dateipruefer: kontext=None, ) - # 2. Standarddatei verfügbar → Nutzer fragen, ob sie verwendet werden soll if self.standarddatei: return pruef_ergebnis( ok=False, meldung=( - f"Es wurde keine Datei angegeben. " + "Es wurde keine Datei angegeben. " f"Soll die Standarddatei '{self.standarddatei}' verwendet werden?" ), aktion="standarddatei_vorschlagen", kontext=self._pfad(self.standarddatei), ) - # 3. Temporäre Datei erlaubt → Nutzer fragen, ob temporär gearbeitet werden soll if self.temporaer_erlaubt: return pruef_ergebnis( ok=False, @@ -125,7 +200,6 @@ class Dateipruefer: kontext=None, ) - # 4. Leereingabe nicht erlaubt → Fehler return pruef_ergebnis( ok=False, meldung="Es wurde keine Datei angegeben.", diff --git a/modules/Pruefmanager.py b/modules/Pruefmanager.py index 12d8669..3ea40ec 100644 --- a/modules/Pruefmanager.py +++ b/modules/Pruefmanager.py @@ -1,66 +1,45 @@ +""" +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, - set_layer_visible, -) - +from sn_basis.functions import ask_yes_no, info, warning, error from sn_basis.modules.pruef_ergebnis import pruef_ergebnis, PruefAktion - +print("DEBUG: Pruefmanager DATEI GELADEN:", __file__) class Pruefmanager: - """ - Zentrale Verarbeitung von pruef_ergebnis-Objekten. - - Erwartete öffentliche API (verwendet von Core-Komponenten wie DataGrabber): - - report_error(thema, meldung, *, aktion: Optional[PruefAktion]=None, kontext=None) -> None - - request_decision(pruef_res) -> str - - report_summary(summary: dict) -> None - - verarbeite(ergebnis: pruef_ergebnis) -> pruef_ergebnis - """ - - def __init__(self, ui_modus: str = "qgis", parent: Optional[Any] = None): + def __init__(self, ui_modus: str = "qgis", parent: Optional[Any] = None) -> None: self.ui_modus = ui_modus self.parent = parent - # --------------------------------------------------------------------- - # Basis-API: Meldungen / Zusammenfassungen - # --------------------------------------------------------------------- - def report_error(self, thema: str, meldung: str, *, aktion: Optional[PruefAktion] = None, kontext: Optional[Any] = None) -> None: - """ - Einheitliche Meldung für Fehler/Warnungen aus dem Core. - Keine Rückgabe; dient als zentraler Hook für Logging/UI. - """ + # ------------------------------------------------------------------ + # 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", + "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", + "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 - - # Default: informative Warnung warning(thema, meldung) def report_summary(self, summary: dict) -> None: @@ -75,202 +54,159 @@ class Pruefmanager: f"Dienste ausserhalb: {len(ausserhalb)}\n" f"Fehler: {len(fehler)}" ) - info("DataGrabber Zusammenfassung", message) - # --------------------------------------------------------------------- - # Entscheidungs-API - # --------------------------------------------------------------------- - def request_decision(self, pruef_res: Any) -> str: - """ - Synchronously request a decision from the user (or return a default in headless mode). + # ------------------------------------------------------------------ + # VERFAHRENS-DB-spezifische Entscheidungen + # ------------------------------------------------------------------ + def _handle_datei_existiert(self, ergebnis: pruef_ergebnis) -> pruef_ergebnis: + if self.ui_modus != "qgis": + return ergebnis - Returns one of: - - "abort" - - "continue" - - "temporaer_erzeugen" - - "ignore" - """ - aktion = getattr(pruef_res, "aktion", None) - meldung = getattr(pruef_res, "meldung", str(pruef_res)) + 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" + "• **Überspringen**: Nur temporäre Layer erzeugen" + ) + + # Vereinfacht: Erst Überschreiben? → Dann Anhängen? → Überspringen + if ask_yes_no( + titel, + f"{meldung}\n\n**Überschreiben** (alle Layer ersetzen)?", + default=False, + parent=self.parent + ): + return pruef_ergebnis( + ok=True, + aktion="datei_existiert_ueberschreiben", + kontext=ergebnis.kontext, + ) + + if ask_yes_no( + titel, + f"{meldung}\n\n**Anhängen** (neue Layer hinzufügen)?", + default=False, + parent=self.parent + ): + return pruef_ergebnis( + ok=True, + aktion="datei_existiert_anhaengen", + kontext=ergebnis.kontext, + ) + + if ask_yes_no( + titel, + f"{meldung}\n\n**Überspringen** (nur temporäre Layer)?", + default=True, + parent=self.parent + ): + return pruef_ergebnis( + ok=True, + aktion="datei_existiert_ueberspringen", + kontext=ergebnis.kontext, + ) + + return ergebnis + + # ------------------------------------------------------------------ + # Basis-Entscheidungen (KORREKT: → pruef_ergebnis) + # ------------------------------------------------------------------ + def _handle_basic_decision(self, ergebnis: pruef_ergebnis) -> pruef_ergebnis: + """Basis-Entscheidung für einfache Ja/Nein-Fragen.""" + print(f"DEBUG _handle_basic_decision: aktion='{ergebnis.aktion}', ui_modus='{self.ui_modus}'") + + if self.ui_modus != "qgis": + print("DEBUG: Nicht QGIS → ergebnis unverändert") + 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: + print(f"DEBUG ask_yes_no: title='{title}', meldung='{meldung[:50]}...'") + yes = ask_yes_no(title, meldung, default=False, parent=self.parent) + print(f"DEBUG ask_yes_no: yes={yes}") + except Exception as e: + print(f"DEBUG ask_yes_no Exception: {e}") + return ergebnis + + if not yes: + print("DEBUG: Nutzer sagte Nein → ok=False") + return ergebnis + + # Nutzer sagte Ja + if ergebnis.aktion == "temporaer_erlaubt": + print("DEBUG: temporaer_erlaubt bestätigt → ok=True") + return pruef_ergebnis( + ok=True, + aktion="temporaer_erlaubt", + kontext=ergebnis.kontext + ) + + print("DEBUG: Andere Aktion bestätigt → ok=True, aktion='ok'") + return pruef_ergebnis( + ok=True, + aktion="ok", + kontext=ergebnis.kontext + ) + + # ------------------------------------------------------------------ + # Hauptlogik: verarbeite() (KORRIGIERT!) + # ------------------------------------------------------------------ + def verarbeite(self, ergebnis: pruef_ergebnis) -> pruef_ergebnis: + print("🔥 verarbeite() START") + print("DEBUG Pruefmanager:", ergebnis.ok, ergebnis.aktion) + print("DEBUG ergebnis.aktion TYPE:", type(ergebnis.aktion), repr(ergebnis.aktion)) + + # 1. Erfolg → direkt weiter + print("🔍 Schritt 1: Prüfe ergebnis.ok =", ergebnis.ok) + if ergebnis.ok: + print("✅ Schritt 1: ok=True → return") + return ergebnis + + # 2. VERFAHRENS-DB: Bestehende Datei + print("🔍 Schritt 2: Prüfe datei_existiert =", ergebnis.aktion == "datei_existiert") + if ergebnis.aktion == "datei_existiert": + print("✅ Schritt 2: _handle_datei_existiert") + return self._handle_datei_existiert(ergebnis) + + # 3. Basis interaktive Aktionen + print("🔍 Schritt 3: Definiere interactive_actions") interactive_actions = { "leereingabe_erlaubt", - "standarddatei_vorschlagen", + "standarddatei_vorschlagen", "temporaer_erlaubt", "layer_unsichtbar", } + print("DEBUG interactive_actions:", repr(interactive_actions)) + print("DEBUG ergebnis.aktion in interactive_actions?", ergebnis.aktion in interactive_actions) + + if ergebnis.aktion in interactive_actions: + print("✅ Schritt 3: Interaktive Aktion → _handle_basic_decision") + decision = self._handle_basic_decision(ergebnis) + print(f"DEBUG: _handle_basic_decision Ergebnis: ok={decision.ok}, aktion='{decision.aktion}'") + return decision - if aktion in interactive_actions: - if self.ui_modus == "qgis": - title_map = { - "leereingabe_erlaubt": "Ohne Eingabe fortfahren", - "standarddatei_vorschlagen": "Standarddatei verwenden", - "temporaer_erlaubt": "Temporäre Datei erzeugen", - "layer_unsichtbar": "Layer einblenden", - } - title = title_map.get(aktion, "Entscheidung erforderlich") - try: - yes = ask_yes_no(title, meldung, default=False, parent=self.parent) - except Exception: - return "abort" - if yes: - if aktion == "temporaer_erlaubt": - return "temporaer_erzeugen" - return "continue" - return "abort" - - if self.ui_modus == "headless": - return "abort" - - informational_actions = { - "leer", - "datei_nicht_gefunden", - "pfad_nicht_gefunden", - "url_nicht_erreichbar", - "netzwerkfehler", - "falscher_geotyp", - "layer_leer", - "falscher_layertyp", - "falsches_crs", - "felder_fehlen", - "datenquelle_unerwartet", - "layer_nicht_editierbar", - "kein_header", - "kein_arbeitsblatt", - "read_error", - "open_error", - "pflichtfelder_fehlen", - } - if aktion in informational_actions: - return "abort" - - return "abort" - - # --------------------------------------------------------------------- - # Höhere Abstraktion: verarbeite - # --------------------------------------------------------------------- - def verarbeite(self, ergebnis: pruef_ergebnis) -> pruef_ergebnis: - """ - Verarbeitet ein pruef_ergebnis-Objekt und führt ggf. Nutzerinteraktion durch. - Liefert ein ggf. modifiziertes pruef_ergebnis zurück. - """ - if ergebnis.ok: - return ergebnis - - aktion = ergebnis.aktion - kontext = ergebnis.kontext - meldung = ergebnis.meldung - - # Zentrale Meldung - self.report_error(aktion or "pruefung", meldung or "", aktion=aktion, kontext=kontext) - - # Interaktive Entscheidungen - if aktion in ("leereingabe_erlaubt", "standarddatei_vorschlagen", "temporaer_erlaubt", "layer_unsichtbar"): - decision = self.request_decision(ergebnis) - if decision == "temporaer_erzeugen": - return pruef_ergebnis(ok=True, meldung="Temporäre Datei soll erzeugt werden.", aktion="temporaer_erzeugen", kontext=None) - if decision == "continue": - return pruef_ergebnis(ok=True, meldung="Fortgefahren.", aktion="ok", kontext=kontext) - return ergebnis # abort / unverändert - - # Spezielle Excel/Importer-Fälle: klare Meldungen, keine interaktive Entscheidung - if aktion == "kein_header": - warning("Excel-Import", meldung or "") - return ergebnis - - if aktion == "kein_arbeitsblatt": - warning("Excel-Import", meldung or "") - return ergebnis - - if aktion in ("read_error", "open_error"): - error("Excel-Import", meldung or "") - return ergebnis - - if aktion == "datei_nicht_gefunden": - warning("Datei nicht gefunden", meldung or "") - return ergebnis - - # Spezieller Fall: layer_unsichtbar (falls nicht interaktiv behandelt) - if aktion == "layer_unsichtbar": - if kontext is not None: - try: - set_layer_visible(kontext, True) - return pruef_ergebnis(ok=True, meldung="Layer wurde eingeblendet.", aktion="ok", kontext=kontext) - except Exception: - return ergebnis - return ergebnis - - # Standard: keine Änderung + # 4. Fehler behandeln + print("❌ Schritt 4: FEHLER BEHANDELN") + self.report_error( + thema=ergebnis.aktion or "pruefung", + meldung=ergebnis.meldung or "", + aktion=ergebnis.aktion, + kontext=ergebnis.kontext, + ) + print("🔥 verarbeite() ENDE mit ok=False") return ergebnis - def ask_overwrite_append_cancel(self, layer_name: str, default: str = "overwrite") -> str: - """ - Zeigt dem Nutzer eine Auswahl für einen bereits existierenden Layer an. - - Rückgabe - ------- - str - Einer der Werte: "overwrite", "append", "cancel". - - Verhalten - -------- - - Verwendet bevorzugt die UI-Wrapper-Funktion `qt_wrapper` / `qgisui_wrapper`, - falls vorhanden (z. B. ein QMessageBox-Dialog mit drei Buttons). - - Im Mock- oder Headless-Modus (kein Qt/QGIS verfügbar) wird der übergebene - `default`-Wert zurückgegeben. - - Alle Nutzerinteraktionen laufen über diese zentrale Methode, damit das - Plugin an einer Stelle gesteuert und ggf. getested werden kann. - - Parameter - --------- - layer_name: - Anzeigename des Layers, der bereits existiert (wird im Dialog angezeigt). - default: - Rückgabewert im Headless/Mock-Modus oder wenn der Dialog nicht verfügbar ist. - Gültige Werte: "overwrite", "append", "cancel". Standard: "overwrite". - """ - # Validierung des Defaults - if default not in ("overwrite", "append", "cancel"): - default = "overwrite" - - # Versuche, eine UI-Wrapper-Funktion zu verwenden, falls vorhanden - try: - # qgisui_wrapper kann eine spezialisierte Dialogfunktion bereitstellen - from sn_basis.functions import qgisui_wrapper as qgisui - ask_fn = getattr(qgisui, "ask_overwrite_append_cancel", None) - if callable(ask_fn): - # Die Wrapper-Funktion soll genau die drei Strings zurückgeben - choice = ask_fn(layer_name) - if choice in ("overwrite", "append", "cancel"): - return choice - except Exception: - # Falls Import/Wrapper fehlschlägt, weiter zum Qt-Fallback - pass - - # Fallback: direkte Qt-Dialoge über qt_wrapper (wenn verfügbar) - try: - from sn_basis.functions import qt_wrapper as qt - QMessageBox = getattr(qt, "QMessageBox", None) - if QMessageBox is not None: - # Erzeuge und konfiguriere Dialog - msg = QMessageBox() - msg.setWindowTitle("Layer bereits vorhanden") - msg.setText(f"Der Layer '{layer_name}' existiert bereits. Was möchten Sie tun?") - overwrite_btn = msg.addButton("Überschreiben", QMessageBox.AcceptRole) - append_btn = msg.addButton("Anhängen", QMessageBox.AcceptRole) - cancel_btn = msg.addButton("Abbrechen", QMessageBox.RejectRole) - msg.setDefaultButton(overwrite_btn) - # Blockierend anzeigen - msg.exec_() - clicked = msg.clickedButton() - if clicked == overwrite_btn: - return "overwrite" - if clicked == append_btn: - return "append" - return "cancel" - except Exception: - # Qt nicht verfügbar oder Fehler beim Dialogaufbau - pass - - # Headless / Mock: gib Default zurück - return default diff --git a/modules/pruef_ergebnis.py b/modules/pruef_ergebnis.py index af0054d..78eb423 100644 --- a/modules/pruef_ergebnis.py +++ b/modules/pruef_ergebnis.py @@ -1,9 +1,21 @@ +""" +sn_basis/modules/pruef_ergebnis.py + +Erweitertes Ergebnisobjekt für Dateiprüfungen mit Verfahrens-DB-spezifischen Aktionen. +""" + from __future__ import annotations from dataclasses import dataclass from typing import Any, Optional, Literal +from pathlib import Path + + +# ============================================================================= +# Erweiterte PruefAktionen für Verfahrens-DB-Workflow +# ============================================================================= -# Erweitertes Literal mit allen erlaubten Aktionen (PruefAktion) PruefAktion = Literal[ + # Basis-Aktionen (bestehend) "ok", "leer", "leereingabe_erlaubt", @@ -16,6 +28,8 @@ PruefAktion = Literal[ "pfad_nicht_gefunden", "url_nicht_erreichbar", "netzwerkfehler", + + # Layer-spezifisch "layer_nicht_gefunden", "layer_unsichtbar", "falscher_geotyp", @@ -25,15 +39,26 @@ PruefAktion = Literal[ "felder_fehlen", "datenquelle_unerwartet", "layer_nicht_editierbar", + + # Dateiendung/Format "falsche_endung", "pflichtfelder_fehlen", - # Excel / Import-spezifische Aktionen + + # Excel/Import "kein_header", "kein_arbeitsblatt", "read_error", "open_error", "datenabruf", - # Generische Prüf-/Speicher-Aktionen + + # 🆕 VERFAHRENS-DB SPEZIFISCH (deine Anforderungen 2.d, 2.e) + "datei_wird_erzeugt", # 2.d: Pfad gültig, Datei fehlt → weiter + "datei_existiert", # Datei vorhanden → Layer-Entscheidung + "datei_existiert_ueberschreiben", # 2.e: Nutzer wählt "Überschreiben" + "datei_existiert_anhaengen", # 2.e: Nutzer wählt "Anhängen" + "datei_existiert_ueberspringen", # 2.e: Nutzer wählt "Überspringen" + + # Generisch "pruefe_exception", "save_exception", "save_not_implemented", @@ -42,22 +67,113 @@ PruefAktion = Literal[ "needs_user_action", ] + @dataclass class pruef_ergebnis: """ - Einheitliches Ergebnisobjekt für Prüfer. - - ok: True wenn Prüfung bestanden - - meldung: menschenlesbare Meldung - - aktion: maschinenlesbarer Aktionscode (PruefAktion) - - kontext: optionaler Zusatzkontext (z. B. Pfad, Layer-Objekt) + Einheitliches Ergebnisobjekt für Prüfer im Verfahrens-DB-Workflow. + + Attributes + ---------- + ok : bool + True wenn Prüfung bestanden und Pipeline fortgesetzt werden kann. + False signalisiert Fehler oder Nutzerentscheidung erforderlich. + meldung : Optional[str], optional + Menschenlesbare Meldung für UI-Dialoge (BY: Pruefmanager). + aktion : Optional[PruefAktion], optional + Maschinenlesbarer Aktionscode für nachfolgende Pipeline-Schritte. + kontext : Optional[Any], optional + Zusatzkontext: meist `pathlib.Path` für Dateipfade oder Layer-Objekte. + + Verfahrens-DB-spezifische Aktionen: + + +-----------------------------+-------------------------------------------------+ + | Aktion | Bedeutung | + +=============================+=================================================+ + | ``datei_wird_erzeugt`` | 2.d: Neues GPKG wird angelegt (Pfad gültig) | + +-----------------------------+-------------------------------------------------+ + | ``datei_existiert`` | Datei vorhanden → Layer-Überschreibung prüfen | + +-----------------------------+-------------------------------------------------+ + | ``datei_existiert_*`` | 2.e: Nutzerentscheidung für bestehende Datei | + +-----------------------------+-------------------------------------------------+ """ + ok: bool meldung: Optional[str] = None aktion: Optional[PruefAktion] = None kontext: Optional[Any] = None - def __init__(self, ok: bool, meldung: Optional[str] = None, aktion: Optional[PruefAktion] = None, kontext: Optional[Any] = None): + def __init__( + self, + ok: bool, + meldung: Optional[str] = None, + aktion: Optional[PruefAktion] = None, + kontext: Optional[Any] = None, + ) -> None: + """ + Erstellt ein neues Prüfergebnis. + + Parameters + ---------- + ok : bool + True für "weiter mit Pipeline", False für "Entscheidung/Fehler". + meldung : Optional[str] + UI-Text für Nutzerdialoge. + aktion : Optional[PruefAktion] + Maschinenaktion für nachfolgende Verarbeitung. + kontext : Optional[Any] + Typischerweise `pathlib.Path` (Dateipfad) oder `QgsVectorLayer`. + """ self.ok = ok self.meldung = meldung self.aktion = aktion self.kontext = kontext + + @property + def ist_verfahrens_db_aktion(self) -> bool: + """ + Prüft, ob es sich um eine Verfahrens-DB-spezifische Aktion handelt. + + Returns + ------- + bool + True für ``datei_wird_erzeugt`` oder ``datei_existiert*``. + """ + return self.aktion in { + "datei_wird_erzeugt", + "datei_existiert", + "datei_existiert_ueberschreiben", + "datei_existiert_anhaengen", + "datei_existiert_ueberspringen", + } + + @property + def dateipfad(self) -> Optional[Path]: + """ + Extrahiert den Dateipfad aus dem Kontext (falls vorhanden). + + Returns + ------- + Optional[Path] + `Path`-Objekt oder None. + """ + if isinstance(self.kontext, Path): + return self.kontext + return None + + @property + def erlaubte_persistierung(self) -> bool: + """ + Prüft, ob die Pipeline Daten persistieren darf. + + Returns + ------- + bool + True für ``datei_wird_erzeugt``, ``datei_existiert_ueberschreiben``, + ``datei_existiert_anhaengen``. + """ + return self.aktion in { + "datei_wird_erzeugt", + "datei_existiert_ueberschreiben", + "datei_existiert_anhaengen", + }