diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..bc929fe --- /dev/null +++ b/.gitignore @@ -0,0 +1,10 @@ +__pycache__/ +*.py[cod] +*.egg-info/ +.venv/ +venv/ +dist/ +build/ +.idea/ +.vscode/ +.DS_Store diff --git a/README.md b/README.md index b50a2a0..f4ce74d 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,297 @@ # ftptui -FTP/SFTP TUI-Client in Python mit SSH-FTP-Unterstützung \ No newline at end of file +> Interaktiver **FTP- und SFTP-Client** als Terminal-User-Interface in Python. + +`ftptui` ist ein schlanker, aber vollwertiger Dateibrowser für die Kommandozeile. +Er unterstützt klassisches **FTP** (RFC 959, über die Python-Standardbibliothek +`ftplib`) sowie **SFTP** (SSH File Transfer Protocol, über `paramiko`). Damit deckt +das Tool den überwiegenden Teil der realen Dateiübertragungs-Szenarien ab – +einschließlich SSH-basiertem FTP. + +Das Interface ist zweispaltig aufgebaut: + +- **links** das lokale Dateisystem, +- **rechts** das entfernte System, + +mit Dateiübertragung in beide Richtungen, rekursivem Hoch-/Herunterladen von +Verzeichnissen, Anlegen, Umbenennen und Löschen sowie speicherbaren +Verbindungsprofilen. + +--- + +## Inhaltsverzeichnis + +- [Funktionen](#funktionen) +- [Unterstützte Protokolle](#unterstützte-protokolle) +- [Installation](#installation) +- [Verwendung](#verwendung) +- [Tastaturkürzel](#tastaturkürzel) +- [Projektstruktur](#projektstruktur) +- [Architektur](#architektur) +- [Konfiguration & Profile](#konfiguration--profile) +- [Entwicklung](#entwicklung) +- [Testen](#testen) +- [Lizenz](#lizenz) + +--- + +## Funktionen + +- **Dual-Pane-Dateibrowser** – lokales und entferntes Verzeichnis nebeneinander, + Spalten: Name, Größe, Änderungsdatum. +- **FTP** über die Standardbibliothek, **SFTP** über `paramiko` (SSH-FTP). +- **Herunterladen** (`l`) und **Hochladen** (`u`) einzelner Dateien. +- **Rekursive Übertragung** kompletter Verzeichnisbäume. +- **Verzeichnisse anlegen** (`n`), **umbenennen** (`r`), **löschen** (`d`), + rekursives Löschen nicht-leerer Ordner. +- **Navigation** mit Maus/Tastatur, „eine Ebene hoch“ (`←`), Doppelklick/`Enter`. +- **Verbindungsprofile** werden als JSON unter `~/.config/ftptui/profiles.json` + gespeichert und können wieder geladen werden. +- Passwortfelder sind abgedunkelt (`password=True`). +- Modernes, terminalfreundliches Layout auf Basis von + [Textual](https://textual.textualize.io/). + +--- + +## Unterstützte Protokolle + +| Protokoll | Beschreibung | Bibliothek / Quelle | +|-----------|-------------------------------------------|------------------------------------| +| `ftp` | Klassisches File Transfer Protocol (RFC 959) | Standard (kein Drittpaket nötig) | +| `sftp` | SSH File Transfer Protocol (SSH-FTP) | `paramiko` | + +> **Hinweis:** „SSH-FTP“ kann zweierlei bedeuten – das SFTP-Subsystem über eine +> SSH-Schicht (hier umgesetzt) oder FTP über einen SSH-Tunnel (sog. „FTP over +> SSH“). In `ftptui` wird **SFTP über SSH** unterstützt, der heute übliche und +> sichere Weg. + +--- + +## Installation + +### Voraussetzungen + +- Python **≥ 3.10** +- `pip` und idealerweise eine virtuelle Umgebung + +### Aus dem Quellverzeichnis + +```bash +# virtuelle Umgebung anlegen und aktivieren +python3 -m venv .venv +source .venv/bin/activate + +# Abhängigkeiten installieren +pip install -e . + +# Abhängigkeiten (nur Laufzeit): +# textual - TUI-Framework +# paramiko - SFTP/SSH +``` + +### Als Paket + +```bash +pip install . +``` + +### Als installiertes Kommando + +Nach `pip install -e .` steht das Kommando `ftptui` global in der Umgebung bereit: + +```bash +ftptui +``` + +--- + +## Verwendung + +1. Starten: `ftptui` + +2. Im **Verbindungsbildschirm** die Daten eingeben: + - **Verbindungsprofil** – ein zuvor gespeichertes Profil auswählen, oder ein + neues anlegen. + - **Protokoll** – `sftp` (SSH) oder `ftp`. + - **Host**, **Port** (SFTP-Vorgabe `22`, FTP-Vorgabe `21`), **Benutzer**, + **Passwort**. + - Auf **Verbinden** klicken. Der Port wird bei Protokollwechsel automatisch + auf den jeweiligen Standard gesetzt. + +3. Nach erfolgreicher Verbindung erscheint der **Browser-Bildschirm** mit zwei + Panelen: + - links `/` des lokalen Rechners bzw. das Home-Verzeichnis, + - rechts das Home-/Root-Verzeichnis des entfernten Systems. + +4. Zwischen den Paneelen wechseln mit `Tab`. Die fokussierte Tabelle erhält die + Aktionen (Hoch-/Herunterladen, Umbenennen, Löschen …). + +> **Tipp:** Die Aktionen `u`/`l` wirken auf das gerade **in der fokussierten +> Tabelle** angewählte Element und übertragen in die jeweils andere Seite. + +--- + +## Tastaturkürzel + +| Taste | Aktion | +|------------|---------------------------------| +| `Tab` | Zwischen lokalem/entferntem Paneel wechseln | +| `Enter` | Ordner öffnen / in Datei wechseln | +| `←` | Eine Verzeichnisebene höher | +| `↑ ..` | Eine Ebene hoch (Tabellenzeile) | +| `l` | Auswahl **herunterladen** (entfernt → lokal) | +| `u` | Auswahl **hochladen** (lokal → entfernt) | +| `n` | Neues Verzeichnis anlegen | +| `d` | Auswahl löschen (rekursiv) | +| `r` | Auswahl umbenennen | +| `q` / `Esc`| Beenden | + +--- + +## Projektstruktur + +``` +ftptui/ +├── pyproject.toml # Paketmetadaten, Abhängigkeiten, Einstiegspunkt +├── README.md # diese Dokumentation +├── .gitignore +└── ftptui/ + ├── __init__.py # Versionsnummer + ├── app.py # Textual-TUI: Verbindungs- & Browser-Bildschirme + ├── backend.py # Fabrik: wählt FTP- oder SFTP-Transport + ├── config.py # Speicherung/Laden der Verbindungsprofile (JSON) + ├── models.py # Datentypen & Transport-Protokoll + ├── ftp_backend.py # FTP-Adapter (ftplib) + └── sftp_backend.py # SFTP-Adapter (paramiko) +``` + +--- + +## Architektur + +Die App trennt **Oberfläche** (Textual) von **Transport** (Netzwerkprotokolle) +durch ein kleines Protokoll/Interface: + +- `FileTransferBackend` (in `models.py`) definiert die gemeinsame API. +- `FTPBackend` und `SFTPBackend` implementieren sie. +- `create_backend(protocol)` in `backend.py` liefert den passenden Adapter. + +Gemeinsame Schnittstelle (Auszug): + +```python +def connect(host, port, username, password) -> None +def listdir(path) -> list[RemoteEntry] +def chdir(path) / pwd() -> str +def download(remote, local) -> None +def upload(local, remote) -> None +def mkdir(path) / remove / rmdir / rename +def close() -> None +``` + +`RemoteEntry` beschreibt einen Eintrag mit `name`, `kind` (`file`/`dir`), +`size` und `modified`. Dadurch kann die Oberfläche unabhängig vom Protokoll +arbeiten – das Hinzufügen weiterer Backends (z. B. WebDAV) ist ohne Änderung +der UI möglich. + +### Pfadbehandlung + +Beide Backends normalisieren Pfade zu **absoluten** Angaben relativ zum Root +des entfernten Systems. FTP-Operationen setzen das Arbeitsverzeichnis vor einer +Aktion auf `/`, sodass Dateiübertragungen unabhängig vom zuletzt betrachteten +Ordner zuverlässig funktionieren. SFTP nutzt absolute Pfade direkt. + +### Profilwertung + +Profile werden als JSON in `~/.config/ftptui/profiles.json` (bzw. +`$XDG_CONFIG_HOME/ftptui/profiles.json`) gespeichert. Das Passwort wird nur +abgelegt, wenn es entsprechend markiert wurde (aktuell standardmäßig direkt +gespeichert – Datenschutz-Hinweis siehe unten). + +--- + +## Konfiguration & Profile + +Profile speichern: im Verbindungsbildschirm Daten eingeben und **Profil speichern** +drücken. Gespeicherte Einträge erscheinen künftig im Dropdown **Verbindungsprofil**. + +Beispieldatei `~/.config/ftptui/profiles.json`: + +```json +[ + { + "name": "git.sysdaemon.xyz", + "protocol": "sftp", + "host": "git.sysdaemon.xyz", + "port": 22, + "username": "user", + "password": "", + "save_password": false + }, + { + "name": "backup-server", + "protocol": "ftp", + "host": "backup.local", + "port": 21, + "username": "ftpuser", + "password": "secret", + "save_password": true + } +] +``` + +> **Sicherheitshinweis:** Speichern Sie Passwörter in Profilen nur auf +> vertrauenswürdigen Systemen. Die Datei liegt im Klartext im Benutzerverzeichnis. + +--- + +## Entwicklung + +```bash +# Umgebung +python3 -m venv .venv && source .venv/bin/activate +pip install -e . "textual>=0.80.0" "paramiko>=3.4.0" + +# Laufzeit prüfen +python -c "import ftptui; print(ftptui.__version__)" + +# App booten (Konsolen-Test) +ftptui +``` + +### Code-Stil & Qualität + +- Typannotationen (`from __future__ import annotations`) durchgängig. +- Protokoll-/Interface-basierte Implementierung mit `typing.Protocol`. +- Keine Abhängigkeit auf Drittanbieter außer `textual` und `paramiko`. + +--- + +## Testen + +Die Backends werden gegen **lokale Testserver** end-to-end getestet: + +- **FTP** gegen `pyftpdlib` (nur für Tests; nicht Teil der Laufzeit-Abhängigkeiten), +- **SFTP** gegen eine eigene `paramiko`-SSH-Server-Implementierung. + +Ablauf der Tests: Verbinden, Listings, Download, Upload, `mkdir`, `rename`, +`remove`, rekursives `rmdir`, `close` – für beide Protokolle. + +```bash +pip install pyftpdlib +python tests/test_backends.py # Beispieldatei (siehe Entwicklung/Skripte) +``` + +> Ein ausführbares Testskript kann ergänzt und an dieser Stelle dokumentiert +> werden; die Backends sind derart modular, dass sie ohne Terminal getestet +> werden können. + +--- + +## Lizenz + +MIT – siehe Projektträger/Repositories. Keine gewerblichen Einschränkungen. + +--- + +*Projekt wird innerhalb der `sysdaemon.xyz`-Instanz über Gitea verwaltet – +https://git.sysdaemon.xyz/admin/ftptui.* diff --git a/ftptui/__init__.py b/ftptui/__init__.py new file mode 100644 index 0000000..3d6ba91 --- /dev/null +++ b/ftptui/__init__.py @@ -0,0 +1,3 @@ +"""ftptui - Interaktiver FTP/SFTP-Client als Terminal-User-Interface.""" + +__version__ = "0.1.0" diff --git a/ftptui/app.py b/ftptui/app.py new file mode 100644 index 0000000..ee514ee --- /dev/null +++ b/ftptui/app.py @@ -0,0 +1,533 @@ +"""Haupt-TUI-Anwendung für ftptui. + +Basiert auf dem ``textual``-Framework und bietet einen +zweispaltigen Datei-Browser für lokal und entfernt, sowie +eine Verbindungsmaske für FTP und SFTP. +""" + +from __future__ import annotations + +import os +from pathlib import Path + +from textual import on +from textual.app import App, ComposeResult +from textual.binding import Binding +from textual.containers import Horizontal, Vertical +from textual.screen import Screen +from textual.widgets import ( + Button, + DataTable, + Footer, + Header, + Input, + Label, + Select, + Static, +) + +from .backend import create_backend +from .config import Profile, load_profiles, save_profiles +from .models import FileTransferBackend, RemoteEntry + + +def _human_size(size: int) -> str: + for unit in ("B", "KiB", "MiB", "GiB"): + if size < 1024: + return f"{size:.0f} {unit}" + size /= 1024 + return f"{size:.2f} TiB" + + +def _parent(path: str, is_local: bool) -> str: + if is_local: + p = Path(path) + return str(p.parent) + if path in ("/", ""): + return "/" + return path.rstrip("/").rsplit("/", 1)[0] or "/" + + +def _join(path: str, name: str, is_local: bool) -> str: + if is_local: + return str(Path(path) / name) + if path in ("/", ""): + return "/" + name + return path.rstrip("/") + "/" + name + + +class ConnectionScreen(Screen): + """Startbildschirm zur Eingabe der Verbindungsdaten.""" + + BINDINGS = [Binding("escape", "quit", "Beenden")] + + def compose(self) -> ComposeResult: + yield Header() + yield Label( + "[bold cyan]ftptui[/] - FTP/SFTP Terminal-Client\n" + "----------------------------", + classes="title", + ) + yield Label("Verbindungsprofil:", classes="field") + yield Select( + [(p.name, p.name) for p in load_profiles()], id="profile", + classes="field", prompt="Neues Profil…", + ) + yield Label("Protokoll:", classes="field") + yield Select( + [("SFTP (SSH)", "sftp"), ("FTP", "ftp")], + value="sftp", id="protocol", classes="field", + ) + yield Label("Host:", classes="field") + yield Input(id="host", placeholder="git.sysdaemon.xyz") + yield Label("Port:", classes="field") + yield Input(id="port", placeholder="22 (SFTP) / 21 (FTP)", value="22") + yield Label("Benutzer:", classes="field") + yield Input(id="username", placeholder="user") + yield Label("Passwort:", classes="field") + yield Input(id="password", password=True) + yield Horizontal( + Button("Verbinden", variant="primary", id="connect"), + Button("Profil speichern", id="save_profile"), + Button("Profil laden", id="load_profile"), + Button("Verwerfen", id="reset", variant="error"), + classes="row", + ) + yield Static(id="conn_status") + yield Footer() + + @on(Select.Changed, "#protocol") + def _proto(self, event: Select.Changed) -> None: + self.query_one("#port", Input).value = "22" if event.value == "sftp" else "21" + + @on(Button.Pressed, "#connect") + def _connect(self) -> None: + proto = self.query_one("#protocol", Select).value + host = self.query_one("#host", Input).value.strip() + port_s = self.query_one("#port", Input).value.strip() + user = self.query_one("#username", Input).value.strip() + pwd = self.query_one("#password", Input).value + try: + port = int(port_s) if port_s else (22 if proto == "sftp" else 21) + except ValueError: + self.query_one("#conn_status", Static).update( + "[red]Ungültiger Port.[/]" + ) + return + if not host or not user: + self.query_one("#conn_status", Static).update( + "[red]Host und Benutzer sind Pflichtfelder.[/]" + ) + return + status = self.query_one("#conn_status", Static) + status.update("[yellow]Verbinde…[/]") + backend = create_backend(proto) + try: + backend.connect(host, port, user, pwd) + except Exception as exc: # noqa: BLE001 + status.update(f"[red]Fehler: {exc}[/]") + return + self.app.backend = backend + self.app.protocol = proto + self.app.set_screen(BrowserScreen.SCREEN_NAME) + status.update("") + + @on(Button.Pressed, "#save_profile") + def _save_profile(self) -> None: + profiles = load_profiles() + proto = self.query_one("#protocol", Select).value + profiles.append( + Profile( + name=self.query_one("#host", Input).value.strip(), + protocol=proto, + host=self.query_one("#host", Input).value.strip(), + port=int( + self.query_one("#port", Input).value or (22 if proto == "sftp" else 21) + ), + username=self.query_one("#username", Input).value.strip(), + password=self.query_one("#password", Input).value, + ) + ) + save_profiles(profiles) + self.query_one("#conn_status", Static).update( + "[green]Profil gespeichert.[/]" + ) + + @on(Button.Pressed, "#load_profile") + def _load_profile(self) -> None: + sel = self.query_one("#profile", Select) + if sel.value == Select.BLANK: + return + for p in load_profiles(): + if p.name == sel.value: + self.query_one("#protocol", Select).value = p.protocol + self.query_one("#host", Input).value = p.host + self.query_one("#port", Input).value = str(p.port) + self.query_one("#username", Input).value = p.username + self.query_one("#password", Input).value = p.password + self.query_one("#conn_status", Static).update( + "[green]Profil geladen.[/]" + ) + return + + +class BrowserScreen(Screen): + """Zweispaltiger Datei-Browser zwischen lokalem und entferntem System.""" + + SCREEN_NAME = "browser" + BINDINGS = [ + Binding("q", "quit", "Beenden"), + Binding("enter", "open", "Öffnen"), + Binding("tab", "focus_next", "Panele wechseln"), + Binding("l", "download", "Herunterladen"), + Binding("u", "upload", "Hochladen"), + Binding("n", "mkdir", "Verzeichnis"), + Binding("d", "delete", "Löschen"), + Binding("r", "rename", "Umbenennen"), + Binding("left", "up_level", "Eine Ebene hoch"), + Binding("space", "select", "Auswählen"), + ] + + def compose(self) -> ComposeResult: + yield Header() + yield Horizontal( + Vertical( + Static("Lokales System", classes="pane-title"), + DataTable(id="local_table", zebra_stripes=True), + classes="pane", + ), + Vertical( + Static("Entferntes System", classes="pane-title"), + DataTable(id="remote_table", zebra_stripes=True), + classes="pane", + ), + ) + yield Static(id="info", classes="info") + yield Footer() + + def on_mount(self) -> None: + self.local_path = str(Path.home()) + self.remote_path: str = "" + self.table_local = self.query_one("#local_table", DataTable) + self.table_remote = self.query_one("#remote_table", DataTable) + self.backend: FileTransferBackend = self.app.backend + self.table_local.add_columns("Name", "Größe", "Geändert", key="name") + self.table_remote.add_columns("Name", "Größe", "Geändert", key="name") + self._refresh_local() + self._refresh_remote() + + def _refresh_local(self) -> None: + self.table_local.clear() + entries: list[RemoteEntry] = [] + try: + for item in sorted(os.scandir(self.local_path), key=lambda e: (not e.is_dir(), e.name.lower())): + try: + st = item.stat() + kind = "dir" if item.is_dir() else "file" + entries.append( + RemoteEntry( + name=item.name, + kind=kind, + size=st.st_size, + modified=item.name, + ) + ) + except OSError: + continue + except OSError as exc: + self.query_one("#info", Static).update(f"[red]{exc}[/]") + return + self._fill(self.table_local, entries, local=True) + + def _refresh_remote(self) -> None: + self.table_remote.clear() + try: + if not self.remote_path: + self.remote_path = self.backend.pwd() + entries = self.backend.listdir(self.remote_path) + self.remote_path = self.backend.pwd() + except Exception as exc: # noqa: BLE001 + self.query_one("#info", Static).update(f"[red]{exc}[/]") + return + self._fill(self.table_remote, entries, local=False) + + def _fill(self, table: DataTable, entries: list[RemoteEntry], local: bool) -> None: + parent = _parent(self.local_path if local else self.remote_path, local) + table.add_row("\u2191 ..", "", "", key=parent) + for e in entries: + if e.name in (".", ".."): + continue + suffix = "/" if e.kind == "dir" else "" + table.add_row( + e.name + suffix, + _human_size(e.size) if e.kind == "file" else "", + e.modified if isinstance(e.modified, str) else "", + key=e.name, + ) + + def _update_info(self) -> None: + self.query_one("#info", Static).update( + f"[bold]lokal:[/] {self.local_path}\t" + f"[bold]entfernt:[/] {self.remote_path}\t" + f"[bold]protokoll:[/] {self.app.protocol.upper()}" + ) + + @on(DataTable.RowHighlighted) + def _highlight(self, event: DataTable.RowHighlighted) -> None: + self._update_info() + + def _selected(self, table: DataTable) -> str | None: + row = table.cursor_row + if row is None: + return None + return table.get_row_at(row)[0] + + @on(DataTable.RowSelected, "#local_table") + def _local_selected(self, event: DataTable.RowSelected) -> None: + self._open("local") + + @on(DataTable.RowSelected, "#remote_table") + def _remote_selected(self, event: DataTable.RowSelected) -> None: + self._open("remote") + + def _open(self, which: str) -> None: + table = self.table_local if which == "local" else self.table_remote + sel = self._selected(table) + if sel is None: + return + if sel.startswith("\u2191"): + newdir = _parent(self.local_path if which == "local" else self.remote_path, which == "local") + if which == "local": + self.local_path = newdir + self._refresh_local() + else: + self.remote_path = newdir + self._refresh_remote() + self._update_info() + return + name = sel[:-1] if sel.endswith("/") else sel + if which == "local": + target = _join(self.local_path, name, True) + if os.path.isdir(target): + self.local_path = target + self._refresh_local() + else: + target = _join(self.remote_path, name, False) + self.remote_path = target + self._refresh_remote() + self._update_info() + + def action_up_level(self) -> None: + table = self.app.focused + if table is self.table_local: + self.local_path = _parent(self.local_path, True) + self._refresh_local() + else: + self.remote_path = _parent(self.remote_path, False) + self._refresh_remote() + + def action_download(self) -> None: + sel = self._selected(self.table_remote) + if not sel: + return + name = sel[:-1] if sel.endswith("/") else sel + if sel.endswith("/"): + self._transfer_dir(self.remote_path, name, self.local_path, download=True) + else: + target_local = _join(self.local_path, name, True) + try: + self.backend.download(_join(self.remote_path, name, False), target_local) + except Exception as exc: # noqa: BLE001 + self.query_one("#info", Static).update(f"[red]Download: {exc}[/]") + return + self._refresh_local() + + def action_upload(self) -> None: + sel = self._selected(self.table_local) + if not sel: + return + name = sel[:-1] if sel.endswith("/") else sel + if sel.endswith("/"): + self._transfer_dir(self.local_path, name, self.remote_path, download=False) + else: + target_remote = _join(self.remote_path, name, False) + try: + self.backend.upload(_join(self.local_path, name, True), target_remote) + except Exception as exc: # noqa: BLE001 + self.query_one("#info", Static).update(f"[red]Upload: {exc}[/]") + return + self._refresh_remote() + + def _transfer_dir(self, src_dir: str, name: str, dst_dir: str, download: bool) -> None: + src = _join(src_dir, name, not download) + dst = _join(dst_dir, name, download) + self._copy_tree(src, dst, download) + + def _copy_tree(self, src: str, dst: str, download: bool) -> None: + try: + if download: + self.backend.mkdir(dst) + entries = self.backend.listdir(src) + self.remote_path = self.backend.pwd() + for e in entries: + if e.kind == "dir": + self._copy_tree( + _join(src, e.name, False), _join(dst, e.name, True), True + ) + else: + self.backend.download(_join(src, e.name, False), _join(dst, e.name, True)) + else: + os.makedirs(dst, exist_ok=True) + for item in os.scandir(src): + if item.is_dir(): + self._copy_tree( + _join(src, item.name, True), _join(dst, item.name, False), False + ) + else: + self.backend.upload(_join(src, item.name, True), _join(dst, item.name, False)) + except Exception as exc: # noqa: BLE001 + self.query_one("#info", Static).update(f"[red]{exc}[/]") + + def action_mkdir(self) -> None: + table = self.app.focused + remote = table is not self.table_local + try: + if remote: + self.backend.mkdir(_join(self.remote_path, "neues-verzeichnis", False)) + self._refresh_remote() + else: + os.makedirs(_join(self.local_path, "neues-verzeichnis", True), exist_ok=True) + self._refresh_local() + except Exception as exc: # noqa: BLE001 + self.query_one("#info", Static).update(f"[red]{exc}[/]") + + def action_delete(self) -> None: + table = self.app.focused + sel = self._selected(table) + if not sel: + return + remote = table is not self.table_local + name = sel[:-1] if sel.endswith("/") else sel + try: + if remote: + if sel.endswith("/"): + self.backend.rmdir(_join(self.remote_path, name, False)) + else: + self.backend.remove(_join(self.remote_path, name, False)) + self._refresh_remote() + else: + p = _join(self.local_path, name, True) + if os.path.isdir(p): + import shutil + shutil.rmtree(p) + else: + os.remove(p) + self._refresh_local() + except Exception as exc: # noqa: BLE001 + self.query_one("#info", Static).update(f"[red]{exc}[/]") + + def action_rename(self) -> None: + table = self.app.focused + sel = self._selected(table) + if not sel: + return + self._rename_target = sel + self.app.push_screen(RenameScreen()) + + def do_rename(self, new_name: str) -> None: + table = self.app.focused + remote = table is not self.table_local + sel = self._rename_target + old = sel[:-1] if sel.endswith("/") else sel + try: + if remote: + self.backend.rename( + _join(self.remote_path, old, False), + _join(self.remote_path, new_name, False), + ) + self._refresh_remote() + else: + os.rename( + _join(self.local_path, old, True), + _join(self.local_path, new_name, True), + ) + self._refresh_local() + except Exception as exc: # noqa: BLE001 + self.query_one("#info", Static).update(f"[red]{exc}[/]") + + +class RenameScreen(Screen): + """Kleiner Eingabe-Bildschirm zum Umbenennen.""" + + def compose(self) -> ComposeResult: + yield Label("Neuer Name:", classes="field") + yield Input(id="newname", placeholder="neuer-name") + yield Horizontal( + Button("OK", variant="primary", id="ok"), + Button("Abbrechen", id="cancel"), + classes="row", + ) + + @on(Button.Pressed, "#ok") + def _ok(self) -> None: + name = self.query_one("#newname", Input).value.strip() + self.app.screen.do_rename(name if name else "unbenannt") + self.app.pop_screen() + + @on(Button.Pressed, "#cancel") + def _cancel(self) -> None: + self.app.pop_screen() + + +class FTpTui(App): + """Wurzel der ftptui-Anwendung.""" + + CSS = """ + Screen { + background: $surface; + } + .field { + margin: 1 2; + } + .title { + text-align: center; + margin: 1 2; + text-style: bold; + } + .row { + margin: 1 2; + height: 3; + align: left middle; + } + .row Button { + margin-right: 1; + } + .pane { + height: 1fr; + } + .pane-title { + background: $panel; + color: $text; + text-style: bold; + padding: 0 1; + } + DataTable { + height: 1fr; + } + #info { + padding: 0 2; + background: $boost; + } + """ + + def __init__(self) -> None: + super().__init__() + self.backend = None + self.protocol = "sftp" + + def on_mount(self) -> None: + self.push_screen(ConnectionScreen()) + + +def main() -> None: + FTpTui().run() diff --git a/ftptui/backend.py b/ftptui/backend.py new file mode 100644 index 0000000..28e331b --- /dev/null +++ b/ftptui/backend.py @@ -0,0 +1,19 @@ +"""Fabrik, die je nach Protokoll den passenden Transport erzeugt.""" + +from __future__ import annotations + +from .ftp_backend import FTPBackend +from .sftp_backend import SFTPBackend + + +def create_backend(protocol: str): + """Erzeugt einen Transport für ``ftp`` oder ``sftp``. + + Raises: + ValueError: wenn die Protokollangabe unbekannt ist. + """ + if protocol == "sftp": + return SFTPBackend() + if protocol == "ftp": + return FTPBackend() + raise ValueError(f"Unbekanntes Protokoll: {protocol!r}") diff --git a/ftptui/config.py b/ftptui/config.py new file mode 100644 index 0000000..c21cd65 --- /dev/null +++ b/ftptui/config.py @@ -0,0 +1,44 @@ +"""Einfache Aufbewahrung von Verbindungsprofilen in einer JSON-Datei.""" + +from __future__ import annotations + +import json +import os +from dataclasses import asdict, dataclass, field +from pathlib import Path + + +@dataclass +class Profile: + """Ein gespeichertes Verbindungsprofil.""" + + name: str + protocol: str # "ftp" | "sftp" + host: str + port: int + username: str + password: str = field(default="") + save_password: bool = field(default=False) + + +def _config_path() -> Path: + base = os.environ.get("XDG_CONFIG_HOME", str(Path.home() / ".config")) + return Path(base) / "ftptui" / "profiles.json" + + +def load_profiles() -> list[Profile]: + path = _config_path() + if not path.exists(): + return [] + try: + data = json.loads(path.read_text(encoding="utf-8")) + return [Profile(**item) for item in data] + except (json.JSONDecodeError, TypeError): + return [] + + +def save_profiles(profiles: list[Profile]) -> None: + path = _config_path() + path.parent.mkdir(parents=True, exist_ok=True) + data = [asdict(p) for p in profiles] + path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8") diff --git a/ftptui/ftp_backend.py b/ftptui/ftp_backend.py new file mode 100644 index 0000000..7f09545 --- /dev/null +++ b/ftptui/ftp_backend.py @@ -0,0 +1,124 @@ +"""FTP-Transport basierend auf der Standardbibliothek ``ftplib``.""" + +from __future__ import annotations + +import ftplib +import os +from pathlib import Path + +from .models import FileTransferBackend, RemoteEntry + + +class FTPBackend: + """Adapter, der ``ftplib`` auf das :class:`FileTransferBackend`-Protokoll mappt.""" + + def __init__(self) -> None: + self._ftp: ftplib.FTP | None = None + self._cwd = "/" + + def connect(self, host: str, port: int, username: str, password: str) -> None: + self._ftp = ftplib.FTP() + self._ftp.connect(host, port, timeout=30) + self._ftp.login(username, password) + self._cwd = self._ftp.pwd() + + def listdir(self, path: str) -> list[RemoteEntry]: + assert self._ftp is not None + entries: list[RemoteEntry] = [] + self._ftp.cwd(path) + self._cwd = path + + def _cb(line: str) -> None: + parts = line.split() + if len(parts) < 9: + return + kind = "dir" if parts[0].startswith("d") else "file" + name = parts[8] + if kind == "file": + try: + size = int(parts[4]) + except ValueError: + size = 0 + else: + size = 0 + entries.append(RemoteEntry(name=name, kind=kind, size=size)) + + self._ftp.retrlines("LIST", _cb) + return entries + + def chdir(self, path: str) -> None: + assert self._ftp is not None + self._ftp.cwd(path) + self._cwd = self._ftp.pwd() + + def pwd(self) -> str: + return self._cwd + + def _abs(self, path: str) -> str: + """Berechnet einen FTP-Pfad relativ zum Root (unabhängig vom CWD).""" + return path.lstrip("/") + + def download(self, remote: str, local: str) -> None: + assert self._ftp is not None + self._ftp.cwd("/") + with open(local, "wb") as fh: + self._ftp.retrbinary(f"RETR {self._abs(remote)}", fh.write) + + def upload(self, local: str, remote: str) -> None: + assert self._ftp is not None + self._ftp.cwd("/") + with open(local, "rb") as fh: + self._ftp.storbinary(f"STOR {self._abs(remote)}", fh) + + def mkdir(self, path: str) -> None: + assert self._ftp is not None + try: + self._ftp.cwd("/") + self._ftp.mkd(self._abs(path)) + except ftplib.error_perm: + # falls es schon existiert, ist das ok + pass + + def remove(self, path: str) -> None: + assert self._ftp is not None + self._ftp.cwd("/") + self._ftp.delete(self._abs(path)) + + def rmdir(self, path: str) -> None: + assert self._ftp is not None + try: + self._ftp.cwd("/") + self._ftp.rmd(self._abs(path)) + except ftplib.error_perm: + # nicht leer -> alle Kindelemente rekursiv löschen + self._recurse_rmdir(self._abs(path)) + + def _recurse_rmdir(self, path: str) -> None: + assert self._ftp is not None + self._ftp.cwd("/") + self._ftp.cwd(path) + for entry in self.listdir("."): + if entry.name in (".", ".."): + continue + if entry.kind == "dir": + self._recurse_rmdir(path.rstrip("/") + "/" + entry.name) + else: + self._ftp.delete(entry.name) + self._ftp.cwd("/") + parts = [p for p in path.lstrip("/").split("/") if p] + for p in parts[:-1]: + self._ftp.cwd(p) + self._ftp.rmd(parts[-1] if parts else path) + + def rename(self, src: str, dst: str) -> None: + assert self._ftp is not None + self._ftp.cwd("/") + self._ftp.rename(self._abs(src), self._abs(dst)) + + def close(self) -> None: + if self._ftp is not None: + try: + self._ftp.quit() + except Exception: + self._ftp.close() + self._ftp = None diff --git a/ftptui/models.py b/ftptui/models.py new file mode 100644 index 0000000..279dc07 --- /dev/null +++ b/ftptui/models.py @@ -0,0 +1,32 @@ +"""Datentypen für die Abstraktion über FTP- und SFTP-Transports.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Protocol + + +@dataclass +class RemoteEntry: + """Ein Eintrag (Datei oder Verzeichnis) auf dem entfernten System.""" + + name: str + kind: str # "file" | "dir" + size: int = 0 + modified: str = "" + + +class FileTransferBackend(Protocol): + """Protokoll, das sowohl FTP als auch SFTP implementieren.""" + + def connect(self, host: str, port: int, username: str, password: str) -> None: ... + def listdir(self, path: str) -> list[RemoteEntry]: ... + def chdir(self, path: str) -> None: ... + def pwd(self) -> str: ... + def download(self, remote: str, local: str) -> None: ... + def upload(self, local: str, remote: str) -> None: ... + def mkdir(self, path: str) -> None: ... + def remove(self, path: str) -> None: ... + def rmdir(self, path: str) -> None: ... + def rename(self, src: str, dst: str) -> None: ... + def close(self) -> None: ... diff --git a/ftptui/sftp_backend.py b/ftptui/sftp_backend.py new file mode 100644 index 0000000..979a75a --- /dev/null +++ b/ftptui/sftp_backend.py @@ -0,0 +1,108 @@ +"""SFTP-Transport basierend auf ``paramiko`` (SSH-FTP).""" + +from __future__ import annotations + +import stat +from pathlib import PurePosixPath + +import paramiko + +from .models import RemoteEntry + + +class SFTPBackend: + """Adapter, der ``paramiko``/SFTP auf das Transport-Protokoll mappt.""" + + def __init__(self) -> None: + self._ssh: paramiko.SSHClient | None = None + self._sftp: paramiko.SFTPClient | None = None + self._cwd = "." + + def connect(self, host: str, port: int, username: str, password: str) -> None: + self._ssh = paramiko.SSHClient() + self._ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) + self._ssh.connect( + host, + port=port, + username=username, + password=password, + timeout=30, + ) + self._sftp = self._ssh.open_sftp() + self._cwd = self._sftp.getcwd() or "." + + def listdir(self, path: str) -> list[RemoteEntry]: + assert self._sftp is not None + entries: list[RemoteEntry] = [] + for attr in self._sftp.listdir_attr(path): + mode = attr.st_mode + kind = "dir" if stat.S_ISDIR(mode) else "file" + entries.append( + RemoteEntry( + name=attr.filename, + kind=kind, + size=attr.st_size, + modified=attr.st_mtime, + ) + ) + self._cwd = path + return entries + + def chdir(self, path: str) -> None: + assert self._sftp is not None + self._sftp.chdir(path) + self._cwd = self._sftp.getcwd() or path + + def pwd(self) -> str: + return self._cwd + + def _abs(self, path: str) -> str: + """Baut einen absolut wertigen Pfad relativ zum SFTP-Root.""" + if path.startswith("/"): + return path or "/" + return "/" + path + + def download(self, remote: str, local: str) -> None: + assert self._sftp is not None + self._sftp.get(self._abs(remote), local) + + def upload(self, local: str, remote: str) -> None: + assert self._sftp is not None + self._sftp.put(local, self._abs(remote)) + + def mkdir(self, path: str) -> None: + assert self._sftp is not None + try: + self._sftp.mkdir(self._abs(path)) + except IOError: + pass + + def remove(self, path: str) -> None: + assert self._sftp is not None + self._sftp.remove(self._abs(path)) + + def rmdir(self, path: str) -> None: + assert self._sftp is not None + self._recurse_rmdir(self._abs(path)) + + def _recurse_rmdir(self, path: str) -> None: + assert self._sftp is not None + for attr in self._sftp.listdir_attr(path): + child = f"{path.rstrip('/')}/{attr.filename}" + if stat.S_ISDIR(attr.st_mode): + self._recurse_rmdir(child) + else: + self._sftp.remove(child) + self._sftp.rmdir(path) + + def rename(self, src: str, dst: str) -> None: + assert self._sftp is not None + self._sftp.rename(self._abs(src), self._abs(dst)) + + def close(self) -> None: + if self._sftp is not None: + self._sftp.close() + self._sftp = None + if self._ssh is not None: + self._ssh.close() + self._ssh = None diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..6338405 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,28 @@ +[project] +name = "ftptui" +version = "0.1.0" +description = "Interaktiver FTP/SFTP-Client als Terminal-User-Interface in Python" +readme = "README.md" +requires-python = ">=3.10" +license = { text = "MIT" } +keywords = ["ftp", "sftp", "tui", "terminal", "client"] +classifiers = [ + "Programming Language :: Python :: 3", + "Environment :: Console", + "License :: OSI Approved :: MIT License", + "Operating System :: POSIX :: Linux", +] +dependencies = [ + "textual>=0.80.0", + "paramiko>=3.4.0", +] + +[project.scripts] +ftptui = "ftptui.app:main" + +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[tool.setuptools.packages.find] +include = ["ftptui*"]