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("

Animation – Connexion BDD

") title.setAlignment(Qt.AlignCenter) layout.addWidget(title) info = QLabel( "Sélectionnez une connexion PostgreSQL déjà enregistrée dans QGIS " "(idéalement avec une configuration d’authentification associée)." ) 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("Fallback manuel (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)