forked from AG_QGIS/Plugin_SN_Basis
78 lines
2.5 KiB
Python
78 lines
2.5 KiB
Python
# Linkpruefer.py – Qt5/Qt6-kompatibel über qt_compat
|
||
|
||
from modules.qt_compat import (
|
||
QEventLoop,
|
||
QUrl,
|
||
QNetworkRequest,
|
||
QNetworkReply
|
||
)
|
||
|
||
from qgis.core import QgsNetworkAccessManager
|
||
from modules.pruef_ergebnis import PruefErgebnis
|
||
|
||
|
||
class Linkpruefer:
|
||
"""Prüft den Link mit QgsNetworkAccessManager und klassifiziert Anbieter nach Attribut."""
|
||
|
||
ANBIETER_TYPEN: dict[str, str] = {
|
||
"REST": "REST",
|
||
"WFS": "WFS",
|
||
"WMS": "WMS",
|
||
"OGR": "OGR"
|
||
}
|
||
|
||
def __init__(self, link: str, anbieter: str):
|
||
self.link = link
|
||
self.anbieter = anbieter.upper().strip() if anbieter else ""
|
||
self.network_manager = QgsNetworkAccessManager()
|
||
|
||
def klassifiziere_anbieter(self):
|
||
typ = self.ANBIETER_TYPEN.get(self.anbieter, self.anbieter)
|
||
quelle = "remote" if self.link.startswith(("http://", "https://")) else "local"
|
||
return {"typ": typ, "quelle": quelle}
|
||
|
||
def pruefe_link(self):
|
||
fehler = []
|
||
warnungen = []
|
||
|
||
if not self.link:
|
||
fehler.append("Link fehlt.")
|
||
return PruefErgebnis(False, fehler=fehler, warnungen=warnungen)
|
||
|
||
if not self.anbieter or not self.anbieter.strip():
|
||
fehler.append("Anbieter muss gesetzt werden und darf nicht leer sein.")
|
||
|
||
# Remote-Links prüfen
|
||
if self.link.startswith(("http://", "https://")):
|
||
request = QNetworkRequest(QUrl(self.link))
|
||
reply = self.network_manager.head(request)
|
||
|
||
loop = QEventLoop()
|
||
reply.finished.connect(loop.quit)
|
||
loop.exec() # Qt5/Qt6-kompatibel über qt_compat
|
||
|
||
# Fehlerprüfung Qt5/Qt6-kompatibel
|
||
if reply.error() != QNetworkReply.NetworkError.NoError:
|
||
fehler.append(f"Verbindungsfehler: {reply.errorString()}")
|
||
else:
|
||
status = reply.attribute(QNetworkRequest.HttpStatusCodeAttribute)
|
||
if status is None or status < 200 or status >= 400:
|
||
fehler.append(f"Link nicht erreichbar: HTTP {status}")
|
||
|
||
reply.deleteLater()
|
||
|
||
else:
|
||
# Lokale Pfade: Plausibilitätscheck
|
||
if "." not in self.link.split("/")[-1]:
|
||
warnungen.append("Der lokale Link sieht ungewöhnlich aus.")
|
||
|
||
return PruefErgebnis(
|
||
len(fehler) == 0,
|
||
daten=self.klassifiziere_anbieter(),
|
||
fehler=fehler,
|
||
warnungen=warnungen
|
||
)
|
||
|
||
def ausfuehren(self):
|
||
return self.pruefe_link()
|