Daniel/Plugin_SN_Basis
1
0
Fork 0
forked from AG_QGIS/Plugin_SN_Basis
Code Pull Requests Activity

Angefangen, DataGrabber anzulegen (Grundlagen gelegt, noch nicht lauffähig)

Browse Source
This commit is contained in:
Daniel Helbig
2026-02-13 21:39:12 +01:00
parent 039c614592
commit e6ffab1c10
12 changed files with 733 additions and 150 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
assets/datagrabber.jpeg Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 100 KiB

38
assets/datagrabber.md Normal file
View File

@@ -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
```

BIN
assets/datagrabber.pdf Normal file
View File

Binary file not shown.

68
functions/qt_wrapper.py
View File

@@ -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

14
functions/test.md Normal file
View File

@@ -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

324
modules/DataGrabber.py Normal file
View File

@@ -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

4
modules/Dateipruefer.py
View File

@@ -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,

282
modules/Pruefmanager.py
View File

@@ -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

91
modules/excel_importer.py Normal file
View File

@@ -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()

4
modules/layerpruefer.py
View File

@@ -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,

4
modules/linkpruefer.py
View File

@@ -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)

54
modules/pruef_ergebnis.py
View File

@@ -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