pysftp.py

# file: sftp_gui.py
"""
PyQt SFTP Front-End (OpenSSH sftp)

Features
- Connect using OpenSSH `sftp` (key auth recommended)
- Local file browser (native filesystem model)
- Remote directory listing + navigation
- Upload/download with queued transfers
- Logs and basic progress tracking via QProcess output

Notes
- Password auth is intentionally not implemented because OpenSSH sftp is interactive and
  not safely automatable without external helpers; use key auth or ssh-agent.
"""

from __future__ import annotations

import os
import re
import shlex
import sys
import tempfile
from dataclasses import dataclass
from pathlib import Path
from typing import Optional, List, Tuple

from PyQt6.QtCore import Qt, QProcess, QModelIndex, pyqtSignal, QObject
from PyQt6.QtGui import (
    QAction,
    QFileSystemModel
)
from PyQt6.QtWidgets import (
    QApplication,
    QComboBox,
    QFileDialog,
    QFormLayout,
    QHBoxLayout,
    QLabel,
    QLineEdit,
    QMainWindow,
    QMessageBox,
    QPushButton,
    QSplitter,
    QTableWidget,
    QTableWidgetItem,
    QTreeWidget,
    QTreeWidgetItem,
    QVBoxLayout,
    QWidget,
    QPlainTextEdit,
    QHeaderView,
)


@dataclass(frozen=True)
class SftpConnection:
    host: str
    port: int
    username: str
    key_path: Optional[str] = None

    def target(self) -> str:
        return f"{self.username}@{self.host}"


@dataclass
class RemoteEntry:
    name: str
    is_dir: bool
    size: Optional[int] = None
    raw: str = ""


class LogBus(QObject):
    message = pyqtSignal(str)


class SftpCommandRunner(QObject):
    """
    Runs OpenSSH `sftp` commands in batch mode, producing stdout/stderr lines.
    """

    finished = pyqtSignal(int, str, str)  # exit_code, stdout, stderr
    line = pyqtSignal(str)
    error = pyqtSignal(str)

    def __init__(self, conn: SftpConnection, parent: Optional[QObject] = None):
        super().__init__(parent)
        self._conn = conn
        self._proc: Optional[QProcess] = None
        self._stdout: List[str] = []
        self._stderr: List[str] = []

    def is_running(self) -> bool:
        return self._proc is not None and self._proc.state() != QProcess.ProcessState.NotRunning

    def run_batch(self, commands: List[str], verbose: bool = False) -> None:
        if self.is_running():
            self.error.emit("Process already running.")
            return

        batch_fd, batch_path = tempfile.mkstemp(prefix="sftp_batch_", suffix=".txt")
        os.close(batch_fd)
        Path(batch_path).write_text("\n".join(commands) + "\n", encoding="utf-8")

        args = []
        if verbose:
            args.append("-v")
        args += ["-b", batch_path, "-P", str(self._conn.port)]
        if self._conn.key_path:
            args += ["-i", self._conn.key_path]
        args.append(self._conn.target())

        self._proc = QProcess(self)
        self._proc.setProgram("sftp")
        self._proc.setArguments(args)
        self._proc.setProcessChannelMode(QProcess.ProcessChannelMode.SeparateChannels)

        self._stdout.clear()
        self._stderr.clear()

        self._proc.readyReadStandardOutput.connect(self._on_stdout)
        self._proc.readyReadStandardError.connect(self._on_stderr)
        self._proc.finished.connect(lambda code, status: self._on_finished(code, status, batch_path))

        self._proc.start()

        if not self._proc.waitForStarted(3000):
            self.error.emit("Failed to start `sftp`. Is OpenSSH installed and in PATH?")
            self._cleanup(batch_path)

    def _cleanup(self, batch_path: str) -> None:
        try:
            Path(batch_path).unlink(missing_ok=True)
        except Exception:
            pass

    def _on_stdout(self) -> None:
        if not self._proc:
            return
        data = bytes(self._proc.readAllStandardOutput()).decode(errors="replace")
        for ln in data.splitlines():
            self._stdout.append(ln)
            self.line.emit(ln)

    def _on_stderr(self) -> None:
        if not self._proc:
            return
        data = bytes(self._proc.readAllStandardError()).decode(errors="replace")
        for ln in data.splitlines():
            self._stderr.append(ln)
            self.line.emit(ln)

    def _on_finished(self, exit_code: int, _status: QProcess.ExitStatus, batch_path: str) -> None:
        out = "\n".join(self._stdout)
        err = "\n".join(self._stderr)
        self._cleanup(batch_path)
        self._proc = None
        self.finished.emit(exit_code, out, err)


class SftpService(QObject):
    """
    High-level SFTP operations: list, cd/pwd, upload/download via batch commands.
    """

    listing_ready = pyqtSignal(str, list)  # remote_cwd, entries[RemoteEntry]
    transfer_progress = pyqtSignal(str, int)  # transfer_id, percent
    transfer_finished = pyqtSignal(str, bool, str)  # transfer_id, ok, message
    log = pyqtSignal(str)

    def __init__(self, parent: Optional[QObject] = None):
        super().__init__(parent)
        self._conn: Optional[SftpConnection] = None
        self._cwd: str = "."
        self._runner: Optional[SftpCommandRunner] = None

    def set_connection(self, conn: SftpConnection) -> None:
        self._conn = conn
        self._cwd = "."
        self._runner = SftpCommandRunner(conn, self)
        self._runner.line.connect(self.log.emit)
        self._runner.error.connect(self.log.emit)

    def connected(self) -> bool:
        return self._conn is not None

    def refresh_listing(self) -> None:
        if not self._runner:
            return
        cmds = [f"cd {shlex.quote(self._cwd)}", "pwd", "ls -la"]
        self._runner.finished.connect(self._on_list_finished)
        self._runner.run_batch(cmds, verbose=False)

    def cd(self, remote_path: str) -> None:
        self._cwd = remote_path
        self.refresh_listing()

    def mkdir(self, name: str) -> None:
        if not self._runner:
            return
        cmds = [f"cd {shlex.quote(self._cwd)}", f"mkdir {shlex.quote(name)}", "ls -la"]
        self._runner.finished.connect(self._on_list_finished)
        self._runner.run_batch(cmds)

    def upload(self, local_path: str, remote_dir: str, transfer_id: str) -> None:
        self._transfer(
            transfer_id=transfer_id,
            commands=[
                f"cd {shlex.quote(remote_dir)}",
                f'put "{local_path}"',
            ],
        )

    def download(self, remote_path: str, local_dir: str, remote_dir: str, transfer_id: str) -> None:
        self._transfer(
            transfer_id=transfer_id,
            commands=[
                f"cd {shlex.quote(remote_dir)}",
                f'get "{remote_path}" "{local_dir}"',
            ],
        )

    def _transfer(self, transfer_id: str, commands: List[str]) -> None:
        if not self._runner:
            self.transfer_finished.emit(transfer_id, False, "Not connected.")
            return

        runner = SftpCommandRunner(self._conn, self)  # dedicated runner for transfer
        stdout_lines: List[str] = []
        stderr_lines: List[str] = []

        # Heuristic progress: OpenSSH sftp sometimes emits lines with percent like "100%"
        percent_re = re.compile(r"(\d{1,3})%")

        def on_line(ln: str) -> None:
            self.log.emit(ln)
            m = percent_re.search(ln)
            if m:
                pct = max(0, min(100, int(m.group(1))))
                self.transfer_progress.emit(transfer_id, pct)

        def on_finished(code: int, out: str, err: str) -> None:
            stdout_lines.append(out)
            stderr_lines.append(err)
            ok = (code == 0) and ("Failure" not in err)
            msg = "OK" if ok else (err.strip() or out.strip() or f"Exit code {code}")
            self.transfer_finished.emit(transfer_id, ok, msg)

        runner.line.connect(on_line)
        runner.finished.connect(on_finished)
        runner.error.connect(lambda e: self.transfer_finished.emit(transfer_id, False, e))
        runner.run_batch(commands, verbose=True)


def parse_sftp_ls_la(output: str) -> Tuple[str, List[RemoteEntry]]:
    """
    Parse output from `pwd` and `ls -la`.

    Returns:
        (cwd, entries)
    """
    cwd = "."
    entries: List[RemoteEntry] = []

    lines = [ln.strip("\r") for ln in output.splitlines() if ln.strip()]
    for ln in lines:
        # Example pwd line: "Remote working directory: /home/user"
        if "Remote working directory:" in ln:
            cwd = ln.split("Remote working directory:", 1)[1].strip()
            continue

    # Parse typical `ls -la` listing lines.
    # Format: drwxr-xr-x  2 user group 4096 Jan  1 12:34 dirname
    #         -rw-r--r--  1 user group  123 Jan  1 12:34 file.txt
    ls_re = re.compile(
        r"^(?P<mode>[d\-][rwx\-]{9})\s+\d+\s+\S+\s+\S+\s+(?P<size>\d+)\s+"
        r"(?P<mon>\w+)\s+(?P<day>\d+)\s+(?P<timeyear>[\d:]+)\s+(?P<name>.+)$"
    )

    for ln in lines:
        m = ls_re.match(ln)
        if not m:
            continue
        name = m.group("name")
        if name in (".", ".."):
            continue
        is_dir = m.group("mode").startswith("d")
        size = int(m.group("size"))
        entries.append(RemoteEntry(name=name, is_dir=is_dir, size=size, raw=ln))

    # If nothing parsed, fallback: show raw lines as files
    if not entries:
        for ln in lines:
            if ln.startswith("sftp>") or "Remote working directory:" in ln:
                continue
            if ln.lower().startswith("total"):
                continue
            entries.append(RemoteEntry(name=ln, is_dir=False, size=None, raw=ln))

    return cwd, entries


class MainWindow(QMainWindow):
    def __init__(self) -> None:
        super().__init__()
        self.setWindowTitle("PyQt SFTP Front-End (OpenSSH)")
        self.resize(1200, 720)

        self.bus = LogBus()
        self.sftp = SftpService(self)
        self.sftp.log.connect(self._log)

        self.sftp.listing_ready.connect(self._on_listing_ready)
        self.sftp.transfer_progress.connect(self._on_transfer_progress)
        self.sftp.transfer_finished.connect(self._on_transfer_finished)

        self._transfer_seq = 0
        self._transfer_rows: dict[str, int] = {}
        self._remote_entries: List[RemoteEntry] = []
        self._remote_cwd = "."

        root = QWidget()
        self.setCentralWidget(root)
        layout = QVBoxLayout(root)

        layout.addLayout(self._build_connection_bar())
        layout.addWidget(self._build_splitter())
        layout.addWidget(self._build_transfer_table())
        layout.addWidget(QLabel("Log"))
        layout.addWidget(self._build_log_box())

        self._build_menu()

    def _build_menu(self) -> None:
        act_quit = QAction("Quit", self)
        act_quit.triggered.connect(self.close)
        self.menuBar().addAction(act_quit)

    def _build_connection_bar(self) -> QHBoxLayout:
        bar = QHBoxLayout()

        self.host = QLineEdit()
        self.host.setPlaceholderText("Host (e.g. sftp.example.com)")
        self.port = QLineEdit("22")
        self.user = QLineEdit()
        self.user.setPlaceholderText("Username")
        self.key = QLineEdit()
        self.key.setPlaceholderText("Path to private key (recommended)")

        btn_key = QPushButton("Browse Key…")
        btn_key.clicked.connect(self._pick_key)

        self.btn_connect = QPushButton("Connect")
        self.btn_connect.clicked.connect(self._connect)

        self.btn_refresh = QPushButton("Refresh Remote")
        self.btn_refresh.clicked.connect(self._refresh_remote)
        self.btn_refresh.setEnabled(False)

        bar.addWidget(QLabel("Host"))
        bar.addWidget(self.host, 2)
        bar.addWidget(QLabel("Port"))
        bar.addWidget(self.port, 1)
        bar.addWidget(QLabel("User"))
        bar.addWidget(self.user, 1)
        bar.addWidget(QLabel("Key"))
        bar.addWidget(self.key, 2)
        bar.addWidget(btn_key)
        bar.addWidget(self.btn_connect)
        bar.addWidget(self.btn_refresh)
        return bar

    def _build_splitter(self) -> QSplitter:
        splitter = QSplitter(Qt.Orientation.Horizontal)

        # Local browser
        local_wrap = QWidget()
        local_layout = QVBoxLayout(local_wrap)
        local_layout.addWidget(QLabel("Local files"))

        self.local_model = QFileSystemModel(self)
        self.local_model.setRootPath(str(Path.home()))
        self.local_view = self._make_local_view()
        local_layout.addWidget(self.local_view)

        local_btns = QHBoxLayout()
        self.btn_upload = QPushButton("Upload →")
        self.btn_upload.clicked.connect(self._upload_selected)
        self.btn_upload.setEnabled(False)

        self.local_root = QComboBox()
        self.local_root.setEditable(True)
        self.local_root.addItem(str(Path.home()))
        self.local_root.setCurrentText(str(Path.home()))
        self.local_root.currentTextChanged.connect(self._set_local_root)

        local_btns.addWidget(QLabel("Local root"))
        local_btns.addWidget(self.local_root, 1)
        local_btns.addWidget(self.btn_upload)
        local_layout.addLayout(local_btns)

        # Remote browser
        remote_wrap = QWidget()
        remote_layout = QVBoxLayout(remote_wrap)
        self.remote_label = QLabel("Remote files (not connected)")
        remote_layout.addWidget(self.remote_label)

        self.remote_tree = QTreeWidget()
        self.remote_tree.setColumnCount(3)
        self.remote_tree.setHeaderLabels(["Name", "Type", "Size"])
        self.remote_tree.header().setSectionResizeMode(0, QHeaderView.ResizeMode.Stretch)
        self.remote_tree.header().setSectionResizeMode(1, QHeaderView.ResizeMode.ResizeToContents)
        self.remote_tree.header().setSectionResizeMode(2, QHeaderView.ResizeMode.ResizeToContents)
        self.remote_tree.itemDoubleClicked.connect(self._remote_double_clicked)
        remote_layout.addWidget(self.remote_tree)

        remote_btns = QHBoxLayout()
        self.btn_download = QPushButton("← Download")
        self.btn_download.clicked.connect(self._download_selected)
        self.btn_download.setEnabled(False)

        self.btn_mkdir = QPushButton("New Folder…")
        self.btn_mkdir.clicked.connect(self._remote_mkdir)
        self.btn_mkdir.setEnabled(False)

        remote_btns.addWidget(self.btn_download)
        remote_btns.addWidget(self.btn_mkdir)
        remote_layout.addLayout(remote_btns)

        splitter.addWidget(local_wrap)
        splitter.addWidget(remote_wrap)
        splitter.setStretchFactor(0, 1)
        splitter.setStretchFactor(1, 1)
        return splitter

    def _make_local_view(self):
        from PyQt6.QtWidgets import QTreeView

        view = QTreeView()
        view.setModel(self.local_model)
        view.setRootIndex(self.local_model.index(str(Path.home())))
        view.setSelectionMode(QTreeView.SelectionMode.ExtendedSelection)
        view.setSortingEnabled(True)
        return view

    def _build_transfer_table(self) -> QWidget:
        wrap = QWidget()
        lay = QVBoxLayout(wrap)
        lay.addWidget(QLabel("Transfers"))

        self.transfers = QTableWidget(0, 5)
        self.transfers.setHorizontalHeaderLabels(["ID", "Direction", "Path", "Status", "Progress"])
        self.transfers.horizontalHeader().setSectionResizeMode(0, QHeaderView.ResizeMode.ResizeToContents)
        self.transfers.horizontalHeader().setSectionResizeMode(1, QHeaderView.ResizeMode.ResizeToContents)
        self.transfers.horizontalHeader().setSectionResizeMode(2, QHeaderView.ResizeMode.Stretch)
        self.transfers.horizontalHeader().setSectionResizeMode(3, QHeaderView.ResizeMode.ResizeToContents)
        self.transfers.horizontalHeader().setSectionResizeMode(4, QHeaderView.ResizeMode.ResizeToContents)
        lay.addWidget(self.transfers)
        return wrap

    def _build_log_box(self) -> QPlainTextEdit:
        self.log_box = QPlainTextEdit()
        self.log_box.setReadOnly(True)
        self.log_box.setMaximumBlockCount(5000)
        return self.log_box

    def _log(self, msg: str) -> None:
        self.log_box.appendPlainText(msg)

    def _pick_key(self) -> None:
        path, _ = QFileDialog.getOpenFileName(self, "Select Private Key", str(Path.home()))
        if path:
            self.key.setText(path)

    def _connect(self) -> None:
        host = self.host.text().strip()
        user = self.user.text().strip()
        port_txt = self.port.text().strip()
        key_path = self.key.text().strip() or None

        if not host or not user or not port_txt:
            QMessageBox.warning(self, "Missing", "Host, Port, and User are required.")
            return
        try:
            port = int(port_txt)
        except ValueError:
            QMessageBox.warning(self, "Invalid", "Port must be an integer.")
            return

        if key_path and not Path(key_path).expanduser().exists():
            QMessageBox.warning(self, "Invalid", "Key file does not exist.")
            return

        if not key_path:
            QMessageBox.information(
                self,
                "Key recommended",
                "Password auth is not implemented. Use a private key or ssh-agent.",
            )

        conn = SftpConnection(host=host, port=port, username=user, key_path=key_path)
        self.sftp.set_connection(conn)

        self._log(f"Connecting to {conn.target()}:{conn.port} ...")
        self._refresh_remote()

        self.btn_refresh.setEnabled(True)
        self.btn_upload.setEnabled(True)
        self.btn_download.setEnabled(True)
        self.btn_mkdir.setEnabled(True)

    def _refresh_remote(self) -> None:
        if not self.sftp.connected():
            return

        runner = self.sftp._runner
        if not runner:
            return

        def finished_once(code: int, out: str, err: str) -> None:
            runner.finished.disconnect(finished_once)
            if code != 0:
                QMessageBox.critical(self, "SFTP error", err or out or f"Exit code {code}")
                return
            cwd, entries = parse_sftp_ls_la(out + "\n" + err)
            self._remote_cwd = cwd
            self._remote_entries = entries
            self._populate_remote()

        runner.finished.connect(finished_once)
        runner.run_batch([f"cd {shlex.quote(self._remote_cwd)}", "pwd", "ls -la"], verbose=False)

    def _on_listing_ready(self, remote_cwd: str, entries: list) -> None:
        self._remote_cwd = remote_cwd
        self._remote_entries = entries
        self._populate_remote()

    def _populate_remote(self) -> None:
        self.remote_label.setText(f"Remote files: {self._remote_cwd}")
        self.remote_tree.clear()

        parent = QTreeWidgetItem(["..", "dir", ""])
        parent.setData(0, Qt.ItemDataRole.UserRole, {"kind": "up"})
        self.remote_tree.addTopLevelItem(parent)

        for ent in self._remote_entries:
            typ = "dir" if ent.is_dir else "file"
            size = "" if ent.size is None or ent.is_dir else str(ent.size)
            item = QTreeWidgetItem([ent.name, typ, size])
            item.setData(0, Qt.ItemDataRole.UserRole, {"kind": "entry", "entry": ent})
            self.remote_tree.addTopLevelItem(item)

    def _remote_double_clicked(self, item: QTreeWidgetItem) -> None:
        data = item.data(0, Qt.ItemDataRole.UserRole) or {}
        kind = data.get("kind")

        if kind == "up":
            # naive parent dir handling
            new = str(Path(self._remote_cwd).parent) if self._remote_cwd not in (".", "/") else "/"
            self._remote_cwd = new
            self._refresh_remote()
            return

        ent: RemoteEntry = data.get("entry")
        if ent and ent.is_dir:
            if self._remote_cwd in (".", "/"):
                new = f"/{ent.name}" if self._remote_cwd == "/" else ent.name
            else:
                new = f"{self._remote_cwd.rstrip('/')}/{ent.name}"
            self._remote_cwd = new
            self._refresh_remote()

    def _set_local_root(self, path: str) -> None:
        p = Path(path).expanduser()
        if p.exists() and p.is_dir():
            self.local_view.setRootIndex(self.local_model.index(str(p)))

    def _selected_local_files(self) -> List[str]:
        idxs: List[QModelIndex] = self.local_view.selectionModel().selectedRows()
        paths: List[str] = []
        for idx in idxs:
            p = self.local_model.filePath(idx)
            if Path(p).is_file():
                paths.append(p)
        return paths

    def _selected_remote_entry(self) -> Optional[RemoteEntry]:
        item = self.remote_tree.currentItem()
        if not item:
            return None
        data = item.data(0, Qt.ItemDataRole.UserRole) or {}
        ent = data.get("entry")
        if isinstance(ent, RemoteEntry):
            return ent
        return None

    def _next_transfer_id(self) -> str:
        self._transfer_seq += 1
        return f"T{self._transfer_seq:04d}"

    def _add_transfer_row(self, transfer_id: str, direction: str, path: str) -> None:
        row = self.transfers.rowCount()
        self.transfers.insertRow(row)
        self.transfers.setItem(row, 0, QTableWidgetItem(transfer_id))
        self.transfers.setItem(row, 1, QTableWidgetItem(direction))
        self.transfers.setItem(row, 2, QTableWidgetItem(path))
        self.transfers.setItem(row, 3, QTableWidgetItem("Running"))
        self.transfers.setItem(row, 4, QTableWidgetItem("0%"))
        self._transfer_rows[transfer_id] = row

    def _on_transfer_progress(self, transfer_id: str, percent: int) -> None:
        row = self._transfer_rows.get(transfer_id)
        if row is None:
            return
        self.transfers.item(row, 4).setText(f"{percent}%")

    def _on_transfer_finished(self, transfer_id: str, ok: bool, message: str) -> None:
        row = self._transfer_rows.get(transfer_id)
        if row is None:
            return
        self.transfers.item(row, 3).setText("OK" if ok else "Failed")
        if ok:
            self.transfers.item(row, 4).setText("100%")
            self._refresh_remote()
        else:
            self._log(f"{transfer_id} failed: {message}")

    def _upload_selected(self) -> None:
        if not self.sftp.connected():
            return
        local_files = self._selected_local_files()
        if not local_files:
            QMessageBox.information(self, "No selection", "Select one or more local files to upload.")
            return

        for lf in local_files:
            tid = self._next_transfer_id()
            self._add_transfer_row(tid, "upload", lf)
            self.sftp.upload(lf, self._remote_cwd, tid)

    def _download_selected(self) -> None:
        if not self.sftp.connected():
            return
        ent = self._selected_remote_entry()
        if not ent:
            QMessageBox.information(self, "No selection", "Select a remote file to download.")
            return
        if ent.is_dir:
            QMessageBox.information(self, "Folder", "Downloading folders is not implemented (files only).")
            return

        local_dir = QFileDialog.getExistingDirectory(self, "Select Download Folder", str(Path.home()))
        if not local_dir:
            return

        tid = self._next_transfer_id()
        remote_path = ent.name
        display = f"{self._remote_cwd.rstrip('/')}/{remote_path} -> {local_dir}"
        self._add_transfer_row(tid, "download", display)
        self.sftp.download(remote_path, local_dir, self._remote_cwd, tid)

    def _remote_mkdir(self) -> None:
        from PyQt6.QtWidgets import QInputDialog

        name, ok = QInputDialog.getText(self, "New Folder", "Folder name:")
        if ok and name.strip():
            self.sftp.mkdir(name.strip())
            self._refresh_remote()


def main() -> int:
    app = QApplication(sys.argv)
    w = MainWindow()
    w.show()
    return app.exec()


if __name__ == "__main__":
    raise SystemExit(main())