258 lines
9.0 KiB
Python
258 lines
9.0 KiB
Python
from qgis.PyQt.QtWidgets import (
|
||
QDialog,
|
||
QVBoxLayout,
|
||
QHBoxLayout,
|
||
QFormLayout,
|
||
QLabel,
|
||
QLineEdit,
|
||
QPushButton,
|
||
QMessageBox,
|
||
QSpinBox,
|
||
QComboBox,
|
||
QWidget,
|
||
)
|
||
from qgis.PyQt.QtCore import Qt
|
||
from qgis.core import QgsApplication, QgsAuthMethodConfig, QgsSettings
|
||
from .db_manager import DBManager
|
||
|
||
SETTINGS_SELECTED_CONNECTION = "000"
|
||
|
||
|
||
def try_connect_from_keychain():
|
||
"""
|
||
Tente de se connecter via une connexion PostGIS déjà enregistrée dans QGIS
|
||
(et potentiellement associée à une config d'authentification 'authcfg').
|
||
Retourne un DBManager connecté ou None.
|
||
"""
|
||
try:
|
||
settings = QgsSettings()
|
||
connection_name = settings.value(SETTINGS_SELECTED_CONNECTION, "", type=str).strip()
|
||
if not connection_name:
|
||
return None
|
||
|
||
params = read_postgis_connection(connection_name)
|
||
if not params:
|
||
return None
|
||
|
||
return connect_from_params(params)
|
||
except Exception as e:
|
||
print(f"Keychain connection failed: {e}")
|
||
return None
|
||
|
||
|
||
def parse_uri(uri_string):
|
||
"""Parse QGIS URI format: host=value port=value dbname=value..."""
|
||
params = {}
|
||
# Remplacer les espaces de séparation par du split plus robuste
|
||
parts = uri_string.split()
|
||
for part in parts:
|
||
if '=' in part:
|
||
key, value = part.split('=', 1)
|
||
params[key.strip()] = value.strip("'\"")
|
||
return params
|
||
|
||
|
||
def list_postgis_connections():
|
||
"""Retourne les noms des connexions PostgreSQL enregistrées dans QGIS."""
|
||
settings = QgsSettings()
|
||
settings.beginGroup("PostgreSQL/connections")
|
||
names = settings.childGroups()
|
||
settings.endGroup()
|
||
return sorted(names)
|
||
|
||
|
||
def read_postgis_connection(name):
|
||
"""
|
||
Lit une connexion PostGIS depuis les settings QGIS.
|
||
Retourne un dict avec host/port/dbname/user/password/authcfg (password peut être vide).
|
||
"""
|
||
settings = QgsSettings()
|
||
base = f"PostgreSQL/connections/{name}"
|
||
host = settings.value(f"{base}/host", "", type=str)
|
||
port = settings.value(f"{base}/port", "5432", type=str)
|
||
dbname = settings.value(f"{base}/database", "", type=str)
|
||
user = settings.value(f"{base}/username", "", type=str)
|
||
password = settings.value(f"{base}/password", "", type=str)
|
||
authcfg = settings.value(f"{base}/authcfg", "", type=str)
|
||
|
||
if not host and not dbname:
|
||
return None
|
||
|
||
return {
|
||
"name": name,
|
||
"host": host or "localhost",
|
||
"port": int(port) if str(port).strip() else 5432,
|
||
"dbname": dbname,
|
||
"user": user,
|
||
"password": password,
|
||
"authcfg": authcfg,
|
||
}
|
||
|
||
|
||
def load_auth_config(authcfg):
|
||
"""Charge une config d'auth QGIS et retourne un dict utile."""
|
||
if not authcfg:
|
||
return None
|
||
auth_mgr = QgsApplication.authManager()
|
||
cfg = QgsAuthMethodConfig()
|
||
auth_mgr.loadAuthenticationConfig(authcfg, cfg, True)
|
||
config_map = cfg.configMap() or {}
|
||
uri = cfg.uri() or ""
|
||
return {
|
||
"id": cfg.id(),
|
||
"name": cfg.name(),
|
||
"method": cfg.method(),
|
||
"config_map": config_map,
|
||
"uri": uri,
|
||
}
|
||
|
||
|
||
def connect_from_params(params):
|
||
"""Construit un DBManager à partir des paramètres + authcfg si présent."""
|
||
authcfg = (params.get("authcfg") or "").strip()
|
||
user = params.get("user") or ""
|
||
password = params.get("password") or ""
|
||
|
||
if authcfg:
|
||
auth = load_auth_config(authcfg)
|
||
if auth:
|
||
# Pour les configs de type Basic, on trouve souvent username/password dans configMap
|
||
user = user or auth["config_map"].get("username") or auth["config_map"].get("user") or ""
|
||
password = password or auth["config_map"].get("password") or ""
|
||
if not password and auth.get("uri"):
|
||
uri_params = parse_uri(auth["uri"])
|
||
user = user or uri_params.get("user", "")
|
||
password = password or uri_params.get("password", "")
|
||
|
||
if not params.get("dbname"):
|
||
raise ValueError("Base de données manquante dans la connexion PostGIS QGIS.")
|
||
if not user:
|
||
raise ValueError("Utilisateur manquant (connexion QGIS + authcfg n'ont pas fourni de user).")
|
||
if not password:
|
||
raise ValueError("Mot de passe manquant (connexion QGIS + authcfg n'ont pas fourni de password).")
|
||
|
||
return DBManager(
|
||
host=params.get("host") or "localhost",
|
||
port=int(params.get("port") or 5432),
|
||
dbname=params.get("dbname") or "",
|
||
user=user,
|
||
password=password,
|
||
)
|
||
|
||
class AuthDialog(QDialog):
|
||
def __init__(self, parent=None):
|
||
super().__init__(parent)
|
||
self.setWindowTitle("Connexion à la base de données – Animation")
|
||
self.setMinimumWidth(420)
|
||
self.db = None
|
||
self._build_ui()
|
||
|
||
def _build_ui(self):
|
||
layout = QVBoxLayout(self)
|
||
|
||
title = QLabel("<h2>Animation – Connexion BDD</h2>")
|
||
title.setAlignment(Qt.AlignCenter)
|
||
layout.addWidget(title)
|
||
|
||
info = QLabel(
|
||
"<small>Sélectionnez une connexion PostgreSQL déjà enregistrée dans QGIS "
|
||
"(idéalement avec une configuration d’authentification associée).</small>"
|
||
)
|
||
info.setWordWrap(True)
|
||
layout.addWidget(info)
|
||
|
||
form = QFormLayout()
|
||
self.connection_combo = QComboBox()
|
||
self.refresh_button = QPushButton("Actualiser")
|
||
refresh_layout = QHBoxLayout()
|
||
refresh_layout.addWidget(self.connection_combo, 1)
|
||
refresh_layout.addWidget(self.refresh_button)
|
||
refresh_widget = QWidget()
|
||
refresh_widget.setLayout(refresh_layout)
|
||
form.addRow("Connexion PostGIS :", refresh_widget)
|
||
|
||
self.manual_host_edit = QLineEdit()
|
||
self.manual_port_spin = QSpinBox()
|
||
self.manual_port_spin.setRange(1, 65535)
|
||
self.manual_port_spin.setValue(5432)
|
||
self.manual_db_edit = QLineEdit()
|
||
self.manual_user_edit = QLineEdit()
|
||
self.manual_password_edit = QLineEdit()
|
||
self.manual_password_edit.setEchoMode(QLineEdit.Password)
|
||
|
||
form.addRow(QLabel("<b>Fallback manuel</b> (si la connexion QGIS ne contient pas le secret)"), QLabel(""))
|
||
form.addRow("Hôte :", self.manual_host_edit)
|
||
form.addRow("Port :", self.manual_port_spin)
|
||
form.addRow("Base de données :", self.manual_db_edit)
|
||
form.addRow("Utilisateur :", self.manual_user_edit)
|
||
form.addRow("Mot de passe :", self.manual_password_edit)
|
||
|
||
layout.addLayout(form)
|
||
|
||
btn_layout = QHBoxLayout()
|
||
self.btn_cancel = QPushButton("Annuler")
|
||
self.btn_cancel.clicked.connect(self.reject)
|
||
self.btn_connect = QPushButton("Se connecter")
|
||
self.btn_connect.setDefault(True)
|
||
self.btn_connect.clicked.connect(self.connect)
|
||
btn_layout.addWidget(self.btn_cancel)
|
||
btn_layout.addWidget(self.btn_connect)
|
||
layout.addLayout(btn_layout)
|
||
|
||
self.refresh_button.clicked.connect(self.refresh_connections)
|
||
self.connection_combo.currentIndexChanged.connect(self._on_connection_changed)
|
||
self.refresh_connections()
|
||
|
||
def connect(self):
|
||
try:
|
||
params = self._params_from_selection()
|
||
if params:
|
||
self.db = connect_from_params(params)
|
||
QgsSettings().setValue(SETTINGS_SELECTED_CONNECTION, params.get("name", ""))
|
||
self.accept()
|
||
return
|
||
|
||
# fallback manuel
|
||
self.db = DBManager(
|
||
host=self.manual_host_edit.text().strip() or "localhost",
|
||
port=self.manual_port_spin.value(),
|
||
dbname=self.manual_db_edit.text().strip(),
|
||
user=self.manual_user_edit.text().strip(),
|
||
password=self.manual_password_edit.text(),
|
||
)
|
||
self.accept()
|
||
except Exception as e:
|
||
QMessageBox.critical(self, "Erreur de connexion", str(e))
|
||
|
||
def refresh_connections(self):
|
||
self.connection_combo.clear()
|
||
self.connection_combo.addItem("", "")
|
||
for name in list_postgis_connections():
|
||
self.connection_combo.addItem(name, name)
|
||
|
||
# Pré-sélection du dernier choix
|
||
last = QgsSettings().value(SETTINGS_SELECTED_CONNECTION, "", type=str)
|
||
if last:
|
||
idx = self.connection_combo.findData(last)
|
||
if idx >= 0:
|
||
self.connection_combo.setCurrentIndex(idx)
|
||
else:
|
||
self._on_connection_changed()
|
||
|
||
def _on_connection_changed(self):
|
||
params = self._params_from_selection()
|
||
if not params:
|
||
return
|
||
# Pré-remplir le fallback manuel avec les valeurs non sensibles
|
||
self.manual_host_edit.setText(params.get("host", "") or "")
|
||
self.manual_port_spin.setValue(int(params.get("port") or 5432))
|
||
self.manual_db_edit.setText(params.get("dbname", "") or "")
|
||
self.manual_user_edit.setText(params.get("user", "") or "")
|
||
self.manual_password_edit.clear()
|
||
|
||
def _params_from_selection(self):
|
||
name = (self.connection_combo.currentData() or "").strip()
|
||
if not name:
|
||
return None
|
||
return read_postgis_connection(name)
|