#!/usr/bin/env python3 """ Zetta Desktop Assistant — Python Server ========================================= - Serves static HTML chat UI - POST /chat → forwards to Zetta API (deepseek-v3) - POST /execute → local system commands (Windows/macOS/Linux) - System tray icon (pystray, optional) - Auto-opens browser on start - Config persistence: ~/.zetta-desktop/config.json - Security: file ops restricted to user home directory """ import os import sys import json import shutil import subprocess import platform import tempfile import mimetypes import webbrowser import threading import datetime import logging from pathlib import Path from http.server import HTTPServer, SimpleHTTPRequestHandler from urllib.parse import urlparse, parse_qs # ── Auto-install missing deps ── def _ensure_deps(): """Auto-install missing packages so non-technical users don't need pip.""" missing = [] for pkg in ['requests', 'websocket-client']: try: __import__(pkg) except ImportError: missing.append(pkg) if missing: print(f"正在安装依赖包: {', '.join(missing)}... (慢的话请稍等)") import subprocess as _sp for pkg in list(missing): try: print(f" → {pkg}...") _sp.check_call([sys.executable, '-m', 'pip', 'install', '--default-timeout=120'] + [pkg], timeout=180) print(f" ✓ {pkg} 安装完成") except Exception as e: print(f" ⚠ {pkg} 安装失败: {e}") if not missing: print("Done!") # Optional: pyautogui for mouse/keyboard control global PYAUTOGUI try: import pyautogui as _pag PYAUTOGUI = _pag except ImportError: print("⚠ pyautogui 未安装 — 鼠标/键盘控制不可用。") print(" 需要的话运行: pip install pyautogui") _ensure_deps() try: import requests as http_requests _HAS_REQUESTS = True except ImportError: http_requests = None _HAS_REQUESTS = False # Platform detection IS_WINDOWS = platform.system() == "Windows" IS_MACOS = platform.system() == "Darwin" IS_LINUX = platform.system() == "Linux" # Optional imports PYSTRAY = None if 'PYAUTOGUI' not in dir(): PYAUTOGUI = None def _try_imports(): global PYSTRAY, PYAUTOGUI try: import pystray from PIL import Image as PILImage, ImageDraw PYSTRAY = (pystray, PILImage, ImageDraw) except ImportError: pass try: import pyautogui PYAUTOGUI = pyautogui except Exception: PYAUTOGUI = None _try_imports() # ── Config ── CONFIG_DIR = Path.home() / ".zetta-desktop" CONFIG_FILE = CONFIG_DIR / "config.json" STATIC_DIR = Path(__file__).parent / "static" SCREENSHOTS_DIR = Path.home() / "Zetta_Screenshots" LOGS_DIR = CONFIG_DIR / "logs" HOST = "127.0.0.1" # v2.2.2: localhost only (0.0.0.0 breaks browser open URL) PORT = 8750 EMBEDDED_HTML = r''' Zetta Desktop Assistant
⚡

Zetta Desktop Assistant

Your AI assistant running locally on your PC. Ask me to open apps, manage files, search the web, or automate tasks.

⚙️ Settings

Zetta Desktop Assistant v2.2.1 · FULL CONTROL
Serving at localhost:PORT_PLACEHOLDER

''' # Inject actual port into EMBEDDED_HTML at startup EMBEDDED_HTML = EMBEDDED_HTML.replace('PORT_PLACEHOLDER', str(PORT)) ZETTA_API_URL = "https://zetta-api.com/api/v1/chat/completions" ZETTA_MODEL = "deepseek/deepseek-chat-v3" # ── Logging ── os.makedirs(LOGS_DIR, exist_ok=True) logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[ logging.FileHandler(LOGS_DIR / "assistant.log"), logging.StreamHandler() ] ) log = logging.getLogger("zetta-desktop") # ── Config helpers ── def load_config(): """Load config from ~/.zetta-desktop/config.json""" defaults = { "api_key": "", "device_id": "", "first_run": True, "version": "2.0.0", "created_at": datetime.datetime.now(datetime.UTC).isoformat() } try: if CONFIG_FILE.exists(): with open(CONFIG_FILE) as f: data = json.load(f) defaults.update(data) return defaults except Exception as e: log.warning(f"Error loading config: {e}") # Generate device_id if missing if not defaults.get("device_id"): import uuid defaults["device_id"] = f"desktop-{uuid.uuid4().hex[:10]}" return defaults def save_config(cfg): """Save config to disk""" try: os.makedirs(CONFIG_DIR, exist_ok=True) with open(CONFIG_FILE, 'w') as f: json.dump(cfg, f, indent=2) return True except Exception as e: log.error(f"Error saving config: {e}") return False # ── Security: path validation ── def is_allowed_path(path_str: str) -> bool: """Ensure the path is within user home directory or temp dir""" try: p = Path(path_str).resolve() home = Path.home().resolve() # Allow within home, temp, or the static/screenshots dirs allowed_parents = [ home, Path(tempfile.gettempdir()).resolve(), SCREENSHOTS_DIR.resolve(), STATIC_DIR.resolve(), ] for parent in allowed_parents: try: p.relative_to(parent) return True except ValueError: continue return False except Exception: return False def sanitize_path(user_path: str) -> str: """Convert a user-provided relative path to a safe absolute path in home dir""" p = Path(user_path) if p.is_absolute(): resolved = p.resolve() if is_allowed_path(str(resolved)): return str(resolved) # Fallback: put it in home with basename only return str(Path.home() / p.name) return str((Path.home() / p).resolve()) # ── Execute handlers ── def execute_list_dir(path: str) -> dict: """List directory contents""" safe = sanitize_path(path or str(Path.home())) try: if not os.path.isdir(safe): return {"success": False, "error": f"Not a directory: {safe}"} items = [] with os.scandir(safe) as it: for entry in sorted(it, key=lambda e: (not e.is_dir(), e.name.lower())): try: stat = entry.stat() size = stat.st_size if entry.is_file() else 0 items.append({ "name": entry.name, "type": "dir" if entry.is_dir() else "file", "size": size, "modified": datetime.datetime.fromtimestamp(stat.st_mtime).isoformat() }) except OSError: pass return { "success": True, "path": safe, "items": items[:500], # cap "count": len(items) } except PermissionError: return {"success": False, "error": f"Permission denied: {safe}"} except Exception as e: return {"success": False, "error": str(e)} def execute_read_file(path: str) -> dict: """Read a file's contents (text files only, max 100KB)""" safe = sanitize_path(path) if not is_allowed_path(safe): return {"success": False, "error": "Access denied: path outside allowed directories"} try: if not os.path.isfile(safe): return {"success": False, "error": f"Not a file: {safe}"} size = os.path.getsize(safe) if size > 102400: # 100KB limit return {"success": False, "error": f"File too large ({size} bytes). Max 100KB for reading."} # Check if likely text mime, _ = mimetypes.guess_type(safe) if mime and not mime.startswith(("text/", "application/json", "application/xml", "application/javascript")): return {"success": False, "error": f"Cannot read binary file: {mime}"} with open(safe, 'r', encoding='utf-8', errors='replace') as f: content = f.read() return {"success": True, "path": safe, "content": content, "size": len(content)} except PermissionError: return {"success": False, "error": f"Permission denied: {safe}"} except Exception as e: return {"success": False, "error": str(e)} def execute_write_file(path: str, content: str) -> dict: """Write content to a file""" safe = sanitize_path(path) if not is_allowed_path(safe): return {"success": False, "error": "Access denied: path outside allowed directories"} try: os.makedirs(os.path.dirname(safe), exist_ok=True) with open(safe, 'w', encoding='utf-8') as f: f.write(content) return {"success": True, "path": safe, "bytes_written": len(content.encode('utf-8'))} except PermissionError: return {"success": False, "error": f"Permission denied: {safe}"} except Exception as e: return {"success": False, "error": str(e)} def _is_installer_name(path: str) -> bool: """Check if a file path looks like an installer""" name = os.path.basename(path).lower() bad = ['setup', 'installer', 'install', 'update', 'uninstall', 'unins', 'setup64', 'setup32'] return any(b in name for b in bad) def execute_open_app(app_name: str) -> dict: """Open an application by name""" app_lower = app_name.lower().strip() if IS_WINDOWS: # Common Windows apps (English + Chinese) app_map = { "calculator": "calc.exe", "calc": "calc.exe", "notepad": "notepad.exe", "记事本": "notepad.exe", "paint": "mspaint.exe", "画图": "mspaint.exe", "explorer": "explorer.exe", "文件资源管理器": "explorer.exe", "cmd": "cmd.exe", "命令提示符": "cmd.exe", "powershell": "powershell.exe", "wordpad": "write.exe", "snipping tool": "snippingtool.exe", "截图工具": "snippingtool.exe", "task manager": "taskmgr.exe", "任务管理器": "taskmgr.exe", "control panel": "control.exe", "控制面板": "control.exe", "settings": "ms-settings:", "设置": "ms-settings:", "chrome": "chrome.exe", "谷歌浏览器": "chrome.exe", "edge": "msedge.exe", "microsoft edge": "msedge.exe", "firefox": "firefox.exe", "火狐": "firefox.exe", "微信": "wechat.exe", "wechat": "wechat.exe", "wps": "wps.exe", "wps office": "wps.exe", "excel": "excel.exe", "word": "winword.exe", "ppt": "powerpnt.exe", "powerpoint": "powerpnt.exe", "vscode": "code.exe", "visual studio code": "code.exe", "spotify": "spotify.exe", "vlc": "vlc.exe", "whatsapp": "whatsapp.exe", "telegram": "telegram.exe", "电报": "telegram.exe", } # Search keywords for Start Menu .lnk lookup start_keywords = { "wps": ["wps", "wps office", "kingsoft"], "wps office": ["wps", "wps office", "kingsoft"], "微信": ["wechat", "微信"], "wechat": ["wechat", "微信"], "chrome": ["chrome", "google chrome"], "谷歌浏览器": ["chrome", "google chrome"], "edge": ["edge", "microsoft edge"], "microsoft edge": ["edge", "microsoft edge"], "firefox": ["firefox"], "火狐": ["firefox"], "excel": ["excel"], "word": ["word", "winword"], "ppt": ["powerpoint"], "powerpoint": ["powerpoint"], "vscode": ["vscode", "visual studio code"], "visual studio code": ["vscode", "visual studio code"], } keywords = start_keywords.get(app_lower, [app_lower]) cmd = app_map.get(app_lower, None) try: if cmd and cmd.startswith("ms-"): subprocess.Popen(["start", "", cmd], shell=True) else: # Step 1: try 'where ' to find in PATH exe_name = cmd or app_lower found = None import subprocess as sp try: r = sp.run(['where', exe_name], capture_output=True, text=True, shell=True, timeout=3) if r.returncode == 0 and r.stdout.strip(): for line in r.stdout.strip().split('\n'): if not _is_installer_name(line): found = line.strip() break except: pass # Step 2: search Start Menu for .lnk shortcut if not found: sm_dirs = [ os.path.join(os.environ.get('APPDATA',''), r'Microsoft\Windows\Start Menu\Programs'), r'C:\ProgramData\Microsoft\Windows\Start Menu\Programs', ] for sm_dir in sm_dirs: if not os.path.isdir(sm_dir): continue for root, dirs, files in os.walk(sm_dir): for f in files: if not f.lower().endswith('.lnk'): continue fn_lower = f.lower().replace('.lnk','') for kw in keywords: if kw.lower() in fn_lower: found = os.path.join(root, f) break if found: break if found: break if found: break # Step 2.5: search Program Files for .exe directly if not found and exe_name.endswith('.exe'): # Known app paths first (faster) known_dirs = { 'wechat.exe': ['C:/Program Files (x86)/Tencent/WeChat', 'C:/Program Files/Tencent/WeChat'], 'whatsapp.exe': ['C:/Program Files/WindowsApps', 'C:/Users/'+os.environ.get('USERNAME','')+'/AppData/Local/WhatsApp'], 'telegram.exe': ['C:/Users/'+os.environ.get('USERNAME','')+'/AppData/Roaming/Telegram Desktop'], 'chrome.exe': ['C:/Program Files/Google/Chrome/Application', 'C:/Program Files (x86)/Google/Chrome/Application'], 'firefox.exe': ['C:/Program Files/Mozilla Firefox', 'C:/Program Files (x86)/Mozilla Firefox'], } for kdir in known_dirs.get(exe_name.lower(), []): candidate = os.path.join(kdir, exe_name) if os.path.isfile(candidate): found = candidate break import glob as _glob search_dirs = [ os.path.join(os.environ.get('LOCALAPPDATA',''), 'Programs'), os.path.join(os.environ.get('ProgramFiles',''), ''), os.path.join(os.environ.get('ProgramFiles(x86)',''), ''), r'C:\Program Files', r'C:\Program Files (x86)', ] for _sd in search_dirs: if not os.path.isdir(_sd): continue try: for _root, _dirs, _files in os.walk(_sd): for _f in _files: if _f.lower() == exe_name.lower() and not _is_installer_name(_f): found = os.path.join(_root, _f) break if found: break # Limit depth to 3 levels for speed if _root.count(os.sep) - _sd.count(os.sep) > 2: _dirs.clear() except: pass if found: break if found: sp.Popen(["start", "", found], shell=True) return {"success": True, "action": f"Opened {app_name}", "command": found} # Step 3: search Windows Store / UWP apps via PowerShell if not found: try: ps_cmd = f'Get-AppxPackage *{app_name}* | Select -ExpandProperty PackageFamilyName -First 1' r = sp.run(['powershell', '-NoProfile', '-Command', ps_cmd], capture_output=True, text=True, timeout=10, shell=True) pfn = r.stdout.strip() if pfn and not pfn.startswith('Get-AppxPackage'): sp.Popen(f"explorer shell:appsFolder\\{pfn}!App", shell=True) return {"success": True, "action": f"Opened {app_name} (Store app)", "command": f"AppX: {pfn}"} except: pass # Step 4: try Windows App Execution Aliases if not found: alias_dir = os.path.join(os.environ.get('LOCALAPPDATA',''), 'Microsoft', 'WindowsApps') if os.path.isdir(alias_dir): try: for f in os.listdir(alias_dir): if f.lower() == exe_name.lower(): found = os.path.join(alias_dir, f) break except: pass if found: sp.Popen(["start", "", found], shell=True) return {"success": True, "action": f"Opened {app_name} (alias)", "command": found} # Step 5: last resort — launch via friendly name if not found: try: sp.Popen(["start", "", app_name], shell=True) return {"success": True, "action": f"Opened {app_name}", "command": f"start {app_name}"} except: pass return {"success": False, "error": f"App '{app_name}' not found."} except Exception as e: return {"success": False, "error": str(e)} elif IS_MACOS: app_map = { "calculator": "Calculator", "calc": "Calculator", "textedit": "TextEdit", "safari": "Safari", "chrome": "Google Chrome", "finder": "Finder", "terminal": "Terminal", "settings": "System Settings", } app = app_map.get(app_lower, app_name) try: subprocess.Popen(["open", "-a", app]) return {"success": True, "action": f"Opened {app_name}"} except Exception as e: return {"success": False, "error": str(e)} else: # Linux app_map = { "calculator": "gnome-calculator", "calc": "gnome-calculator", "gedit": "gedit", "nautilus": "nautilus", "files": "nautilus", "terminal": "gnome-terminal", "settings": "gnome-control-center", } cmd = app_map.get(app_lower, app_name) try: subprocess.Popen([cmd], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) return {"success": True, "action": f"Opened {app_name}", "command": cmd} except FileNotFoundError: # Try xdg-open try: subprocess.Popen(["xdg-open", cmd], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) return {"success": True, "action": f"Opened {app_name} via xdg-open"} except: return {"success": False, "error": f"Application not found: {app_name}"} except Exception as e: return {"success": False, "error": str(e)} def execute_screenshot() -> dict: """Take a screenshot using pyautogui""" global PYAUTOGUI if PYAUTOGUI is None: # Fallback: try importing now try: import pyautogui PYAUTOGUI = pyautogui except ImportError: return {"success": False, "error": "pyautogui not installed. Run: pip install pyautogui"} try: os.makedirs(SCREENSHOTS_DIR, exist_ok=True) timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"screenshot_{timestamp}.png" filepath = SCREENSHOTS_DIR / filename screenshot = PYAUTOGUI.screenshot() screenshot.save(str(filepath)) return { "success": True, "path": str(filepath), "filename": filename, "message": f"Screenshot saved to {filepath}" } except Exception as e: return {"success": False, "error": str(e)} def execute_shell(command: str) -> dict: """Execute a shell command (RESTRICTED)""" DANGEROUS = ["rm -rf", "format", "del /f", "shutdown", "reboot", "mkfs", "dd if=", ":(){", "chmod 777", "> /dev/sda"] cmd_lower = command.lower() for d in DANGEROUS: if d in cmd_lower: return {"success": False, "error": f"Blocked dangerous command pattern: '{d}'"} try: result = subprocess.run( command, shell=True, capture_output=True, text=True, timeout=30, cwd=str(Path.home()) ) return { "success": result.returncode == 0, "stdout": result.stdout[-5000:], "stderr": result.stderr[-2000:], "returncode": result.returncode } except subprocess.TimeoutExpired: return {"success": False, "error": "Command timed out (30s limit)"} except Exception as e: return {"success": False, "error": str(e)} def execute_download(url, save_path=None): if not _HAS_REQUESTS: return {"success": False, "error": "requests library not installed. Run: pip install requests"} """Download a file from the internet""" try: _req = http_requests if not save_path: fname = url.split('/')[-1].split('?')[0] or 'download' save_path = os.path.join(str(Path.home()), 'Downloads', fname) os.makedirs(os.path.dirname(save_path) or '.', exist_ok=True) r = _req.get(url, stream=True, timeout=120) r.raise_for_status() total = int(r.headers.get('content-length', 0)) downloaded = 0 with open(save_path, 'wb') as f: for chunk in r.iter_content(8192): f.write(chunk) downloaded += len(chunk) size_mb = downloaded / (1024*1024) return {"success": True, "action": "Downloaded {:.1f}MB to {}".format(size_mb, save_path), "path": save_path, "size": downloaded} except Exception as e: return {"success": False, "error": str(e)} def execute_click(x=None, y=None): """Click at coordinates or current position""" try: import pyautogui as pag if x is not None and y is not None: pag.click(int(x), int(y)) else: pag.click() return {"success": True, "action": f"Clicked ({x},{y})"} except Exception as e: return {"success": False, "error": str(e)} def execute_type(text): """Type text using keyboard""" try: import pyautogui as pag pag.write(str(text), interval=0.05) return {"success": True, "action": f"Typed: {text[:80]}"} except Exception as e: return {"success": False, "error": str(e)} def execute_press(key): """Press a key (enter, tab, escape, backspace, etc.)""" try: import pyautogui as pag pag.press(str(key)) return {"success": True, "action": f"Pressed: {key}"} except Exception as e: return {"success": False, "error": str(e)} def execute_hotkey(keys_str): """Press key combo like ctrl+v, alt+tab""" try: import pyautogui as pag keys = [k.strip().lower() for k in keys_str.split('+')] pag.hotkey(*keys) return {"success": True, "action": f"Hotkey: {keys_str}"} except Exception as e: return {"success": False, "error": str(e)} def execute_scroll(amount): """Scroll mouse wheel. Positive = up, negative = down.""" try: import pyautogui as pag pag.scroll(int(amount)) return {"success": True, "action": f"Scrolled {amount}"} except Exception as e: return {"success": False, "error": str(e)} def execute_move(x, y): """Move mouse to coordinates""" try: import pyautogui as pag pag.moveTo(int(x), int(y), duration=0.3) return {"success": True, "action": f"Moved to ({x},{y})"} except Exception as e: return {"success": False, "error": str(e)} def execute_drag(x1, y1, x2, y2): """Drag from (x1,y1) to (x2,y2)""" try: import pyautogui as pag pag.moveTo(int(x1), int(y1), duration=0.2) pag.drag(int(x2)-int(x1), int(y2)-int(y1), duration=0.5) return {"success": True, "action": f"Dragged ({x1},{y1})->({x2},{y2})"} except Exception as e: return {"success": False, "error": str(e)} # Conversation memory (keeps last 20 messages = 10 exchanges) CONVERSATION_HISTORY = [] EXECUTE_ROUTES = { "list_dir": execute_list_dir, "read_file": execute_read_file, "write_file": execute_write_file, "open_app": execute_open_app, "screenshot": execute_screenshot, "shell": execute_shell, "click": execute_click, "type": execute_type, "press": execute_press, "hotkey": execute_hotkey, "scroll": execute_scroll, "move": execute_move, "drag": execute_drag, "download": execute_download, } # ── HTTP Request Handler ── class ZettaHandler(SimpleHTTPRequestHandler): """Custom HTTP handler for Zetta Desktop Assistant""" def __init__(self, *args, **kwargs): self.config = load_config() super().__init__(*args, directory=str(STATIC_DIR), **kwargs) def log_message(self, format, *args): log.info(f"HTTP {self.client_address[0]} - {format % args}") def _send_json(self, data, status=200): body = json.dumps(data, ensure_ascii=False).encode('utf-8') self.send_response(status) self.send_header('Content-Type', 'application/json; charset=utf-8') self.send_header('Content-Length', str(len(body))) self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() self.wfile.write(body) def _read_body(self) -> dict: length = int(self.headers.get('Content-Length', 0)) if length == 0: return {} body = self.rfile.read(length) return json.loads(body.decode('utf-8')) def do_OPTIONS(self): self.send_response(200) self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS') self.send_header('Access-Control-Allow-Headers', 'Content-Type, Authorization') self.end_headers() def do_GET(self): global PYAUTOGUI parsed = urlparse(self.path) if parsed.path == '/health': self._send_json({ "status": "ok", "version": "2.0.0", "platform": platform.system(), "timestamp": datetime.datetime.now(datetime.UTC).isoformat(), "has_api_key": bool(self.config.get("api_key")), "pyautogui": bool(PYAUTOGUI) }) return if parsed.path == '/install_pyautogui': import subprocess as _sp2 try: _sp2.check_call([sys.executable, '-m', 'pip', 'install', '--default-timeout=180', 'pyautogui'], timeout=300) import pyautogui as _pag PYAUTOGUI = _pag self._send_json({"success": True, "message": "pyautogui installed"}) except Exception as e: self._send_json({"success": False, "error": str(e)[:200]}) return if parsed.path == '/' or parsed.path == '/desktop-assistant.html' or parsed.path == '/desktop-assistant': self.send_response(200) self.send_header('Content-Type', 'text/html; charset=utf-8') self.end_headers() self.wfile.write(EMBEDDED_HTML.encode('utf-8')) return if parsed.path == '/config': cfg = self.config.copy() if 'api_key' in cfg and cfg['api_key']: cfg['api_key'] = cfg['api_key'][:8] + '...' + cfg['api_key'][-4:] if len(cfg['api_key']) > 12 else '***' self._send_json(cfg) return # Serve static files super().do_GET() def do_POST(self): global PYAUTOGUI parsed = urlparse(self.path) if parsed.path == '/chat': self._handle_chat() elif parsed.path == '/execute': self._handle_execute() elif parsed.path == '/config': self._handle_config_update() else: self._send_json({"error": "Not found"}, 404) def _handle_chat(self): """Forward message to Zetta API with conversation history""" global CONVERSATION_HISTORY try: body = self._read_body() user_message = body.get('message', '') api_key = body.get('api_key') or self.config.get('api_key', '') if not user_message: self._send_json({"error": "No message provided"}, 400) return if not api_key: self._send_json({"error": "API key not configured. Please set your Zetta API key in settings."}, 401) return # Commands to reset/clear history if user_message.strip().lower() in ('/reset', '/clear', '重置', '清空记忆'): CONVERSATION_HISTORY = [] self._send_json({"reply": "Memory cleared. Starting fresh!", "actions": []}) return # Build system prompt with ACTION capability available_actions = ", ".join(EXECUTE_ROUTES.keys()) mouse_actions = """ For mouse click: [[ACTION:click]]x,y[[/ACTION]] (e.g. [[ACTION:click]]500,300[[/ACTION]] for coordinates) For typing text: [[ACTION:type]]hello world[[/ACTION]] For pressing keys: [[ACTION:press]]enter[[/ACTION]] or [[ACTION:press]]tab[[/ACTION]] or [[ACTION:press]]escape[[/ACTION]] For key combos: [[ACTION:hotkey]]ctrl+v[[/ACTION]] or [[ACTION:hotkey]]alt+tab[[/ACTION]] For scrolling: [[ACTION:scroll]]-500[[/ACTION]] (down) or [[ACTION:scroll]]300[[/ACTION]] (up) For moving mouse: [[ACTION:move]]x,y[[/ACTION]] For dragging: [[ACTION:drag]]x1,y1,x2,y2[[/ACTION]] IMPORTANT: When doing multi-step tasks (e.g. "open WhatsApp and send a message to Mom"): 1. First use open_app 2. Wait for app to load 3. Use click, type, press to interact 4. Combine multiple actions in your response as needed """ if PYAUTOGUI else "" now = datetime.datetime.now().isoformat() home = str(Path.home()) system_prompt = ( f"You are Zetta Desktop Assistant, running locally on the user's PC ({platform.system()}).\n" f"You have FULL CONTROL of this computer. You CAN open apps, take screenshots, list files, read/write files, and run shell commands.\n\n" f"⚠️ IMPORTANT: Only use action tags when the user EXPLICITLY asks you to perform an action on their computer. If they say hello, ask a question, or chat casually — just reply normally WITHOUT any action tags.\n\n" f"When the user DOES ask you to do something, use these tags:\n\n" f"For opening apps: [[ACTION:open_app]]app_name[[/ACTION]]\n" f"Examples: [[ACTION:open_app]]notepad[[/ACTION]] [[ACTION:open_app]]calculator[[/ACTION]] [[ACTION:open_app]]chrome[[/ACTION]]\n\n" f"For screenshots: [[ACTION:screenshot]][[/ACTION]]\n\n" f"For listing files: [[ACTION:list_dir]]path[[/ACTION]]\n" f"Example: [[ACTION:list_dir]]Desktop[[/ACTION]] [[ACTION:list_dir]]Downloads[[/ACTION]]\n\n" f"For reading a file: [[ACTION:read_file]]filepath[[/ACTION]]\n\n" f"For writing a file: [[ACTION:write_file]]filepath|content[[/ACTION]]\n\n" f"For shell commands: [[ACTION:shell]]command[[/ACTION]]\n" f"Example: [[ACTION:shell]]dir[[/ACTION]] [[ACTION:shell]]tasklist[[/ACTION]]\n" f"{mouse_actions}\n" f"After the action tag, provide a brief helpful message. Use the user's language (Chinese/Spanish/English).\n" f"If the user is just chatting, respond normally without action tags.\n\n" f"Available actions: {available_actions}\n" f"Current time: {now}\n" f"User's home: {home}" ) # Build messages with conversation history (keep last 20 = 10 exchanges) messages = [{"role": "system", "content": system_prompt}] for msg in CONVERSATION_HISTORY[-20:]: messages.append(msg) messages.append({"role": "user", "content": user_message}) # Add to history CONVERSATION_HISTORY.append({"role": "user", "content": user_message}) payload = { "model": ZETTA_MODEL, "messages": messages, "temperature": 0.7, "max_tokens": 2048 } headers = { "Content-Type": "application/json", "Authorization": f"Bearer {api_key}" } log.info(f"Chat request: {user_message[:80]}...") resp = http_requests.post( ZETTA_API_URL, json=payload, headers=headers, timeout=60 ) if resp.status_code == 200: data = resp.json() choices = data.get('choices', []) if choices: reply = choices[0].get('message', {}).get('content', '') # Save assistant reply to history CONVERSATION_HISTORY.append({"role": "assistant", "content": reply}) # Parse and execute ACTION tags import re as regex actions_executed = [] action_pattern = r'\[\[ACTION:(\w+)\]\](.*?)\[\[/ACTION\]\]' action_matches = regex.findall(action_pattern, reply, regex.DOTALL) for act_type, act_params in action_matches: act_type = act_type.strip() act_params = act_params.strip() if act_type in EXECUTE_ROUTES: try: if act_type == "open_app": app_name = act_params.strip() result = execute_open_app(app_name) elif act_type == "screenshot": result = execute_screenshot() elif act_type == "list_dir": result = execute_list_dir(act_params) elif act_type == "read_file": result = execute_read_file(act_params) elif act_type == "write_file": parts = act_params.split('|', 1) fpath = parts[0].strip() if parts else '' content = parts[1] if len(parts) > 1 else '' result = execute_write_file(fpath, content) elif act_type == "shell": result = execute_shell(act_params) elif act_type == "click": parts = act_params.split(',') x = int(parts[0].strip()) if len(parts) > 0 and parts[0].strip() else None y = int(parts[1].strip()) if len(parts) > 1 and parts[1].strip() else None result = execute_click(x, y) elif act_type == "type": result = execute_type(act_params) elif act_type == "press": result = execute_press(act_params) elif act_type == "hotkey": result = execute_hotkey(act_params) elif act_type == "scroll": result = execute_scroll(act_params) elif act_type == "move": parts = act_params.split(',') x = int(parts[0].strip()) if parts else 0 y = int(parts[1].strip()) if len(parts) > 1 else 0 result = execute_move(x, y) elif act_type == "drag": parts = act_params.split(',') if len(parts) >= 4: result = execute_drag(int(parts[0]), int(parts[1]), int(parts[2]), int(parts[3])) else: result = {"success": False, "error": "drag needs 4 coords: x1,y1,x2,y2"} elif act_type == "download": result = execute_download(act_params) else: result = {"success": False, "error": f"Unknown action: {act_type}"} actions_executed.append({"action": act_type, "params": act_params[:50], "result": result}) log.info(f"Executed: {act_type} -> {result.get('success', False)}") except Exception as e: actions_executed.append({"action": act_type, "params": act_params[:50], "result": {"success": False, "error": str(e)}}) # Clean up action tags from reply for display clean_reply = regex.sub(action_pattern, '', reply, flags=regex.DOTALL).strip() # Add execution results if actions_executed: result_lines = [] for a in actions_executed: r = a['result'] if r.get('success'): if a['action'] == 'open_app': result_lines.append(f"Opened: {a['params']}") elif a['action'] == 'screenshot': result_lines.append(f"Screenshot: {r.get('message', 'taken')}") elif a['action'] == 'list_dir': files = r.get('files', []) preview = ', '.join(files[:15]) result_lines.append(f"Files in {r.get('path', '')}: {preview}{'...' if len(files) > 15 else ''}") elif a['action'] == 'shell': out = r.get('stdout', '')[:200] result_lines.append(f"Command output: {out}") else: result_lines.append(f"{a['action']}: done") else: result_lines.append(f"{a['action']} failed: {r.get('error', 'unknown error')}") if clean_reply: clean_reply += "\n\n" + "\n".join(result_lines) else: clean_reply = "\n".join(result_lines) self._send_json({ "reply": clean_reply or reply, "model": data.get('model', ZETTA_MODEL), "usage": data.get('usage', {}), "actions": [a['action'] for a in actions_executed] if actions_executed else [] }) else: self._send_json({"reply": "No response from Zetta API.", "error": "empty_choices"}) elif resp.status_code == 401: self._send_json({"error": "Invalid API key. Please check your Zetta API key.", "detail": resp.text[:500]}, 401) elif resp.status_code == 429: self._send_json({"error": "Rate limited. Please wait a moment and try again."}, 429) else: log.error(f"Zetta API error {resp.status_code}: {resp.text[:500]}") self._send_json({ "error": f"Zetta API returned error {resp.status_code}", "detail": resp.text[:500] }, resp.status_code) except http_requests.Timeout: self._send_json({"error": "Request to Zetta API timed out. Please try again."}, 504) except http_requests.ConnectionError: self._send_json({"error": "Cannot connect to Zetta API. Check your internet connection."}, 502) except Exception as e: log.exception("Chat error") self._send_json({"error": f"Internal error: {str(e)}"}, 500) def _handle_execute(self): """Execute local system commands""" try: body = self._read_body() action = body.get('action', '') params = body.get('params', {}) if action not in EXECUTE_ROUTES: self._send_json({ "success": False, "error": f"Unknown action: {action}. Available: {list(EXECUTE_ROUTES.keys())}" }, 400) return handler = EXECUTE_ROUTES[action] log.info(f"Execute: {action} params={params}") if action == "list_dir": result = handler(params.get('path', '')) elif action == "read_file": result = handler(params.get('path', '')) elif action == "write_file": result = handler(params.get('path', ''), params.get('content', '')) elif action == "open_app": result = handler(params.get('app_name', '')) elif action == "screenshot": result = handler() elif action == "shell": result = handler(params.get('command', '')) elif action == "click": result = handler(params.get('x'), params.get('y')) elif action == "type": result = handler(params.get('text', '')) elif action == "press": result = handler(params.get('key', '')) elif action == "hotkey": result = handler(params.get('keys', '')) elif action == "scroll": result = handler(params.get('amount', 0)) elif action == "move": result = handler(params.get('x', 0), params.get('y', 0)) elif action == "drag": result = handler(params.get('x1', 0), params.get('y1', 0), params.get('x2', 0), params.get('y2', 0)) elif action == "download": result = handler(params.get('url', ''), params.get('save_path')) else: result = {"success": False, "error": "Not implemented"} self._send_json(result) except Exception as e: log.exception("Execute error") self._send_json({"success": False, "error": str(e)}, 500) def _handle_config_update(self): """Update config (API key, etc.)""" try: body = self._read_body() cfg = self.config.copy() if 'api_key' in body: cfg['api_key'] = body['api_key'] if 'first_run' in body: cfg['first_run'] = body['first_run'] if save_config(cfg): self.config = cfg self._send_json({"success": True, "message": "Config saved"}) else: self._send_json({"error": "Failed to save config"}, 500) except Exception as e: log.exception("Config update error") self._send_json({"error": str(e)}, 500) # ── Tray Icon ── def create_tray_icon(server_thread): """Create system tray icon (optional, graceful fallback)""" if PYSTRAY is None: log.info("pystray not available — skipping tray icon") return pystray_mod, PILImage, ImageDraw = PYSTRAY def make_icon_image(): """Create a simple Z icon""" size = 64 img = PILImage.new('RGBA', (size, size), (0, 0, 0, 0)) draw = ImageDraw.Draw(img) # Draw rounded rect background draw.rounded_rectangle([2, 2, size-2, size-2], radius=14, fill=(0, 176, 255)) # Draw Z draw.text((size//2 - 12, size//2 - 16), "Z", fill=(255, 255, 255)) return img def on_open(icon, item): webbrowser.open(f"http://{HOST}:{PORT}") def on_quit(icon, item): log.info("Tray quit requested — shutting down") icon.stop() os._exit(0) icon = pystray_mod.Icon( "zetta_desktop", make_icon_image(), "Zetta Desktop Assistant", menu=pystray_mod.Menu( pystray_mod.MenuItem("Open Zetta", on_open, default=True), pystray_mod.Menu.SEPARATOR, pystray_mod.MenuItem("Quit", on_quit) ) ) log.info("Starting system tray icon") icon.run() # ── WebSocket Remote Client ───────────────────────────── WEBSOCKET_SERVER = "wss://zetta-api.com" def run_websocket_client(config): """Connect to Zetta API WebSocket for remote control. Runs in a background thread with exponential backoff reconnect. """ import websocket import time as time_mod device_id = config.get("device_id", "") api_key = config.get("api_key", "") if not api_key: log.info("[Remote] No API key configured — skipping remote connection") return if not device_id: log.error("[Remote] No device_id — skipping") return ws_url = f"{WEBSOCKET_SERVER}/ws/desktop/{device_id}" reconnect_delay = 1.0 max_delay = 60.0 log.info(f"[Remote] Starting WebSocket client → {ws_url}") while True: try: ws = websocket.create_connection(ws_url, timeout=15) log.info(f"[Remote] Connected to {ws_url}") # Send auth auth_msg = json.dumps({ "type": "auth", "api_key": api_key, "device_name": platform.node() or device_id, "platform": f"{platform.system()} {platform.release()}", }) ws.send(auth_msg) # Wait for auth response ws.settimeout(10) auth_resp_raw = ws.recv() auth_resp = json.loads(auth_resp_raw) if auth_resp.get("type") == "error": log.error(f"[Remote] Auth failed: {auth_resp.get('message')}") ws.close() time_mod.sleep(reconnect_delay * 3) reconnect_delay = min(reconnect_delay * 2, max_delay) continue log.info(f"[Remote] Authenticated: {auth_resp.get('user_id', 'unknown')}") reconnect_delay = 1.0 # reset on success # Main command loop ws.settimeout(60) while True: try: raw = ws.recv() msg = json.loads(raw) if msg.get("type") == "ping": ws.send(json.dumps({"type": "pong"})) continue if msg.get("type") == "command": cmd = msg.get("command", "") params = msg.get("params", "") request_id = msg.get("request_id", "") result = execute_remote_command(cmd, params) response = json.dumps({ "type": "result", "request_id": request_id, "command": cmd, "success": result.get("success", False), "data": result, }) ws.send(response) log.info(f"[Remote] Executed: {cmd} -> success={result.get('success')}") except websocket.WebSocketTimeoutException: continue # Normal, keep waiting for commands except websocket.WebSocketConnectionClosedException: log.warning("[Remote] Connection closed by server") break except Exception as e: log.error(f"[Remote] Error in command loop: {e}") break ws.close() except (websocket.WebSocketException, ConnectionRefusedError, OSError, Exception) as e: log.warning(f"[Remote] Connection failed: {e} — retrying in {reconnect_delay:.1f}s") time_mod.sleep(reconnect_delay) reconnect_delay = min(reconnect_delay * 2, max_delay) def execute_remote_command(command: str, params: str) -> dict: """Execute a remote command from the server.""" try: if command == "open_app": return execute_open_app(params) elif command == "screenshot": result = execute_screenshot() # For remote screenshots, include base64 data if result.get("success") and os.path.exists(result.get("path", "")): try: import base64 with open(result["path"], "rb") as f: result["image_base64"] = base64.b64encode(f.read()).decode() except Exception: pass return result elif command == "list_dir": return execute_list_dir(params) elif command == "read_file": return execute_read_file(params) elif command == "write_file": parts = params.split("|", 1) fpath = parts[0].strip() if parts else "" content = parts[1] if len(parts) > 1 else "" return execute_write_file(fpath, content) elif command == "shell": return execute_shell(params) elif command == "chat": # Forward chat to Zetta API with local system context return remote_chat(params) else: return {"success": False, "error": f"Unknown remote command: {command}"} except Exception as e: log.exception(f"[Remote] Command error: {command}") return {"success": False, "error": str(e)} def remote_chat(message: str) -> dict: """Process a chat message via Zetta API, same as local /chat endpoint.""" config = load_config() api_key = config.get("api_key", "") if not api_key: return {"success": False, "error": "API key not configured"} available_actions = ", ".join(EXECUTE_ROUTES.keys()) mouse_actions = """For mouse click: [[ACTION:click]]x,y[[/ACTION]] For typing text: [[ACTION:type]]text[[/ACTION]] For pressing keys: [[ACTION:press]]enter[[/ACTION]] For key combos: [[ACTION:hotkey]]ctrl+v[[/ACTION]] For scrolling: [[ACTION:scroll]]-500[[/ACTION]] For moving mouse: [[ACTION:move]]x,y[[/ACTION]] For dragging: [[ACTION:drag]]x1,y1,x2,y2[[/ACTION]] IMPORTANT: Combine multiple actions for multi-step tasks. """ if PYAUTOGUI else "" now = datetime.datetime.now().isoformat() home = str(Path.home()) system_prompt = ( f"You are Zetta Desktop Assistant, running locally on the user's PC ({platform.system()}).\n" f"You have FULL CONTROL of this computer. You CAN open apps, take screenshots, list files, read/write files, and run shell commands.\n\n" f"⚠️ IMPORTANT: Only use action tags when the user EXPLICITLY asks you to perform an action. If they chat casually, reply WITHOUT action tags.\n\n" f"When action is needed, use:\n" f"[[ACTION:open_app]]app_name[[/ACTION]]\n" f"[[ACTION:screenshot]][[/ACTION]]\n" f"[[ACTION:list_dir]]path[[/ACTION]]\n" f"[[ACTION:read_file]]filepath[[/ACTION]]\n" f"[[ACTION:write_file]]filepath|content[[/ACTION]]\n" f"[[ACTION:shell]]command[[/ACTION]]\n" f"{mouse_actions}\n" f"Available actions: {available_actions}\n" f"Current time: {now}\n" f"User's home: {home}\n\n" f"After the action tag, provide a brief helpful message in the user's language.\n" f"If the user is just chatting, respond normally without action tags." ) try: resp = http_requests.post( ZETTA_API_URL, json={ "model": ZETTA_MODEL, "messages": [ {"role": "system", "content": system_prompt}, {"role": "user", "content": message} ], "temperature": 0.7, "max_tokens": 2048 }, headers={ "Content-Type": "application/json", "Authorization": f"Bearer {api_key}" }, timeout=60 ) if resp.status_code == 200: data = resp.json() choices = data.get('choices', []) if choices: reply = choices[0].get('message', {}).get('content', '') # Parse and execute ACTION tags import re as regex actions_executed = [] action_pattern = r'\[\[ACTION:(\w+)\]\](.*?)\[\[/ACTION\]\]' action_matches = regex.findall(action_pattern, reply, regex.DOTALL) for act_type, act_params in action_matches: act_type = act_type.strip() act_params = act_params.strip() if act_type in EXECUTE_ROUTES: try: if act_type == "open_app": result = execute_open_app(act_params) elif act_type == "screenshot": result = execute_screenshot() elif act_type == "list_dir": result = execute_list_dir(act_params) elif act_type == "read_file": result = execute_read_file(act_params) elif act_type == "write_file": parts = act_params.split('|', 1) fpath = parts[0].strip() if parts else '' content = parts[1] if len(parts) > 1 else '' result = execute_write_file(fpath, content) elif act_type == "shell": result = execute_shell(act_params) elif act_type == "click": parts = act_params.split(',') x = int(parts[0].strip()) if len(parts) > 0 and parts[0].strip() else None y = int(parts[1].strip()) if len(parts) > 1 and parts[1].strip() else None result = execute_click(x, y) elif act_type == "type": result = execute_type(act_params) elif act_type == "press": result = execute_press(act_params) elif act_type == "hotkey": result = execute_hotkey(act_params) elif act_type == "scroll": result = execute_scroll(act_params) elif act_type == "move": parts = act_params.split(',') x = int(parts[0].strip()) if parts else 0 y = int(parts[1].strip()) if len(parts) > 1 else 0 result = execute_move(x, y) elif act_type == "drag": parts = act_params.split(',') if len(parts) >= 4: result = execute_drag(int(parts[0]), int(parts[1]), int(parts[2]), int(parts[3])) else: result = {"success": False, "error": "drag needs 4 coords: x1,y1,x2,y2"} elif act_type == "download": result = execute_download(act_params) else: result = {"success": False, "error": f"Unknown action: {act_type}"} actions_executed.append({"action": act_type, "params": act_params[:50], "result": result}) except Exception as e: actions_executed.append({"action": act_type, "error": str(e)}) clean_reply = regex.sub(action_pattern, '', reply, flags=regex.DOTALL).strip() if actions_executed: result_lines = [] for a in actions_executed: r = a['result'] if r.get('success'): result_lines.append(f"✅ {a['action']}: {r.get('action', 'done')}") else: result_lines.append(f"❌ {a['action']}: {r.get('error', 'failed')}") if clean_reply: clean_reply += "\n\n" + "\n".join(result_lines) else: clean_reply = "\n".join(result_lines) return {"success": True, "reply": clean_reply or reply, "actions": [a['action'] for a in actions_executed]} return {"success": False, "error": f"API error: {resp.status_code}"} except Exception as e: return {"success": False, "error": f"Chat error: {str(e)}"} # ── Main ── def main(): log.info("=" * 60) log.info("Zetta Desktop Assistant v2.2.1") log.info(f"Platform: {platform.system()} {platform.release()}") log.info(f"Python: {sys.version}") log.info(f"Server: http://{HOST}:{PORT}") log.info("=" * 60) # Load config config = load_config() if config.get('first_run'): log.info("First run — API key will be requested via UI") config['first_run'] = False save_config(config) # Ensure static dir exists os.makedirs(STATIC_DIR, exist_ok=True) os.makedirs(SCREENSHOTS_DIR, exist_ok=True) # HTML is embedded in this file # Create server server = HTTPServer((HOST, PORT), ZettaHandler) log.info(f"Server listening on http://{HOST}:{PORT}") # Start server thread server_thread = threading.Thread(target=server.serve_forever, daemon=True) server_thread.start() # Start WebSocket remote client thread (connects to Zetta API) if config.get("api_key"): ws_thread = threading.Thread( target=run_websocket_client, args=(config,), daemon=True ) ws_thread.start() log.info("WebSocket remote client started") # Open browser url = f"http://{HOST}:{PORT}/" log.info(f"Opening browser: {url}") webbrowser.open(url) # Tray icon in main thread (blocking) or keep-alive if PYSTRAY is not None: create_tray_icon(server_thread) else: log.info("Server running. Press Ctrl+C to stop.") try: import time while True: time.sleep(1) except KeyboardInterrupt: log.info("Shutting down...") server.shutdown() if __name__ == "__main__": main()