diff --git a/assets/datagrabber.jpeg b/assets/datagrabber.jpeg new file mode 100644 index 0000000..e38d84b Binary files /dev/null and b/assets/datagrabber.jpeg differ diff --git a/assets/datagrabber.md b/assets/datagrabber.md new file mode 100644 index 0000000..a3161f6 --- /dev/null +++ b/assets/datagrabber.md @@ -0,0 +1,38 @@ +```mermaid +flowchart TD + subgraph Plugin + P[sn_plan41 Fachplugin] + A[Adapter Plan41LinklistAdapter] + PM[Pruefmanager] + LP[Layerpruefer] + KP[Linkpruefer] + SP[Stilpruefer] + end + + subgraph Core + DG[DataGrabber] + NL[normalized entries] + LL[Layer Loader Provider Dispatch] + SM[Spatial Matcher] + ST[Storage GPKG / PostGIS] + PR[Project QGIS - addMapLayer] + LOG[Log / Ergebnisstruktur] + end + + P -->|gibt Adapter, Prüfer, Pruefmanager| DG + A -->|load liefert Rohdaten| DG + DG -->|adapter.normalize| NL + NL --> DG + DG -->|für jeden Eintrag: _check_link -> KP.check| KP + DG -->|für jeden Eintrag: _check_style -> SP.check| SP + DG -->|prüfe vorhandene Layer| LP + DG -->|lade Layer via provider| LL + LL -->|Features| SM + SM -->|Abgleich| DG + DG -->|speichern| ST + ST --> PR + DG --> PR + DG -->|Ergebnis/Fehler| LOG + LOG --> PM + DG --> PM +``` \ No newline at end of file diff --git a/assets/datagrabber.pdf b/assets/datagrabber.pdf new file mode 100644 index 0000000..f79edb2 Binary files /dev/null and b/assets/datagrabber.pdf differ diff --git a/functions/qt_wrapper.py b/functions/qt_wrapper.py index 9e116dc..8346f8b 100644 --- a/functions/qt_wrapper.py +++ b/functions/qt_wrapper.py @@ -32,6 +32,7 @@ QTabWidget: type QToolButton: Type[Any] QSizePolicy: Type[Any] Qt: Type[Any] +ComboBox: Type[Any] YES: Optional[Any] = None NO: Optional[Any] = None @@ -68,6 +69,7 @@ try: QTabWidget as _QTabWidget,# type: ignore QToolButton as _QToolButton,#type:ignore QSizePolicy as _QSizePolicy,#type:ignore + QComboBox as _QComboBox, ) @@ -107,6 +109,7 @@ try: QTabWidget = _QTabWidget QToolButton=_QToolButton QSizePolicy=_QSizePolicy + QComboBox=_QComboBox YES = QMessageBox.StandardButton.Yes NO = QMessageBox.StandardButton.No @@ -148,7 +151,7 @@ try: except Exception: try: - from PyQt5.QtWidgets import ( + from PyQt5.QtWidgets import (# type: ignore QMessageBox as _QMessageBox, QFileDialog as _QFileDialog, QWidget as _QWidget, @@ -166,18 +169,20 @@ except Exception: QTabWidget as _QTabWidget, QToolButton as _QToolButton, QSizePolicy as _QSizePolicy, + QComboBox as _QComboBox, ) - from PyQt5.QtCore import ( + from PyQt5.QtCore import (# type: ignore QEventLoop as _QEventLoop, QUrl as _QUrl, QCoreApplication as _QCoreApplication, Qt as _Qt, ) - from PyQt5.QtNetwork import ( + from PyQt5.QtNetwork import (# type: ignore QNetworkRequest as _QNetworkRequest, QNetworkReply as _QNetworkReply, ) + QMessageBox = _QMessageBox QFileDialog = _QFileDialog QEventLoop = _QEventLoop @@ -203,6 +208,7 @@ except Exception: QTabWidget = _QTabWidget QToolButton=_QToolButton QSizePolicy=_QSizePolicy + ComboBox=_QComboBox YES = QMessageBox.Yes NO = QMessageBox.No @@ -210,6 +216,8 @@ except Exception: ICON_QUESTION = QMessageBox.Question QT_VERSION = 5 + + # then try next backend # --------------------------------------------------------- # Qt5 Enum-Aliase (vereinheitlicht) # --------------------------------------------------------- @@ -246,7 +254,7 @@ except Exception: QT_VERSION = 0 class FakeEnum(int): - def __or__(self, other: "FakeEnum") -> "FakeEnum": + def __or__(self, other: int) -> "FakeEnum": return FakeEnum(int(self) | int(other)) YES = FakeEnum(1) @@ -518,3 +526,55 @@ except Exception: 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 + + 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) + + def currentText(self) -> str: + if 0 <= self._index < len(self._items): + return self._items[self._index] + return "" + + ComboBox = _MockComboBox diff --git a/functions/test.md b/functions/test.md new file mode 100644 index 0000000..84240dc --- /dev/null +++ b/functions/test.md @@ -0,0 +1,14 @@ +mermaid´´´ +flowchart TD + A[Projekt] + + subgraph children[ ] + direction TB + B[src] + C[docs] + D[README.md] + end + + A --> B + A --> C + A --> D diff --git a/modules/DataGrabber.py b/modules/DataGrabber.py new file mode 100644 index 0000000..85ba2fe --- /dev/null +++ b/modules/DataGrabber.py @@ -0,0 +1,324 @@ +# 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. + +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. +""" + +from __future__ import annotations + +from typing import ( + Optional, + Any, + Mapping, + Iterable, + Dict, + Protocol, + Literal, + Tuple, + List, +) +from pathlib import Path +import sqlite3 + +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 +from sn_basis.modules.stilpruefer import Stilpruefer +from sn_basis.modules.excel_importer import ExcelImporter + + +SourceType = Literal["file", "link", "database", "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]: + ... + + +class DataGrabber: + """ + DataGrabber orchestriert das Einlesen einer Quelle und die Übergabe an Prüfer. + + 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. + """ + + def __init__( + self, + pruefmanager: Pruefmanager, + *, + datei_pruefer_cls=Dateipruefer, + link_pruefer: Linkpruefer, + layer_pruefer: Layerpruefer, + stil_pruefer: Stilpruefer, + ) -> None: + # Pruefmanager ist verpflichtend + self.pruefmanager: Pruefmanager = pruefmanager + + # Dateipruefer-Klasse (wird zur Laufzeit mit einem Pfad instanziert) + self._datei_pruefer_cls = datei_pruefer_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 + + # Quelle (wird später gesetzt) + self.source: Optional[str] = None + + # ------------------------------------------------------------------ # + # Source Management + # ------------------------------------------------------------------ # + def set_source(self, source: str) -> None: + """ + Setzt die Quelle für den DataGrabber. + + Die Quelle ist ein String, der entweder ein lokaler Dateipfad, + ein Einzellink (URL/URI) oder ein Pfad zu einer Datenbank/GeoPackage ist. + """ + 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 ohne Prüfung der Links, Pfade oder Stile geladen, da verschiedene Plugins verschiedene xlsx-Strukturen haben können + # ------------------------------------------------------------------ # + def process_excel_source(self, filepath: str) -> Tuple[Optional[Dict[str, List[Mapping[str, Any]]]], pruef_ergebnis]: + """ + Liest eine Excel-Datei (.xlsx/.xls) mit dem ExcelImporter und gibt ein Dict + mit den Zeilen zurück sowie das vom Pruefmanager verarbeitete pruef_ergebnis. + + Rückgabe + ------- + - (data_dict, processed_pruef_ergebnis) + data_dict: {'rows': [Mapping,...]} oder None bei Fehlern + processed_pruef_ergebnis: das Ergebnis, nachdem der Pruefmanager das + interne pruef_ergebnis verarbeitet hat. + """ + importer = ExcelImporter(filepath=filepath, pruefmanager=self.pruefmanager) + rows = importer.import_xlsx() # erwartet: List[Mapping[str, Any]] + data = {"rows": rows} + pe_ok = pruef_ergebnis(ok=True, meldung="Excel erfolgreich gelesen", aktion="ok", kontext=filepath) + processed = self.pruefmanager.verarbeite(pe_ok) + return data, processed + + # ------------------------------------------------------------------ # + # Einzellink-Verarbeitung + # ------------------------------------------------------------------ # + def process_single_link(self, link: str) -> Tuple[Optional[Dict[str, List[Mapping[str, Any]]]], pruef_ergebnis]: + """ + Verarbeitet einen Einzellink. + + Ablauf + ------ + 1. Führt die fachliche Prüfung über self.link_pruefer.pruefe(link) aus. + 2. Übergibt das Ergebnis an den Pruefmanager (self.pruefmanager.verarbeite). + 3. Wenn die Prüfung nicht OK ist, wird nur das verarbeitete pruef_ergebnis zurückgegeben. + 4. Wenn die Prüfung OK ist, erwartet diese Implementierung, dass der Prüfer + die Link-Parameter im pruef_ergebnis.kontext als Mapping bereitstellt. + Dieses Mapping wird unverändert in ein Dict {'rows': [kontext]} überführt + und zusammen mit dem verarbeiteten pruef_ergebnis zurückgegeben. + + Hinweis + ------ + Diese Funktion enthält keine Fallbacks, keine normalize-/load-Aufrufe und + keine zusätzlichen Validierungen. Der Linkpruefer ist verantwortlich dafür, + bei OK ein geeignetes Mapping im pruef_ergebnis.kontext bereitzustellen. + """ + # 1) Fachliche Prüfung durch den Linkpruefer + pe = self.link_pruefer.pruefe(link) + + # 2) Pruefmanager verarbeiten lassen (Logging / UI / Entscheidung) + processed = self.pruefmanager.verarbeite(pe) + + # 3) Wenn Prüfung nicht OK -> nur das verarbeitete pruef_ergebnis zurückgeben + if not getattr(processed, "ok", False): + return None, processed + + # 4) Prüfung OK -> Prüfer liefert die Link-Parameter im pruef_ergebnis.kontext + kontext = getattr(pe, "kontext", None) + data = {"rows": [kontext]} + # Erwartung: kontext ist ein Mapping mit den Link-Parametern. + # Wir übergeben es unverändert in das rows-Format. + 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 diff --git a/modules/Dateipruefer.py b/modules/Dateipruefer.py index cb4e6af..31e5c25 100644 --- a/modules/Dateipruefer.py +++ b/modules/Dateipruefer.py @@ -39,7 +39,7 @@ class Dateipruefer: def _pfad(self, relativer_pfad: str) -> Path: """ - Erzeugt einen OS‑unabhängigen Pfad relativ zum Basisverzeichnis. + Erzeugt einen OS-unabhängigen Pfad relativ zum Basisverzeichnis. """ return join_path(self.basis_pfad, relativer_pfad) @@ -119,7 +119,7 @@ class Dateipruefer: ok=False, meldung=( "Es wurde keine Datei angegeben. " - "Soll eine temporäre Datei erzeugt werden?" + "Sollen temporäre Layer erzeugt werden?" ), aktion="temporaer_erlaubt", kontext=None, diff --git a/modules/Pruefmanager.py b/modules/Pruefmanager.py index 4a0a85c..a7a8910 100644 --- a/modules/Pruefmanager.py +++ b/modules/Pruefmanager.py @@ -1,7 +1,5 @@ -""" -sn_basis/modules/Pruefmanager.py – zentrale Verarbeitung von pruef_ergebnis-Objekten. -Steuert die Nutzerinteraktion über Wrapper. -""" +from __future__ import annotations +from typing import Optional, Any from sn_basis.functions import ( ask_yes_no, @@ -11,150 +9,196 @@ from sn_basis.functions import ( set_layer_visible, ) -from sn_basis.modules.pruef_ergebnis import pruef_ergebnis +from sn_basis.modules.pruef_ergebnis import pruef_ergebnis, PruefAktion class Pruefmanager: """ - Verarbeitet pruef_ergebnis-Objekte und steuert die Nutzerinteraktion. + 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"): + def __init__(self, ui_modus: str = "qgis", parent: Optional[Any] = None): self.ui_modus = ui_modus + self.parent = parent - # --------------------------------------------------------- - # Hauptfunktion - # --------------------------------------------------------- + # --------------------------------------------------------------------- + # 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. + """ + 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 + + # Default: informative Warnung + 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) + + # --------------------------------------------------------------------- + # Entscheidungs-API + # --------------------------------------------------------------------- + def request_decision(self, pruef_res: Any) -> str: + """ + Synchronously request a decision from the user (or return a default in headless mode). + + Returns one of: + - "abort" + - "continue" + - "temporaer_erzeugen" + - "ignore" + """ + aktion = getattr(pruef_res, "aktion", None) + meldung = getattr(pruef_res, "meldung", str(pruef_res)) + + interactive_actions = { + "leereingabe_erlaubt", + "standarddatei_vorschlagen", + "temporaer_erlaubt", + "layer_unsichtbar", + } + + 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", + } + 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 und führt ggf. Nutzerinteraktion durch. - Rückgabe: neues oder unverändertes 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 - # ----------------------------------------------------- - # Allgemeine Aktionen - # ----------------------------------------------------- + # Zentrale Meldung + self.report_error(aktion or "pruefung", meldung or "", aktion=aktion, kontext=kontext) - if aktion == "leer": - warning("Eingabe fehlt", ergebnis.meldung) + # 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 == "leereingabe_erlaubt": - if ask_yes_no("Ohne Eingabe fortfahren", ergebnis.meldung): - return pruef_ergebnis( - ok=True, - meldung="Ohne Eingabe fortgefahren.", - aktion="ok", - kontext=None, - ) + if aktion == "kein_arbeitsblatt": + warning("Excel-Import", meldung or "") return ergebnis - if aktion == "leereingabe_nicht_erlaubt": - warning("Eingabe erforderlich", ergebnis.meldung) - return ergebnis - - if aktion == "standarddatei_vorschlagen": - if ask_yes_no("Standarddatei verwenden", ergebnis.meldung): - return pruef_ergebnis( - ok=True, - meldung="Standarddatei wird verwendet.", - aktion="ok", - kontext=kontext, - ) - return ergebnis - - if aktion == "temporaer_erlaubt": - if ask_yes_no("Temporäre Datei erzeugen", ergebnis.meldung): - return pruef_ergebnis( - ok=True, - meldung="Temporäre Datei soll erzeugt werden.", - aktion="temporaer_erzeugen", - kontext=None, - ) + if aktion in ("read_error", "open_error"): + error("Excel-Import", meldung or "") return ergebnis if aktion == "datei_nicht_gefunden": - warning("Datei nicht gefunden", ergebnis.meldung) - return ergebnis - - if aktion == "kein_dateipfad": - warning("Ungültiger Pfad", ergebnis.meldung) - return ergebnis - - if aktion == "pfad_nicht_gefunden": - warning("Pfad nicht gefunden", ergebnis.meldung) - return ergebnis - - if aktion == "url_nicht_erreichbar": - warning("URL nicht erreichbar", ergebnis.meldung) - return ergebnis - - if aktion == "netzwerkfehler": - error("Netzwerkfehler", ergebnis.meldung) - return ergebnis - - # ----------------------------------------------------- - # Layer-Aktionen - # ----------------------------------------------------- - - if aktion == "layer_nicht_gefunden": - error("Layer fehlt", ergebnis.meldung) + warning("Datei nicht gefunden", meldung or "") return ergebnis + # Spezieller Fall: layer_unsichtbar (falls nicht interaktiv behandelt) if aktion == "layer_unsichtbar": - if ask_yes_no("Layer einblenden", ergebnis.meldung): - if kontext is not None: - try: - set_layer_visible(kontext, True) - except Exception: - pass - - return pruef_ergebnis( - ok=True, - meldung="Layer wurde eingeblendet.", - aktion="ok", - kontext=kontext, - ) + 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 - if aktion == "falscher_geotyp": - warning("Falscher Geometrietyp", ergebnis.meldung) - return ergebnis - - if aktion == "layer_leer": - warning("Layer enthält keine Objekte", ergebnis.meldung) - return ergebnis - - if aktion == "falscher_layertyp": - warning("Falscher Layertyp", ergebnis.meldung) - return ergebnis - - if aktion == "falsches_crs": - warning("Falsches CRS", ergebnis.meldung) - return ergebnis - - if aktion == "felder_fehlen": - warning("Fehlende Felder", ergebnis.meldung) - return ergebnis - - if aktion == "datenquelle_unerwartet": - warning("Unerwartete Datenquelle", ergebnis.meldung) - return ergebnis - - if aktion == "layer_nicht_editierbar": - warning("Layer nicht editierbar", ergebnis.meldung) - return ergebnis - - # ----------------------------------------------------- - # Fallback - # ----------------------------------------------------- - - warning("Unbekannte Aktion", f"Unbekannte Aktion: {aktion}") + # Standard: keine Änderung return ergebnis diff --git a/modules/excel_importer.py b/modules/excel_importer.py new file mode 100644 index 0000000..50f8913 --- /dev/null +++ b/modules/excel_importer.py @@ -0,0 +1,91 @@ +# sn_plan41/modules/excel_importer.py +import os +from typing import Optional, Iterable, Mapping, Any, List, cast + +from openpyxl import load_workbook +from openpyxl.workbook.workbook import Workbook +from openpyxl.worksheet.worksheet import Worksheet + +from sn_basis.modules.Dateipruefer import Dateipruefer +from sn_basis.modules.Pruefmanager import Pruefmanager +from sn_basis.modules.pruef_ergebnis import pruef_ergebnis + + +class ExcelImporter: + """ + Excel-Importer für Linklisten, verwendet Dateipruefer und Pruefmanager zur Meldungsbehandlung. + + - Der Aufrufer übergibt einen konkreten Dateipfad. + - Vor dem Öffnen wird der Pfad mit Dateipruefer geprüft. + - Link- und Stilprüfungen erfolgen nicht hier, sondern im DataGrabber. + - Nach dem Ladevorgang wird die Arbeitsmappe geschlossen, damit die Datei vom OS freigegeben wird. + """ + + def __init__(self, filepath: str, pruefmanager: Pruefmanager): + if not filepath: + raise ValueError("ExcelImporter benötigt einen gültigen Dateipfad.") + if pruefmanager is None: + raise ValueError("ExcelImporter benötigt einen Pruefmanager.") + self.filepath = filepath + self.pruefmanager = pruefmanager + + def import_xlsx(self) -> List[Mapping[str, Any]]: + """ + Liest die Excel-Datei und gibt eine Liste von Dicts (Zeilen) zurück. + Bei Prüf- oder Leseproblemen wird der Pruefmanager zur Verarbeitung des pruef_ergebnis aufgerufen. + Im Fehlerfall wird eine leere Liste zurückgegeben. + """ + # 1) Dateiprüfung über Dateipruefer + datei_pruefer = Dateipruefer(pfad=self.filepath, temporaer_erlaubt=False) + ergebnis: pruef_ergebnis = datei_pruefer.pruefe() + ergebnis = self.pruefmanager.verarbeite(ergebnis) + + if not ergebnis.ok: + return [] + + workbook: Optional[Workbook] = None + try: + workbook = load_workbook(filename=self.filepath, data_only=True) + + # workbook.active kann typmäßig als Optional angesehen werden; cast/prüfen, damit Pylance weiß, dass sheet ein Worksheet ist + sheet = workbook.active + if sheet is None: + pe = pruef_ergebnis(ok=False, meldung=f"Kein aktives Blatt in der Arbeitsmappe: {self.filepath}", aktion="kein_arbeitsblatt", kontext=self.filepath) + self.pruefmanager.verarbeite(pe) + return [] + + # Typengranularität für den Linter + sheet = cast(Worksheet, sheet) + + # Header aus erster Zeile (als Werte) + header_row = next(sheet.iter_rows(min_row=1, max_row=1, values_only=True), None) + if not header_row: + pe = pruef_ergebnis(ok=False, meldung=f"Excel-Datei enthält keine Header-Zeile: {self.filepath}", aktion="kein_header", kontext=self.filepath) + self.pruefmanager.verarbeite(pe) + return [] + + header = list(header_row) + if not header or all(h is None for h in header): + pe = pruef_ergebnis(ok=False, meldung=f"Excel-Header ist leer oder ungültig: {self.filepath}", aktion="kein_header", kontext=self.filepath) + self.pruefmanager.verarbeite(pe) + return [] + + ergebnis_list: List[Mapping[str, Any]] = [] + # Werte-only lesen für Performance und Einfachheit + for row in sheet.iter_rows(min_row=2, values_only=True): + if row is None: + continue + # zip stoppt bei kürzerer Länge; das ist beabsichtigt + attributes = dict(zip(header, row)) + ergebnis_list.append(attributes) + + return ergebnis_list + + except Exception as exc: + pe = pruef_ergebnis(ok=False, meldung=f"Fehler beim Lesen der Excel-Datei '{self.filepath}': {exc}", aktion="read_error", kontext=self.filepath) + self.pruefmanager.verarbeite(pe) + return [] + + finally: + if workbook is not None: + workbook.close() diff --git a/modules/layerpruefer.py b/modules/layerpruefer.py index 164f9cf..3718a31 100644 --- a/modules/layerpruefer.py +++ b/modules/layerpruefer.py @@ -2,7 +2,7 @@ sn_basis/modules/layerpruefer.py – Prüfung von QGIS-Layern. Verwendet ausschließlich Wrapper und gibt pruef_ergebnis zurück. """ - +from typing import Optional, Any from sn_basis.functions import ( layer_exists, get_layer_geometry_type, @@ -26,7 +26,7 @@ class Layerpruefer: def __init__( self, - layer, + layer:Optional[Any]=None, erwarteter_geotyp: str | None = None, muss_sichtbar_sein: bool = False, erwarteter_layertyp: str | None = None, diff --git a/modules/linkpruefer.py b/modules/linkpruefer.py index 6f59306..a94e863 100644 --- a/modules/linkpruefer.py +++ b/modules/linkpruefer.py @@ -32,7 +32,7 @@ class Linkpruefer: def _pfad(self, relativer_pfad: str) -> Path: """ - Erzeugt einen OS‑unabhängigen Pfad relativ zum Basisverzeichnis. + Erzeugt einen OS-unabhängigen Pfad relativ zum Basisverzeichnis. """ if not self.basis: return Path(relativer_pfad) @@ -79,7 +79,7 @@ class Linkpruefer: def _pruefe_url(self, url: str) -> pruef_ergebnis: """ - Prüft eine URL über einen HEAD‑Request. + Prüft eine URL über einen HEAD-Request. """ reply = network_head(url) diff --git a/modules/pruef_ergebnis.py b/modules/pruef_ergebnis.py index 084f314..c7313be 100644 --- a/modules/pruef_ergebnis.py +++ b/modules/pruef_ergebnis.py @@ -1,14 +1,8 @@ -""" -sn_basis/modules/pruef_ergebnis.py – Ergebnisobjekt für alle Prüfer. -""" - +from __future__ import annotations from dataclasses import dataclass -from pathlib import Path from typing import Any, Optional, Literal - -# Alle möglichen Aktionen, die ein Prüfer auslösen kann. -# Erweiterbar ohne Umbau der Klasse. +# Erweitertes Literal mit allen erlaubten Aktionen (PruefAktion) PruefAktion = Literal[ "ok", "leer", @@ -16,34 +10,52 @@ PruefAktion = Literal[ "leereingabe_nicht_erlaubt", "standarddatei_vorschlagen", "temporaer_erlaubt", + "temporaer_erzeugen", "datei_nicht_gefunden", "kein_dateipfad", "pfad_nicht_gefunden", "url_nicht_erreichbar", "netzwerkfehler", - "falscher_layertyp", + "layer_nicht_gefunden", + "layer_unsichtbar", "falscher_geotyp", "layer_leer", + "falscher_layertyp", "falsches_crs", "felder_fehlen", "datenquelle_unerwartet", "layer_nicht_editierbar", - "temporaer_erzeugen", - "stil_nicht_anwendbar", - "layer_unsichtbar", - "layer_nicht_gefunden", - "unbekannt", - "stil_anwendbar", "falsche_endung", + # Excel / Import-spezifische Aktionen + "kein_header", + "kein_arbeitsblatt", + "read_error", + "open_error", + # Generische Prüf-/Speicher-Aktionen + "pruefe_exception", + "save_exception", + "save_not_implemented", + "stil_not_implemented", + "datei_unbekannt", + "needs_user_action", ] - - -@dataclass(slots=True) +@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) + """ ok: bool - meldung: str - aktion: PruefAktion + 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): + self.ok = ok + self.meldung = meldung + self.aktion = aktion + self.kontext = kontext