#!/usr/bin/env python3 import os import sys import zipfile import subprocess import threading import tkinter as tk from tkinter import filedialog, messagebox, ttk from pathlib import Path from queue import Queue from concurrent.futures import ThreadPoolExecutor, as_completed from collections import defaultdict try: import ttkbootstrap as tb from ttkbootstrap.constants import * TTKBOOTSTRAP_AVAILABLE = True except ImportError: TTKBOOTSTRAP_AVAILABLE = False print("⚠️ ttkbootstrap not installed. Install with: pip install ttkbootstrap") # ===== CONFIG — EDIT THESE ===== REMOTE_USER = "sourav" REMOTE_HOST = "100.124.92.67" REMOTE_DIR = "shared_files" MAX_SIZE_GB = 2.5 # =============================== # Modern font configuration if sys.platform == "win32": DEFAULT_FONT = ("Segoe UI", 10) else: DEFAULT_FONT = ("Inter", 10) # Fallback for other platforms def format_size(size_bytes): """Format file size in human-readable format""" for unit in ['B', 'KB', 'MB', 'GB']: if size_bytes < 1024.0: return f"{size_bytes:.1f} {unit}" size_bytes /= 1024.0 return f"{size_bytes:.1f} TB" def is_scp_available(): try: subprocess.run(["scp", "-V"], capture_output=True, timeout=5) return True except FileNotFoundError: return False def get_file_icon(filename): ext = Path(filename).suffix.lower() if ext in ('.jpg', '.jpeg', '.png', '.gif', '.bmp', '.webp'): return "🖼️" elif ext in ('.zip', '.tar', '.gz'): return "📦" elif ext in ('.mp4', '.mov', '.avi'): return "🎥" elif ext in ('.pdf',): return "📕" else: return "📄" class FileManagerApp: def __init__(self, root): self.root = root root.title("Tailscale File Manager") root.geometry("780x540") # Set default font root.option_add("*Font", DEFAULT_FONT) # Configure ttk style if using standard ttk if not TTKBOOTSTRAP_AVAILABLE: style = ttk.Style() if sys.platform == "win32": style.theme_use("vista") else: style.theme_use("clam") self.progress_win = None self.cancel_event = threading.Event() self.current_path = REMOTE_DIR # Track current directory # Queue system for uploads/downloads self.task_queue = Queue() self.task_lock = threading.Lock() self.active_tasks = defaultdict(dict) # task_id -> task_info self.task_counter = 0 # Start queue worker thread self.worker_thread = threading.Thread(target=self._queue_worker, daemon=True) self.worker_thread.start() # Progress tracking self.progress_items = {} # task_id -> progress_widgets # Upload section top = ttk.Frame(root) top.pack(pady=10, fill="x", padx=15) ButtonClass = tb.Button if TTKBOOTSTRAP_AVAILABLE else ttk.Button CheckbuttonClass = tb.Checkbutton if TTKBOOTSTRAP_AVAILABLE else ttk.Checkbutton upload_btn = ButtonClass(top, text="Upload Files", command=self.browse_files, width=14) if TTKBOOTSTRAP_AVAILABLE: upload_btn.config(bootstyle="primary") upload_btn.pack(side="left", padx=5) upload_folder_btn = ButtonClass(top, text="Upload Folder", command=self.browse_folders, width=14) if TTKBOOTSTRAP_AVAILABLE: upload_folder_btn.config(bootstyle="primary") upload_folder_btn.pack(side="left", padx=5) self.make_public = tk.BooleanVar() CheckbuttonClass(top, text="Public", variable=self.make_public).pack(side="left", padx=10) refresh_btn = ButtonClass(top, text="Refresh", command=lambda: threading.Thread(target=self._refresh_list_async, daemon=True).start(), width=10) if TTKBOOTSTRAP_AVAILABLE: refresh_btn.config(bootstyle="secondary") refresh_btn.pack(side="left", padx=5) # File list with Treeview list_frame = ttk.Frame(root) list_frame.pack(fill="both", expand=True, padx=15, pady=5) # Create Treeview with columns tree_frame = ttk.Frame(list_frame) tree_frame.pack(side="left", fill="both", expand=True) # Scrollbar for treeview scrollbar = ttk.Scrollbar(tree_frame) scrollbar.pack(side="right", fill="y") self.file_tree = ttk.Treeview( tree_frame, columns=("Select", "Type", "Size", "Public"), show="tree headings", height=16, yscrollcommand=scrollbar.set ) scrollbar.config(command=self.file_tree.yview) # Configure columns self.file_tree.heading("#0", text="Name") self.file_tree.heading("Select", text="☐") self.file_tree.heading("Type", text="Type") self.file_tree.heading("Size", text="Size") self.file_tree.heading("Public", text="Public") self.file_tree.column("#0", width=250, anchor="w") self.file_tree.column("Select", width=50, anchor="center") self.file_tree.column("Type", width=100, anchor="w") self.file_tree.column("Size", width=100, anchor="e") self.file_tree.column("Public", width=80, anchor="center") # Track checked items self.checked_items = set() self.file_tree.pack(fill="both", expand=True) self.file_tree.bind("", self.on_double_click) self.file_tree.bind("", self.on_tree_click) # Bind header click for select all self.file_tree.heading("Select", command=self.toggle_select_all) # Action buttons btn_frame = ttk.Frame(list_frame) btn_frame.pack(side="right", padx=(10, 0)) self.download_btn = ButtonClass(btn_frame, text="Download", command=self.download_selected, width=12) if TTKBOOTSTRAP_AVAILABLE: self.download_btn.config(bootstyle="secondary") self.download_btn.pack(pady=5, fill="x") self.delete_btn = ButtonClass(btn_frame, text="Delete", command=self.delete_selected, width=12) if TTKBOOTSTRAP_AVAILABLE: self.delete_btn.config(bootstyle="danger") self.delete_btn.pack(pady=5, fill="x") self.link_btn = ButtonClass(btn_frame, text="Copy Link", command=self.copy_public_link, width=12) if TTKBOOTSTRAP_AVAILABLE: self.link_btn.config(bootstyle="info") self.link_btn.pack(pady=5, fill="x") # Select all/none buttons select_all_btn = ButtonClass(btn_frame, text="Select All", command=self.select_all_items, width=12) if TTKBOOTSTRAP_AVAILABLE: select_all_btn.config(bootstyle="secondary") select_all_btn.pack(pady=5, fill="x") select_none_btn = ButtonClass(btn_frame, text="Select None", command=self.deselect_all_items, width=12) if TTKBOOTSTRAP_AVAILABLE: select_none_btn.config(bootstyle="secondary") select_none_btn.pack(pady=5, fill="x") # Status self.status = ttk.Label(root, text="Ready", relief="sunken", anchor="w", padding=5) self.status.pack(side="bottom", fill="x") if not is_scp_available(): messagebox.showerror("Error", "scp not found. Install Git for Windows with Unix tools.") root.destroy() else: # Run initial refresh in background to avoid blocking UI startup threading.Thread(target=self._refresh_list_async, daemon=True).start() def _ensure_progress_window(self): """Ensure progress window exists (non-modal, supports multiple tasks)""" if self.progress_win is None or not self.progress_win.winfo_exists(): ToplevelClass = tb.Toplevel if TTKBOOTSTRAP_AVAILABLE else tk.Toplevel self.progress_win = ToplevelClass(self.root) self.progress_win.title("Upload/Download Progress") self.progress_win.geometry("500x400") self.progress_win.transient(self.root) # Don't use grab_set() - allow multiple operations self.progress_win.resizable(True, True) # Scrollable frame for multiple progress items self.progress_canvas = tk.Canvas(self.progress_win) scrollbar = ttk.Scrollbar(self.progress_win, orient="vertical", command=self.progress_canvas.yview) self.progress_frame = ttk.Frame(self.progress_canvas) def on_frame_configure(event): self.progress_canvas.configure(scrollregion=self.progress_canvas.bbox("all")) self.progress_frame.bind("", on_frame_configure) self.progress_canvas.create_window((0, 0), window=self.progress_frame, anchor="nw") self.progress_canvas.configure(yscrollcommand=scrollbar.set) self.progress_canvas.pack(side="left", fill="both", expand=True) scrollbar.pack(side="right", fill="y") return self.progress_win def _add_progress_item(self, task_id, filename): """Add a new progress bar item for a task""" def _add(): self._ensure_progress_window() item_frame = ttk.Frame(self.progress_frame) item_frame.pack(fill="x", padx=10, pady=5) LabelClass = ttk.Label if not TTKBOOTSTRAP_AVAILABLE else tb.Label ProgressbarClass = ttk.Progressbar if not TTKBOOTSTRAP_AVAILABLE else tb.Progressbar label = LabelClass(item_frame, text=filename[:50], anchor="w") label.pack(fill="x") progress_bar = ProgressbarClass(item_frame, mode='determinate', length=400) progress_bar.pack(fill="x", pady=2) status_label = LabelClass(item_frame, text="Queued...", anchor="w", font=("", 8)) status_label.pack(fill="x") self.progress_items[task_id] = { 'frame': item_frame, 'label': label, 'progress': progress_bar, 'status': status_label } # Auto-scroll to bottom if hasattr(self, 'progress_canvas'): self.progress_canvas.update_idletasks() self.progress_canvas.yview_moveto(1.0) if threading.current_thread() == threading.main_thread(): _add() else: self.root.after(0, _add) def _update_progress_item(self, task_id, percentage=None, status=None): """Update progress for a specific task""" def _update(): if task_id in self.progress_items: item = self.progress_items[task_id] if percentage is not None: item['progress']['value'] = percentage if status: item['status'].config(text=status) if threading.current_thread() == threading.main_thread(): _update() else: self.root.after(0, _update) def _remove_progress_item(self, task_id): """Remove progress item when task completes""" def _remove(): if task_id in self.progress_items: self.progress_items[task_id]['frame'].destroy() del self.progress_items[task_id] # Close progress window if no more items if not self.progress_items and self.progress_win: self.progress_win.destroy() self.progress_win = None if threading.current_thread() == threading.main_thread(): _remove() else: self.root.after(0, _remove) def _queue_worker(self): """Worker thread that processes tasks from the queue sequentially""" while True: task = self.task_queue.get() if task is None: # Shutdown signal break task_type = task['type'] task_id = task['id'] try: if task_type == 'upload': self._process_upload_task(task) elif task_type == 'download': self._process_download_task(task) except Exception as e: self._update_progress_item(task_id, status=f"Error: {str(e)}") finally: self.task_queue.task_done() def _process_upload_task(self, task): """Process a single upload task""" task_id = task['id'] file_path = task['file_path'] make_public = task.get('make_public', False) try: src = Path(file_path) temp_file = src item_name = src.name # Update status self._update_progress_item(task_id, percentage=0, status="Preparing...") # Handle directories - always zip them if src.is_dir(): self._update_progress_item(task_id, percentage=10, status=f"Zipping folder {src.name[:30]}...") zip_path = src.parent / (src.name + ".zip") with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED, compresslevel=1) as zf: files_in_dir = list(src.rglob('*')) total_files = len([f for f in files_in_dir if f.is_file()]) for idx, file_path_obj in enumerate(files_in_dir): if file_path_obj.is_file(): arcname = file_path_obj.relative_to(src) zf.write(file_path_obj, arcname=arcname) progress = 10 + int((idx / total_files) * 20) if total_files > 0 else 10 self._update_progress_item(task_id, percentage=progress, status=f"Zipping {idx+1}/{total_files} files...") temp_file = zip_path item_name = zip_path.name elif src.is_file(): # For files, check size and zip if needed size_gb = src.stat().st_size / (1024**3) if size_gb > MAX_SIZE_GB: self._update_progress_item(task_id, percentage=10, status=f"Zipping {src.name[:30]}...") zip_path = src.parent / (src.stem + ".zip") with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED, compresslevel=1) as zf: zf.write(src, arcname=src.name) temp_file = zip_path item_name = zip_path.name else: raise Exception(f"Path does not exist: {src}") # Upload remote_dest = f"{REMOTE_USER}@{REMOTE_HOST}:{self.current_path}/{item_name}" file_size = temp_file.stat().st_size if temp_file.exists() else 0 self._update_progress_item(task_id, percentage=30, status=f"Uploading {item_name[:30]}...") cmd = ["scp", str(temp_file), remote_dest] process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) # Simulate progress during upload (we can't get real progress from scp easily) # Use a progress simulation thread def simulate_progress(): import time progress = 30 while process.poll() is None and progress < 85: time.sleep(0.5) # Update every 500ms progress += 2 if progress > 85: progress = 85 self._update_progress_item(task_id, percentage=progress, status="Uploading...") progress_thread = threading.Thread(target=simulate_progress, daemon=True) progress_thread.start() stdout, stderr = process.communicate(timeout=600) if process.returncode != 0: raise Exception(stderr.decode() if stderr else "SCP failed") self._update_progress_item(task_id, percentage=90, status="Finalizing...") if make_public: marker = f".public.{item_name}" subprocess.run([ "ssh", f"{REMOTE_USER}@{REMOTE_HOST}", f"touch", f"{self.current_path}/{marker}" ], check=True, timeout=10) if temp_file != src: temp_file.unlink() self._update_progress_item(task_id, percentage=100, status="Completed!") # Auto-remove after 2 seconds threading.Timer(2.0, lambda: self._remove_progress_item(task_id)).start() except Exception as e: self._update_progress_item(task_id, status=f"Failed: {str(e)[:50]}") raise def _process_download_task(self, task): """Process a single download task""" task_id = task['id'] filename = task['filename'] save_path = task['save_path'] try: self._update_progress_item(task_id, percentage=0, status="Starting download...") remote = f"{REMOTE_USER}@{REMOTE_HOST}:{self.current_path}/{filename}" self._update_progress_item(task_id, percentage=10, status="Connecting...") cmd = ["scp", remote, save_path] process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) # Simulate progress during download def simulate_progress(): import time progress = 10 while process.poll() is None and progress < 90: time.sleep(0.5) # Update every 500ms progress += 3 if progress > 90: progress = 90 self._update_progress_item(task_id, percentage=progress, status="Downloading...") progress_thread = threading.Thread(target=simulate_progress, daemon=True) progress_thread.start() stdout, stderr = process.communicate(timeout=600) if process.returncode != 0: raise Exception(stderr.decode() if stderr else "Download failed") self._update_progress_item(task_id, percentage=100, status="Completed!") # Auto-remove after 2 seconds threading.Timer(2.0, lambda: self._remove_progress_item(task_id)).start() except Exception as e: self._update_progress_item(task_id, status=f"Failed: {str(e)[:50]}") raise def browse_files(self): # Allow selecting both files and directories paths = filedialog.askopenfilenames(title="Select files or folders to upload") if paths: self.queue_uploads(paths, parallel=True) def browse_folders(self): # Alternative method to select folders specifically folder_path = filedialog.askdirectory(title="Select folder to upload") if folder_path: self.queue_uploads([folder_path], parallel=False) def queue_uploads(self, file_paths, parallel=False): """Queue files/folders for upload. If parallel=True and multiple files, upload in parallel.""" make_public = self.make_public.get() if parallel and len(file_paths) > 1: # Parallel upload for multiple files self._upload_parallel(file_paths, make_public) else: # Sequential upload via queue for file_path in file_paths: with self.task_lock: self.task_counter += 1 task_id = self.task_counter task = { 'type': 'upload', 'id': task_id, 'file_path': file_path, 'make_public': make_public } filename = Path(file_path).name self._add_progress_item(task_id, filename) self.task_queue.put(task) # Refresh list when all tasks in this batch complete if file_path == file_paths[-1]: threading.Thread( target=self._wait_and_refresh, args=(len(file_paths),), daemon=True ).start() def _upload_parallel(self, file_paths, make_public): """Upload multiple files in parallel using ThreadPoolExecutor - runs in background thread""" def _do_parallel_upload(): with ThreadPoolExecutor(max_workers=min(5, len(file_paths))) as executor: futures = {} for file_path in file_paths: with self.task_lock: self.task_counter += 1 task_id = self.task_counter filename = Path(file_path).name self.root.after(0, lambda tid=task_id, fn=filename: self._add_progress_item(tid, fn)) task = { 'type': 'upload', 'id': task_id, 'file_path': file_path, 'make_public': make_public } # Submit to executor for parallel execution future = executor.submit(self._process_upload_task, task) futures[future] = task_id # Wait for all to complete (non-blocking for UI) for future in as_completed(futures): task_id = futures[future] try: future.result() except Exception as e: self._update_progress_item(task_id, status=f"Error: {str(e)[:50]}") # Refresh list when all parallel uploads complete threading.Thread(target=self._wait_and_refresh, args=(1,), daemon=True).start() # Run in background thread to avoid blocking threading.Thread(target=_do_parallel_upload, daemon=True).start() def _wait_and_refresh(self, delay_seconds=1): """Wait a bit then refresh the file list""" import time time.sleep(delay_seconds) self.root.after(0, lambda: threading.Thread(target=self._refresh_list_async, daemon=True).start()) def refresh_list(self, async_mode=False): """Refresh file list. If async_mode=True, runs in background thread.""" if async_mode: threading.Thread(target=self._refresh_list_async, daemon=True).start() return # Synchronous version for direct calls # Clear existing items and checked items for item in self.file_tree.get_children(): self.file_tree.delete(item) self.checked_items.clear() try: # Show parent directory option if not at root if self.current_path != REMOTE_DIR: self.file_tree.insert("", "end", values=("☐", "Directory", "", ""), text="..", tags=("dir",)) # List files and directories - use ls -1p which shows directories with trailing / safe_path = self.current_path.replace("'", "'\\''") cmd = ["ssh", f"{REMOTE_USER}@{REMOTE_HOST}", f"ls -1p '{safe_path}' 2>/dev/null"] result = subprocess.run(cmd, capture_output=True, text=True, timeout=10) items = [] for line in result.stdout.strip().split('\n'): if not line.strip() or line.strip().startswith('.public.'): continue name = line.rstrip('/').strip() is_dir = line.endswith('/') items.append((name, is_dir)) # Sort: directories first, then files items.sort(key=lambda x: (not x[1], x[0].lower())) # Get file info (sizes and public status) - skip sizes initially to avoid blocking # We'll show files first, then update sizes in background if needed file_names = [name for name, is_dir in items if not is_dir] file_info = {} # Get file sizes and public status in batch if file_names: safe_names = [name.replace("'", "'\\''") for name in file_names] # Build a command that gets both size and public status for each file script_parts = [] for safe_name in safe_names: file_path = f"'{safe_path}/{safe_name}'" marker_path = f"'{safe_path}/.public.{safe_name}'" script_parts.append( f"if [ -f {file_path} ]; then " f"SIZE=$(stat -c%s {file_path} 2>/dev/null || stat -f%z {file_path} 2>/dev/null || echo 0); " f"PUB=$([ -f {marker_path} ] && echo 1 || echo 0); " f"echo '{safe_name}|$SIZE|$PUB'; " f"fi" ) # Use semicolon to ensure all commands run even if one fails batch_cmd = "; ".join(script_parts) try: batch_res = subprocess.run( ["ssh", f"{REMOTE_USER}@{REMOTE_HOST}", batch_cmd], capture_output=True, text=True, timeout=15 ) if batch_res.returncode == 0 and batch_res.stdout.strip(): for line in batch_res.stdout.strip().split('\n'): if '|' in line and line.strip(): parts = line.split('|', 2) if len(parts) >= 3: name = parts[0].strip() size_str = parts[1].strip() pub_str = parts[2].strip() try: # Handle size string - remove any whitespace and check if it's a number if size_str and size_str.isdigit(): size_bytes = int(size_str) file_info[name] = { "size": format_size(size_bytes), "public": pub_str == "1" } else: file_info[name] = {"size": "", "public": pub_str == "1"} except (ValueError, AttributeError, IndexError) as e: file_info[name] = {"size": "", "public": False} # Fill in any missing files with individual queries for name in file_names: if name not in file_info: safe_name = name.replace("'", "'\\''") file_path = f"'{safe_path}/{safe_name}'" marker_path = f"'{safe_path}/.public.{safe_name}'" try: size_cmd = f"stat -c%s {file_path} 2>/dev/null || stat -f%z {file_path} 2>/dev/null || echo 0" size_res = subprocess.run( ["ssh", f"{REMOTE_USER}@{REMOTE_HOST}", size_cmd], capture_output=True, text=True, timeout=3 ) size_bytes = int(size_res.stdout.strip()) if size_res.stdout.strip().isdigit() else 0 size = format_size(size_bytes) if size_bytes >= 0 else "" pub_cmd = f"[ -f {marker_path} ] && echo 1 || echo 0" pub_res = subprocess.run( ["ssh", f"{REMOTE_USER}@{REMOTE_HOST}", pub_cmd], capture_output=True, text=True, timeout=3 ) is_public = pub_res.stdout.strip() == "1" file_info[name] = {"size": size, "public": is_public} except: file_info[name] = {"size": "", "public": False} except (subprocess.TimeoutExpired, Exception) as e: # If batch fails, try individual calls for name in file_names: if name not in file_info: safe_name = name.replace("'", "'\\''") file_path = f"'{safe_path}/{safe_name}'" marker_path = f"'{safe_path}/.public.{safe_name}'" try: size_cmd = f"stat -c%s {file_path} 2>/dev/null || stat -f%z {file_path} 2>/dev/null || echo 0" size_res = subprocess.run( ["ssh", f"{REMOTE_USER}@{REMOTE_HOST}", size_cmd], capture_output=True, text=True, timeout=3 ) size_bytes = int(size_res.stdout.strip()) if size_res.stdout.strip().isdigit() else 0 size = format_size(size_bytes) if size_bytes >= 0 else "" pub_cmd = f"[ -f {marker_path} ] && echo 1 || echo 0" pub_res = subprocess.run( ["ssh", f"{REMOTE_USER}@{REMOTE_HOST}", pub_cmd], capture_output=True, text=True, timeout=3 ) is_public = pub_res.stdout.strip() == "1" file_info[name] = {"size": size, "public": is_public} except: file_info[name] = {"size": "", "public": False} # Insert items into treeview for name, is_dir in items: if is_dir: item_type = "Directory" size = "" public = "" tags = ("dir",) display_name = "📁 " + name else: # Get file extension for type ext = Path(name).suffix.lower() if ext: item_type = ext[1:].upper() + " File" else: item_type = "File" info = file_info.get(name, {}) size = info.get("size", "") if info else "" # If size is still empty and file not in file_info, try one more time if not size and name not in file_info: # Initialize with empty, will be populated if async refresh is used size = "" file_info[name] = {"size": "", "public": False} is_public = info.get("public", False) if info else False public = "Yes" if is_public else "No" tags = ("public",) if is_public else ("private",) display_name = get_file_icon(name) + " " + name self.file_tree.insert("", "end", values=("☐", item_type, size, public), text=display_name, tags=tags) # Configure tag colors if using ttkbootstrap if TTKBOOTSTRAP_AVAILABLE: self.file_tree.tag_configure("public", foreground="#28a745") self.file_tree.tag_configure("dir", foreground="#007bff") path_display = self.current_path if self.current_path == REMOTE_DIR else f"{REMOTE_DIR}/.../{self.current_path.split('/')[-1]}" self.status.config(text=f"{len(items)} items in {path_display}") except Exception as e: self.file_tree.insert("", "end", values=("☐", "Error", "", ""), text=f"⚠️ {e}", tags=("error",)) self.status.config(text="Refresh failed") def _refresh_list_async(self): """Wrapper to run refresh_list in a thread and update UI on main thread""" try: # Run the actual refresh items_data = [] try: # Show parent directory option if not at root if self.current_path != REMOTE_DIR: items_data.append(("..", True, "Directory", "", "", ("dir",))) # List files and directories safe_path = self.current_path.replace("'", "'\\''") cmd = ["ssh", f"{REMOTE_USER}@{REMOTE_HOST}", f"ls -1p '{safe_path}' 2>/dev/null"] result = subprocess.run(cmd, capture_output=True, text=True, timeout=10) items = [] for line in result.stdout.strip().split('\n'): if not line.strip() or line.strip().startswith('.public.'): continue name = line.rstrip('/').strip() is_dir = line.endswith('/') items.append((name, is_dir)) items.sort(key=lambda x: (not x[1], x[0].lower())) # Get file info file_names = [name for name, is_dir in items if not is_dir] file_info = {} if file_names: safe_names = [name.replace("'", "'\\''") for name in file_names] script_parts = [] for safe_name in safe_names: file_path = f"'{safe_path}/{safe_name}'" marker_path = f"'{safe_path}/.public.{safe_name}'" script_parts.append( f"if [ -f {file_path} ]; then " f"SIZE=$(stat -c%s {file_path} 2>/dev/null || stat -f%z {file_path} 2>/dev/null || echo 0); " f"PUB=$([ -f {marker_path} ] && echo 1 || echo 0); " f"echo '{safe_name}|$SIZE|$PUB'; " f"fi" ) # Use semicolon to ensure all commands run even if one fails batch_cmd = "; ".join(script_parts) try: batch_res = subprocess.run( ["ssh", f"{REMOTE_USER}@{REMOTE_HOST}", batch_cmd], capture_output=True, text=True, timeout=15 ) if batch_res.returncode == 0 and batch_res.stdout.strip(): for line in batch_res.stdout.strip().split('\n'): if '|' in line and line.strip(): parts = line.split('|', 2) if len(parts) >= 3: name = parts[0].strip() size_str = parts[1].strip() pub_str = parts[2].strip() try: # Handle size string - remove any whitespace and check if it's a number if size_str and size_str.isdigit(): size_bytes = int(size_str) file_info[name] = { "size": format_size(size_bytes), "public": pub_str == "1" } else: file_info[name] = {"size": "", "public": pub_str == "1"} except (ValueError, AttributeError, IndexError) as e: file_info[name] = {"size": "", "public": False} # Fill in any missing files with individual queries for name in file_names: if name not in file_info: safe_name = name.replace("'", "'\\''") file_path = f"'{safe_path}/{safe_name}'" marker_path = f"'{safe_path}/.public.{safe_name}'" try: size_cmd = f"stat -c%s {file_path} 2>/dev/null || stat -f%z {file_path} 2>/dev/null || echo 0" size_res = subprocess.run( ["ssh", f"{REMOTE_USER}@{REMOTE_HOST}", size_cmd], capture_output=True, text=True, timeout=3 ) size_bytes = int(size_res.stdout.strip()) if size_res.stdout.strip().isdigit() else 0 size = format_size(size_bytes) if size_bytes > 0 else "" pub_cmd = f"[ -f {marker_path} ] && echo 1 || echo 0" pub_res = subprocess.run( ["ssh", f"{REMOTE_USER}@{REMOTE_HOST}", pub_cmd], capture_output=True, text=True, timeout=3 ) is_public = pub_res.stdout.strip() == "1" file_info[name] = {"size": size, "public": is_public} except: file_info[name] = {"size": "", "public": False} except (subprocess.TimeoutExpired, Exception): # If batch fails, try individual calls for name in file_names: if name not in file_info: safe_name = name.replace("'", "'\\''") file_path = f"'{safe_path}/{safe_name}'" marker_path = f"'{safe_path}/.public.{safe_name}'" try: size_cmd = f"stat -c%s {file_path} 2>/dev/null || stat -f%z {file_path} 2>/dev/null || echo 0" size_res = subprocess.run( ["ssh", f"{REMOTE_USER}@{REMOTE_HOST}", size_cmd], capture_output=True, text=True, timeout=3 ) size_bytes = int(size_res.stdout.strip()) if size_res.stdout.strip().isdigit() else 0 size = format_size(size_bytes) if size_bytes > 0 else "" pub_cmd = f"[ -f {marker_path} ] && echo 1 || echo 0" pub_res = subprocess.run( ["ssh", f"{REMOTE_USER}@{REMOTE_HOST}", pub_cmd], capture_output=True, text=True, timeout=3 ) is_public = pub_res.stdout.strip() == "1" file_info[name] = {"size": size, "public": is_public} except: file_info[name] = {"size": "", "public": False} # Build items data for name, is_dir in items: if is_dir: items_data.append((name, is_dir, "☐", "Directory", "", "", ("dir",))) else: ext = Path(name).suffix.lower() item_type = ext[1:].upper() + " File" if ext else "File" info = file_info.get(name, {}) size = info.get("size", "") is_public = info.get("public", False) public = "Yes" if is_public else "No" tags = ("public",) if is_public else ("private",) items_data.append((name, is_dir, "☐", item_type, size, public, tags)) # Update UI on main thread def update_ui(): for item in self.file_tree.get_children(): self.file_tree.delete(item) for name, is_dir, checkbox, item_type, size, public, tags in items_data: if is_dir: display_name = "📁 " + name else: display_name = get_file_icon(name) + " " + name self.file_tree.insert("", "end", values=(checkbox, item_type, size, public), text=display_name, tags=tags) if TTKBOOTSTRAP_AVAILABLE: self.file_tree.tag_configure("public", foreground="#28a745") self.file_tree.tag_configure("dir", foreground="#007bff") path_display = self.current_path if self.current_path == REMOTE_DIR else f"{REMOTE_DIR}/.../{self.current_path.split('/')[-1]}" self.status.config(text=f"{len(items)} items in {path_display}") self.root.after(0, update_ui) except Exception as e: def show_error(): for item in self.file_tree.get_children(): self.file_tree.delete(item) self.file_tree.insert("", "end", values=("☐", "Error", "", ""), text=f"⚠️ {e}", tags=("error",)) self.status.config(text="Refresh failed") self.root.after(0, show_error) except Exception as e: def show_error(): self.status.config(text=f"Refresh error: {e}") self.root.after(0, show_error) def on_double_click(self, event): sel = self.file_tree.selection() if not sel: return item = sel[0] values = self.file_tree.item(item, "values") text = self.file_tree.item(item, "text") # Extract name (remove icon emoji) name = text.split(" ", 1)[1] if " " in text else text if name == "..": # Navigate up if self.current_path != REMOTE_DIR: parts = self.current_path.split('/') self.current_path = '/'.join(parts[:-1]) if len(parts) > 1 else REMOTE_DIR self.status.config(text="Loading...") threading.Thread(target=self._refresh_list_async, daemon=True).start() elif len(values) > 1 and values[1] == "Directory": # Navigate into folder new_path = f"{self.current_path}/{name}" self.current_path = new_path self.status.config(text="Loading...") threading.Thread(target=self._refresh_list_async, daemon=True).start() def on_tree_click(self, event): """Handle clicks on the treeview, specifically checkbox column""" region = self.file_tree.identify_region(event.x, event.y) if region == "cell": column = self.file_tree.identify_column(event.x) item = self.file_tree.identify_row(event.y) # Column index: #1 is checkbox column (0-indexed: 1) if column == "#1" and item: values = list(self.file_tree.item(item, "values")) if len(values) > 0: current_check = values[0] new_check = "☑" if current_check == "☐" else "☐" values[0] = new_check self.file_tree.item(item, values=values) # Update checked items set text = self.file_tree.item(item, "text") name = text.split(" ", 1)[1] if " " in text else text if new_check == "☑": self.checked_items.add(item) else: self.checked_items.discard(item) def toggle_select_all(self): """Toggle select all items""" all_items = self.file_tree.get_children() if not all_items: return # Check if all are selected all_selected = all(len(self.file_tree.item(item, "values")) > 0 and self.file_tree.item(item, "values")[0] == "☑" for item in all_items) new_check = "☐" if all_selected else "☑" for item in all_items: values = list(self.file_tree.item(item, "values")) if len(values) > 0: text = self.file_tree.item(item, "text") name = text.split(" ", 1)[1] if " " in text else text # Don't select ".." directory if name != "..": values[0] = new_check self.file_tree.item(item, values=values) if new_check == "☑": self.checked_items.add(item) else: self.checked_items.discard(item) def select_all_items(self): """Select all items""" all_items = self.file_tree.get_children() for item in all_items: values = list(self.file_tree.item(item, "values")) if len(values) > 0: text = self.file_tree.item(item, "text") name = text.split(" ", 1)[1] if " " in text else text if name != "..": values[0] = "☑" self.file_tree.item(item, values=values) self.checked_items.add(item) def deselect_all_items(self): """Deselect all items""" all_items = self.file_tree.get_children() for item in all_items: values = list(self.file_tree.item(item, "values")) if len(values) > 0: values[0] = "☐" self.file_tree.item(item, values=values) self.checked_items.clear() def get_checked_items(self): """Get all checked items. Returns list of (name, is_directory, item_id) tuples.""" checked = [] for item in self.checked_items: if item in self.file_tree.get_children() or item in self.file_tree.get_children(""): values = self.file_tree.item(item, "values") text = self.file_tree.item(item, "text") # Extract name (remove icon emoji) name = text.split(" ", 1)[1] if " " in text else text # Skip ".." and check if actually checked if name == ".." or (len(values) > 0 and values[0] != "☑"): continue is_directory = len(values) > 1 and values[1] == "Directory" checked.append((name, is_directory, item)) # Also check current selection if no checkboxes are checked if not checked: sel = self.file_tree.selection() if sel: item = sel[0] values = self.file_tree.item(item, "values") text = self.file_tree.item(item, "text") name = text.split(" ", 1)[1] if " " in text else text if name != "..": is_directory = len(values) > 1 and values[1] == "Directory" checked.append((name, is_directory, item)) return checked def get_selected_item(self): """Get selected item name and type. Returns (name, is_directory) or None.""" checked = self.get_checked_items() if checked: # Return first checked item for backward compatibility return (checked[0][0], checked[0][1]) sel = self.file_tree.selection() if not sel: return None item = sel[0] values = self.file_tree.item(item, "values") text = self.file_tree.item(item, "text") # Extract name (remove icon emoji) name = text.split(" ", 1)[1] if " " in text else text # Don't allow operations on ".." if name == "..": return None is_directory = len(values) > 1 and values[1] == "Directory" return (name, is_directory) def get_selected_filename(self): """Get selected filename (for backward compatibility). Returns None for directories.""" result = self.get_selected_item() if result is None: messagebox.showwarning("No item selected", "Please select a file or folder first.") return None name, is_directory = result if is_directory: messagebox.showwarning("Invalid operation", "Please select a file for this operation.") return None return name def download_selected(self): """Download selected files (multiple if checkboxes are used)""" checked = self.get_checked_items() files_to_download = [(name, is_dir) for name, is_dir, _ in checked if not is_dir] if not files_to_download: # Fallback to single selection filename = self.get_selected_filename() if not filename: messagebox.showwarning("No selection", "Please select files to download (use checkboxes for multiple).") return files_to_download = [(filename, False)] # For single file, use save dialog; for multiple, use directory if len(files_to_download) == 1: filename = files_to_download[0][0] save_path = filedialog.asksaveasfilename(initialfile=filename) if not save_path: return save_paths = [save_path] else: # Multiple files - ask for directory save_dir = filedialog.askdirectory(title="Select folder to save files") if not save_dir: return save_paths = [Path(save_dir) / filename for filename, _ in files_to_download] # Queue all download tasks for (filename, _), save_path in zip(files_to_download, save_paths): with self.task_lock: self.task_counter += 1 task_id = self.task_counter task = { 'type': 'download', 'id': task_id, 'filename': filename, 'save_path': str(save_path) } self._add_progress_item(task_id, f"Download: {filename}") self.task_queue.put(task) # Clear checkboxes after queueing self.deselect_all_items() def delete_selected(self): """Delete selected items (multiple if checkboxes are used)""" checked = self.get_checked_items() if not checked: # Fallback to single selection result = self.get_selected_item() if not result: messagebox.showwarning("No selection", "Please select items to delete (use checkboxes for multiple).") return checked = [(result[0], result[1], None)] # Separate files and directories items_to_delete = [(name, is_dir) for name, is_dir, _ in checked] if not items_to_delete: return # Build confirmation message file_count = sum(1 for _, is_dir in items_to_delete if not is_dir) dir_count = sum(1 for _, is_dir in items_to_delete if is_dir) if file_count > 0 and dir_count > 0: confirm_msg = f"Delete {file_count} file(s) and {dir_count} folder(s) from server?\n\nThis action cannot be undone." elif file_count > 1: confirm_msg = f"Delete {file_count} files from server?\n\nThis action cannot be undone." elif dir_count > 1: confirm_msg = f"Delete {dir_count} folders from server?\n\nThis action cannot be undone." elif file_count == 1: confirm_msg = f"Delete file '{items_to_delete[0][0]}' from server?\n\nThis action cannot be undone." else: confirm_msg = f"Delete folder '{items_to_delete[0][0]}' from server?\n\nThis action cannot be undone." if not messagebox.askyesno("Confirm Deletion", confirm_msg): return # Delete all items try: safe_path = self.current_path.replace("'", "'\\''") commands = [] for name, is_directory in items_to_delete: safe_name = name.replace("'", "'\\''") item_path = f"'{safe_path}/{safe_name}'" marker_path = f"'{safe_path}/.public.{safe_name}'" if is_directory: # Delete directory recursively cmd = f"rm -rf {item_path}; rm -f {marker_path}" else: # Delete file cmd = f"rm -f {item_path} {marker_path}" commands.append(cmd) # Execute all delete commands combined_cmd = "; ".join(commands) result = subprocess.run( ["ssh", f"{REMOTE_USER}@{REMOTE_HOST}", combined_cmd], capture_output=True, text=True, timeout=60 ) if result.returncode != 0: raise Exception(result.stderr.strip() or "Delete failed") deleted_count = len(items_to_delete) if deleted_count > 1: messagebox.showinfo("Deleted", f"{deleted_count} items removed.") else: item_type = "Folder" if items_to_delete[0][1] else "File" messagebox.showinfo("Deleted", f"{item_type} removed.") except Exception as e: messagebox.showerror("Delete Failed", str(e)) finally: # Clear checkboxes and refresh self.deselect_all_items() threading.Thread(target=self._refresh_list_async, daemon=True).start() def copy_public_link(self): filename = self.get_selected_filename() if not filename: return # Check if public cmd = ["ssh", f"{REMOTE_USER}@{REMOTE_HOST}", f"[ -f {self.current_path}/.public.{filename} ] && echo 1"] res = subprocess.run(cmd, capture_output=True, text=True) if res.stdout.strip() != "1": messagebox.showwarning("Not Public", "File is not public.") return # Generate link with relative path from REMOTE_DIR rel_path = self.current_path.replace(REMOTE_DIR + '/', '') if self.current_path != REMOTE_DIR else '' link_path = f"{rel_path}/{filename}" if rel_path else filename link = f"http://{REMOTE_HOST}:8080/{link_path}" self.root.clipboard_clear() self.root.clipboard_append(link) messagebox.showinfo("Copied", "Public link copied to clipboard.") # === RUN === if __name__ == "__main__": if TTKBOOTSTRAP_AVAILABLE: root = tb.Window( title="Tailscale File Manager", themename="flatly", # Options: flatly, darkly, solar, morph, etc. size=(780, 540) ) else: root = tk.Tk() root.title("Tailscale File Manager") root.geometry("780x540") app = FileManagerApp(root) root.mainloop()