Spaces:
Paused
Paused
| import gradio as gr | |
| import subprocess | |
| import os | |
| import sys | |
| import shutil | |
| import signal | |
| import time | |
| import threading | |
| import json | |
| import secrets | |
| from pathlib import Path | |
| from datetime import datetime | |
| from collections import deque | |
| WORKSPACE = os.getcwd() | |
| LOGS_DIR = os.path.join(WORKSPACE, ".system_logs") | |
| SESSIONS_DIR = os.path.join(WORKSPACE, ".sessions") | |
| BACKUP_DIR = os.path.join(WORKSPACE, ".backups") | |
| FILE_LOGS_DIR = os.path.join(WORKSPACE, ".file_logs") | |
| OPS_LOG_FILE = os.path.join(LOGS_DIR, "operations.json") | |
| for d in [WORKSPACE, LOGS_DIR, SESSIONS_DIR, BACKUP_DIR, FILE_LOGS_DIR]: | |
| os.makedirs(d, exist_ok=True) | |
| bg_processes = {} | |
| proc_lock = threading.Lock() | |
| cmd_history = deque(maxlen=500) | |
| ops_log = deque(maxlen=2000) | |
| if os.path.exists(OPS_LOG_FILE): | |
| try: | |
| with open(OPS_LOG_FILE, "r", encoding="utf-8") as f: | |
| data = json.load(f) | |
| if isinstance(data, list): | |
| ops_log.extend(data) | |
| except Exception: | |
| pass | |
| def save_ops_log(): | |
| try: | |
| with open(OPS_LOG_FILE, "w", encoding="utf-8") as f: | |
| json.dump(list(ops_log), f, ensure_ascii=False, indent=2) | |
| except Exception: | |
| pass | |
| def log_operation(op_type, target, details="", status="success"): | |
| entry = { | |
| "time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), | |
| "type": op_type, | |
| "target": str(target), | |
| "details": str(details), | |
| "status": status | |
| } | |
| ops_log.append(entry) | |
| save_ops_log() | |
| if target: | |
| safe_name = str(target).replace("/", "_").replace("\\", "_").replace(" ", "_") | |
| if len(safe_name) > 100: | |
| safe_name = safe_name[:100] | |
| file_log_path = os.path.join(FILE_LOGS_DIR, f"{safe_name}.log") | |
| try: | |
| with open(file_log_path, "a", encoding="utf-8") as f: | |
| f.write(f"[{entry['time']}] {op_type}: {details} ({status})\n") | |
| except Exception: | |
| pass | |
| def resolve_path(path): | |
| if not path: | |
| return WORKSPACE | |
| path = path.strip() | |
| if path == "~": | |
| return WORKSPACE | |
| if not os.path.isabs(path): | |
| return os.path.normpath(os.path.join(WORKSPACE, path)) | |
| return os.path.normpath(path) | |
| def get_history_text(): | |
| lines = [] | |
| for item in list(cmd_history)[-30:]: | |
| lines.append(f"[{item['time']}] $ {item['cmd']}") | |
| return "\n".join(lines) if lines else "No commands yet" | |
| def execute_terminal(command, work_dir): | |
| if not command or not command.strip(): | |
| return "", work_dir or WORKSPACE, get_history_text() | |
| command = command.strip() | |
| wd = resolve_path(work_dir) if work_dir else WORKSPACE | |
| if not os.path.isdir(wd): | |
| wd = WORKSPACE | |
| cmd_history.append({"time": datetime.now().strftime("%H:%M:%S"), "cmd": command}) | |
| log_operation("COMMAND", command, f"dir: {wd}") | |
| if command == "clear": | |
| return "", wd, get_history_text() | |
| if command == "pwd": | |
| return wd, wd, get_history_text() | |
| if command.startswith("cd "): | |
| target = command[3:].strip().strip('"').strip("'") | |
| if target == "~": | |
| return f"📁 {WORKSPACE}", WORKSPACE, get_history_text() | |
| if target == "-": | |
| return f"📁 {wd}", wd, get_history_text() | |
| if os.path.isabs(target): | |
| new_path = target | |
| else: | |
| new_path = os.path.normpath(os.path.join(wd, target)) | |
| if os.path.isdir(new_path): | |
| return f"📁 {new_path}", new_path, get_history_text() | |
| return f"❌ Directory not found: {target}", wd, get_history_text() | |
| if command == "cd": | |
| return f"📁 {WORKSPACE}", WORKSPACE, get_history_text() | |
| try: | |
| env = os.environ.copy() | |
| env["HOME"] = WORKSPACE | |
| env["TERM"] = "xterm-256color" | |
| env["PYTHONUNBUFFERED"] = "1" | |
| result = subprocess.run( | |
| command, | |
| shell=True, | |
| capture_output=True, | |
| text=True, | |
| timeout=180, | |
| cwd=wd, | |
| executable="/bin/bash", | |
| env=env | |
| ) | |
| output = "" | |
| if result.stdout: | |
| output += result.stdout | |
| if result.stderr: | |
| if output: | |
| output += "\n--- STDERR ---\n" | |
| output += result.stderr | |
| if not output and result.returncode == 0: | |
| output = "✅ Command completed successfully" | |
| if result.returncode != 0 and not output: | |
| output = f"❌ Exit code: {result.returncode}" | |
| return output.rstrip(), wd, get_history_text() | |
| except subprocess.TimeoutExpired: | |
| return "⏰ Command timed out (180s limit)", wd, get_history_text() | |
| except Exception as e: | |
| return f"❌ Error: {str(e)}", wd, get_history_text() | |
| def execute_background(command, work_dir): | |
| if not command or not command.strip(): | |
| return "❌ No command specified" | |
| wd = resolve_path(work_dir) if work_dir else WORKSPACE | |
| if not os.path.isdir(wd): | |
| wd = WORKSPACE | |
| try: | |
| ts = datetime.now().strftime("%Y%m%d_%H%M%S") | |
| proc_id = secrets.token_hex(4) | |
| log_path = os.path.join(LOGS_DIR, f"bg_{ts}_{proc_id}.log") | |
| log_f = open(log_path, "w", buffering=1, encoding="utf-8") | |
| log_f.write(f"Command: {command}\n") | |
| log_f.write(f"Started: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n") | |
| log_f.write(f"Directory: {wd}\n") | |
| log_f.write(f"{'=' * 60}\n\n") | |
| env = os.environ.copy() | |
| env["PYTHONUNBUFFERED"] = "1" | |
| proc = subprocess.Popen( | |
| command, | |
| shell=True, | |
| stdout=log_f, | |
| stderr=subprocess.STDOUT, | |
| cwd=wd, | |
| preexec_fn=os.setsid, | |
| executable="/bin/bash", | |
| env=env | |
| ) | |
| with proc_lock: | |
| bg_processes[proc.pid] = { | |
| "cmd": command, | |
| "pid": proc.pid, | |
| "start": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), | |
| "log": log_path, | |
| "proc": proc, | |
| "log_f": log_f, | |
| "id": proc_id, | |
| "dir": wd, | |
| } | |
| log_operation("BG_START", command, f"PID: {proc.pid} | ID: {proc_id}") | |
| return f"✅ Process Started\n PID: {proc.pid}\n ID: {proc_id}\n Log: {log_path}" | |
| except Exception as e: | |
| log_operation("BG_START", command, str(e), "error") | |
| return f"❌ Failed to start: {str(e)}" | |
| def get_process_list(): | |
| lines = [] | |
| with proc_lock: | |
| if not bg_processes: | |
| return "📭 No background processes running" | |
| running = 0 | |
| stopped = 0 | |
| for pid, info in bg_processes.items(): | |
| poll = info["proc"].poll() | |
| if poll is None: | |
| status = "🟢 RUNNING" | |
| running += 1 | |
| else: | |
| status = f"🔴 EXITED (code: {poll})" | |
| stopped += 1 | |
| lines.append(f"{'─' * 50}") | |
| lines.append(f" PID: {pid} | ID: {info['id']} | {status}") | |
| lines.append(f" CMD: {info['cmd']}") | |
| lines.append(f" DIR: {info.get('dir', 'N/A')}") | |
| lines.append(f" Started: {info['start']}") | |
| lines.append(f" Log: {info['log']}") | |
| lines.insert(0, f"📊 Total: {len(bg_processes)} | Running: {running} | Stopped: {stopped}") | |
| lines.append(f"{'─' * 50}") | |
| return "\n".join(lines) | |
| def stop_process(pid_input): | |
| if not pid_input or not pid_input.strip(): | |
| return "❌ Enter PID or ID" | |
| pid_input = pid_input.strip() | |
| target_pid = None | |
| with proc_lock: | |
| try: | |
| val = int(pid_input) | |
| if val in bg_processes: | |
| target_pid = val | |
| except ValueError: | |
| for pid, info in bg_processes.items(): | |
| if info["id"] == pid_input: | |
| target_pid = pid | |
| break | |
| if target_pid is None: | |
| return f"❌ Process not found: {pid_input}" | |
| try: | |
| try: | |
| os.killpg(os.getpgid(target_pid), signal.SIGTERM) | |
| except ProcessLookupError: | |
| pass | |
| time.sleep(1) | |
| with proc_lock: | |
| if target_pid in bg_processes: | |
| if bg_processes[target_pid]["proc"].poll() is None: | |
| try: | |
| os.killpg(os.getpgid(target_pid), signal.SIGKILL) | |
| except ProcessLookupError: | |
| pass | |
| log_operation("BG_STOP", str(target_pid), "Process terminated") | |
| return f"✅ Process {target_pid} terminated" | |
| except Exception as e: | |
| return f"❌ Error stopping process: {e}" | |
| def remove_from_list(pid_input): | |
| if not pid_input or not pid_input.strip(): | |
| return "❌ Enter PID or ID", get_process_list() | |
| pid_input = pid_input.strip() | |
| target_pid = None | |
| with proc_lock: | |
| try: | |
| val = int(pid_input) | |
| if val in bg_processes: | |
| target_pid = val | |
| except ValueError: | |
| for pid, info in bg_processes.items(): | |
| if info["id"] == pid_input: | |
| target_pid = pid | |
| break | |
| if target_pid is not None and target_pid in bg_processes: | |
| info = bg_processes[target_pid] | |
| try: | |
| info["log_f"].close() | |
| except Exception: | |
| pass | |
| del bg_processes[target_pid] | |
| return f"✅ Removed {target_pid} from list", get_process_list() | |
| return f"❌ Process not found: {pid_input}", get_process_list() | |
| def read_bg_log(pid_input, tail_lines=100): | |
| if not pid_input or not pid_input.strip(): | |
| return "❌ Enter PID or ID" | |
| pid_input = pid_input.strip() | |
| log_path = None | |
| with proc_lock: | |
| try: | |
| val = int(pid_input) | |
| if val in bg_processes: | |
| log_path = bg_processes[val]["log"] | |
| except ValueError: | |
| for pid, info in bg_processes.items(): | |
| if info["id"] == pid_input: | |
| log_path = info["log"] | |
| break | |
| if log_path is None: | |
| return f"❌ Process not found: {pid_input}" | |
| try: | |
| if not os.path.exists(log_path): | |
| return "❌ Log file not found" | |
| with open(log_path, "r", encoding="utf-8", errors="replace") as f: | |
| content = f.read() | |
| if not content.strip(): | |
| return "📭 Log is empty" | |
| lines = content.split("\n") | |
| if len(lines) > tail_lines: | |
| content = f"... (showing last {tail_lines} lines of {len(lines)}) ...\n\n" | |
| content += "\n".join(lines[-tail_lines:]) | |
| return content | |
| except Exception as e: | |
| return f"❌ Error reading log: {e}" | |
| def stop_all_processes(): | |
| results = [] | |
| pids_to_stop = [] | |
| with proc_lock: | |
| for pid, info in bg_processes.items(): | |
| if info["proc"].poll() is None: | |
| pids_to_stop.append(pid) | |
| if not pids_to_stop: | |
| return "📭 No running processes to stop" | |
| for pid in pids_to_stop: | |
| try: | |
| os.killpg(os.getpgid(pid), signal.SIGTERM) | |
| results.append(f"✅ Sent SIGTERM to {pid}") | |
| except ProcessLookupError: | |
| results.append(f"⚠️ Process {pid} already finished") | |
| except Exception as e: | |
| results.append(f"❌ Failed to stop {pid}: {e}") | |
| time.sleep(1) | |
| for pid in pids_to_stop: | |
| with proc_lock: | |
| if pid in bg_processes and bg_processes[pid]["proc"].poll() is None: | |
| try: | |
| os.killpg(os.getpgid(pid), signal.SIGKILL) | |
| results.append(f"🔪 Sent SIGKILL to {pid}") | |
| except Exception: | |
| pass | |
| log_operation("BG_STOP_ALL", "all", f"Stopped {len(pids_to_stop)} processes") | |
| return "\n".join(results) | |
| def cleanup_finished(): | |
| removed = [] | |
| with proc_lock: | |
| to_remove = [] | |
| for pid, info in bg_processes.items(): | |
| if info["proc"].poll() is not None: | |
| to_remove.append(pid) | |
| for pid in to_remove: | |
| try: | |
| bg_processes[pid]["log_f"].close() | |
| except Exception: | |
| pass | |
| removed.append(str(pid)) | |
| del bg_processes[pid] | |
| if removed: | |
| return f"🧹 Removed {len(removed)} finished processes: {', '.join(removed)}", get_process_list() | |
| return "📭 No finished processes to clean", get_process_list() | |
| def list_files(directory): | |
| d = resolve_path(directory) if directory else WORKSPACE | |
| if not os.path.isdir(d): | |
| d = WORKSPACE | |
| try: | |
| items = [] | |
| items.append(f"📍 {d}\n") | |
| hidden_dirs = {".system_logs", ".sessions", ".backups", ".file_logs", "__pycache__", ".git"} | |
| entries = [] | |
| try: | |
| entries = sorted(os.listdir(d), key=lambda x: (not os.path.isdir(os.path.join(d, x)), x.lower())) | |
| except PermissionError: | |
| return f"❌ Permission denied: {d}", d | |
| if d != "/": | |
| items.append("📁 .. (parent directory)") | |
| dir_count = 0 | |
| file_count = 0 | |
| for entry in entries: | |
| if entry in hidden_dirs: | |
| continue | |
| full = os.path.join(d, entry) | |
| if os.path.isdir(full): | |
| dir_count += 1 | |
| try: | |
| sub_count = len(os.listdir(full)) | |
| items.append(f"📁 {entry}/ ({sub_count} items)") | |
| except Exception: | |
| items.append(f"📁 {entry}/") | |
| else: | |
| file_count += 1 | |
| try: | |
| sz = os.path.getsize(full) | |
| if sz < 1024: | |
| size = f"{sz} B" | |
| elif sz < 1024 * 1024: | |
| size = f"{sz / 1024:.1f} KB" | |
| else: | |
| size = f"{sz / (1024 * 1024):.1f} MB" | |
| except Exception: | |
| size = "?" | |
| ext = Path(entry).suffix.lower() | |
| icons = { | |
| ".py": "🐍", ".js": "📜", ".ts": "📘", ".json": "📋", | |
| ".md": "📝", ".txt": "📝", ".sh": "⚙️", ".bash": "⚙️", | |
| ".html": "🌐", ".css": "🎨", ".xml": "📰", | |
| ".jpg": "🖼️", ".png": "🖼️", ".gif": "🖼️", ".svg": "🖼️", | |
| ".zip": "📦", ".tar": "📦", ".gz": "📦", ".rar": "📦", | |
| ".mp3": "🎵", ".wav": "🎵", ".mp4": "🎬", | |
| ".pdf": "📕", ".doc": "📄", ".docx": "📄", | |
| ".csv": "📊", ".xlsx": "📊", | |
| ".yml": "⚙️", ".yaml": "⚙️", ".toml": "⚙️", | |
| ".env": "🔐", ".cfg": "⚙️", ".ini": "⚙️", | |
| ".log": "📋", ".db": "🗄️", ".sql": "🗄️", | |
| ".c": "🔧", ".cpp": "🔧", ".h": "🔧", | |
| ".java": "☕", ".go": "🔵", ".rs": "🦀", ".rb": "💎", | |
| } | |
| icon = icons.get(ext, "📄") | |
| mod_time = "" | |
| try: | |
| mtime = os.path.getmtime(full) | |
| mod_time = datetime.fromtimestamp(mtime).strftime("%m-%d %H:%M") | |
| except Exception: | |
| pass | |
| items.append(f"{icon} {entry} [{size}] {mod_time}") | |
| items.append(f"\n📊 {dir_count} folders, {file_count} files") | |
| return "\n".join(items), d | |
| except Exception as e: | |
| return f"❌ Error: {e}", d | |
| def open_directory(target, base_dir): | |
| if not target or not target.strip(): | |
| return list_files(base_dir) | |
| target = target.strip().rstrip("/") | |
| d = resolve_path(base_dir) if base_dir else WORKSPACE | |
| if target == "..": | |
| new_path = os.path.dirname(d) | |
| elif os.path.isabs(target): | |
| new_path = target | |
| else: | |
| new_path = os.path.join(d, target) | |
| if os.path.isdir(new_path): | |
| return list_files(new_path) | |
| return f"❌ Not a directory: {new_path}", d | |
| def read_file(filepath, base_dir): | |
| if not filepath or not filepath.strip(): | |
| return "", "❌ No file specified" | |
| filepath = filepath.strip() | |
| if os.path.isabs(filepath): | |
| fp = filepath | |
| else: | |
| fp = os.path.join(resolve_path(base_dir), filepath) | |
| if not os.path.exists(fp): | |
| return "", f"❌ File not found: {fp}" | |
| if not os.path.isfile(fp): | |
| return "", f"❌ Not a file: {fp}" | |
| try: | |
| sz = os.path.getsize(fp) | |
| if sz > 10 * 1024 * 1024: | |
| return "", f"❌ File too large: {sz / (1024*1024):.1f} MB (limit 10 MB)" | |
| with open(fp, "r", encoding="utf-8", errors="replace") as f: | |
| content = f.read() | |
| log_operation("READ", fp, f"Size: {sz} bytes") | |
| if sz < 1024: | |
| size_str = f"{sz} B" | |
| elif sz < 1024 * 1024: | |
| size_str = f"{sz / 1024:.1f} KB" | |
| else: | |
| size_str = f"{sz / (1024*1024):.1f} MB" | |
| return content, f"✅ Loaded: {fp} ({size_str}, {len(content.splitlines())} lines)" | |
| except Exception as e: | |
| log_operation("READ", fp, str(e), "error") | |
| return "", f"❌ Error reading file: {e}" | |
| def write_file(filepath, content, base_dir): | |
| if not filepath or not filepath.strip(): | |
| return "❌ No filename specified" | |
| filepath = filepath.strip() | |
| if os.path.isabs(filepath): | |
| fp = filepath | |
| else: | |
| fp = os.path.join(resolve_path(base_dir), filepath) | |
| try: | |
| parent = os.path.dirname(fp) | |
| if parent and not os.path.exists(parent): | |
| os.makedirs(parent, exist_ok=True) | |
| if os.path.exists(fp) and os.path.isfile(fp): | |
| ts = datetime.now().strftime("%Y%m%d_%H%M%S") | |
| backup_name = f"{os.path.basename(fp)}.{ts}.bak" | |
| backup_path = os.path.join(BACKUP_DIR, backup_name) | |
| try: | |
| shutil.copy2(fp, backup_path) | |
| except Exception: | |
| pass | |
| if content is None: | |
| content = "" | |
| with open(fp, "w", encoding="utf-8") as f: | |
| f.write(content) | |
| log_operation("WRITE", fp, f"Size: {len(content)} chars, {len(content.splitlines())} lines") | |
| return f"✅ Saved: {fp} ({len(content)} chars)" | |
| except Exception as e: | |
| log_operation("WRITE", fp, str(e), "error") | |
| return f"❌ Error saving file: {e}" | |
| def create_new_file(filename, base_dir): | |
| if not filename or not filename.strip(): | |
| return "❌ No filename specified", "" | |
| filename = filename.strip() | |
| fp = os.path.join(resolve_path(base_dir), filename) | |
| try: | |
| parent = os.path.dirname(fp) | |
| if parent and not os.path.exists(parent): | |
| os.makedirs(parent, exist_ok=True) | |
| if os.path.exists(fp): | |
| return f"⚠️ File already exists: {fp}", "" | |
| with open(fp, "w", encoding="utf-8") as f: | |
| f.write("") | |
| log_operation("CREATE", fp) | |
| return f"✅ Created: {fp}", list_files(base_dir)[0] | |
| except Exception as e: | |
| log_operation("CREATE", fp, str(e), "error") | |
| return f"❌ Error: {e}", "" | |
| def create_new_folder(foldername, base_dir): | |
| if not foldername or not foldername.strip(): | |
| return "❌ No folder name specified", "" | |
| foldername = foldername.strip() | |
| fp = os.path.join(resolve_path(base_dir), foldername) | |
| try: | |
| if os.path.exists(fp): | |
| return f"⚠️ Already exists: {fp}", list_files(base_dir)[0] | |
| os.makedirs(fp, exist_ok=True) | |
| log_operation("CREATE_DIR", fp) | |
| return f"✅ Created folder: {fp}", list_files(base_dir)[0] | |
| except Exception as e: | |
| log_operation("CREATE_DIR", fp, str(e), "error") | |
| return f"❌ Error: {e}", "" | |
| def delete_item(target_name, base_dir): | |
| if not target_name or not target_name.strip(): | |
| return "❌ No target specified", "" | |
| target_name = target_name.strip() | |
| fp = os.path.join(resolve_path(base_dir), target_name) | |
| if not os.path.exists(fp): | |
| return f"❌ Not found: {fp}", list_files(base_dir)[0] | |
| try: | |
| if os.path.isfile(fp): | |
| ts = datetime.now().strftime("%Y%m%d_%H%M%S") | |
| backup_name = f"{os.path.basename(fp)}.{ts}.deleted" | |
| try: | |
| shutil.copy2(fp, os.path.join(BACKUP_DIR, backup_name)) | |
| except Exception: | |
| pass | |
| os.remove(fp) | |
| log_operation("DELETE", fp) | |
| return f"✅ File deleted: {fp}", list_files(base_dir)[0] | |
| elif os.path.isdir(fp): | |
| shutil.rmtree(fp) | |
| log_operation("DELETE_DIR", fp) | |
| return f"✅ Folder deleted: {fp}", list_files(base_dir)[0] | |
| except Exception as e: | |
| log_operation("DELETE", fp, str(e), "error") | |
| return f"❌ Error: {e}", list_files(base_dir)[0] | |
| return f"❌ Unknown type: {fp}", list_files(base_dir)[0] | |
| def rename_item(old_name, new_name, base_dir): | |
| if not old_name or not new_name: | |
| return "❌ Specify both old and new names", "" | |
| d = resolve_path(base_dir) | |
| old_fp = os.path.join(d, old_name.strip()) | |
| new_fp = os.path.join(d, new_name.strip()) | |
| if not os.path.exists(old_fp): | |
| return f"❌ Not found: {old_fp}", list_files(base_dir)[0] | |
| if os.path.exists(new_fp): | |
| return f"❌ Target already exists: {new_fp}", list_files(base_dir)[0] | |
| try: | |
| os.rename(old_fp, new_fp) | |
| log_operation("RENAME", old_fp, f"to {new_fp}") | |
| return f"✅ Renamed: {old_name} to {new_name}", list_files(base_dir)[0] | |
| except Exception as e: | |
| log_operation("RENAME", old_fp, str(e), "error") | |
| return f"❌ Error: {e}", list_files(base_dir)[0] | |
| def copy_item(source, dest, base_dir): | |
| if not source or not dest: | |
| return "❌ Specify both source and destination", "" | |
| d = resolve_path(base_dir) | |
| src = os.path.join(d, source.strip()) if not os.path.isabs(source.strip()) else source.strip() | |
| dst = os.path.join(d, dest.strip()) if not os.path.isabs(dest.strip()) else dest.strip() | |
| if not os.path.exists(src): | |
| return f"❌ Source not found: {src}", list_files(base_dir)[0] | |
| try: | |
| if os.path.isfile(src): | |
| shutil.copy2(src, dst) | |
| elif os.path.isdir(src): | |
| shutil.copytree(src, dst) | |
| log_operation("COPY", src, f"to {dst}") | |
| return f"✅ Copied: {source} to {dest}", list_files(base_dir)[0] | |
| except Exception as e: | |
| log_operation("COPY", src, str(e), "error") | |
| return f"❌ Error: {e}", list_files(base_dir)[0] | |
| def move_item(source, dest, base_dir): | |
| if not source or not dest: | |
| return "❌ Specify both source and destination", "" | |
| d = resolve_path(base_dir) | |
| src = os.path.join(d, source.strip()) if not os.path.isabs(source.strip()) else source.strip() | |
| dst = os.path.join(d, dest.strip()) if not os.path.isabs(dest.strip()) else dest.strip() | |
| if not os.path.exists(src): | |
| return f"❌ Source not found: {src}", list_files(base_dir)[0] | |
| try: | |
| shutil.move(src, dst) | |
| log_operation("MOVE", src, f"to {dst}") | |
| return f"✅ Moved: {source} to {dest}", list_files(base_dir)[0] | |
| except Exception as e: | |
| log_operation("MOVE", src, str(e), "error") | |
| return f"❌ Error: {e}", list_files(base_dir)[0] | |
| def run_file_bg(filepath, base_dir): | |
| if not filepath or not filepath.strip(): | |
| return "❌ No file specified" | |
| filepath = filepath.strip() | |
| if os.path.isabs(filepath): | |
| fp = filepath | |
| else: | |
| fp = os.path.join(resolve_path(base_dir), filepath) | |
| if not os.path.exists(fp): | |
| return f"❌ File not found: {fp}" | |
| ext = Path(fp).suffix.lower() | |
| runners = { | |
| ".py": f"python3 \"{fp}\"", | |
| ".js": f"node \"{fp}\"", | |
| ".ts": f"npx ts-node \"{fp}\"", | |
| ".sh": f"bash \"{fp}\"", | |
| ".rb": f"ruby \"{fp}\"", | |
| ".go": f"go run \"{fp}\"", | |
| } | |
| cmd = runners.get(ext, f"python3 \"{fp}\"") | |
| return execute_background(cmd, os.path.dirname(fp)) | |
| def save_and_run(filepath, content, base_dir): | |
| if not filepath or not filepath.strip(): | |
| return "❌ No file specified" | |
| save_result = write_file(filepath, content, base_dir) | |
| if not save_result.startswith("✅"): | |
| return save_result | |
| run_result = run_file_bg(filepath, base_dir) | |
| return f"{save_result}\n{run_result}" | |
| def install_packages(names): | |
| if not names or not names.strip(): | |
| return "❌ No package names specified" | |
| packages = [p.strip() for p in names.replace(",", " ").split() if p.strip()] | |
| results = [] | |
| for pkg in packages: | |
| log_operation("INSTALL", pkg) | |
| output, _ = execute_terminal(f"pip install \"{pkg}\"", WORKSPACE)[:2] | |
| results.append(f"{'═' * 40}\n📦 {pkg}\n{'═' * 40}\n{output}\n") | |
| return "\n".join(results) | |
| def uninstall_packages(names): | |
| if not names or not names.strip(): | |
| return "❌ No package names specified" | |
| packages = [p.strip() for p in names.replace(",", " ").split() if p.strip()] | |
| results = [] | |
| for pkg in packages: | |
| log_operation("UNINSTALL", pkg) | |
| output, _ = execute_terminal(f"pip uninstall -y \"{pkg}\"", WORKSPACE)[:2] | |
| results.append(f"📦 {pkg}: {output}\n") | |
| return "\n".join(results) | |
| def upgrade_packages(names): | |
| if not names or not names.strip(): | |
| return "❌ No package names specified" | |
| packages = [p.strip() for p in names.replace(",", " ").split() if p.strip()] | |
| results = [] | |
| for pkg in packages: | |
| log_operation("UPGRADE", pkg) | |
| output, _ = execute_terminal(f"pip install --upgrade \"{pkg}\"", WORKSPACE)[:2] | |
| results.append(f"📦 {pkg}: {output}\n") | |
| return "\n".join(results) | |
| def get_installed_packages(): | |
| try: | |
| result = subprocess.run( | |
| "pip list --format=columns", | |
| shell=True, capture_output=True, text=True, timeout=30 | |
| ) | |
| return result.stdout if result.stdout else "No packages found" | |
| except Exception as e: | |
| return f"Error: {e}" | |
| def search_package(name): | |
| if not name or not name.strip(): | |
| return "❌ No package name" | |
| try: | |
| result = subprocess.run( | |
| f"pip show {name.strip()}", | |
| shell=True, capture_output=True, text=True, timeout=15 | |
| ) | |
| output = result.stdout if result.stdout else "" | |
| if result.stderr: | |
| output += result.stderr | |
| return output if output.strip() else f"Package '{name}' not found" | |
| except Exception as e: | |
| return f"Error: {e}" | |
| def get_system_info(): | |
| parts = [] | |
| commands = [ | |
| ("OS", "uname -a"), | |
| ("Python", "python3 --version 2>&1"), | |
| ("Pip", "pip --version 2>&1"), | |
| ("Node", "node --version 2>&1 || echo 'Not installed'"), | |
| ("Git", "git --version 2>&1 || echo 'Not installed'"), | |
| ] | |
| parts.append("+" + "=" * 42 + "+") | |
| parts.append("| SYSTEM INFORMATION |") | |
| parts.append("+" + "=" * 42 + "+\n") | |
| for label, cmd in commands: | |
| try: | |
| r = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=5) | |
| parts.append(f" {label}: {r.stdout.strip()}") | |
| except Exception: | |
| parts.append(f" {label}: N/A") | |
| parts.append("") | |
| try: | |
| r = subprocess.run("df -h . 2>/dev/null", shell=True, capture_output=True, text=True, timeout=5, cwd=WORKSPACE) | |
| parts.append(f"Disk Usage:\n{r.stdout.strip()}") | |
| except Exception: | |
| pass | |
| parts.append("") | |
| try: | |
| r = subprocess.run("free -h 2>/dev/null || echo 'N/A'", shell=True, capture_output=True, text=True, timeout=5) | |
| parts.append(f"Memory:\n{r.stdout.strip()}") | |
| except Exception: | |
| pass | |
| parts.append("") | |
| try: | |
| r = subprocess.run("nproc 2>/dev/null || echo 'N/A'", shell=True, capture_output=True, text=True, timeout=5) | |
| parts.append(f"CPU Cores: {r.stdout.strip()}") | |
| except Exception: | |
| pass | |
| parts.append(f"\nWorkspace: {WORKSPACE}") | |
| with proc_lock: | |
| running = sum(1 for p in bg_processes.values() if p["proc"].poll() is None) | |
| total = len(bg_processes) | |
| parts.append(f"Background Processes: {running} running / {total} total") | |
| parts.append(f"Operations Logged: {len(ops_log)}") | |
| parts.append(f"Server Time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") | |
| try: | |
| r = subprocess.run("uptime -p 2>/dev/null || uptime 2>/dev/null || echo 'N/A'", shell=True, capture_output=True, text=True, timeout=5) | |
| parts.append(f"Uptime: {r.stdout.strip()}") | |
| except Exception: | |
| pass | |
| return "\n".join(parts) | |
| def get_operations_log(filter_type="ALL", search_text=""): | |
| entries = list(ops_log) | |
| if filter_type and filter_type != "ALL": | |
| entries = [e for e in entries if e.get("type") == filter_type] | |
| if search_text and search_text.strip(): | |
| search_lower = search_text.strip().lower() | |
| entries = [e for e in entries if | |
| search_lower in e.get("target", "").lower() or | |
| search_lower in e.get("details", "").lower()] | |
| if not entries: | |
| return "No operations found" | |
| lines = [] | |
| lines.append(f"Showing {min(len(entries), 200)} of {len(entries)} operations\n") | |
| for entry in reversed(entries[-200:]): | |
| status_icon = "[OK]" if entry.get("status") == "success" else "[ERR]" | |
| lines.append(f"{status_icon} [{entry['time']}] {entry['type']}") | |
| lines.append(f" Target: {entry['target']}") | |
| if entry.get("details"): | |
| lines.append(f" Details: {entry['details']}") | |
| lines.append("") | |
| return "\n".join(lines) | |
| def get_file_history(filepath, base_dir): | |
| if not filepath or not filepath.strip(): | |
| return "❌ No file specified" | |
| filepath = filepath.strip() | |
| if os.path.isabs(filepath): | |
| fp = filepath | |
| else: | |
| fp = os.path.join(resolve_path(base_dir), filepath) | |
| safe_name = fp.replace("/", "_").replace("\\", "_").replace(" ", "_") | |
| if len(safe_name) > 100: | |
| safe_name = safe_name[:100] | |
| log_path = os.path.join(FILE_LOGS_DIR, f"{safe_name}.log") | |
| if os.path.exists(log_path): | |
| try: | |
| with open(log_path, "r", encoding="utf-8") as f: | |
| content = f.read() | |
| if content.strip(): | |
| return f"Operations log for: {fp}\n{'=' * 50}\n{content}" | |
| return f"No operations recorded for: {filepath}" | |
| except Exception as e: | |
| return f"❌ Error: {e}" | |
| return f"No operations recorded for: {filepath}" | |
| def clear_operations_log(): | |
| ops_log.clear() | |
| save_ops_log() | |
| return "✅ Operations log cleared" | |
| def get_backups_list(): | |
| try: | |
| if not os.path.exists(BACKUP_DIR): | |
| return "No backups" | |
| files = sorted(os.listdir(BACKUP_DIR), reverse=True) | |
| if not files: | |
| return "No backups" | |
| lines = [f"Backups Directory: {BACKUP_DIR}\n"] | |
| for f in files[:50]: | |
| fp = os.path.join(BACKUP_DIR, f) | |
| try: | |
| sz = os.path.getsize(fp) | |
| if sz < 1024: | |
| size = f"{sz} B" | |
| else: | |
| size = f"{sz / 1024:.1f} KB" | |
| mt = datetime.fromtimestamp(os.path.getmtime(fp)).strftime("%Y-%m-%d %H:%M") | |
| lines.append(f" {f} [{size}] {mt}") | |
| except Exception: | |
| lines.append(f" {f}") | |
| return "\n".join(lines) | |
| except Exception as e: | |
| return f"❌ Error: {e}" | |
| def restore_backup(backup_name, dest_path, base_dir): | |
| if not backup_name or not dest_path: | |
| return "❌ Specify backup name and destination" | |
| src = os.path.join(BACKUP_DIR, backup_name.strip()) | |
| if not os.path.exists(src): | |
| return f"❌ Backup not found: {backup_name}" | |
| dst = os.path.join(resolve_path(base_dir), dest_path.strip()) | |
| try: | |
| shutil.copy2(src, dst) | |
| log_operation("RESTORE", dst, f"From backup: {backup_name}") | |
| return f"✅ Restored: {backup_name} to {dst}" | |
| except Exception as e: | |
| return f"❌ Error: {e}" | |
| CSS = """ | |
| .terminal textarea { | |
| font-family: 'JetBrains Mono', 'Fira Code', 'Consolas', 'Monaco', monospace !important; | |
| background: #0d1117 !important; | |
| color: #58a6ff !important; | |
| font-size: 13px !important; | |
| line-height: 1.6 !important; | |
| border: 1px solid #30363d !important; | |
| border-radius: 8px !important; | |
| padding: 12px !important; | |
| } | |
| .file-browser textarea { | |
| font-family: 'JetBrains Mono', 'Consolas', monospace !important; | |
| background: #161b22 !important; | |
| color: #c9d1d9 !important; | |
| font-size: 13px !important; | |
| line-height: 1.5 !important; | |
| border: 1px solid #21262d !important; | |
| border-radius: 8px !important; | |
| } | |
| .status-display textarea { | |
| font-family: 'Consolas', monospace !important; | |
| background: #1c1e26 !important; | |
| color: #e0e0e0 !important; | |
| font-size: 13px !important; | |
| border-radius: 8px !important; | |
| } | |
| .log-display textarea { | |
| font-family: 'Consolas', monospace !important; | |
| background: #0d1117 !important; | |
| color: #8b949e !important; | |
| font-size: 12px !important; | |
| line-height: 1.4 !important; | |
| border-radius: 8px !important; | |
| } | |
| """ | |
| with gr.Blocks(title="Cloud VPS Pro") as app: | |
| gr.Markdown("# Cloud VPS Pro\n**Terminal - File Manager - Code Editor - Process Manager - Packages - System Monitor**") | |
| with gr.Tabs() as main_tabs: | |
| with gr.Tab("Terminal"): | |
| with gr.Row(): | |
| t_cmd = gr.Textbox( | |
| label="Command Input", | |
| placeholder="Enter command... (e.g. ls -la, python3 script.py)", | |
| lines=1, | |
| scale=5 | |
| ) | |
| t_dir = gr.Textbox(label="Working Directory", value=WORKSPACE, scale=2) | |
| with gr.Row(): | |
| t_run = gr.Button("Execute", variant="primary", scale=2) | |
| t_bg = gr.Button("Run in Background", variant="secondary", scale=2) | |
| t_clear = gr.Button("Clear", scale=1) | |
| t_out = gr.Textbox( | |
| label="Output", | |
| lines=22, | |
| max_lines=60, | |
| elem_classes=["terminal"] | |
| ) | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| gr.Markdown("**Quick Commands**") | |
| with gr.Row(): | |
| qb1 = gr.Button("ls -la", size="sm") | |
| qb2 = gr.Button("pwd", size="sm") | |
| qb3 = gr.Button("df -h", size="sm") | |
| qb4 = gr.Button("free -h", size="sm") | |
| with gr.Row(): | |
| qb5 = gr.Button("top snapshot", size="sm") | |
| qb6 = gr.Button("ps aux", size="sm") | |
| qb7 = gr.Button("env", size="sm") | |
| qb8 = gr.Button("pip list", size="sm") | |
| with gr.Accordion("Command History", open=False): | |
| t_history = gr.Textbox(label="Recent Commands", lines=12, elem_classes=["log-display"]) | |
| t_hist_btn = gr.Button("Refresh History") | |
| t_hist_btn.click(get_history_text, None, [t_history]) | |
| t_run.click(execute_terminal, [t_cmd, t_dir], [t_out, t_dir, t_history]) | |
| t_cmd.submit(execute_terminal, [t_cmd, t_dir], [t_out, t_dir, t_history]) | |
| t_bg.click(execute_background, [t_cmd, t_dir], [t_out]) | |
| t_clear.click(lambda: "", None, [t_out]) | |
| qb1.click(lambda d: execute_terminal("ls -la", d), [t_dir], [t_out, t_dir, t_history]) | |
| qb2.click(lambda d: execute_terminal("pwd", d), [t_dir], [t_out, t_dir, t_history]) | |
| qb3.click(lambda d: execute_terminal("df -h", d), [t_dir], [t_out, t_dir, t_history]) | |
| qb4.click(lambda d: execute_terminal("free -h", d), [t_dir], [t_out, t_dir, t_history]) | |
| qb5.click(lambda d: execute_terminal("top -bn1 | head -20", d), [t_dir], [t_out, t_dir, t_history]) | |
| qb6.click(lambda d: execute_terminal("ps aux", d), [t_dir], [t_out, t_dir, t_history]) | |
| qb7.click(lambda d: execute_terminal("env | sort", d), [t_dir], [t_out, t_dir, t_history]) | |
| qb8.click(lambda d: execute_terminal("pip list", d), [t_dir], [t_out, t_dir, t_history]) | |
| with gr.Tab("File Manager"): | |
| with gr.Row(): | |
| fm_dir = gr.Textbox(label="Current Directory", value=WORKSPACE, scale=4) | |
| fm_go = gr.Button("Go", scale=1) | |
| fm_up = gr.Button("Up", scale=1) | |
| fm_home = gr.Button("Home", scale=1) | |
| fm_refresh = gr.Button("Refresh", scale=1) | |
| with gr.Row(): | |
| with gr.Column(scale=3): | |
| fm_list = gr.Textbox( | |
| label="Files and Folders", | |
| lines=24, | |
| elem_classes=["file-browser"] | |
| ) | |
| with gr.Column(scale=2): | |
| fm_status = gr.Textbox(label="Status", lines=2, elem_classes=["status-display"]) | |
| with gr.Accordion("Create", open=True): | |
| fm_new_name = gr.Textbox(label="Name", placeholder="new_file.py or folder_name") | |
| with gr.Row(): | |
| fm_create_file_btn = gr.Button("New File", variant="primary") | |
| fm_create_dir_btn = gr.Button("New Folder", variant="primary") | |
| with gr.Accordion("Operations", open=True): | |
| fm_target = gr.Textbox(label="Target", placeholder="file or folder name") | |
| with gr.Row(): | |
| fm_open_btn = gr.Button("Open") | |
| fm_delete_btn = gr.Button("Delete", variant="stop") | |
| with gr.Accordion("Rename / Copy / Move", open=False): | |
| fm_source = gr.Textbox(label="Source Name") | |
| fm_dest = gr.Textbox(label="Destination Name") | |
| with gr.Row(): | |
| fm_rename_btn = gr.Button("Rename") | |
| fm_copy_btn = gr.Button("Copy") | |
| fm_move_btn = gr.Button("Move") | |
| fm_go.click(list_files, [fm_dir], [fm_list, fm_dir]) | |
| fm_refresh.click(list_files, [fm_dir], [fm_list, fm_dir]) | |
| fm_up.click(lambda d: list_files(os.path.dirname(resolve_path(d))), [fm_dir], [fm_list, fm_dir]) | |
| fm_home.click(lambda: list_files(WORKSPACE), None, [fm_list, fm_dir]) | |
| fm_open_btn.click(open_directory, [fm_target, fm_dir], [fm_list, fm_dir]) | |
| fm_create_file_btn.click(create_new_file, [fm_new_name, fm_dir], [fm_status, fm_list]) | |
| fm_create_dir_btn.click(create_new_folder, [fm_new_name, fm_dir], [fm_status, fm_list]) | |
| fm_delete_btn.click(delete_item, [fm_target, fm_dir], [fm_status, fm_list]) | |
| fm_rename_btn.click(rename_item, [fm_source, fm_dest, fm_dir], [fm_status, fm_list]) | |
| fm_copy_btn.click(copy_item, [fm_source, fm_dest, fm_dir], [fm_status, fm_list]) | |
| fm_move_btn.click(move_item, [fm_source, fm_dest, fm_dir], [fm_status, fm_list]) | |
| app.load(list_files, [fm_dir], [fm_list, fm_dir]) | |
| with gr.Tab("Code Editor"): | |
| with gr.Row(): | |
| e_path = gr.Textbox(label="File Path", placeholder="script.py", scale=3) | |
| e_base = gr.Textbox(label="Base Directory", value=WORKSPACE, scale=2) | |
| with gr.Row(): | |
| e_load_btn = gr.Button("Load File", variant="secondary", scale=1) | |
| e_save_btn = gr.Button("Save", variant="primary", scale=1) | |
| e_run_btn = gr.Button("Run", variant="primary", scale=1) | |
| e_saverun_btn = gr.Button("Save and Run", variant="primary", scale=1) | |
| e_new_btn = gr.Button("New", scale=1) | |
| e_editor = gr.Code(label="Code Editor", language="python", lines=28) | |
| e_status = gr.Textbox(label="Status", lines=3, elem_classes=["status-display"]) | |
| with gr.Accordion("File Operations History", open=False): | |
| e_file_log = gr.Textbox(label="Operations on this file", lines=10, elem_classes=["log-display"]) | |
| with gr.Row(): | |
| e_show_hist = gr.Button("Show File History") | |
| e_show_backups = gr.Button("Show Backups") | |
| e_backups_display = gr.Textbox(label="Available Backups", lines=8, elem_classes=["log-display"]) | |
| with gr.Accordion("Restore from Backup", open=False): | |
| with gr.Row(): | |
| e_backup_name = gr.Textbox(label="Backup Filename", scale=2) | |
| e_restore_dest = gr.Textbox(label="Restore To", scale=2) | |
| e_restore_btn = gr.Button("Restore", scale=1) | |
| e_restore_status = gr.Textbox(label="Restore Status", elem_classes=["status-display"]) | |
| e_load_btn.click(read_file, [e_path, e_base], [e_editor, e_status]) | |
| e_save_btn.click(write_file, [e_path, e_editor, e_base], [e_status]) | |
| e_run_btn.click(run_file_bg, [e_path, e_base], [e_status]) | |
| e_saverun_btn.click(save_and_run, [e_path, e_editor, e_base], [e_status]) | |
| e_new_btn.click(lambda: ("", "New file ready"), None, [e_editor, e_status]) | |
| e_show_hist.click(get_file_history, [e_path, e_base], [e_file_log]) | |
| e_show_backups.click(get_backups_list, None, [e_backups_display]) | |
| e_restore_btn.click(restore_backup, [e_backup_name, e_restore_dest, e_base], [e_restore_status]) | |
| with gr.Tab("Process Manager"): | |
| with gr.Row(): | |
| p_refresh_btn = gr.Button("Refresh", variant="secondary", scale=2) | |
| p_cleanup_btn = gr.Button("Cleanup Finished", scale=2) | |
| p_stopall_btn = gr.Button("Stop All", variant="stop", scale=2) | |
| p_list_display = gr.Textbox( | |
| label="Background Processes", | |
| lines=16, | |
| elem_classes=["terminal"] | |
| ) | |
| with gr.Row(): | |
| with gr.Column(scale=2): | |
| gr.Markdown("### Process Control") | |
| p_pid_input = gr.Textbox(label="PID or Process ID", placeholder="Enter PID number or process ID") | |
| with gr.Row(): | |
| p_stop_btn = gr.Button("Stop Process", variant="stop") | |
| p_remove_btn = gr.Button("Remove from List") | |
| p_log_btn = gr.Button("View Log") | |
| with gr.Column(scale=2): | |
| p_action_result = gr.Textbox(label="Action Result", lines=4, elem_classes=["status-display"]) | |
| with gr.Accordion("Process Log Output", open=False): | |
| p_log_content = gr.Textbox( | |
| label="Log Content", | |
| lines=18, | |
| elem_classes=["log-display"] | |
| ) | |
| gr.Markdown("---") | |
| gr.Markdown("### Start New Background Process") | |
| with gr.Row(): | |
| p_new_cmd = gr.Textbox(label="Command", placeholder="python3 bot.py", scale=4) | |
| p_new_dir = gr.Textbox(label="Working Directory", value=WORKSPACE, scale=2) | |
| p_new_run = gr.Button("Start", variant="primary", scale=1) | |
| p_new_result = gr.Textbox(label="Result", lines=3, elem_classes=["status-display"]) | |
| p_refresh_btn.click(get_process_list, None, [p_list_display]) | |
| p_stop_btn.click(stop_process, [p_pid_input], [p_action_result]) | |
| p_remove_btn.click(remove_from_list, [p_pid_input], [p_action_result, p_list_display]) | |
| p_log_btn.click(read_bg_log, [p_pid_input], [p_log_content]) | |
| p_cleanup_btn.click(cleanup_finished, None, [p_action_result, p_list_display]) | |
| p_stopall_btn.click(stop_all_processes, None, [p_action_result]) | |
| p_new_run.click(execute_background, [p_new_cmd, p_new_dir], [p_new_result]) | |
| app.load(get_process_list, None, [p_list_display]) | |
| with gr.Tab("Package Manager"): | |
| with gr.Row(): | |
| with gr.Column(scale=3): | |
| pk_input = gr.Textbox( | |
| label="Package Names (space or comma separated)", | |
| placeholder="requests flask numpy pandas", | |
| lines=1 | |
| ) | |
| with gr.Row(): | |
| pk_install_btn = gr.Button("Install", variant="primary", scale=2) | |
| pk_upgrade_btn = gr.Button("Upgrade", variant="secondary", scale=2) | |
| pk_uninstall_btn = gr.Button("Uninstall", variant="stop", scale=2) | |
| pk_output = gr.Textbox( | |
| label="Output", | |
| lines=12, | |
| elem_classes=["terminal"] | |
| ) | |
| with gr.Column(scale=2): | |
| with gr.Row(): | |
| pk_listall_btn = gr.Button("List All Packages", variant="secondary") | |
| pk_search_btn = gr.Button("Package Info") | |
| pk_search_input = gr.Textbox(label="Package to Search", placeholder="requests") | |
| pk_search_result = gr.Textbox(label="Package Info", lines=6, elem_classes=["status-display"]) | |
| pk_all_list = gr.Textbox( | |
| label="Installed Packages", | |
| lines=16, | |
| elem_classes=["file-browser"] | |
| ) | |
| gr.Markdown("### Quick Install") | |
| with gr.Row(): | |
| qi1 = gr.Button("requests", size="sm") | |
| qi2 = gr.Button("flask", size="sm") | |
| qi3 = gr.Button("numpy", size="sm") | |
| qi4 = gr.Button("pandas", size="sm") | |
| qi5 = gr.Button("discord.py", size="sm") | |
| qi6 = gr.Button("python-telegram-bot", size="sm") | |
| qi7 = gr.Button("aiohttp", size="sm") | |
| qi8 = gr.Button("beautifulsoup4", size="sm") | |
| pk_install_btn.click(install_packages, [pk_input], [pk_output]) | |
| pk_upgrade_btn.click(upgrade_packages, [pk_input], [pk_output]) | |
| pk_uninstall_btn.click(uninstall_packages, [pk_input], [pk_output]) | |
| pk_listall_btn.click(get_installed_packages, None, [pk_all_list]) | |
| pk_search_btn.click(search_package, [pk_search_input], [pk_search_result]) | |
| qi1.click(lambda: install_packages("requests"), None, [pk_output]) | |
| qi2.click(lambda: install_packages("flask"), None, [pk_output]) | |
| qi3.click(lambda: install_packages("numpy"), None, [pk_output]) | |
| qi4.click(lambda: install_packages("pandas"), None, [pk_output]) | |
| qi5.click(lambda: install_packages("discord.py"), None, [pk_output]) | |
| qi6.click(lambda: install_packages("python-telegram-bot"), None, [pk_output]) | |
| qi7.click(lambda: install_packages("aiohttp"), None, [pk_output]) | |
| qi8.click(lambda: install_packages("beautifulsoup4"), None, [pk_output]) | |
| with gr.Tab("System and Logs"): | |
| sys_refresh_btn = gr.Button("Refresh System Info", variant="secondary") | |
| sys_display = gr.Textbox( | |
| label="System Information", | |
| lines=20, | |
| elem_classes=["terminal"] | |
| ) | |
| gr.Markdown("---") | |
| gr.Markdown("### Operations Log") | |
| with gr.Row(): | |
| log_filter = gr.Dropdown( | |
| choices=["ALL", "COMMAND", "READ", "WRITE", "CREATE", "CREATE_DIR", | |
| "DELETE", "DELETE_DIR", "RENAME", "COPY", "MOVE", | |
| "BG_START", "BG_STOP", "BG_STOP_ALL", | |
| "INSTALL", "UNINSTALL", "UPGRADE", "RESTORE"], | |
| value="ALL", | |
| label="Filter by Type", | |
| scale=2 | |
| ) | |
| log_search = gr.Textbox(label="Search", placeholder="Search in logs...", scale=2) | |
| log_refresh_btn = gr.Button("Refresh", scale=1) | |
| log_clear_btn = gr.Button("Clear All Logs", variant="stop", scale=1) | |
| ops_display = gr.Textbox( | |
| label="Operations Log", | |
| lines=22, | |
| elem_classes=["log-display"] | |
| ) | |
| with gr.Accordion("Backups", open=False): | |
| bk_refresh_btn = gr.Button("Refresh Backups List") | |
| bk_display = gr.Textbox(label="Backups", lines=10, elem_classes=["file-browser"]) | |
| sys_refresh_btn.click(get_system_info, None, [sys_display]) | |
| log_refresh_btn.click(get_operations_log, [log_filter, log_search], [ops_display]) | |
| log_filter.change(get_operations_log, [log_filter, log_search], [ops_display]) | |
| log_search.submit(get_operations_log, [log_filter, log_search], [ops_display]) | |
| log_clear_btn.click(clear_operations_log, None, [ops_display]) | |
| bk_refresh_btn.click(get_backups_list, None, [bk_display]) | |
| app.load(get_system_info, None, [sys_display]) | |
| with gr.Tab("Help"): | |
| gr.Markdown(""" | |
| ## User Guide | |
| ### Terminal | |
| - Type any Linux/bash command and press Execute or Enter | |
| - Use cd to change directories | |
| - Run in Background starts long-running commands without blocking | |
| - Quick command buttons for common operations | |
| - Full command history tracking | |
| ### File Manager | |
| - Browse files and folders visually | |
| - Create, delete, rename, copy, and move files/folders | |
| - Navigate using Open, Up, and Home buttons | |
| - All file operations are automatically logged and backed up | |
| ### Code Editor | |
| - Load and edit any text file | |
| - Syntax highlighting for Python and other languages | |
| - Save and Run saves the file and runs it in background | |
| - View file operation history | |
| - Automatic backup before every save | |
| - Restore from backups when needed | |
| ### Process Manager | |
| - View all background processes with status | |
| - Stop individual processes or all at once | |
| - View real-time process log output | |
| - Cleanup finished processes | |
| - Start new background processes directly | |
| ### Package Manager | |
| - Install, upgrade, and uninstall Python packages | |
| - Install multiple packages at once (space separated) | |
| - View package info and list all installed packages | |
| - Quick install buttons for popular packages | |
| ### System and Logs | |
| - View system information (OS, Python, disk, memory, CPU) | |
| - Complete operations log with filtering and search | |
| - View and manage file backups | |
| - Track all activities performed in the VPS | |
| ### Tips | |
| - All file modifications create automatic backups | |
| - Every operation is logged with timestamp | |
| - Each file has its own operations history | |
| - Use the search function in logs to find specific operations | |
| - Background processes continue running even if you close the browser | |
| """) | |
| if __name__ == "__main__": | |
| app.queue(default_concurrency_limit=20) | |
| app.launch( | |
| server_name="0.0.0.0", | |
| server_port=7860, | |
| show_error=True, | |
| share=False, | |
| css=CSS, | |
| theme=gr.themes.Soft() | |
| ) |