ftptui: FTP/SFTP TUI-Client für das Terminal
This commit is contained in:
+10
@@ -0,0 +1,10 @@
|
|||||||
|
__pycache__/
|
||||||
|
*.py[cod]
|
||||||
|
*.egg-info/
|
||||||
|
.venv/
|
||||||
|
venv/
|
||||||
|
dist/
|
||||||
|
build/
|
||||||
|
.idea/
|
||||||
|
.vscode/
|
||||||
|
.DS_Store
|
||||||
@@ -1,3 +1,297 @@
|
|||||||
# ftptui
|
# ftptui
|
||||||
|
|
||||||
FTP/SFTP TUI-Client in Python mit SSH-FTP-Unterstützung
|
> 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.*
|
||||||
|
|||||||
@@ -0,0 +1,3 @@
|
|||||||
|
"""ftptui - Interaktiver FTP/SFTP-Client als Terminal-User-Interface."""
|
||||||
|
|
||||||
|
__version__ = "0.1.0"
|
||||||
+533
@@ -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 "<dir>",
|
||||||
|
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()
|
||||||
@@ -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}")
|
||||||
@@ -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")
|
||||||
@@ -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
|
||||||
@@ -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: ...
|
||||||
@@ -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
|
||||||
@@ -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*"]
|
||||||
Reference in New Issue
Block a user