"""Client SFTP integre facon WinSCP : double panneau, colonnes triables, glisser-deposer, renommage, permissions, et file de transferts avec progression globale.""" from __future__ import annotations import json import os import posixpath import stat from datetime import datetime from pathlib import Path import paramiko from PySide6.QtCore import Qt, QThread, Signal, QMimeData, QPoint from PySide6.QtGui import QDrag, QKeySequence, QShortcut from PySide6.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QLineEdit, QToolButton, QTreeWidget, QTreeWidgetItem, QSplitter, QLabel, QMessageBox, QInputDialog, QProgressDialog, QAbstractItemView, QMenu, QHeaderView, ) MIME_TYPE = "application/x-pyssh-files" NAME_ROLE = Qt.UserRole ISDIR_ROLE = Qt.UserRole + 1 SIZE_ROLE = Qt.UserRole + 2 MTIME_ROLE = Qt.UserRole + 3 PARENT_ROLE = Qt.UserRole + 4 # marque l'entree ".." def human_size(n: int) -> str: value = float(n) for unit in ("o", "Ko", "Mo", "Go", "To"): if value < 1024: return f"{value:.0f} {unit}" if unit == "o" else f"{value:.1f} {unit}" value /= 1024 return f"{value:.1f} Po" def format_time(ts: float) -> str: try: return datetime.fromtimestamp(ts).strftime("%d/%m/%Y %H:%M") except (OSError, ValueError, OverflowError): return "" # ---------------------------------------------------------------------------- # Transferts # ---------------------------------------------------------------------------- class TransferJob: def __init__(self, direction: str, local_path: str, remote_path: str, is_dir: bool): self.direction = direction # "upload" | "download" self.local_path = local_path self.remote_path = remote_path self.is_dir = is_dir class TransferWorker(QThread): """Execute une file de transferts sequentiellement, dans son propre thread.""" 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]): super().__init__() self.sftp = sftp self.jobs = jobs self._cancel = False self._done = 0 self._errors: list[str] = [] def cancel(self) -> None: 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) 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) def _cb(self, transferred: int, total: int) -> None: self.file_progress.emit(transferred, total) def _upload_dir(self, local_dir: str, remote_dir: str) -> None: try: self.sftp.mkdir(remote_dir) except IOError: pass for entry in os.listdir(local_dir): if self._cancel: return lp = os.path.join(local_dir, entry) rp = posixpath.join(remote_dir, entry) if os.path.isdir(lp): self._upload_dir(lp, rp) else: self.sftp.put(lp, rp, callback=self._cb) def _download_dir(self, remote_dir: str, local_dir: str) -> None: Path(local_dir).mkdir(parents=True, exist_ok=True) for attr in self.sftp.listdir_attr(remote_dir): if self._cancel: return rp = posixpath.join(remote_dir, attr.filename) lp = os.path.join(local_dir, attr.filename) if stat.S_ISDIR(attr.st_mode): self._download_dir(rp, lp) else: self.sftp.get(rp, lp, callback=self._cb) # ---------------------------------------------------------------------------- # Panneaux de fichiers # ---------------------------------------------------------------------------- class FileItem(QTreeWidgetItem): """Ligne de fichier avec le tri attendu d'un gestionnaire de fichiers : ".." toujours en tete, puis les dossiers, puis les fichiers ; taille et date comparees numeriquement (et non comme du texte).""" SORT_VALUE_ROLE = Qt.UserRole + 10 def __lt__(self, other: QTreeWidgetItem) -> bool: tree = self.treeWidget() column = tree.sortColumn() if tree else 0 descending = bool( tree and tree.header().sortIndicatorOrder() == Qt.DescendingOrder ) def pinned(result: bool) -> bool: # Qt inverse le resultat en tri descendant : on le compense pour que # ".." et les dossiers gardent leur place quel que soit le sens du tri. return not result if descending else result self_parent = bool(self.data(0, PARENT_ROLE)) other_parent = bool(other.data(0, PARENT_ROLE)) if self_parent != other_parent: return pinned(self_parent) self_dir = bool(self.data(0, ISDIR_ROLE)) other_dir = bool(other.data(0, ISDIR_ROLE)) if self_dir != other_dir: return pinned(self_dir) self_value = self.data(column, self.SORT_VALUE_ROLE) other_value = other.data(column, self.SORT_VALUE_ROLE) if self_value is not None and other_value is not None: return self_value < other_value return self.text(column).lower() < other.text(column).lower() class FileTree(QTreeWidget): """Arbre de fichiers avec glisser-deposer entre panneaux.""" files_dropped = Signal(dict) def __init__(self, side: str, parent=None): super().__init__(parent) self.side = side self.setSelectionMode(QAbstractItemView.ExtendedSelection) self.setRootIsDecorated(False) self.setSortingEnabled(True) self.setDragEnabled(True) self.setAcceptDrops(True) self.setDropIndicatorShown(True) self.setDragDropMode(QAbstractItemView.DragDrop) self.setHeaderLabels(["Nom", "Taille", "Modifie", "Droits"]) # sans tri initial explicite, les lignes resteraient dans l'ordre d'insertion self.sortByColumn(0, Qt.AscendingOrder) self.header().setSectionResizeMode(0, QHeaderView.Stretch) self.setColumnWidth(1, 90) self.setColumnWidth(2, 130) self.setColumnWidth(3, 100) def startDrag(self, actions) -> None: entries = [] for item in self.selectedItems(): if item.data(0, PARENT_ROLE): continue entries.append({"name": item.data(0, NAME_ROLE), "is_dir": bool(item.data(0, ISDIR_ROLE))}) if not entries: return payload = {"side": self.side, "entries": entries} mime = QMimeData() mime.setData(MIME_TYPE, json.dumps(payload).encode("utf-8")) drag = QDrag(self) drag.setMimeData(mime) drag.exec(Qt.CopyAction) def dragEnterEvent(self, event) -> None: if event.mimeData().hasFormat(MIME_TYPE): event.acceptProposedAction() def dragMoveEvent(self, event) -> None: if event.mimeData().hasFormat(MIME_TYPE): event.acceptProposedAction() def dropEvent(self, event) -> None: if not event.mimeData().hasFormat(MIME_TYPE): return payload = json.loads(bytes(event.mimeData().data(MIME_TYPE)).decode("utf-8")) if payload.get("side") == self.side: return # depose sur lui-meme : rien a faire event.acceptProposedAction() self.files_dropped.emit(payload) class FilePanel(QWidget): """Panneau generique (decline en local et distant).""" path_changed = Signal(str) transfer_here = Signal(dict) def __init__(self, title: str, side: str, parent=None): super().__init__(parent) self.side = side layout = QVBoxLayout(self) layout.setContentsMargins(4, 4, 4, 4) layout.setSpacing(4) header = QHBoxLayout() self.title_label = QLabel(f"{title}") header.addWidget(self.title_label) header.addStretch() layout.addLayout(header) nav = QHBoxLayout() self.up_btn = QToolButton(); self.up_btn.setText("⬆"); self.up_btn.setToolTip("Dossier parent") self.refresh_btn = QToolButton(); self.refresh_btn.setText("⟳"); self.refresh_btn.setToolTip("Actualiser (F5)") self.mkdir_btn = QToolButton(); self.mkdir_btn.setText("📁+"); self.mkdir_btn.setToolTip("Nouveau dossier") self.path_edit = QLineEdit() nav.addWidget(self.up_btn) nav.addWidget(self.refresh_btn) nav.addWidget(self.mkdir_btn) nav.addWidget(self.path_edit, 1) layout.addLayout(nav) self.tree = FileTree(side) layout.addWidget(self.tree) self.status = QLabel("") self.status.setObjectName("statusLabel") layout.addWidget(self.status) self.up_btn.clicked.connect(self.go_up) self.refresh_btn.clicked.connect(self.refresh) self.mkdir_btn.clicked.connect(self.make_dir) self.path_edit.returnPressed.connect(lambda: self.navigate(self.path_edit.text())) self.tree.itemDoubleClicked.connect(self._on_double_click) self.tree.files_dropped.connect(self.transfer_here.emit) self.tree.setContextMenuPolicy(Qt.CustomContextMenu) self.tree.customContextMenuRequested.connect(self._context_menu) QShortcut(QKeySequence("F2"), self.tree, activated=self.rename_selected) QShortcut(QKeySequence("F5"), self.tree, activated=self.refresh) QShortcut(QKeySequence(Qt.Key_Delete), self.tree, activated=self.delete_selected) self.current_path = "" # ---------- helpers d'affichage ---------- def _add_row(self, name: str, is_dir: bool, size: int, mtime: float, perms: str) -> None: item = FileItem([ ("📁 " if is_dir else "📄 ") + name, "" if is_dir else human_size(size), format_time(mtime) if mtime else "", perms, ]) item.setData(0, NAME_ROLE, name) item.setData(0, ISDIR_ROLE, is_dir) # tri numerique correct sur la taille et la date item.setData(1, FileItem.SORT_VALUE_ROLE, size) item.setData(2, FileItem.SORT_VALUE_ROLE, mtime) self.tree.addTopLevelItem(item) def _add_parent_row(self) -> None: item = FileItem(["📁 ..", "", "", ""]) item.setData(0, PARENT_ROLE, True) item.setData(0, ISDIR_ROLE, True) self.tree.addTopLevelItem(item) def _on_double_click(self, item: QTreeWidgetItem) -> None: if item.data(0, PARENT_ROLE): self.go_up() return if item.data(0, ISDIR_ROLE): self.navigate(self.join(self.current_path, item.data(0, NAME_ROLE))) def go_up(self) -> None: parent = self.parent_of(self.current_path) if parent and parent != self.current_path: self.navigate(parent) def selected_entries(self) -> list[tuple[str, bool]]: out = [] for item in self.tree.selectedItems(): if item.data(0, PARENT_ROLE): continue out.append((item.data(0, NAME_ROLE), bool(item.data(0, ISDIR_ROLE)))) return out def _context_menu(self, pos: QPoint) -> None: menu = QMenu(self) has_selection = bool(self.selected_entries()) act_transfer = menu.addAction("⬆ Envoyer" if self.side == "local" else "⬇ Recevoir") act_transfer.setEnabled(has_selection) menu.addSeparator() act_rename = menu.addAction("Renommer (F2)") act_rename.setEnabled(has_selection) act_delete = menu.addAction("Supprimer (Suppr)") act_delete.setEnabled(has_selection) act_perms = None if self.side == "remote": menu.addSeparator() act_perms = menu.addAction("Permissions...") act_perms.setEnabled(has_selection) menu.addSeparator() act_mkdir = menu.addAction("Nouveau dossier") act_refresh = menu.addAction("Actualiser (F5)") action = menu.exec(self.tree.viewport().mapToGlobal(pos)) if action == act_transfer: self.transfer_requested() elif action == act_rename: self.rename_selected() elif action == act_delete: self.delete_selected() elif act_perms is not None and action == act_perms: self.change_permissions() elif action == act_mkdir: self.make_dir() elif action == act_refresh: self.refresh() def transfer_requested(self) -> None: entries = [{"name": n, "is_dir": d} for n, d in self.selected_entries()] if entries: self.transfer_here.emit({"side": self.side, "entries": entries, "_from_button": True}) # ---------- a surcharger ---------- def navigate(self, path: str) -> None: ... def refresh(self) -> None: ... def join(self, a: str, b: str) -> str: ... def parent_of(self, path: str) -> str: ... def make_dir(self) -> None: ... def rename_selected(self) -> None: ... def delete_selected(self) -> None: ... def change_permissions(self) -> None: ... class LocalFilePanel(FilePanel): def __init__(self, parent=None): super().__init__("💻 Local", "local", parent) self.navigate(str(Path.home())) def join(self, a: str, b: str) -> str: return str(Path(a) / b) def parent_of(self, path: str) -> str: return str(Path(path).parent) def navigate(self, path: str) -> None: p = Path(path) if not p.exists() or not p.is_dir(): QMessageBox.warning(self, "Chemin invalide", f"Dossier introuvable : {path}") return self.current_path = str(p) self.path_edit.setText(self.current_path) self.refresh() self.path_changed.emit(self.current_path) def refresh(self) -> None: self.tree.setSortingEnabled(False) self.tree.clear() self._add_parent_row() count = 0 try: for entry in Path(self.current_path).iterdir(): try: st = entry.stat() is_dir = entry.is_dir() self._add_row(entry.name, is_dir, st.st_size, st.st_mtime, stat.filemode(st.st_mode)) count += 1 except OSError: continue except OSError as exc: QMessageBox.warning(self, "Erreur", str(exc)) self.tree.setSortingEnabled(True) self.status.setText(f"{count} element(s)") def make_dir(self) -> None: name, ok = QInputDialog.getText(self, "Nouveau dossier", "Nom du dossier :") if not ok or not name: return try: Path(self.current_path, name).mkdir() self.refresh() except OSError as exc: QMessageBox.critical(self, "Erreur", str(exc)) def rename_selected(self) -> None: entries = self.selected_entries() if not entries: return old_name = entries[0][0] 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: Path(self.current_path, old_name).rename(Path(self.current_path, new_name)) self.refresh() except OSError as exc: QMessageBox.critical(self, "Erreur", str(exc)) def delete_selected(self) -> None: entries = self.selected_entries() if not entries: return if QMessageBox.question( self, "Supprimer", f"Supprimer definitivement {len(entries)} element(s) en local ?", ) != QMessageBox.Yes: return import shutil for name, is_dir in entries: target = Path(self.current_path, name) try: if is_dir: shutil.rmtree(target) else: target.unlink() except OSError as exc: QMessageBox.critical(self, "Erreur", f"{name} : {exc}") self.refresh() class RemoteFilePanel(FilePanel): def __init__(self, sftp: paramiko.SFTPClient, parent=None): super().__init__("🌐 Distant", "remote", parent) self.sftp = sftp try: home = self.sftp.normalize(".") except Exception: home = "/" self.navigate(home) def join(self, a: str, b: str) -> str: return posixpath.join(a, b) 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}") return self.current_path = path self.path_edit.setText(path) self._populate(attrs) self.path_changed.emit(path) def refresh(self) -> None: if self.current_path: self.navigate(self.current_path) def _populate(self, attrs) -> None: self.tree.setSortingEnabled(False) self.tree.clear() self._add_parent_row() for attr in attrs: is_dir = stat.S_ISDIR(attr.st_mode) self._add_row( attr.filename, is_dir, attr.st_size or 0, attr.st_mtime or 0, stat.filemode(attr.st_mode), ) self.tree.setSortingEnabled(True) self.status.setText(f"{len(attrs)} element(s)") def make_dir(self) -> None: 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: QMessageBox.critical(self, "Erreur", str(exc)) def rename_selected(self) -> None: entries = self.selected_entries() if not entries: return old_name = entries[0][0] 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: QMessageBox.critical(self, "Erreur", str(exc)) def delete_selected(self) -> None: entries = self.selected_entries() if not entries: return if QMessageBox.question( self, "Supprimer", 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() def _rmdir_recursive(self, path: str) -> None: for attr in self.sftp.listdir_attr(path): sub = posixpath.join(path, attr.filename) if stat.S_ISDIR(attr.st_mode): self._rmdir_recursive(sub) else: self.sftp.remove(sub) self.sftp.rmdir(path) def change_permissions(self) -> None: entries = self.selected_entries() if not entries: 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: 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: 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() # ---------------------------------------------------------------------------- # Conteneur double panneau # ---------------------------------------------------------------------------- class SFTPPanel(QWidget): def __init__(self, sftp: paramiko.SFTPClient, parent=None): super().__init__(parent) self.sftp = sftp self._worker: TransferWorker | None = None layout = QVBoxLayout(self) layout.setContentsMargins(4, 4, 4, 4) toolbar = QHBoxLayout() self.upload_btn = QToolButton(); self.upload_btn.setText("⬆ Envoyer") self.download_btn = QToolButton(); self.download_btn.setText("⬇ Recevoir") hint = QLabel(" Glissez-deposez entre les panneaux, ou utilisez F2 (renommer), Suppr, F5.") hint.setObjectName("statusLabel") toolbar.addWidget(self.upload_btn) toolbar.addWidget(self.download_btn) toolbar.addWidget(hint) toolbar.addStretch() layout.addLayout(toolbar) splitter = QSplitter(Qt.Horizontal) self.local = LocalFilePanel() self.remote = RemoteFilePanel(sftp) splitter.addWidget(self.local) splitter.addWidget(self.remote) splitter.setSizes([500, 500]) layout.addWidget(splitter, 1) self.upload_btn.clicked.connect(self.local.transfer_requested) self.download_btn.clicked.connect(self.remote.transfer_requested) # un depot venant du panneau local signifie "envoyer", et inversement self.local.transfer_here.connect(self._on_transfer_payload) self.remote.transfer_here.connect(self._on_transfer_payload) def _on_transfer_payload(self, payload: dict) -> None: source_side = payload.get("side") entries = payload.get("entries", []) if not entries: return jobs = [] for entry in entries: name, is_dir = entry["name"], entry["is_dir"] if source_side == "local": jobs.append(TransferJob( "upload", self.local.join(self.local.current_path, name), self.remote.join(self.remote.current_path, name), is_dir, )) else: jobs.append(TransferJob( "download", self.local.join(self.local.current_path, name), self.remote.join(self.remote.current_path, name), is_dir, )) self._run_jobs(jobs) def _run_jobs(self, jobs: list[TransferJob]) -> None: if not jobs: return if self._worker and self._worker.isRunning(): QMessageBox.information(self, "Transfert en cours", "Attendez la fin du transfert en cours.") return dialog = QProgressDialog("Preparation...", "Annuler", 0, 100, self) dialog.setWindowTitle("Transfert SFTP") dialog.setWindowModality(Qt.WindowModal) dialog.setMinimumDuration(0) dialog.setAutoClose(False) dialog.setAutoReset(False) worker = TransferWorker(self.sftp, jobs) worker.file_started.connect( lambda name, i, total: dialog.setLabelText(f"({i}/{total}) {name}") ) worker.file_progress.connect( lambda done, total: dialog.setValue(int(done * 100 / total) if total else 0) ) def on_finished(done: int, errors: list) -> None: dialog.close() self.local.refresh() self.remote.refresh() if errors: QMessageBox.warning( self, "Transfert termine avec des erreurs", f"{done} element(s) transfere(s).\n\nErreurs :\n" + "\n".join(errors[:10]), ) worker.finished_all.connect(on_finished) dialog.canceled.connect(worker.cancel) self._worker = worker worker.start()