aibot777/server-manager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 36 Wiki Activity

v1.8.0: file manager improvements

Browse Source 
- SFTP cleanup on app close and language switch
- Windows drive selector in local panel
- Browse and Refresh buttons for local panel
- Recursive upload/download/delete of folders
- Drag-and-drop between local and remote panels
- Sudo mode toggle for privileged file operations
- New i18n keys for EN/RU/ZH

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
chrome-storm-c442
2026-02-23 16:30:56 -05:00
parent a77ca6fee7
commit 3e9aeababe
7 changed files with 528 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files

379
gui/tabs/files_tab.py
View File

@@ -3,10 +3,12 @@ Files tab — dual-pane SFTP file manager.
"""
import os
import platform
import stat
import string
import threading
from datetime import datetime
from tkinter import messagebox
from tkinter import messagebox, filedialog
import customtkinter as ctk
@@ -45,6 +47,16 @@ def _format_perm(mode: int) -> str:
return "".join(parts)
def _get_windows_drives() -> list[str]:
"""Return list of available drive letters on Windows (e.g. ['C:\\', 'D:\\'])."""
drives = []
for letter in string.ascii_uppercase:
drive = f"{letter}:\\"
if os.path.exists(drive):
drives.append(drive)
return drives
class FilesTab(ctk.CTkFrame):
def __init__(self, master, store):
super().__init__(master, fg_color="transparent")
@@ -56,6 +68,7 @@ class FilesTab(ctk.CTkFrame):
self._local_history: list[str] = []
self._remote_history: list[str] = []
self._transferring = False
self._sudo_var = ctk.BooleanVar(value=False)
self._build_ui()
self._refresh_local()
@@ -87,6 +100,33 @@ class FilesTab(ctk.CTkFrame):
)
self._local_up_btn.pack(side="left", padx=2)
# Local refresh button
self._local_refresh_btn = ctk.CTkButton(
left_header, text="\u21BB", width=30, height=28,
command=self._refresh_local,
)
self._local_refresh_btn.pack(side="left", padx=2)
# Browse button
self._browse_btn = ctk.CTkButton(
left_header, text=t("browse"), width=60, height=28,
command=self._browse_local,
)
self._browse_btn.pack(side="left", padx=2)
# Windows drive selector
self._drive_menu = None
if platform.system() == "Windows":
drives = _get_windows_drives()
if drives:
current_drive = os.path.splitdrive(self._local_path)[0] + "\\"
self._drive_var = ctk.StringVar(value=current_drive)
self._drive_menu = ctk.CTkOptionMenu(
left_header, values=drives, variable=self._drive_var,
width=65, height=28, command=self._on_drive_change,
)
self._drive_menu.pack(side="left", padx=2)
self._local_path_entry = ctk.CTkEntry(left_header, height=28)
self._local_path_entry.pack(side="left", fill="x", expand=True, padx=(4, 0))
self._local_path_entry.bind("<Return>", lambda e: self._local_go_to_path())
@@ -95,6 +135,7 @@ class FilesTab(ctk.CTkFrame):
left_pane,
columns=[(t("name_col"), 220), (t("size_col"), 80), (t("date_col"), 110)],
on_navigate=self._navigate_local,
on_drop=self._on_drop_to_local,
)
self._local_list.pack(fill="both", expand=True)
@@ -143,6 +184,7 @@ class FilesTab(ctk.CTkFrame):
(t("date_col"), 100), (t("perm_col"), 80),
],
on_navigate=self._navigate_remote,
on_drop=self._on_drop_to_remote,
)
self._remote_list.pack(fill="both", expand=True)
@@ -152,6 +194,10 @@ class FilesTab(ctk.CTkFrame):
)
self._remote_status.pack(fill="x", pady=(2, 0))
# Wire drag-and-drop partners
self._local_list.set_drag_partner(self._remote_list)
self._remote_list.set_drag_partner(self._local_list)
# === Toolbar ===
toolbar = ctk.CTkFrame(self, fg_color="transparent")
toolbar.pack(fill="x", padx=10, pady=4)
@@ -190,6 +236,16 @@ class FilesTab(ctk.CTkFrame):
)
self._rename_btn.pack(side="left", padx=4)
# Sudo toggle
sep2 = ctk.CTkFrame(toolbar, width=2, height=24, fg_color="gray40")
sep2.pack(side="left", padx=8)
self._sudo_switch = ctk.CTkSwitch(
toolbar, text=t("sudo_mode"), variable=self._sudo_var,
width=50, height=24, command=self._on_sudo_toggle,
)
self._sudo_switch.pack(side="left", padx=4)
self._set_remote_buttons_state("disabled")
# === Transfer / Log area ===
@@ -248,6 +304,7 @@ class FilesTab(ctk.CTkFrame):
def _on_sftp_connected(self, sftp: SFTPSession, home: str):
self._sftp = sftp
self._sftp.sudo_mode = self._sudo_var.get()
self._remote_path = home
self._remote_history.clear()
self._remote_status.configure(
@@ -268,11 +325,36 @@ class FilesTab(ctk.CTkFrame):
pass
self._sftp = None
# ── Browse / Drive ──
def _browse_local(self):
path = filedialog.askdirectory(initialdir=self._local_path)
if path:
self._local_history.append(self._local_path)
self._local_path = os.path.normpath(path)
self._refresh_local()
def _on_drive_change(self, drive: str):
self._local_history.append(self._local_path)
self._local_path = drive
self._refresh_local()
def _on_sudo_toggle(self):
if self._sftp:
self._sftp.sudo_mode = self._sudo_var.get()
self._refresh_remote()
# ── Local navigation ──
def _refresh_local(self):
self._local_path_entry.delete(0, "end")
self._local_path_entry.insert(0, self._local_path)
# Sync drive selector
if self._drive_menu and platform.system() == "Windows":
current_drive = os.path.splitdrive(self._local_path)[0] + "\\"
self._drive_var.set(current_drive)
try:
entries = os.listdir(self._local_path)
except PermissionError:
@@ -344,7 +426,13 @@ class FilesTab(ctk.CTkFrame):
def _do():
try:
attrs = self._sftp.listdir_attr(self._remote_path)
if self._sftp.sudo_mode:
try:
attrs = self._sftp.listdir_attr_sudo(self._remote_path)
except Exception:
attrs = self._sftp.listdir_attr(self._remote_path)
else:
attrs = self._sftp.listdir_attr(self._remote_path)
items = []
# Parent dir
if self._remote_path != "/":
@@ -361,6 +449,9 @@ class FilesTab(ctk.CTkFrame):
"is_dir": is_dir,
})
self.after(0, lambda: self._populate_remote(items))
except PermissionError as e:
hint = f"\n{t('try_sudo_hint')}" if not self._sftp.sudo_mode else ""
self.after(0, lambda: self._on_sftp_error(str(e) + hint))
except Exception as e:
self.after(0, lambda: self._on_sftp_error(str(e)))
@@ -405,99 +496,189 @@ class FilesTab(ctk.CTkFrame):
self._remote_path = path
self._refresh_remote()
# ── File operations ──
# ── Remote path helper ──
def _remote_join(self, name: str) -> str:
if self._remote_path == "/":
return "/" + name
return self._remote_path + "/" + name
# ── Upload / Download (shared logic) ──
def _do_upload(self, items: list[dict]):
"""Upload files and folders from local to remote. Runs in background thread."""
if not self._sftp or not self._sftp.connected or self._transferring:
return
items = [s for s in items if s["name"] != ".."]
if not items:
return
self._transferring = True
self._upload_btn.configure(state="disabled")
self._download_btn.configure(state="disabled")
def _do():
for item in items:
name = item["name"]
local = os.path.join(self._local_path, name)
remote = self._remote_join(name)
try:
if item.get("is_dir"):
self.after(0, lambda n=name: self._transfer_label.configure(
text=t("uploading_dir").format(name=n)
))
def _file_cb(cur, total, fname, n=name):
self.after(0, lambda c=cur, tt=total, fn=fname:
self._transfer_label.configure(
text=t("transfer_file_progress").format(cur=c, total=tt, name=fn)
))
def _progress(transferred, total):
if total > 0:
self.after(0, lambda f=transferred/total: self._progress.set(f))
self._sftp.upload_dir(local, remote, progress_cb=_progress, file_cb=_file_cb)
else:
file_size = os.path.getsize(local)
self.after(0, lambda n=name: self._transfer_label.configure(
text=t("uploading").format(name=n)
))
def _progress(transferred, total, n=name):
if total > 0:
frac = transferred / total
size_str = f"{_format_size(transferred)}/{_format_size(total)}"
self.after(0, lambda f=frac, s=size_str, nn=n:
self._update_progress(f, t("uploading").format(name=nn) + f" ({s})"))
if self._sftp.sudo_mode:
try:
self._sftp.upload_sudo(local, remote, progress_cb=_progress)
except Exception:
self._sftp.upload(local, remote, progress_cb=_progress)
else:
self._sftp.upload(local, remote, progress_cb=_progress)
self.after(0, lambda n=name: self._log_msg(
t("transfer_done").format(name=n)
))
except PermissionError as e:
hint = f" - {t('try_sudo_hint')}" if not self._sftp.sudo_mode else ""
self.after(0, lambda e=e, h=hint: self._log_msg(
t("transfer_failed").format(e=str(e)) + h
))
except Exception as e:
self.after(0, lambda e=e: self._log_msg(
t("transfer_failed").format(e=str(e))
))
self.after(0, self._on_transfer_done)
threading.Thread(target=_do, daemon=True).start()
def _do_download(self, items: list[dict]):
"""Download files and folders from remote to local. Runs in background thread."""
if not self._sftp or not self._sftp.connected or self._transferring:
return
items = [s for s in items if s["name"] != ".."]
if not items:
return
self._transferring = True
self._upload_btn.configure(state="disabled")
self._download_btn.configure(state="disabled")
def _do():
for item in items:
name = item["name"]
remote = self._remote_join(name)
local = os.path.join(self._local_path, name)
try:
if item.get("is_dir"):
self.after(0, lambda n=name: self._transfer_label.configure(
text=t("downloading_dir").format(name=n)
))
os.makedirs(local, exist_ok=True)
def _file_cb(cur, total, fname, n=name):
self.after(0, lambda c=cur, tt=total, fn=fname:
self._transfer_label.configure(
text=t("transfer_file_progress").format(cur=c, total=tt, name=fn)
))
def _progress(transferred, total):
if total > 0:
self.after(0, lambda f=transferred/total: self._progress.set(f))
self._sftp.download_dir(remote, local, progress_cb=_progress, file_cb=_file_cb)
else:
self.after(0, lambda n=name: self._transfer_label.configure(
text=t("downloading").format(name=n)
))
def _progress(transferred, total, n=name):
if total > 0:
frac = transferred / total
size_str = f"{_format_size(transferred)}/{_format_size(total)}"
self.after(0, lambda f=frac, s=size_str, nn=n:
self._update_progress(f, t("downloading").format(name=nn) + f" ({s})"))
if self._sftp.sudo_mode:
try:
self._sftp.download_sudo(remote, local, progress_cb=_progress)
except Exception:
self._sftp.download(remote, local, progress_cb=_progress)
else:
self._sftp.download(remote, local, progress_cb=_progress)
self.after(0, lambda n=name: self._log_msg(
t("transfer_done").format(name=n)
))
except PermissionError as e:
hint = f" - {t('try_sudo_hint')}" if not self._sftp.sudo_mode else ""
self.after(0, lambda e=e, h=hint: self._log_msg(
t("transfer_failed").format(e=str(e)) + h
))
except Exception as e:
self.after(0, lambda e=e: self._log_msg(
t("transfer_failed").format(e=str(e))
))
self.after(0, self._on_transfer_done)
threading.Thread(target=_do, daemon=True).start()
# ── Button handlers ──
def _upload_selected(self):
if not self._sftp or not self._sftp.connected or self._transferring:
return
selected = self._local_list.get_selected()
files = [s for s in selected if not s.get("is_dir") and s["name"] != ".."]
if not files:
items = [s for s in selected if s["name"] != ".."]
if not items:
self._log_msg(t("no_server_selected"))
return
self._transferring = True
self._upload_btn.configure(state="disabled")
self._download_btn.configure(state="disabled")
def _do():
for item in files:
name = item["name"]
local = os.path.join(self._local_path, name)
if self._remote_path == "/":
remote = "/" + name
else:
remote = self._remote_path + "/" + name
try:
file_size = os.path.getsize(local)
self.after(0, lambda n=name: self._transfer_label.configure(
text=t("uploading").format(name=n)
))
def _progress(transferred, total, n=name):
if total > 0:
frac = transferred / total
size_str = f"{_format_size(transferred)}/{_format_size(total)}"
self.after(0, lambda f=frac, s=size_str, nn=n:
self._update_progress(f, t("uploading").format(name=nn) + f" ({s})"))
self._sftp.upload(local, remote, progress_cb=_progress)
self.after(0, lambda n=name: self._log_msg(
t("transfer_done").format(name=n)
))
except Exception as e:
self.after(0, lambda e=e: self._log_msg(
t("transfer_failed").format(e=str(e))
))
self.after(0, self._on_transfer_done)
threading.Thread(target=_do, daemon=True).start()
self._do_upload(items)
def _download_selected(self):
if not self._sftp or not self._sftp.connected or self._transferring:
return
selected = self._remote_list.get_selected()
files = [s for s in selected if not s.get("is_dir") and s["name"] != ".."]
if not files:
items = [s for s in selected if s["name"] != ".."]
if not items:
return
self._do_download(items)
self._transferring = True
self._upload_btn.configure(state="disabled")
self._download_btn.configure(state="disabled")
# ── Drag-and-drop callbacks ──
def _do():
for item in files:
name = item["name"]
if self._remote_path == "/":
remote = "/" + name
else:
remote = self._remote_path + "/" + name
local = os.path.join(self._local_path, name)
try:
self.after(0, lambda n=name: self._transfer_label.configure(
text=t("downloading").format(name=n)
))
def _on_drop_to_remote(self, items: list[dict], source):
"""Called when items are dropped onto remote panel (upload)."""
self._do_upload(items)
def _progress(transferred, total, n=name):
if total > 0:
frac = transferred / total
size_str = f"{_format_size(transferred)}/{_format_size(total)}"
self.after(0, lambda f=frac, s=size_str, nn=n:
self._update_progress(f, t("downloading").format(name=nn) + f" ({s})"))
self._sftp.download(remote, local, progress_cb=_progress)
self.after(0, lambda n=name: self._log_msg(
t("transfer_done").format(name=n)
))
except Exception as e:
self.after(0, lambda e=e: self._log_msg(
t("transfer_failed").format(e=str(e))
))
self.after(0, self._on_transfer_done)
threading.Thread(target=_do, daemon=True).start()
def _on_drop_to_local(self, items: list[dict], source):
"""Called when items are dropped onto local panel (download)."""
self._do_download(items)
def _on_transfer_done(self):
self._transferring = False
@@ -522,10 +703,7 @@ class FilesTab(ctk.CTkFrame):
if not name or not name.strip():
return
name = name.strip()
if self._remote_path == "/":
path = "/" + name
else:
path = self._remote_path + "/" + name
path = self._remote_join(name)
def _do():
try:
@@ -543,22 +721,29 @@ class FilesTab(ctk.CTkFrame):
items = [s for s in selected if s["name"] != ".."]
if not items:
return
if not messagebox.askyesno(
t("delete_files"),
t("delete_files_confirm").format(count=len(items)),
):
return
# Check for directories — use recursive delete confirm
has_dirs = any(s.get("is_dir") for s in items)
if has_dirs and len(items) == 1 and items[0].get("is_dir"):
if not messagebox.askyesno(
t("delete_files"),
t("recursive_delete_confirm").format(name=items[0]["name"]),
):
return
else:
if not messagebox.askyesno(
t("delete_files"),
t("delete_files_confirm").format(count=len(items)),
):
return
def _do():
for item in items:
name = item["name"]
if self._remote_path == "/":
path = "/" + name
else:
path = self._remote_path + "/" + name
path = self._remote_join(name)
try:
if item.get("is_dir"):
self._sftp.rmdir(path)
self._sftp.rmdir_recursive(path)
else:
self._sftp.remove(path)
except Exception as e:
@@ -583,12 +768,8 @@ class FilesTab(ctk.CTkFrame):
if not new_name or not new_name.strip() or new_name.strip() == old_name:
return
new_name = new_name.strip()
if self._remote_path == "/":
old_path = "/" + old_name
new_path = "/" + new_name
else:
old_path = self._remote_path + "/" + old_name
new_path = self._remote_path + "/" + new_name
old_path = self._remote_join(old_name)
new_path = self._remote_join(new_name)
def _do():
try: