457 lines
15 KiB
Python
457 lines
15 KiB
Python
#desktop.py
|
|
import os
|
|
import sys
|
|
import json
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
import platform
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
|
|
try:
|
|
import pyautogui
|
|
_PYAUTOGUI = True
|
|
except ImportError:
|
|
_PYAUTOGUI = False
|
|
|
|
_OS = platform.system() # "Windows" | "Darwin" | "Linux"
|
|
|
|
|
|
def _get_base_dir() -> Path:
|
|
if getattr(sys, "frozen", False):
|
|
return Path(sys.executable).parent
|
|
return Path(__file__).resolve().parent.parent
|
|
|
|
def _get_desktop() -> Path:
|
|
if _OS == "Linux":
|
|
xdg = os.environ.get("XDG_DESKTOP_DIR", "")
|
|
if xdg and Path(xdg).exists():
|
|
return Path(xdg)
|
|
return Path.home() / "Desktop"
|
|
|
|
def _build_sandbox() -> dict:
|
|
import time
|
|
|
|
safe_builtins = {
|
|
"print": print,
|
|
"len": len, "str": str, "int": int, "float": float,
|
|
"bool": bool, "list": list, "dict": dict, "tuple": tuple,
|
|
"range": range, "enumerate": enumerate, "sorted": sorted,
|
|
"isinstance": isinstance, "hasattr": hasattr, "getattr": getattr,
|
|
"max": max, "min": min, "sum": sum, "abs": abs,
|
|
"zip": zip, "map": map, "filter": filter,
|
|
}
|
|
|
|
sandbox = {
|
|
"__builtins__": safe_builtins,
|
|
"Path": Path,
|
|
"time": time,
|
|
"shutil": type("shutil", (), {
|
|
"copy2": shutil.copy2,
|
|
"copytree": shutil.copytree,
|
|
"disk_usage": shutil.disk_usage,
|
|
})(),
|
|
"os_path": os.path,
|
|
}
|
|
|
|
if _PYAUTOGUI:
|
|
sandbox["pyautogui"] = pyautogui
|
|
|
|
if _OS == "Windows":
|
|
try:
|
|
import ctypes
|
|
import winreg
|
|
sandbox["ctypes"] = ctypes
|
|
sandbox["winreg"] = type("winreg", (), {
|
|
# Sadece okuma
|
|
"OpenKey": winreg.OpenKey,
|
|
"QueryValueEx": winreg.QueryValueEx,
|
|
"HKEY_CURRENT_USER": winreg.HKEY_CURRENT_USER,
|
|
})()
|
|
except ImportError:
|
|
pass
|
|
|
|
return sandbox
|
|
|
|
|
|
def _execute_generated_code(code: str, player=None) -> str:
|
|
if not code or code.strip() == "UNSAFE":
|
|
return "This action cannot be performed safely."
|
|
|
|
# Kod temizleme
|
|
if code.startswith("```"):
|
|
lines = code.split("\n")
|
|
code = "\n".join(lines[1:-1]).strip()
|
|
|
|
sandbox = _build_sandbox()
|
|
output_lines = []
|
|
sandbox["__builtins__"]["print"] = lambda *a: output_lines.append(" ".join(str(x) for x in a))
|
|
|
|
try:
|
|
exec(compile(code, "<jarvis_desktop>", "exec"), sandbox)
|
|
return "\n".join(output_lines) if output_lines else "Done."
|
|
except Exception as e:
|
|
print(f"[Desktop] Exec error: {e}\nCode:\n{code[:300]}")
|
|
return f"Execution error: {e}"
|
|
|
|
def _ask_gemini_for_desktop_action(task: str) -> str:
|
|
from or_client import client
|
|
|
|
desktop = str(_get_desktop())
|
|
os_specific = {
|
|
"Windows": "- ctypes (Windows API calls, read-only)\n- winreg (registry READ only)",
|
|
"Darwin": "- subprocess is NOT available; use pyautogui or Path only",
|
|
"Linux": "- subprocess is NOT available; use pyautogui or Path only",
|
|
}.get(_OS, "")
|
|
|
|
prompt = f"""You are a desktop automation assistant.
|
|
Current OS: {_OS}
|
|
Desktop path: {desktop}
|
|
Generate safe Python code to accomplish the task below.
|
|
Allowed modules ONLY:
|
|
- pyautogui (mouse, keyboard — if needed)
|
|
- pathlib.Path (file/folder inspection only, no deletion)
|
|
- shutil.copy2, shutil.copytree, shutil.disk_usage (NO move, NO rmtree)
|
|
- os_path (os.path equivalent, read-only)
|
|
- time.sleep
|
|
{os_specific}
|
|
Hard rules:
|
|
- NO file deletion, NO subprocess, NO exec/eval inside the code
|
|
- NO import statements, NO file write except explicitly requested
|
|
- If task cannot be done safely with these tools, output exactly: UNSAFE
|
|
Output ONLY the Python code. No explanation, no markdown, no backticks.
|
|
Task: {task}"""
|
|
|
|
try:
|
|
return client.chat(prompt, system="You are a code generator. Output only raw Python code.")
|
|
except Exception as e:
|
|
return f"ERROR: {e}"
|
|
|
|
def set_wallpaper(image_path: str) -> str:
|
|
path = Path(image_path).expanduser().resolve()
|
|
if not path.exists():
|
|
return f"Image not found: {image_path}"
|
|
if path.suffix.lower() not in {".jpg", ".jpeg", ".png", ".bmp", ".webp"}:
|
|
return f"Unsupported format: {path.suffix}. Use jpg, png, bmp or webp."
|
|
|
|
try:
|
|
if _OS == "Windows":
|
|
import ctypes
|
|
if path.suffix.lower() in {".webp", ".png"}:
|
|
try:
|
|
from PIL import Image
|
|
bmp_path = Path(tempfile.mktemp(suffix=".bmp"))
|
|
Image.open(path).convert("RGB").save(bmp_path, "BMP")
|
|
path = bmp_path
|
|
except ImportError:
|
|
pass
|
|
ctypes.windll.user32.SystemParametersInfoW(20, 0, str(path), 3)
|
|
return f"Wallpaper set: {path.name}"
|
|
|
|
elif _OS == "Darwin":
|
|
script = (
|
|
f'tell application "System Events" to tell every desktop to '
|
|
f'set picture to POSIX file "{path}"'
|
|
)
|
|
subprocess.run(["osascript", "-e", script], capture_output=True)
|
|
return f"Wallpaper set: {path.name}"
|
|
|
|
else:
|
|
desktop_env = os.environ.get("XDG_CURRENT_DESKTOP", "").lower()
|
|
uri = f"file://{path}"
|
|
|
|
if "gnome" in desktop_env or "unity" in desktop_env:
|
|
subprocess.run([
|
|
"gsettings", "set", "org.gnome.desktop.background",
|
|
"picture-uri", uri
|
|
], capture_output=True)
|
|
subprocess.run([
|
|
"gsettings", "set", "org.gnome.desktop.background",
|
|
"picture-uri-dark", uri
|
|
], capture_output=True)
|
|
|
|
elif "kde" in desktop_env:
|
|
# KDE Plasma
|
|
script = f"""
|
|
var allDesktops = desktops();
|
|
for (var i = 0; i < allDesktops.length; i++) {{
|
|
d = allDesktops[i];
|
|
d.wallpaperPlugin = "org.kde.image";
|
|
d.currentConfigGroup = ["Wallpaper", "org.kde.image", "General"];
|
|
d.writeConfig("Image", "file://{path}");
|
|
}}
|
|
"""
|
|
subprocess.run(
|
|
["qdbus", "org.kde.plasmashell", "/PlasmaShell",
|
|
"org.kde.PlasmaShell.evaluateScript", script],
|
|
capture_output=True
|
|
)
|
|
|
|
elif "xfce" in desktop_env:
|
|
subprocess.run([
|
|
"xfconf-query", "-c", "xfce4-desktop",
|
|
"-p", "/backdrop/screen0/monitor0/workspace0/last-image",
|
|
"-s", str(path)
|
|
], capture_output=True)
|
|
|
|
else:
|
|
result = subprocess.run(
|
|
["feh", "--bg-scale", str(path)],
|
|
capture_output=True
|
|
)
|
|
if result.returncode != 0:
|
|
return (
|
|
f"Could not set wallpaper automatically on {desktop_env}. "
|
|
f"Try manually or install 'feh'."
|
|
)
|
|
|
|
return f"Wallpaper set: {path.name}"
|
|
|
|
except Exception as e:
|
|
return f"Could not set wallpaper: {e}"
|
|
|
|
|
|
def set_wallpaper_from_url(url: str) -> str:
|
|
try:
|
|
import urllib.request
|
|
suffix = Path(url.split("?")[0]).suffix or ".jpg"
|
|
tmp = Path(tempfile.mktemp(suffix=suffix))
|
|
urllib.request.urlretrieve(url, str(tmp))
|
|
result = set_wallpaper(str(tmp))
|
|
try:
|
|
tmp.unlink()
|
|
except Exception:
|
|
pass
|
|
return result
|
|
except Exception as e:
|
|
return f"Could not download wallpaper: {e}"
|
|
|
|
|
|
def get_current_wallpaper() -> str:
|
|
try:
|
|
if _OS == "Windows":
|
|
import winreg
|
|
key = winreg.OpenKey(
|
|
winreg.HKEY_CURRENT_USER, r"Control Panel\Desktop"
|
|
)
|
|
val, _ = winreg.QueryValueEx(key, "Wallpaper")
|
|
winreg.CloseKey(key)
|
|
return f"Current wallpaper: {val}"
|
|
|
|
elif _OS == "Darwin":
|
|
script = (
|
|
'tell application "System Events" to get picture of desktop 1'
|
|
)
|
|
result = subprocess.run(
|
|
["osascript", "-e", script],
|
|
capture_output=True, text=True
|
|
)
|
|
return f"Current wallpaper: {result.stdout.strip()}"
|
|
|
|
else:
|
|
desktop_env = os.environ.get("XDG_CURRENT_DESKTOP", "").lower()
|
|
if "gnome" in desktop_env or "unity" in desktop_env:
|
|
result = subprocess.run(
|
|
["gsettings", "get", "org.gnome.desktop.background", "picture-uri"],
|
|
capture_output=True, text=True
|
|
)
|
|
return f"Current wallpaper: {result.stdout.strip()}"
|
|
return "Wallpaper path retrieval not supported for this desktop environment."
|
|
|
|
except Exception as e:
|
|
return f"Could not get wallpaper: {e}"
|
|
|
|
FILE_TYPE_MAP = {
|
|
"Images": {".jpg", ".jpeg", ".png", ".gif", ".bmp", ".webp", ".svg", ".ico", ".heic"},
|
|
"Documents": {".pdf", ".doc", ".docx", ".txt", ".xls", ".xlsx",
|
|
".ppt", ".pptx", ".csv", ".odt", ".ods", ".odp"},
|
|
"Videos": {".mp4", ".avi", ".mkv", ".mov", ".wmv", ".flv", ".webm", ".m4v"},
|
|
"Music": {".mp3", ".wav", ".flac", ".aac", ".ogg", ".wma", ".m4a"},
|
|
"Archives": {".zip", ".rar", ".7z", ".tar", ".gz", ".bz2", ".xz"},
|
|
"Code": {".py", ".js", ".ts", ".html", ".css", ".json", ".xml",
|
|
".cpp", ".java", ".cs", ".go", ".rs", ".sh", ".php"},
|
|
"Executables": {".exe", ".msi", ".bat", ".cmd", ".sh", ".appimage", ".deb", ".rpm"},
|
|
}
|
|
|
|
_SKIP_EXTENSIONS = {
|
|
"Windows": {".lnk", ".url"},
|
|
"Darwin": {".webloc"},
|
|
"Linux": {".desktop"},
|
|
}
|
|
|
|
|
|
def organize_desktop(mode: str = "by_type") -> str:
|
|
desktop = _get_desktop()
|
|
skip_exts = _SKIP_EXTENSIONS.get(_OS, set())
|
|
moved, skipped = [], []
|
|
|
|
for item in desktop.iterdir():
|
|
if item.is_dir() or item.name.startswith("."):
|
|
continue
|
|
if item.suffix.lower() in skip_exts:
|
|
continue
|
|
|
|
if mode == "by_date":
|
|
mtime = datetime.fromtimestamp(item.stat().st_mtime)
|
|
folder_name = mtime.strftime("%Y-%m")
|
|
else:
|
|
ext = item.suffix.lower()
|
|
folder_name = "Others"
|
|
for folder, exts in FILE_TYPE_MAP.items():
|
|
if ext in exts:
|
|
folder_name = folder
|
|
break
|
|
|
|
target_dir = desktop / folder_name
|
|
target_dir.mkdir(exist_ok=True)
|
|
new_path = target_dir / item.name
|
|
|
|
if new_path.exists():
|
|
skipped.append(item.name)
|
|
continue
|
|
|
|
shutil.move(str(item), str(new_path))
|
|
moved.append(f"{item.name} → {folder_name}/")
|
|
|
|
result = f"Desktop organized ({mode}): {len(moved)} files moved."
|
|
if moved:
|
|
result += "\n" + "\n".join(moved[:8])
|
|
if len(moved) > 8:
|
|
result += f"\n... and {len(moved) - 8} more."
|
|
if skipped:
|
|
result += f"\n{len(skipped)} file(s) skipped (name conflict)."
|
|
return result
|
|
|
|
|
|
def list_desktop() -> str:
|
|
desktop = _get_desktop()
|
|
items = []
|
|
for item in sorted(desktop.iterdir()):
|
|
if item.name.startswith("."):
|
|
continue
|
|
if item.is_dir():
|
|
try:
|
|
count = len(list(item.iterdir()))
|
|
except PermissionError:
|
|
count = "?"
|
|
items.append(f"📁 {item.name}/ ({count} items)")
|
|
else:
|
|
size = item.stat().st_size
|
|
size_str = (
|
|
f"{size / 1024:.1f} KB" if size < 1024 * 1024
|
|
else f"{size / 1024 / 1024:.1f} MB"
|
|
)
|
|
items.append(f"📄 {item.name} ({size_str})")
|
|
|
|
if not items:
|
|
return "Desktop is empty."
|
|
return f"Desktop ({len(items)} items):\n" + "\n".join(items)
|
|
|
|
|
|
def clean_desktop() -> str:
|
|
desktop = _get_desktop()
|
|
skip_exts = _SKIP_EXTENSIONS.get(_OS, set())
|
|
today = datetime.now().strftime("%Y-%m-%d")
|
|
archive_dir = desktop / f"Desktop Archive {today}"
|
|
archive_dir.mkdir(exist_ok=True)
|
|
|
|
moved = 0
|
|
for item in desktop.iterdir():
|
|
if item.is_dir() or item.name.startswith("."):
|
|
continue
|
|
if item.suffix.lower() in skip_exts:
|
|
continue
|
|
new_path = archive_dir / item.name
|
|
if not new_path.exists():
|
|
shutil.move(str(item), str(new_path))
|
|
moved += 1
|
|
|
|
return f"Desktop cleaned: {moved} files archived to '{archive_dir.name}'."
|
|
|
|
|
|
def get_desktop_stats() -> str:
|
|
desktop = _get_desktop()
|
|
files = [i for i in desktop.iterdir() if i.is_file()]
|
|
folders = [i for i in desktop.iterdir() if i.is_dir()]
|
|
total_size = sum(f.stat().st_size for f in files if f.exists())
|
|
size_str = (
|
|
f"{total_size / 1024:.1f} KB" if total_size < 1024 * 1024
|
|
else f"{total_size / 1024 / 1024:.1f} MB"
|
|
)
|
|
return (
|
|
f"Desktop stats ({_OS}):\n"
|
|
f" Files : {len(files)}\n"
|
|
f" Folders : {len(folders)}\n"
|
|
f" Size : {size_str}\n"
|
|
f" Path : {desktop}"
|
|
)
|
|
|
|
def desktop_control(
|
|
parameters: dict = None,
|
|
response=None,
|
|
player=None,
|
|
session_memory=None,
|
|
) -> str:
|
|
"""
|
|
parameters:
|
|
action : wallpaper | wallpaper_url | current_wallpaper |
|
|
organize | clean | list | stats |
|
|
task (AI-powered)
|
|
path : image path for 'wallpaper'
|
|
url : image URL for 'wallpaper_url'
|
|
mode : 'by_type' or 'by_date' for 'organize'
|
|
task : natural language description for AI-powered actions
|
|
"""
|
|
params = parameters or {}
|
|
action = params.get("action", "").lower().strip()
|
|
task = params.get("task", "").strip()
|
|
|
|
if player:
|
|
player.write_log(f"[desktop] {action or task[:40]}")
|
|
|
|
try:
|
|
if action == "wallpaper":
|
|
path = params.get("path", "")
|
|
return set_wallpaper(path) if path else "No image path provided."
|
|
|
|
elif action == "wallpaper_url":
|
|
url = params.get("url", "")
|
|
return set_wallpaper_from_url(url) if url else "No URL provided."
|
|
|
|
elif action == "current_wallpaper":
|
|
return get_current_wallpaper()
|
|
|
|
elif action == "organize":
|
|
return organize_desktop(params.get("mode", "by_type"))
|
|
|
|
elif action == "clean":
|
|
return clean_desktop()
|
|
|
|
elif action == "list":
|
|
return list_desktop()
|
|
|
|
elif action == "stats":
|
|
return get_desktop_stats()
|
|
|
|
elif action == "task" or task:
|
|
actual_task = task or params.get("description", "")
|
|
if not actual_task:
|
|
return "Please describe what you want to do on the desktop."
|
|
|
|
print(f"[Desktop] Asking Gemini: {actual_task}")
|
|
if player:
|
|
player.write_log("[Desktop] Generating action...")
|
|
|
|
code = _ask_gemini_for_desktop_action(actual_task)
|
|
return _execute_generated_code(code, player=player)
|
|
|
|
else:
|
|
if action:
|
|
code = _ask_gemini_for_desktop_action(action)
|
|
return _execute_generated_code(code, player=player)
|
|
return "No action or task specified."
|
|
|
|
except Exception as e:
|
|
print(f"[Desktop] Error: {e}")
|
|
return f"Desktop control error: {e}" |