diff --git a/app/sftp/__init__.py b/app/sftp/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/sftp/worker.py b/app/sftp/worker.py new file mode 100644 index 0000000..c84ae1e --- /dev/null +++ b/app/sftp/worker.py @@ -0,0 +1,59 @@ +"""Execution asynchrone des operations SFTP. + +paramiko.SFTPClient n'est pas concu pour etre pilote depuis plusieurs threads a +la fois : ce module fait en sorte qu'un seul thread possede jamais un client SFTP +donne. Toute operation (listage, mkdir, rename, remove, chmod, stat, ...) est +soumise a la file de ce thread sous forme de callable ; le resultat revient sur +le thread appelant (GUI) via un signal Qt, qui les mets automatiquement en file +d'attente quand emetteur et recepteur vivent dans des threads differents. +""" +from __future__ import annotations + +import queue +import threading +from dataclasses import dataclass +from typing import Any, Callable + +from PySide6.QtCore import QThread, Signal + + +@dataclass +class _Job: + request_id: int + fn: Callable[[], Any] + + +class SftpWorker(QThread): + """Un worker = un canal SFTP, jamais partage avec un autre thread.""" + + result_ready = Signal(int, object, object) # request_id, resultat, exception|None + + def __init__(self, parent=None): + super().__init__(parent) + self._queue: queue.Queue[_Job | None] = queue.Queue() + self._next_id = 0 + self._lock = threading.Lock() + + def submit(self, fn: Callable[[], Any]) -> int: + """Met `fn` en file pour execution sur le thread du worker. Renvoie un + identifiant a rapprocher du signal `result_ready`.""" + with self._lock: + self._next_id += 1 + request_id = self._next_id + self._queue.put(_Job(request_id, fn)) + return request_id + + def run(self) -> None: + while True: + job = self._queue.get() + if job is None: # sentinelle posee par stop() + break + try: + result = job.fn() + self.result_ready.emit(job.request_id, result, None) + except Exception as exc: # noqa: BLE001 - remonte a l'appelant via le signal + self.result_ready.emit(job.request_id, None, exc) + + def stop(self) -> None: + """Arrete le worker une fois les jobs deja en file executes.""" + self._queue.put(None) diff --git a/app/sftp_panel.py b/app/sftp_panel.py index 474cf1e..a77dc28 100644 --- a/app/sftp_panel.py +++ b/app/sftp_panel.py @@ -19,6 +19,8 @@ from PySide6.QtWidgets import ( QAbstractItemView, QMenu, QHeaderView, ) +from app.sftp.worker import SftpWorker + MIME_TYPE = "application/x-pyssh-files" NAME_ROLE = Qt.UserRole @@ -56,16 +58,25 @@ class TransferJob: class TransferWorker(QThread): - """Execute une file de transferts sequentiellement, dans son propre thread.""" + """Execute une file de transferts sequentiellement, dans son propre thread. + + Ouvre son propre canal SFTP (via `sftp_factory`) plutot que de reutiliser celui + de la navigation : un gros transfert ne doit jamais faire attendre un listage + de dossier, et paramiko.SFTPClient n'est pas concu pour etre partage entre + threads. Quand `close_when_done` est faux (aucun canal dedie disponible), le + canal fourni est reutilise tel quel et n'est pas ferme en fin de transfert. + """ file_started = Signal(str, int, int) # nom, index, total file_progress = Signal(int, int) # octets transferes, taille totale finished_all = Signal(int, list) # nb reussis, liste d'erreurs - def __init__(self, sftp: paramiko.SFTPClient, jobs: list[TransferJob]): + def __init__(self, sftp_factory, jobs: list[TransferJob], close_when_done: bool = False): super().__init__() - self.sftp = sftp + self.sftp_factory = sftp_factory + self.sftp: paramiko.SFTPClient | None = None self.jobs = jobs + self.close_when_done = close_when_done self._cancel = False self._done = 0 self._errors: list[str] = [] @@ -74,27 +85,39 @@ class TransferWorker(QThread): self._cancel = True def run(self) -> None: - total = len(self.jobs) - for index, job in enumerate(self.jobs, start=1): - if self._cancel: - break - label = os.path.basename(job.local_path) or job.local_path - self.file_started.emit(label, index, total) - try: - if job.direction == "upload": - if job.is_dir: - self._upload_dir(job.local_path, job.remote_path) + try: + self.sftp = self.sftp_factory() + except Exception as exc: # noqa: BLE001 + self.finished_all.emit(0, [f"Impossible d'ouvrir le canal SFTP : {exc}"]) + return + try: + total = len(self.jobs) + for index, job in enumerate(self.jobs, start=1): + if self._cancel: + break + label = os.path.basename(job.local_path) or job.local_path + self.file_started.emit(label, index, total) + try: + if job.direction == "upload": + if job.is_dir: + self._upload_dir(job.local_path, job.remote_path) + else: + self.sftp.put(job.local_path, job.remote_path, callback=self._cb) else: - self.sftp.put(job.local_path, job.remote_path, callback=self._cb) - else: - if job.is_dir: - self._download_dir(job.remote_path, job.local_path) - else: - self.sftp.get(job.remote_path, job.local_path, callback=self._cb) - self._done += 1 - except Exception as exc: # noqa: BLE001 - self._errors.append(f"{label} : {exc}") - self.finished_all.emit(self._done, self._errors) + if job.is_dir: + self._download_dir(job.remote_path, job.local_path) + else: + self.sftp.get(job.remote_path, job.local_path, callback=self._cb) + self._done += 1 + except Exception as exc: # noqa: BLE001 + self._errors.append(f"{label} : {exc}") + self.finished_all.emit(self._done, self._errors) + finally: + if self.close_when_done: + try: + self.sftp.close() + except Exception: + pass def _cb(self, transferred: int, total: int) -> None: self.file_progress.emit(transferred, total) @@ -456,18 +479,32 @@ class LocalFilePanel(FilePanel): class RemoteFilePanel(FilePanel): - def __init__(self, sftp: paramiko.SFTPClient, parent=None): + """Toutes les operations reseau passent par `worker` (un `SftpWorker` dedie a ce + panneau) : aucun appel a `self.sftp` ne doit avoir lieu directement depuis le + thread GUI, pour ne jamais geler l'interface et pour respecter le fait qu'un + SFTPClient paramiko ne doit etre pilote que par un seul thread a la fois.""" + + def __init__(self, sftp: paramiko.SFTPClient, worker: SftpWorker, parent=None): super().__init__("🌐 Distant", "remote", parent) self.sftp = sftp + self.worker = worker + self.worker.result_ready.connect(self._on_result) + self._callbacks: dict[int, tuple] = {} + self._nav_generation = 0 # fichiers ouverts pour edition : chemin local temporaire -> chemin distant self._edited: dict[str, str] = {} self._watcher = QFileSystemWatcher(self) self._watcher.fileChanged.connect(self._on_edited_file_changed) - try: - home = self.sftp.normalize(".") - except Exception: - home = "/" - self.navigate(home) + + self.status.setText("Connexion...") + + def on_home_success(home) -> None: + self.navigate(home or "/") + + def on_home_error(_exc) -> None: + self.navigate("/") + + self._run(lambda: self.sftp.normalize("."), on_home_success, on_home_error) def join(self, a: str, b: str) -> str: return posixpath.join(a, b) @@ -475,16 +512,44 @@ class RemoteFilePanel(FilePanel): def parent_of(self, path: str) -> str: return posixpath.dirname(path.rstrip("/")) or "/" - def navigate(self, path: str) -> None: - try: - attrs = self.sftp.listdir_attr(path) - except IOError as exc: - QMessageBox.warning(self, "Erreur SFTP", f"{path} : {exc}") + # ---------- soumission asynchrone ---------- + def _run(self, fn, on_success=None, on_error=None) -> None: + request_id = self.worker.submit(fn) + self._callbacks[request_id] = (on_success, on_error) + + def _on_result(self, request_id: int, result, error: Exception | None) -> None: + callbacks = self._callbacks.pop(request_id, None) + if callbacks is None: return - self.current_path = path - self.path_edit.setText(path) - self._populate(attrs) - self.path_changed.emit(path) + on_success, on_error = callbacks + if error is not None: + if on_error: + on_error(error) + else: + QMessageBox.critical(self, "Erreur SFTP", str(error)) + elif on_success: + on_success(result) + + def navigate(self, path: str) -> None: + self._nav_generation += 1 + generation = self._nav_generation + self.status.setText("Chargement...") + + def on_success(attrs) -> None: + if generation != self._nav_generation: + return # une navigation plus recente a ete lancee entre-temps + self.current_path = path + self.path_edit.setText(path) + self._populate(attrs) + self.path_changed.emit(path) + + def on_error(exc) -> None: + if generation != self._nav_generation: + return + self.status.setText("") + QMessageBox.warning(self, "Erreur SFTP", f"{path} : {exc}") + + self._run(lambda: self.sftp.listdir_attr(path), on_success, on_error) def refresh(self) -> None: if self.current_path: @@ -507,12 +572,13 @@ class RemoteFilePanel(FilePanel): name, ok = QInputDialog.getText(self, "Nouveau dossier", "Nom du dossier :") if not ok or not name: return - try: - self.sftp.mkdir(self.join(self.current_path, name)) - self.refresh() - except IOError as exc: + path = self.join(self.current_path, name) + + def on_error(exc) -> None: QMessageBox.critical(self, "Erreur", str(exc)) + self._run(lambda: self.sftp.mkdir(path), lambda _r: self.refresh(), on_error) + def rename_selected(self) -> None: entries = self.selected_entries() if not entries: @@ -521,12 +587,14 @@ class RemoteFilePanel(FilePanel): new_name, ok = QInputDialog.getText(self, "Renommer", "Nouveau nom :", text=old_name) if not ok or not new_name or new_name == old_name: return - try: - self.sftp.rename(self.join(self.current_path, old_name), self.join(self.current_path, new_name)) - self.refresh() - except IOError as exc: + old_path = self.join(self.current_path, old_name) + new_path = self.join(self.current_path, new_name) + + def on_error(exc) -> None: QMessageBox.critical(self, "Erreur", str(exc)) + self._run(lambda: self.sftp.rename(old_path, new_path), lambda _r: self.refresh(), on_error) + def delete_selected(self) -> None: entries = self.selected_entries() if not entries: @@ -536,18 +604,31 @@ class RemoteFilePanel(FilePanel): f"Supprimer definitivement {len(entries)} element(s) sur le serveur ?", ) != QMessageBox.Yes: return - for name, is_dir in entries: - path = self.join(self.current_path, name) - try: - if is_dir: - self._rmdir_recursive(path) - else: - self.sftp.remove(path) - except IOError as exc: - QMessageBox.critical(self, "Erreur", f"{name} : {exc}") - self.refresh() + current_path = self.current_path + + def do_delete() -> list[str]: + errors = [] + for name, is_dir in entries: + path = self.join(current_path, name) + try: + if is_dir: + self._rmdir_recursive(path) + else: + self.sftp.remove(path) + except IOError as exc: + errors.append(f"{name} : {exc}") + return errors + + def on_success(errors: list[str]) -> None: + if errors: + QMessageBox.critical(self, "Erreur", "\n".join(errors)) + self.refresh() + + self._run(do_delete, on_success) def _rmdir_recursive(self, path: str) -> None: + """Appele uniquement depuis le thread du worker (jamais directement sur le + thread GUI) : parcourt et supprime un dossier distant recursivement.""" for attr in self.sftp.listdir_attr(path): sub = posixpath.join(path, attr.filename) if stat.S_ISDIR(attr.st_mode): @@ -570,40 +651,65 @@ class RemoteFilePanel(FilePanel): local_dir = Path(tempfile.gettempdir()) / "pyssh-edit" local_dir.mkdir(parents=True, exist_ok=True) local_path = local_dir / name - try: - self.sftp.get(remote_path, str(local_path)) - except (IOError, OSError) as exc: - QMessageBox.critical(self, "Erreur", f"Telechargement impossible : {exc}") - return - self._edited[str(local_path)] = remote_path - self._watcher.addPath(str(local_path)) - self.status.setText(f"Edition de {name} — enregistrez pour renvoyer sur le serveur") - try: - os.startfile(str(local_path)) # ouvre avec l'application associee - except OSError as exc: - QMessageBox.warning( - self, "Ouverture impossible", - f"Le fichier a ete telecharge dans :\n{local_path}\n\n" - f"Ouvrez-le manuellement ; il sera renvoye a chaque enregistrement.\n({exc})", - ) + self.status.setText(f"Telechargement de {name}...") + + def on_success(_result) -> None: + self._edited[str(local_path)] = remote_path + self._watcher.addPath(str(local_path)) + self.status.setText(f"Edition de {name} — enregistrez pour renvoyer sur le serveur") + try: + os.startfile(str(local_path)) # ouvre avec l'application associee + except OSError as exc: + QMessageBox.warning( + self, "Ouverture impossible", + f"Le fichier a ete telecharge dans :\n{local_path}\n\n" + f"Ouvrez-le manuellement ; il sera renvoye a chaque enregistrement.\n({exc})", + ) + + def on_error(exc) -> None: + self.status.setText("") + QMessageBox.critical(self, "Erreur", f"Telechargement impossible : {exc}") + + self._run(lambda: self.sftp.get(remote_path, str(local_path)), on_success, on_error) def _on_edited_file_changed(self, local_path: str) -> None: remote_path = self._edited.get(local_path) if not remote_path: return - try: - self.sftp.put(local_path, remote_path) - except (IOError, OSError) as exc: + + def on_error(exc) -> None: self.status.setText(f"Echec du renvoi : {exc}") - return - # Beaucoup d'editeurs enregistrent en remplacant le fichier : le chemin surveille - # disparait alors du watcher, il faut le reinscrire pour les enregistrements suivants. - if local_path not in self._watcher.files() and Path(local_path).exists(): - self._watcher.addPath(local_path) - self.refresh() - # apres refresh() : sinon le message serait ecrase par le compteur d'elements - self.status.setText(f"↑ {posixpath.basename(remote_path)} renvoye sur le serveur") + + def on_success(_result) -> None: + # Beaucoup d'editeurs enregistrent en remplacant le fichier : le chemin + # surveille disparait alors du watcher, il faut le reinscrire pour les + # enregistrements suivants. + if local_path not in self._watcher.files() and Path(local_path).exists(): + self._watcher.addPath(local_path) + self._refresh_after_edit(remote_path) + + self._run(lambda: self.sftp.put(local_path, remote_path), on_success, on_error) + + def _refresh_after_edit(self, remote_path: str) -> None: + """Comme refresh(), mais affiche un message de confirmation specifique une + fois le listage revenu, au lieu du compteur d'elements habituel.""" + self._nav_generation += 1 + generation = self._nav_generation + path = self.current_path + + def on_success(attrs) -> None: + if generation != self._nav_generation: + return + self._populate(attrs) + self.status.setText(f"↑ {posixpath.basename(remote_path)} renvoye sur le serveur") + + def on_error(exc) -> None: + if generation != self._nav_generation: + return + QMessageBox.warning(self, "Erreur SFTP", f"{path} : {exc}") + + self._run(lambda: self.sftp.listdir_attr(path), on_success, on_error) def change_permissions(self) -> None: entries = self.selected_entries() @@ -611,40 +717,61 @@ class RemoteFilePanel(FilePanel): return name = entries[0][0] path = self.join(self.current_path, name) - try: - current = stat.S_IMODE(self.sftp.stat(path).st_mode) - except IOError as exc: + + def on_stat_error(exc) -> None: QMessageBox.critical(self, "Erreur", str(exc)) - return - text, ok = QInputDialog.getText( - self, "Permissions", f"Droits octaux pour {name} :", text=f"{current:03o}", - ) - if not ok or not text: - return - try: - mode = int(text.strip(), 8) - except ValueError: - QMessageBox.warning(self, "Valeur invalide", "Entrez une valeur octale, par exemple 644 ou 755.") - return - errors = [] - for entry_name, _ in entries: + + def on_stat_success(attr) -> None: + current = stat.S_IMODE(attr.st_mode) + text, ok = QInputDialog.getText( + self, "Permissions", f"Droits octaux pour {name} :", text=f"{current:03o}", + ) + if not ok or not text: + return try: - self.sftp.chmod(self.join(self.current_path, entry_name), mode) - except IOError as exc: - errors.append(f"{entry_name} : {exc}") - if errors: - QMessageBox.critical(self, "Erreur", "\n".join(errors)) - self.refresh() + mode = int(text.strip(), 8) + except ValueError: + QMessageBox.warning(self, "Valeur invalide", "Entrez une valeur octale, par exemple 644 ou 755.") + return + self._apply_chmod(entries, mode) + + self._run(lambda: self.sftp.stat(path), on_stat_success, on_stat_error) + + def _apply_chmod(self, entries: list[tuple[str, bool]], mode: int) -> None: + current_path = self.current_path + + def do_chmod() -> list[str]: + errors = [] + for entry_name, _ in entries: + try: + self.sftp.chmod(self.join(current_path, entry_name), mode) + except IOError as exc: + errors.append(f"{entry_name} : {exc}") + return errors + + def on_success(errors: list[str]) -> None: + if errors: + QMessageBox.critical(self, "Erreur", "\n".join(errors)) + self.refresh() + + self._run(do_chmod, on_success) # ---------------------------------------------------------------------------- # Conteneur double panneau # ---------------------------------------------------------------------------- class SFTPPanel(QWidget): - def __init__(self, sftp: paramiko.SFTPClient, parent=None): + """`transfer_sftp_factory`, si fourni, ouvre un second canal SFTP reserve aux + transferts (upload/download), independant du canal de navigation : un gros + transfert ne bloque alors jamais le listage de dossiers. A defaut (ex. client + SFTP de test sans transport reel), le canal de navigation est reutilise tel quel.""" + + def __init__(self, sftp: paramiko.SFTPClient, transfer_sftp_factory=None, parent=None): super().__init__(parent) self.sftp = sftp self._worker: TransferWorker | None = None + self._dedicated_transfer_channel = transfer_sftp_factory is not None + self._transfer_sftp_factory = transfer_sftp_factory or (lambda: sftp) layout = QVBoxLayout(self) layout.setContentsMargins(4, 4, 4, 4) @@ -662,7 +789,9 @@ class SFTPPanel(QWidget): splitter = QSplitter(Qt.Horizontal) self.local = LocalFilePanel() - self.remote = RemoteFilePanel(sftp) + self.remote_worker = SftpWorker(self) + self.remote_worker.start() + self.remote = RemoteFilePanel(sftp, self.remote_worker) splitter.addWidget(self.local) splitter.addWidget(self.remote) splitter.setSizes([500, 500]) @@ -674,6 +803,15 @@ class SFTPPanel(QWidget): self.local.transfer_here.connect(self._on_transfer_payload) self.remote.transfer_here.connect(self._on_transfer_payload) + def shutdown(self) -> None: + """A appeler avant que le panneau ne soit detruit (fermeture de l'onglet) : + arrete proprement le worker de navigation et tout transfert en cours.""" + if self._worker and self._worker.isRunning(): + self._worker.cancel() + self._worker.wait(2000) + self.remote_worker.stop() + self.remote_worker.wait(2000) + def _on_transfer_payload(self, payload: dict) -> None: source_side = payload.get("side") entries = payload.get("entries", []) @@ -708,7 +846,9 @@ class SFTPPanel(QWidget): dialog.setAutoClose(False) dialog.setAutoReset(False) - worker = TransferWorker(self.sftp, jobs) + worker = TransferWorker( + self._transfer_sftp_factory, jobs, close_when_done=self._dedicated_transfer_channel, + ) worker.file_started.connect( lambda name, i, total: dialog.setLabelText(f"({i}/{total}) {name}") ) diff --git a/tests/test_sftp.py b/tests/test_sftp.py index df3b2d1..16032ac 100644 --- a/tests/test_sftp.py +++ b/tests/test_sftp.py @@ -70,6 +70,19 @@ class FakeSFTP: app = QApplication([]) + +def wait_idle(panel_side, timeout=5.0): + """Pompe la boucle d'evenements Qt jusqu'a ce que toutes les operations SFTP + soumises par ce panneau (navigate/refresh/etc, desormais asynchrones) soient + revenues.""" + deadline = time.time() + timeout + app.processEvents() + while panel_side._callbacks and time.time() < deadline: + app.processEvents() + time.sleep(0.01) + app.processEvents() + + remote_root = Path(tempfile.mkdtemp()) local_root = Path(tempfile.mkdtemp()) @@ -85,6 +98,7 @@ local_root = Path(tempfile.mkdtemp()) panel = SFTPPanel(FakeSFTP(remote_root)) panel.local.navigate(str(local_root)) panel.resize(1000, 600) +wait_idle(panel.remote) # laisse le temps a la navigation initiale (asynchrone) d'arriver print("=== 1. Listing distant avec colonnes ===") tree = panel.remote.tree @@ -135,11 +149,13 @@ assert got.exists() print() print("=== 5. Renommage distant ===") panel.remote.navigate("/") +wait_idle(panel.remote) item = next(panel.remote.tree.topLevelItem(i) for i in range(panel.remote.tree.topLevelItemCount()) if panel.remote.tree.topLevelItem(i).data(0, NAME_ROLE) == "config.yml") item.setSelected(True) panel.remote.sftp.rename("/config.yml", "/config_renomme.yml") panel.remote.refresh() +wait_idle(panel.remote) names = [panel.remote.tree.topLevelItem(i).data(0, NAME_ROLE) for i in range(panel.remote.tree.topLevelItemCount())] print(" entrees apres renommage:", [n for n in names if n]) @@ -155,6 +171,7 @@ print() print("=== 7. Suppression recursive distante ===") panel.remote._rmdir_recursive("/dossier_local") panel.remote.refresh() +wait_idle(panel.remote) print(" dossier supprime du serveur:", not (remote_root / "dossier_local").exists()) assert not (remote_root / "dossier_local").exists() @@ -165,5 +182,7 @@ has_parent = any(panel.remote.tree.topLevelItem(i).data(0, PARENT_ROLE) print(" ligne '..' presente:", has_parent) assert has_parent +panel.shutdown() + print() print("TOUS LES TESTS SFTP OK")