From e92cc701f614f4afd231ab18e031769a88a459c2 Mon Sep 17 00:00:00 2001 From: truetype Date: Fri, 20 Mar 2026 19:31:53 +0100 Subject: [PATCH] Phase 12: Additional plugin examples (File Tools and System Utils) --- README.md | 76 +++++- plugins/file_tools_plugin.py | 441 +++++++++++++++++++++++++++++++++ plugins/system_utils_plugin.py | 421 +++++++++++++++++++++++++++++++ 3 files changed, 937 insertions(+), 1 deletion(-) create mode 100644 plugins/file_tools_plugin.py create mode 100644 plugins/system_utils_plugin.py diff --git a/README.md b/README.md index 5162288..cd068f3 100644 --- a/README.md +++ b/README.md @@ -20,6 +20,7 @@ Un agente configurabile basato su DeepAgents che supporta più provider LLM. - **Monitoraggio**: Sistema di logging e monitoraggio delle performance - **Dashboard performance**: Monitoraggio dettagliato delle performance - **Sistema di plugin**: Estendi le funzionalità con plugin personalizzati +- **Plugin inclusi**: File tools, System utils, Example plugin - **Supporto multilingue**: Interfaccia in italiano e inglese ## Installazione @@ -424,7 +425,17 @@ MaxClaw Agent include un sistema di plugin per estendere le funzionalità. ### Plugin Inclusi -- **Example Plugin**: Plugin di esempio che dimostra come creare plugin +1. **Example Plugin**: Plugin di esempio che dimostra come creare plugin + - Strumenti: `example_hello`, `example_echo`, `example_math` + - Comandi: `/example`, `/hello`, `/math` + +2. **File Tools Plugin**: Strumenti aggiuntivi per la gestione dei file + - Strumenti: `file_find`, `file_size`, `file_copy`, `file_move`, `file_delete`, `file_info`, `dir_list`, `dir_create` + - Comandi: `/find`, `/size`, `/copy`, `/move`, `/delete`, `/info`, `/dir`, `/mkdir` + +3. **System Utils Plugin**: Strumenti per informazioni di sistema + - Strumenti: `system_info`, `system_uptime`, `system_ip`, `system_hostname`, `system_platform`, `system_users`, `system_processes`, `system_disk_usage` + - Comandi: `/sysinfo`, `/uptime`, `/ip`, `/hostname`, `/platform`, `/users`, `/ps`, `/df` ### Creare un Plugin @@ -479,6 +490,69 @@ python -c "from plugins.manager import plugin_manager; plugin_manager.load_all_p /mycommand ``` +### Esempi di Utilizzo dei Plugin + +#### File Tools Plugin +```bash +# Trova file per nome +"Find all .py files in the current directory" +# oppure +/find *.py + +# Ottieni dimensione di un file +"Get the size of main.py" +# oppure +/size main.py + +# Copia un file +"Copy config.json to backup_config.json" +# oppure +/copy config.json backup_config.json + +# Elenca directory con dettagli +"List files in the current directory with details" +# oppure +/dir +``` + +#### System Utils Plugin +```bash +# Ottieni informazioni di sistema +"Show system information" +# oppure +/sysinfo + +# Ottieni uptime del sistema +"How long has the system been running?" +# oppure +/uptime + +# Ottieni indirizzo IP +"What is my IP address?" +# oppure +/ip + +# Ottieni informazioni sulla piattaforma +"What platform am I running on?" +# oppure +/platform + +# Elenca utenti loggati +"Who is logged in?" +# oppure +/users + +# Elenca processi in esecuzione +"Show running processes" +# oppure +/ps + +# Mostra utilizzo disco +"Show disk usage" +# oppure +/df +``` + ## Supporto Multilingue MaxClaw Agent supporta più lingue per l'interfaccia utente. diff --git a/plugins/file_tools_plugin.py b/plugins/file_tools_plugin.py new file mode 100644 index 0000000..f75c5ac --- /dev/null +++ b/plugins/file_tools_plugin.py @@ -0,0 +1,441 @@ +""" +File tools plugin for MaxClaw Agent +Provides additional file management tools +""" +import os +import sys +import shutil +from typing import List, Dict, Any +from pathlib import Path +from langchain.tools import Tool + +# Add parent directory to path +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from plugins.base import BasePlugin + + +class FileToolsPlugin(BasePlugin): + """Plugin providing additional file management tools""" + + def __init__(self): + """Initialize file tools plugin""" + super().__init__( + name="file_tools", + version="1.0.0", + description="Additional file management tools for MaxClaw Agent" + ) + + def get_tools(self) -> List[Tool]: + """Get tools provided by this plugin""" + return [ + Tool( + name="file_find", + func=self.find_files, + description="Find files by name pattern" + ), + Tool( + name="file_size", + func=self.get_file_size, + description="Get file or directory size" + ), + Tool( + name="file_copy", + func=self.copy_file, + description="Copy a file to a new location" + ), + Tool( + name="file_move", + func=self.move_file, + description="Move a file to a new location" + ), + Tool( + name="file_delete", + func=self.delete_file, + description="Delete a file or directory" + ), + Tool( + name="file_info", + func=self.get_file_info, + description="Get detailed file information" + ), + Tool( + name="dir_list", + func=self.list_directory, + description="List directory contents with details" + ), + Tool( + name="dir_create", + func=self.create_directory, + description="Create a new directory" + ) + ] + + def get_commands(self) -> Dict[str, callable]: + """Get commands provided by this plugin""" + return { + "/find": self.find_command, + "/size": self.size_command, + "/copy": self.copy_command, + "/move": self.move_command, + "/delete": self.delete_command, + "/info": self.info_command, + "/dir": self.dir_command, + "/mkdir": self.mkdir_command + } + + def find_files(self, pattern: str, directory: str = ".") -> str: + """Find files by name pattern""" + try: + path = Path(directory) + if not path.exists(): + return f"❌ Directory not found: {directory}" + + matches = list(path.rglob(pattern)) + + if not matches: + return f"No files found matching '{pattern}' in {directory}" + + output = f"🔍 Found {len(matches)} files matching '{pattern}':\n" + output += "=" * 50 + "\n" + + for match in matches[:20]: # Limit to 20 results + if match.is_file(): + size = match.stat().st_size + if size > 1024 * 1024: + size_str = f"{size / (1024 * 1024):.1f} MB" + elif size > 1024: + size_str = f"{size / 1024:.1f} KB" + else: + size_str = f"{size} B" + + output += f"📄 {match.relative_to(path)} ({size_str})\n" + + if len(matches) > 20: + output += f"\n... and {len(matches) - 20} more files" + + return output + + except Exception as e: + return f"❌ Error finding files: {str(e)}" + + def get_file_size(self, path: str) -> str: + """Get file or directory size""" + try: + path_obj = Path(path) + + if not path_obj.exists(): + return f"❌ Path not found: {path}" + + if path_obj.is_file(): + size = path_obj.stat().st_size + if size > 1024 * 1024 * 1024: + size_str = f"{size / (1024 * 1024 * 1024):.2f} GB" + elif size > 1024 * 1024: + size_str = f"{size / (1024 * 1024):.2f} MB" + elif size > 1024: + size_str = f"{size / 1024:.2f} KB" + else: + size_str = f"{size} bytes" + + return f"📄 {path}: {size_str}" + + elif path_obj.is_dir(): + total_size = 0 + file_count = 0 + + for item in path_obj.rglob('*'): + if item.is_file(): + total_size += item.stat().st_size + file_count += 1 + + if total_size > 1024 * 1024 * 1024: + size_str = f"{total_size / (1024 * 1024 * 1024):.2f} GB" + elif total_size > 1024 * 1024: + size_str = f"{total_size / (1024 * 1024):.2f} MB" + elif total_size > 1024: + size_str = f"{total_size / 1024:.2f} KB" + else: + size_str = f"{total_size} bytes" + + return f"📁 {path}: {size_str} ({file_count} files)" + + else: + return f"❌ Unknown path type: {path}" + + except Exception as e: + return f"❌ Error getting size: {str(e)}" + + def copy_file(self, source: str, destination: str) -> str: + """Copy a file to a new location""" + try: + source_path = Path(source) + dest_path = Path(destination) + + if not source_path.exists(): + return f"❌ Source not found: {source}" + + # If destination is a directory, copy into it + if dest_path.is_dir(): + dest_path = dest_path / source_path.name + + # Create parent directories if needed + dest_path.parent.mkdir(parents=True, exist_ok=True) + + if source_path.is_file(): + shutil.copy2(source_path, dest_path) + return f"✅ File copied: {source} → {destination}" + elif source_path.is_dir(): + shutil.copytree(source_path, dest_path) + return f"✅ Directory copied: {source} → {destination}" + else: + return f"❌ Unknown source type: {source}" + + except Exception as e: + return f"❌ Error copying: {str(e)}" + + def move_file(self, source: str, destination: str) -> str: + """Move a file to a new location""" + try: + source_path = Path(source) + dest_path = Path(destination) + + if not source_path.exists(): + return f"❌ Source not found: {source}" + + # If destination is a directory, move into it + if dest_path.is_dir(): + dest_path = dest_path / source_path.name + + # Create parent directories if needed + dest_path.parent.mkdir(parents=True, exist_ok=True) + + shutil.move(str(source_path), str(dest_path)) + return f"✅ Moved: {source} → {destination}" + + except Exception as e: + return f"❌ Error moving: {str(e)}" + + def delete_file(self, path: str, confirm: bool = True) -> str: + """Delete a file or directory""" + try: + path_obj = Path(path) + + if not path_obj.exists(): + return f"❌ Path not found: {path}" + + if confirm: + return f"⚠️ To delete {path}, use the command with confirm=True" + + if path_obj.is_file(): + path_obj.unlink() + return f"✅ File deleted: {path}" + elif path_obj.is_dir(): + shutil.rmtree(path_obj) + return f"✅ Directory deleted: {path}" + else: + return f"❌ Unknown path type: {path}" + + except Exception as e: + return f"❌ Error deleting: {str(e)}" + + def get_file_info(self, path: str) -> str: + """Get detailed file information""" + try: + path_obj = Path(path) + + if not path_obj.exists(): + return f"❌ Path not found: {path}" + + stat = path_obj.stat() + + # Format size + size = stat.st_size + if size > 1024 * 1024 * 1024: + size_str = f"{size / (1024 * 1024 * 1024):.2f} GB" + elif size > 1024 * 1024: + size_str = f"{size / (1024 * 1024):.2f} MB" + elif size > 1024: + size_str = f"{size / 1024:.2f} KB" + else: + size_str = f"{size} bytes" + + # Format dates + from datetime import datetime + created = datetime.fromtimestamp(stat.st_ctime).strftime('%Y-%m-%d %H:%M:%S') + modified = datetime.fromtimestamp(stat.st_mtime).strftime('%Y-%m-%d %H:%M:%S') + accessed = datetime.fromtimestamp(stat.st_atime).strftime('%Y-%m-%d %H:%M:%S') + + # File type + if path_obj.is_file(): + file_type = "File" + elif path_obj.is_dir(): + file_type = "Directory" + elif path_obj.is_symlink(): + file_type = "Symbolic Link" + else: + file_type = "Unknown" + + output = f"📋 File Information: {path}\n" + output += "=" * 50 + "\n" + output += f"Type: {file_type}\n" + output += f"Size: {size_str}\n" + output += f"Created: {created}\n" + output += f"Modified: {modified}\n" + output += f"Accessed: {accessed}\n" + output += f"Permissions: {oct(stat.st_mode)[-3:]}\n" + + if path_obj.is_file(): + output += f"Extension: {path_obj.suffix}\n" + output += f"Name: {path_obj.name}\n" + output += f"Parent: {path_obj.parent}\n" + + return output + + except Exception as e: + return f"❌ Error getting file info: {str(e)}" + + def list_directory(self, path: str = ".", show_hidden: bool = False) -> str: + """List directory contents with details""" + try: + path_obj = Path(path) + + if not path_obj.exists(): + return f"❌ Directory not found: {path}" + + if not path_obj.is_dir(): + return f"❌ Not a directory: {path}" + + items = list(path_obj.iterdir()) + + if not show_hidden: + items = [item for item in items if not item.name.startswith('.')] + + if not items: + return f"📁 Directory is empty: {path}" + + output = f"📁 Directory: {path}\n" + output += "=" * 50 + "\n" + + # Sort items: directories first, then files + dirs = [item for item in items if item.is_dir()] + files = [item for item in items if item.is_file()] + + # List directories + for dir_item in sorted(dirs, key=lambda x: x.name.lower()): + try: + size = sum(f.stat().st_size for f in dir_item.rglob('*') if f.is_file()) + if size > 1024 * 1024: + size_str = f"{size / (1024 * 1024):.1f} MB" + elif size > 1024: + size_str = f"{size / 1024:.1f} KB" + else: + size_str = f"{size} B" + + output += f"📁 {dir_item.name}/ ({size_str})\n" + except: + output += f"📁 {dir_item.name}/\n" + + # List files + for file_item in sorted(files, key=lambda x: x.name.lower()): + try: + size = file_item.stat().st_size + if size > 1024 * 1024: + size_str = f"{size / (1024 * 1024):.1f} MB" + elif size > 1024: + size_str = f"{size / 1024:.1f} KB" + else: + size_str = f"{size} B" + + output += f"📄 {file_item.name} ({size_str})\n" + except: + output += f"📄 {file_item.name}\n" + + output += f"\nTotal: {len(dirs)} directories, {len(files)} files" + + return output + + except Exception as e: + return f"❌ Error listing directory: {str(e)}" + + def create_directory(self, path: str, parents: bool = True) -> str: + """Create a new directory""" + try: + path_obj = Path(path) + + if path_obj.exists(): + return f"⚠️ Path already exists: {path}" + + path_obj.mkdir(parents=parents, exist_ok=True) + return f"✅ Directory created: {path}" + + except Exception as e: + return f"❌ Error creating directory: {str(e)}" + + # Command implementations + def find_command(self, args: List[str] = None) -> str: + """Find command implementation""" + if not args: + return "❌ Usage: /find [directory]" + + pattern = args[0] + directory = args[1] if len(args) > 1 else "." + + return self.find_files(pattern, directory) + + def size_command(self, args: List[str] = None) -> str: + """Size command implementation""" + if not args: + return "❌ Usage: /size " + + return self.get_file_size(args[0]) + + def copy_command(self, args: List[str] = None) -> str: + """Copy command implementation""" + if not args or len(args) < 2: + return "❌ Usage: /copy " + + return self.copy_file(args[0], args[1]) + + def move_command(self, args: List[str] = None) -> str: + """Move command implementation""" + if not args or len(args) < 2: + return "❌ Usage: /move " + + return self.move_file(args[0], args[1]) + + def delete_command(self, args: List[str] = None) -> str: + """Delete command implementation""" + if not args: + return "❌ Usage: /delete " + + return self.delete_file(args[0], confirm=False) + + def info_command(self, args: List[str] = None) -> str: + """Info command implementation""" + if not args: + return "❌ Usage: /info " + + return self.get_file_info(args[0]) + + def dir_command(self, args: List[str] = None) -> str: + """Dir command implementation""" + path = args[0] if args else "." + show_hidden = "-a" in args if args else False + + return self.list_directory(path, show_hidden) + + def mkdir_command(self, args: List[str] = None) -> str: + """Mkdir command implementation""" + if not args: + return "❌ Usage: /mkdir " + + return self.create_directory(args[0]) + + def on_load(self): + """Called when plugin is loaded""" + print(f"📦 File tools plugin loaded: {self.name} v{self.version}") + + def on_unload(self): + """Called when plugin is unloaded""" + print(f"🗑️ File tools plugin unloaded: {self.name}") \ No newline at end of file diff --git a/plugins/system_utils_plugin.py b/plugins/system_utils_plugin.py new file mode 100644 index 0000000..7ae92c5 --- /dev/null +++ b/plugins/system_utils_plugin.py @@ -0,0 +1,421 @@ +""" +System utilities plugin for MaxClaw Agent +Provides system information and utility tools +""" +import os +import sys +import platform +import socket +import subprocess +from typing import List, Dict, Any +from datetime import datetime +from langchain.tools import Tool + +# Add parent directory to path +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from plugins.base import BasePlugin + + +class SystemUtilsPlugin(BasePlugin): + """Plugin providing system utilities and information""" + + def __init__(self): + """Initialize system utilities plugin""" + super().__init__( + name="system_utils", + version="1.0.0", + description="System information and utility tools" + ) + + def get_tools(self) -> List[Tool]: + """Get tools provided by this plugin""" + return [ + Tool( + name="system_info", + func=self.get_system_info, + description="Get comprehensive system information" + ), + Tool( + name="system_uptime", + func=self.get_system_uptime, + description="Get system uptime" + ), + Tool( + name="system_ip", + func=self.get_ip_address, + description="Get system IP addresses" + ), + Tool( + name="system_hostname", + func=self.get_hostname, + description="Get system hostname" + ), + Tool( + name="system_platform", + func=self.get_platform_info, + description="Get platform information" + ), + Tool( + name="system_users", + func=self.get_logged_users, + description="Get logged in users" + ), + Tool( + name="system_processes", + func=self.get_running_processes, + description="Get running processes" + ), + Tool( + name="system_disk_usage", + func=self.get_disk_usage, + description="Get disk usage information" + ) + ] + + def get_commands(self) -> Dict[str, callable]: + """Get commands provided by this plugin""" + return { + "/sysinfo": self.sysinfo_command, + "/uptime": self.uptime_command, + "/ip": self.ip_command, + "/hostname": self.hostname_command, + "/platform": self.platform_command, + "/users": self.users_command, + "/ps": self.ps_command, + "/df": self.df_command + } + + def get_system_info(self) -> str: + """Get comprehensive system information""" + try: + info = { + "System": platform.system(), + "Node Name": platform.node(), + "Release": platform.release(), + "Version": platform.version(), + "Machine": platform.machine(), + "Processor": platform.processor(), + "Python Version": platform.python_version(), + "Hostname": socket.gethostname(), + "IP Address": socket.gethostbyname(socket.gethostname()), + "Current User": os.getlogin(), + "Current Directory": os.getcwd(), + "Home Directory": os.path.expanduser("~"), + "Timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S") + } + + output = "💻 System Information:\n" + output += "=" * 50 + "\n" + + for key, value in info.items(): + output += f"{key}: {value}\n" + + return output + + except Exception as e: + return f"❌ Error getting system info: {str(e)}" + + def get_system_uptime(self) -> str: + """Get system uptime""" + try: + if platform.system() == "Linux": + with open('/proc/uptime', 'r') as f: + uptime_seconds = float(f.readline().split()[0]) + + days = int(uptime_seconds // (24 * 3600)) + hours = int((uptime_seconds % (24 * 3600)) // 3600) + minutes = int((uptime_seconds % 3600) // 60) + seconds = int(uptime_seconds % 60) + + if days > 0: + return f"⏱️ System uptime: {days} days, {hours} hours, {minutes} minutes, {seconds} seconds" + elif hours > 0: + return f"⏱️ System uptime: {hours} hours, {minutes} minutes, {seconds} seconds" + elif minutes > 0: + return f"⏱️ System uptime: {minutes} minutes, {seconds} seconds" + else: + return f"⏱️ System uptime: {seconds} seconds" + + elif platform.system() == "Darwin": # macOS + result = subprocess.run(['uptime'], capture_output=True, text=True) + return f"⏱️ {result.stdout.strip()}" + + elif platform.system() == "Windows": + result = subprocess.run(['net', 'stats', 'srv'], capture_output=True, text=True) + # Parse Windows uptime from output + for line in result.stdout.split('\n'): + if 'Statistics since' in line: + return f"⏱️ {line.strip()}" + + return "⏱️ Uptime information not available" + + else: + return "⏱️ Uptime information not available for this platform" + + except Exception as e: + return f"❌ Error getting uptime: {str(e)}" + + def get_ip_address(self) -> str: + """Get system IP addresses""" + try: + hostname = socket.gethostname() + + # Get all IP addresses + ip_addresses = [] + + # Get hostname IP + try: + host_ip = socket.gethostbyname(hostname) + ip_addresses.append(("Hostname", host_ip)) + except: + pass + + # Get all network interfaces (if available) + try: + import netifaces + for interface in netifaces.interfaces(): + addrs = netifaces.ifaddresses(interface) + if netifaces.AF_INET in addrs: + for addr in addrs[netifaces.AF_INET]: + ip_addresses.append((interface, addr['addr'])) + except ImportError: + # netifaces not available, use alternative method + pass + + if not ip_addresses: + return "❌ No IP addresses found" + + output = "🌐 IP Addresses:\n" + output += "=" * 30 + "\n" + + for interface, ip in ip_addresses: + output += f"{interface}: {ip}\n" + + return output + + except Exception as e: + return f"❌ Error getting IP addresses: {str(e)}" + + def get_hostname(self) -> str: + """Get system hostname""" + try: + hostname = socket.gethostname() + fqdn = socket.getfqdn() + + output = "🏠 Hostname Information:\n" + output += "=" * 30 + "\n" + output += f"Hostname: {hostname}\n" + output += f"FQDN: {fqdn}\n" + + return output + + except Exception as e: + return f"❌ Error getting hostname: {str(e)}" + + def get_platform_info(self) -> str: + """Get platform information""" + try: + info = { + "System": platform.system(), + "Node": platform.node(), + "Release": platform.release(), + "Version": platform.version(), + "Machine": platform.machine(), + "Processor": platform.processor(), + "Python Build": platform.python_build(), + "Python Compiler": platform.python_compiler(), + "Python Implementation": platform.python_implementation(), + "Python Version": platform.python_version() + } + + output = "🖥️ Platform Information:\n" + output += "=" * 50 + "\n" + + for key, value in info.items(): + output += f"{key}: {value}\n" + + return output + + except Exception as e: + return f"❌ Error getting platform info: {str(e)}" + + def get_logged_users(self) -> str: + """Get logged in users""" + try: + if platform.system() in ["Linux", "Darwin"]: + result = subprocess.run(['who'], capture_output=True, text=True) + users = result.stdout.strip() + + if not users: + return "👥 No users currently logged in" + + output = "👥 Logged in users:\n" + output += "=" * 30 + "\n" + output += users + + return output + + elif platform.system() == "Windows": + result = subprocess.run(['query', 'user'], capture_output=True, text=True) + users = result.stdout.strip() + + if not users: + return "👥 No users currently logged in" + + output = "👥 Logged in users:\n" + output += "=" * 30 + "\n" + output += users + + return output + + else: + return "👥 User information not available for this platform" + + except Exception as e: + return f"❌ Error getting logged users: {str(e)}" + + def get_running_processes(self, limit: int = 20) -> str: + """Get running processes""" + try: + if platform.system() in ["Linux", "Darwin"]: + result = subprocess.run(['ps', 'aux'], capture_output=True, text=True) + processes = result.stdout.strip().split('\n') + + if len(processes) <= 1: + return "🔄 No processes found" + + output = f"🔄 Running processes (top {limit}):\n" + output += "=" * 50 + "\n" + + # Header + output += processes[0] + "\n" + output += "-" * 50 + "\n" + + # Processes (limit to specified number) + for process in processes[1:limit+1]: + output += process + "\n" + + if len(processes) > limit+1: + output += f"\n... and {len(processes) - limit - 1} more processes" + + return output + + elif platform.system() == "Windows": + result = subprocess.run(['tasklist'], capture_output=True, text=True) + processes = result.stdout.strip().split('\n') + + if len(processes) <= 3: + return "🔄 No processes found" + + output = f"🔄 Running processes (top {limit}):\n" + output += "=" * 50 + "\n" + + # Header (first 3 lines in Windows) + for i in range(3): + if i < len(processes): + output += processes[i] + "\n" + + output += "-" * 50 + "\n" + + # Processes (limit to specified number) + for process in processes[3:limit+3]: + output += process + "\n" + + if len(processes) > limit+3: + output += f"\n... and {len(processes) - limit - 3} more processes" + + return output + + else: + return "🔄 Process information not available for this platform" + + except Exception as e: + return f"❌ Error getting processes: {str(e)}" + + def get_disk_usage(self, path: str = "/") -> str: + """Get disk usage information""" + try: + if platform.system() == "Windows": + # For Windows, use different drive letters + if path == "/": + path = "C:\\" + + import shutil + total, used, free = shutil.disk_usage(path) + + # Convert to human readable + def format_bytes(bytes): + for unit in ['B', 'KB', 'MB', 'GB', 'TB']: + if bytes < 1024.0: + return f"{bytes:.2f} {unit}" + bytes /= 1024.0 + return f"{bytes:.2f} PB" + + total_str = format_bytes(total) + used_str = format_bytes(used) + free_str = format_bytes(free) + percent_used = (used / total) * 100 + + output = f"💾 Disk Usage for {path}:\n" + output += "=" * 40 + "\n" + output += f"Total: {total_str}\n" + output += f"Used: {used_str} ({percent_used:.1f}%)\n" + output += f"Free: {free_str}\n" + + # Create a simple progress bar + bar_length = 30 + filled_length = int(bar_length * percent_used // 100) + bar = '█' * filled_length + '░' * (bar_length - filled_length) + output += f"\n[{bar}] {percent_used:.1f}%\n" + + return output + + except Exception as e: + return f"❌ Error getting disk usage: {str(e)}" + + # Command implementations + def sysinfo_command(self, args: List[str] = None) -> str: + """Sysinfo command implementation""" + return self.get_system_info() + + def uptime_command(self, args: List[str] = None) -> str: + """Uptime command implementation""" + return self.get_system_uptime() + + def ip_command(self, args: List[str] = None) -> str: + """IP command implementation""" + return self.get_ip_address() + + def hostname_command(self, args: List[str] = None) -> str: + """Hostname command implementation""" + return self.get_hostname() + + def platform_command(self, args: List[str] = None) -> str: + """Platform command implementation""" + return self.get_platform_info() + + def users_command(self, args: List[str] = None) -> str: + """Users command implementation""" + return self.get_logged_users() + + def ps_command(self, args: List[str] = None) -> str: + """PS command implementation""" + limit = 20 + if args and args[0].isdigit(): + limit = int(args[0]) + + return self.get_running_processes(limit) + + def df_command(self, args: List[str] = None) -> str: + """DF command implementation""" + path = args[0] if args else "/" + return self.get_disk_usage(path) + + def on_load(self): + """Called when plugin is loaded""" + print(f"📦 System utils plugin loaded: {self.name} v{self.version}") + + def on_unload(self): + """Called when plugin is unloaded""" + print(f"🗑️ System utils plugin unloaded: {self.name}") \ No newline at end of file