diff --git a/code_helper.py b/code_helper.py new file mode 100644 index 0000000..78de1ca --- /dev/null +++ b/code_helper.py @@ -0,0 +1,520 @@ +import asyncio +import threading +import concurrent.futures +import platform +import shutil +import subprocess +from pathlib import Path +from playwright.async_api import async_playwright, TimeoutError as PlaywrightTimeout + + +def _get_default_browser_id() -> str: + """Returns raw default browser identifier string for current OS.""" + system = platform.system() + try: + if system == "Windows": + import winreg + key = winreg.OpenKey( + winreg.HKEY_CURRENT_USER, + r"Software\Microsoft\Windows\Shell\Associations\UrlAssociations\http\UserChoice" + ) + prog_id = winreg.QueryValueEx(key, "ProgId")[0].lower() + winreg.CloseKey(key) + return prog_id + + elif system == "Darwin": + result = subprocess.run( + ["defaults", "read", + "com.apple.LaunchServices/com.apple.launchservices.secure", + "LSHandlers"], + capture_output=True, text=True, timeout=5 + ) + return result.stdout.lower() + + elif system == "Linux": + result = subprocess.run( + ["xdg-settings", "get", "default-web-browser"], + capture_output=True, text=True, timeout=5 + ) + return result.stdout.lower() + + except Exception: + pass + + return "" + + +_BROWSER_BINARIES = { + "Windows": { + "opera": ["opera.exe"], + "brave": ["brave.exe"], + "vivaldi": ["vivaldi.exe"], + "chrome": ["chrome.exe"], + "firefox": ["firefox.exe"], + }, + "Darwin": { + "opera": ["opera"], + "brave": ["brave browser", "brave"], + "vivaldi": ["vivaldi"], + "chrome": ["google chrome", "google-chrome"], + "firefox": ["firefox"], + }, + "Linux": { + "opera": ["opera", "opera-stable"], + "brave": ["brave-browser", "brave"], + "vivaldi": ["vivaldi-stable", "vivaldi"], + "chrome": ["google-chrome", "google-chrome-stable", "chromium-browser", "chromium"], + "firefox": ["firefox"], + }, +} + + +def _get_opera_executable() -> str | None: + if platform.system() != "Windows": + return None + try: + import winreg + candidate_keys = [ + r"SOFTWARE\Microsoft\Windows\CurrentVersion\App Paths\opera.exe", + r"SOFTWARE\Microsoft\Windows\CurrentVersion\App Paths\launcher.exe", + r"SOFTWARE\Clients\StartMenuInternet\OperaStable\shell\open\command", + r"SOFTWARE\Clients\StartMenuInternet\OperaGXStable\shell\open\command", + ] + for key_path in candidate_keys: + for hive in [winreg.HKEY_LOCAL_MACHINE, winreg.HKEY_CURRENT_USER]: + try: + key = winreg.OpenKey(hive, key_path) + val = winreg.QueryValue(key, None) + winreg.CloseKey(key) + exe = val.strip().strip('"').split('"')[0].split(" --")[0].strip() + if exe and Path(exe).exists(): + print(f"[Browser] 🔍 Opera found via registry: {exe}") + return exe + except Exception: + continue + except Exception: + pass + return None + + +def _find_browser_executable(prog_id: str) -> tuple: + """ + Returns (engine_name, exe_path, channel, is_opera). + is_opera=True → extra args needed to prevent private-mode launch. + """ + system = platform.system() + os_bins = _BROWSER_BINARIES.get(system, {}) + + if any(x in prog_id for x in ["firefox", "mozilla"]): + return "firefox", None, None, False + + if "safari" in prog_id: + return "webkit", None, None, False + + if "edge" in prog_id: + return "chromium", None, "msedge", False + + if "opera" in prog_id: + exe = _get_opera_executable() + if exe: + return "chromium", exe, None, True + for binary in os_bins.get("opera", []): + path = shutil.which(binary) + if path: + return "chromium", path, None, True + + browser_patterns = { + "brave": ["brave"], + "vivaldi": ["vivaldi"], + "chrome": ["chrome"], + } + for browser_name, patterns in browser_patterns.items(): + if not any(p in prog_id for p in patterns): + continue + binaries = os_bins.get(browser_name, []) + for binary in binaries: + path = shutil.which(binary) + if path: + print(f"[Browser] 🔍 Found {browser_name} at: {path}") + return "chromium", path, None, False + + if "chrome" in prog_id or not prog_id: + return "chromium", None, "chrome", False + + return "chromium", None, None, False + + +class _BrowserThread: + + def __init__(self): + self._loop = None + self._thread = None + self._ready = threading.Event() + self._playwright = None + self._browser = None + self._context = None + self._page = None + self._engine_name = "chromium" + self._exe_path = None + self._channel = None + self._is_opera = False + + def start(self): + if self._thread and self._thread.is_alive(): + return + self._thread = threading.Thread( + target=self._run_loop, daemon=True, name="BrowserThread" + ) + self._thread.start() + self._ready.wait(timeout=15) + + def _run_loop(self): + self._loop = asyncio.new_event_loop() + asyncio.set_event_loop(self._loop) + self._loop.run_until_complete(self._init()) + self._ready.set() + self._loop.run_forever() + + async def _init(self): + self._playwright = await async_playwright().start() + + def run(self, coro, timeout: int = 30): + if not self._loop: + raise RuntimeError("BrowserThread not started.") + future = asyncio.run_coroutine_threadsafe(coro, self._loop) + return future.result(timeout=timeout) + + # ── Tarayıcı ve sayfa yönetimi ─────────────────────────────────────────── + + async def _launch_browser_if_needed(self): + """ + Tarayıcıyı başlatır. Zaten açıksa hiçbir şey yapmaz. + Her zaman default tarayıcıyı kullanır, özel sekme açmaz. + """ + if self._browser and self._browser.is_connected(): + return + + prog_id = _get_default_browser_id() + self._engine_name, self._exe_path, self._channel, self._is_opera = _find_browser_executable(prog_id) + engine = getattr(self._playwright, self._engine_name) + + # Temel chromium argümanları + chromium_args = ["--start-maximized"] + + if self._is_opera: + # Opera GX bazı sürümlerde varsayılan olarak private modda başlar. + # Aşağıdaki flag'ler bunu engeller. + chromium_args += [ + "--disable-features=OperaPrivacyMode", + "--no-private", + ] + print("[Browser] 🎭 Opera detected — disabling private-mode flags") + + launch_kwargs = {"headless": False} + if self._engine_name == "chromium": + launch_kwargs["args"] = chromium_args + if self._exe_path: + launch_kwargs["executable_path"] = self._exe_path + elif self._channel: + launch_kwargs["channel"] = self._channel + + try: + self._browser = await engine.launch(**launch_kwargs) + print( + f"[Browser] ✅ Launched ({self._engine_name}" + f"{' / ' + self._channel if self._channel else ''}" + f"{' / ' + self._exe_path if self._exe_path else ''})" + ) + except Exception as e: + print(f"[Browser] ⚠️ Launch failed ({e}), falling back to built-in Chromium") + self._browser = await self._playwright.chromium.launch( + headless=False, + args=["--start-maximized"] + ) + + async def _get_page(self): + """ + Mevcut sayfayı döndürür. + - Tarayıcı kapalıysa açar. + - Context yoksa oluşturur. + - Sayfa kapalıysa yeni sekme açar (aynı pencerede). + - Sayfa zaten açıksa aynı sayfayı döndürür (yeni pencere açmaz). + """ + await self._launch_browser_if_needed() + + if self._context is None: + self._context = await self._browser.new_context( + viewport=None, + user_agent=( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/120.0.0.0 Safari/537.36" + ) + ) + + if self._page is None or self._page.is_closed(): + self._page = await self._context.new_page() + + return self._page + + # ── Eylemler ───────────────────────────────────────────────────────────── + + async def _go_to(self, url: str) -> str: + if not url.startswith("http"): + url = "https://" + url + page = await self._get_page() + try: + await page.goto(url, wait_until="domcontentloaded", timeout=15000) + return f"Opened: {page.url}" + except PlaywrightTimeout: + return f"Timeout loading: {url}" + except Exception as e: + return f"Navigation error: {e}" + + async def _search(self, query: str, engine: str = "google") -> str: + engines = { + "google": f"https://www.google.com/search?q={query.replace(' ', '+')}", + "bing": f"https://www.bing.com/search?q={query.replace(' ', '+')}", + "duckduckgo": f"https://duckduckgo.com/?q={query.replace(' ', '+')}", + } + url = engines.get(engine.lower(), engines["google"]) + return await self._go_to(url) + + async def _click(self, selector=None, text=None) -> str: + page = await self._get_page() + try: + if text: + await page.get_by_text(text, exact=False).first.click(timeout=8000) + return f"Clicked: '{text}'" + elif selector: + await page.click(selector, timeout=8000) + return f"Clicked: {selector}" + return "No selector or text provided." + except PlaywrightTimeout: + return "Element not found or not clickable." + except Exception as e: + return f"Click error: {e}" + + async def _type(self, selector=None, text: str = "", clear_first: bool = True) -> str: + page = await self._get_page() + try: + element = page.locator(selector).first if selector else page.locator(":focus") + if clear_first: + await element.clear() + await element.type(text, delay=50) + return "Text typed." + except Exception as e: + return f"Type error: {e}" + + async def _scroll(self, direction: str = "down", amount: int = 500) -> str: + page = await self._get_page() + try: + y = amount if direction == "down" else -amount + await page.mouse.wheel(0, y) + return f"Scrolled {direction}." + except Exception as e: + return f"Scroll error: {e}" + + async def _press(self, key: str) -> str: + page = await self._get_page() + try: + await page.keyboard.press(key) + return f"Pressed: {key}" + except Exception as e: + return f"Key error: {e}" + + async def _get_text(self) -> str: + page = await self._get_page() + try: + text = await page.inner_text("body") + return text[:4000] if len(text) > 4000 else text + except Exception as e: + return f"Could not get page text: {e}" + + async def _fill_form(self, fields: dict) -> str: + page = await self._get_page() + results = [] + for selector, value in fields.items(): + try: + el = page.locator(selector).first + await el.clear() + await el.type(str(value), delay=40) + results.append(f"✓ {selector}") + except Exception as e: + results.append(f"✗ {selector}: {e}") + return "Form filled: " + ", ".join(results) + + async def _smart_click(self, description: str) -> str: + page = await self._get_page() + desc_lower = description.lower() + + role_hints = { + "button": ["button", "buton", "btn"], + "link": ["link", "bağlantı"], + "searchbox": ["search", "arama"], + "textbox": ["input", "field", "alan"], + } + for role, keywords in role_hints.items(): + if any(k in desc_lower for k in keywords): + try: + await page.get_by_role(role).first.click(timeout=5000) + return f"Clicked ({role}): '{description}'" + except Exception: + pass + + try: + await page.get_by_text(description, exact=False).first.click(timeout=5000) + return f"Clicked (text): '{description}'" + except Exception: + pass + + try: + await page.get_by_placeholder(description, exact=False).first.click(timeout=5000) + return f"Clicked (placeholder): '{description}'" + except Exception: + pass + + return f"Could not find: '{description}'" + + async def _smart_type(self, description: str, text: str) -> str: + page = await self._get_page() + + for method, locator in [ + ("placeholder", page.get_by_placeholder(description, exact=False)), + ("label", page.get_by_label(description, exact=False)), + ("role", page.get_by_role("textbox")), + ]: + try: + el = locator.first + await el.clear() + await el.type(text, delay=50) + return f"Typed into ({method}): '{description}'" + except Exception: + continue + + return f"Could not find input: '{description}'" + + async def _close_browser(self) -> str: + if self._browser: + await self._browser.close() + self._browser = None + self._context = None + self._page = None + + if self._playwright: + await self._playwright.stop() + self._playwright = None + + return "Browser closed." + + +# ── Singleton browser thread ───────────────────────────────────────────────── + +_bt = _BrowserThread() +_bt_started = False +_bt_lock = threading.Lock() + + +def _ensure_started(): + global _bt_started + with _bt_lock: + if not _bt_started: + _bt.start() + _bt_started = True + + +# ── Public API ─────────────────────────────────────────────────────────────── + +def browser_control( + parameters: dict, + response=None, + player=None, + session_memory=None +) -> str: + """ + Browser controller — auto-detects and uses system default browser. + Always reuses the existing browser window/page; never opens incognito. + + parameters: + action : go_to | search | click | type | scroll | fill_form | + smart_click | smart_type | get_text | press | close + url : URL for go_to + query : search query + engine : google | bing | duckduckgo (default: google) + selector : CSS selector for click/type + text : text to click or type + description : element description for smart_click/smart_type + direction : up | down for scroll + amount : scroll amount in pixels (default: 500) + key : key name for press (e.g. Enter, Escape, Tab) + fields : {selector: value} dict for fill_form + clear_first : bool, clear input before typing (default: True) + """ + _ensure_started() + + action = (parameters or {}).get("action", "").lower().strip() + result = "Unknown action." + + try: + if action == "go_to": + result = _bt.run(_bt._go_to(parameters.get("url", ""))) + + elif action == "search": + result = _bt.run(_bt._search( + parameters.get("query", ""), + parameters.get("engine", "google"), + )) + + elif action == "click": + result = _bt.run(_bt._click( + selector=parameters.get("selector"), + text=parameters.get("text"), + )) + + elif action == "type": + result = _bt.run(_bt._type( + selector=parameters.get("selector"), + text=parameters.get("text", ""), + clear_first=parameters.get("clear_first", True), + )) + + elif action == "scroll": + result = _bt.run(_bt._scroll( + direction=parameters.get("direction", "down"), + amount=parameters.get("amount", 500), + )) + + elif action == "fill_form": + result = _bt.run(_bt._fill_form(parameters.get("fields", {}))) + + elif action == "smart_click": + result = _bt.run(_bt._smart_click(parameters.get("description", ""))) + + elif action == "smart_type": + result = _bt.run(_bt._smart_type( + parameters.get("description", ""), + parameters.get("text", ""), + )) + + elif action == "get_text": + result = _bt.run(_bt._get_text()) + + elif action == "press": + result = _bt.run(_bt._press(parameters.get("key", "Enter"))) + + elif action == "close": + result = _bt.run(_bt._close_browser()) + + else: + result = f"Unknown action: {action}" + + except concurrent.futures.TimeoutError: + result = "Browser action timed out." + except Exception as e: + result = f"Browser error: {e}" + + print(f"[Browser] {result[:80]}") + if player: + player.write_log(f"[browser] {result[:60]}") + + return result \ No newline at end of file diff --git a/computer_control.py b/computer_control.py new file mode 100644 index 0000000..7247aa1 --- /dev/null +++ b/computer_control.py @@ -0,0 +1,583 @@ +# actions/code_helper.py +# AI-powered code assistant — writes, edits, explains, runs, builds, debugs, and optimizes code. +# +# Actions: +# write → Describe what you want, Gemini writes it, saves to file +# edit → Read existing file, apply natural language change +# explain → Explain what a piece of code or file does +# run → Execute a script file, return output +# build → Write → Run → Fix loop (max 3 attempts), speaks when done +# screen_debug → Screenshot al, ekrandaki kodu/hatayı Gemini ile analiz et ve düzelt +# optimize → Mevcut kodu Gemini ile optimize et (performans, okunabilirlik, best practices) +# auto → (default) Intent auto-detected from context + +import subprocess +import sys +import json +import re +import time +from pathlib import Path + + +def get_base_dir(): + if getattr(sys, "frozen", False): + return Path(sys.executable).parent + return Path(__file__).resolve().parent.parent + +BASE_DIR = get_base_dir() +API_CONFIG_PATH = BASE_DIR / "config" / "api_keys.json" +DESKTOP = Path.home() / "Desktop" +MAX_BUILD_ATTEMPTS = 3 +GEMINI_MODEL = "gemini-2.5-flash" + + +def _get_api_key() -> str: + with open(API_CONFIG_PATH, "r", encoding="utf-8") as f: + return json.load(f)["gemini_api_key"] + + +def _get_gemini(model: str = GEMINI_MODEL): + import google.generativeai as genai + genai.configure(api_key=_get_api_key()) + return genai.GenerativeModel(model) + + +def _clean_code(text: str) -> str: + text = text.strip() + text = re.sub(r"^```[a-zA-Z]*\n?", "", text) + text = re.sub(r"\n?```$", "", text) + return text.strip() + + +def _resolve_save_path(output_path: str, language: str) -> Path: + ext_map = { + "python": ".py", "py": ".py", + "javascript": ".js", "js": ".js", + "typescript": ".ts", "ts": ".ts", + "html": ".html", "css": ".css", + "java": ".java", "cpp": ".cpp", "c": ".c", + "bash": ".sh", "shell": ".sh", "powershell": ".ps1", + "sql": ".sql", "json": ".json", "rust": ".rs", "go": ".go", + } + if output_path: + p = Path(output_path) + return p if p.is_absolute() else DESKTOP / p + ext = ext_map.get((language or "python").lower(), ".py") + return DESKTOP / f"jarvis_code{ext}" + + +def _read_file(file_path: str) -> tuple[str, str]: + if not file_path: + return "", "No file path provided." + p = Path(file_path) + if not p.exists(): + return "", f"File not found: {file_path}" + try: + return p.read_text(encoding="utf-8"), "" + except Exception as e: + return "", f"Could not read file: {e}" + + +def _save_file(path: Path, content: str) -> str: + try: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(content, encoding="utf-8") + return f"Saved to: {path}" + except Exception as e: + return f"Could not save: {e}" + + +def _preview(code: str, lines: int = 10) -> str: + all_lines = code.splitlines() + preview = "\n".join(all_lines[:lines]) + suffix = f"\n... ({len(all_lines) - lines} more lines)" if len(all_lines) > lines else "" + return preview + suffix + + +def _has_error(output: str) -> bool: + error_signals = ["error", "exception", "traceback", "syntaxerror", + "nameerror", "typeerror", "stderr", "failed", "crash"] + return any(s in output.lower() for s in error_signals) + + +def _take_screenshot() -> Path | None: + try: + import pyautogui + screenshot_path = Path.home() / "Desktop" / f"jarvis_debug_{int(time.time())}.png" + screenshot = pyautogui.screenshot() + screenshot.save(str(screenshot_path)) + print(f"[Code] 📸 Screenshot: {screenshot_path}") + return screenshot_path + except Exception as e: + print(f"[Code] ⚠️ Screenshot failed: {e}") + return None + + +def _image_to_base64(path: Path) -> str: + import base64 + return base64.b64encode(path.read_bytes()).decode("utf-8") + + +def _detect_intent(description: str, file_path: str, code: str) -> str: + desc = (description or "").lower() + + screen_kw = ["ekrandaki", "screen", "ekranda", "bu hatayı", "why am i getting", + "neden hata", "what's wrong", "ne yanlış", "screenshot", "görüntü"] + if any(k in desc for k in screen_kw): + return "screen_debug" + + # Optimize keywords + optimize_kw = ["optimize", "refactor", "clean up", "improve", "temizle", + "iyileştir", "daha iyi", "make it better", "hızlandır"] + if any(k in desc for k in optimize_kw) and (code or file_path): + return "optimize" + + if file_path: + p = Path(file_path) + edit_kw = ["edit", "update", "modify", "change", "add", "remove", + "refactor", "fix", "rename", "replace", "düzenle", "değiştir"] + run_kw = ["run", "execute", "launch", "start", "çalıştır"] + build_kw = ["build", "make it work", "try", "attempt"] + + if p.exists() and any(k in desc for k in edit_kw): + return "edit" + if p.exists() and any(k in desc for k in run_kw): + return "run" + if any(k in desc for k in build_kw): + return "build" + if p.exists(): + return "explain" + + explain_kw = ["explain", "what does", "describe", "analyze", "açıkla", "ne yapıyor"] + if any(k in desc for k in explain_kw) and (code or file_path): + return "explain" + + build_kw = ["build", "make it work", "try and", "attempt"] + if any(k in desc for k in build_kw): + return "build" + + return "write" + +def _write(description: str, language: str, output_path: str, player=None) -> tuple[str, Path]: + lang = language or "python" + model = _get_gemini() + + prompt = f"""You are an expert {lang} developer. +Write clean, working, well-commented {lang} code for the description below. + +Rules: +- Output ONLY the code. No explanation, no markdown, no backticks. +- Add helpful inline comments. +- Handle errors and edge cases properly. +- Use modern best practices. + +Description: {description} + +Code:""" + + response = model.generate_content(prompt) + code = _clean_code(response.text) + path = _resolve_save_path(output_path, lang) + _save_file(path, code) + return code, path + + +def _fix_code(code: str, error_output: str, description: str) -> str: + model = _get_gemini() + prompt = f"""You are an expert debugger. +The code below failed with the following error. Fix it. +Return ONLY the corrected code — no explanation, no markdown, no backticks. + +Original goal: {description} + +Error: +{error_output[:2000]} + +Broken code: +{code} + +Fixed code:""" + + response = model.generate_content(prompt) + return _clean_code(response.text) + + +def _run_file(path: Path, args: list, timeout: int) -> str: + interpreters = { + ".py": [sys.executable], + ".js": ["node"], + ".ts": ["ts-node"], + ".sh": ["bash"], + ".ps1": ["powershell", "-File"], + ".rb": ["ruby"], + ".php": ["php"], + } + interp = interpreters.get(path.suffix.lower()) + if not interp: + return f"No interpreter for {path.suffix}." + + try: + result = subprocess.run( + interp + [str(path)] + (args or []), + capture_output=True, text=True, + encoding="utf-8", errors="replace", + timeout=timeout, cwd=str(path.parent) + ) + output = result.stdout.strip() + error = result.stderr.strip() + parts = [] + if output: parts.append(f"Output:\n{output}") + if error: parts.append(f"Stderr:\n{error}") + return "\n\n".join(parts) if parts else "Executed with no output." + + except subprocess.TimeoutExpired: + return f"Timed out after {timeout}s." + except FileNotFoundError: + return f"Interpreter not found: {interp[0]}." + except Exception as e: + return f"Execution error: {e}" + + +def _build(description, language, output_path, args, timeout, speak=None, player=None) -> str: + if not description: + return "Please describe what you want me to build, sir." + + if player: + player.write_log("[Code] Build started...") + + lang = language or "python" + + try: + code, path = _write(description, lang, output_path, player) + print(f"[Code] ✅ Written: {path}") + except Exception as e: + msg = f"Could not write initial code: {e}" + if speak: speak(msg) + return msg + + last_output = "" + for attempt in range(1, MAX_BUILD_ATTEMPTS + 1): + print(f"[Code] 🔄 Attempt {attempt}/{MAX_BUILD_ATTEMPTS}") + if player: + player.write_log(f"[Code] Attempt {attempt}...") + + last_output = _run_file(path, args, timeout) + + if not _has_error(last_output): + msg = ( + f"Build complete, sir. " + f"The code is working after {attempt} attempt{'s' if attempt > 1 else ''}. " + f"Saved to {path}." + ) + if speak: speak(msg) + return f"{msg}\n\nOutput:\n{last_output}" + + print(f"[Code] ⚠️ Error on attempt {attempt}, fixing...") + if player: + player.write_log(f"[Code] Fixing (attempt {attempt})...") + + try: + code = _fix_code(code, last_output, description) + _save_file(path, code) + except Exception as e: + msg = f"Could not fix code on attempt {attempt}: {e}" + if speak: speak(msg) + return msg + + msg = ( + f"I was unable to build a working version after {MAX_BUILD_ATTEMPTS} attempts, sir. " + f"The last error was: {last_output[:200]}" + ) + if speak: speak(msg) + return f"{msg}\n\nLast code saved to: {path}" + +def _write_action(description, language, output_path, player) -> str: + if not description: + return "Please describe what you want me to write, sir." + if player: + player.write_log("[Code] Writing code...") + try: + code, path = _write(description, language, output_path, player) + print(f"[Code] ✅ Written: {path}") + return f"Code written. Saved to: {path}\n\nPreview:\n{_preview(code)}" + except Exception as e: + return f"Could not generate code: {e}" + + +def _edit_action(file_path, instruction, player) -> str: + if not file_path: + return "Please provide a file path to edit, sir." + if not instruction: + return "Please describe what change to make, sir." + + content, err = _read_file(file_path) + if err: + return err + + if player: + player.write_log("[Code] Editing file...") + + model = _get_gemini() + prompt = f"""You are an expert code editor. +Apply the following change to the code below. +Return ONLY the complete updated code — no explanation, no markdown, no backticks. + +Change: {instruction} + +Original code: +{content} + +Updated code:""" + + try: + response = model.generate_content(prompt) + edited = _clean_code(response.text) + except Exception as e: + return f"Could not edit code: {e}" + + status = _save_file(Path(file_path), edited) + print(f"[Code] ✅ Edited: {file_path}") + return f"File edited. {status}\n\nPreview:\n{_preview(edited)}" + + +def _explain_action(file_path, code, player) -> str: + if file_path and not code: + code, err = _read_file(file_path) + if err: + return err + if not code: + return "Please provide code or a file path to explain, sir." + + if player: + player.write_log("[Code] Analyzing code...") + + model = _get_gemini() + prompt = f"""Explain what this code does in simple, clear language. +Focus on: what it does, how it works, and any important details. +Be concise — 3 to 6 sentences maximum. + +Code: +{code[:4000]} + +Explanation:""" + + try: + response = model.generate_content(prompt) + return response.text.strip() + except Exception as e: + return f"Could not explain code: {e}" + + +def _run_action(file_path, args, timeout, player) -> str: + if not file_path: + return "Please provide a file path to run, sir." + p = Path(file_path) + if not p.exists(): + return f"File not found: {file_path}" + if player: + player.write_log(f"[Code] Running {p.name}...") + return _run_file(p, args, timeout) + + +def _optimize_action(file_path, code, language, output_path, player) -> str: + + if file_path and not code: + code, err = _read_file(file_path) + if err: + return err + if not code: + return "Please provide code or a file path to optimize, sir." + + if player: + player.write_log("[Code] Optimizing code...") + + lang = language or "python" + model = _get_gemini() + + prompt = f"""You are an expert {lang} developer and code reviewer. +Optimize the following code for: +1. Performance — eliminate unnecessary operations, use efficient data structures +2. Readability — clear variable names, proper formatting, logical structure +3. Best practices — modern {lang} patterns, error handling, type hints if applicable +4. Remove dead code, redundant comments, and unnecessary complexity + +Return ONLY the optimized code — no explanation, no markdown, no backticks. + +Original code: +{code[:6000]} + +Optimized code:""" + + try: + response = model.generate_content(prompt) + optimized = _clean_code(response.text) + except Exception as e: + return f"Could not optimize code: {e}" + + # Kaydet + if file_path: + save_path = Path(file_path) + else: + save_path = _resolve_save_path(output_path, lang) + + status = _save_file(save_path, optimized) + print(f"[Code] ✅ Optimized: {save_path}") + + original_lines = len(code.splitlines()) + optimized_lines = len(optimized.splitlines()) + diff = original_lines - optimized_lines + + return ( + f"Code optimized. {status}\n" + f"Lines: {original_lines} → {optimized_lines} " + f"({'−' if diff > 0 else '+'}{abs(diff)} lines)\n\n" + f"Preview:\n{_preview(optimized)}" + ) + + +def _screen_debug_action(description, file_path, player, speak=None) -> str: + + if player: + player.write_log("[Code] Taking screenshot for analysis...") + + print("[Code] 📸 Capturing screen for debug...") + + + screenshot_path = _take_screenshot() + if not screenshot_path: + return "Could not take screenshot, sir. Please make sure PyAutoGUI is installed." + + + file_content = "" + if file_path: + file_content, err = _read_file(file_path) + if err: + print(f"[Code] ⚠️ Could not read file: {err}") + + try: + from google import genai + from google.genai import types + + client = genai.Client(api_key=_get_api_key()) + + image_bytes = screenshot_path.read_bytes() + image_base64 = _image_to_base64(screenshot_path) + + user_question = description or "What error or problem do you see on the screen? How can it be fixed?" + + context = "" + if file_content: + context = f"\n\nAdditionally, here is the related file content:\n```\n{file_content[:4000]}\n```" + + analysis_prompt = f"""You are an expert programmer and debugger analyzing a screenshot. + +User's question: {user_question}{context} + +Please: +1. Identify any errors, exceptions, or problems visible on the screen +2. Explain what is causing the problem in simple terms +3. Provide a concrete fix or solution +4. If there's code visible, show the corrected version + +Be specific and actionable. If you see an error message, quote it exactly.""" + + contents = [ + types.Part.from_bytes(data=image_bytes, mime_type="image/png"), + analysis_prompt, + ] + + response = client.models.generate_content( + model="gemini-2.5-flash", + contents=contents, + ) + + analysis = response.text.strip() + print(f"[Code] ✅ Screen analysis complete") + + try: + screenshot_path.unlink() + except Exception: + pass + + if file_path and file_content: + + code_match = re.search(r"```[a-zA-Z]*\n(.*?)```", analysis, re.DOTALL) + if code_match: + fixed_code = code_match.group(1).strip() + save_path = Path(file_path) + _save_file(save_path, fixed_code) + analysis += f"\n\n✅ Fixed code has been saved to: {file_path}" + print(f"[Code] ✅ Fixed code saved: {file_path}") + + return analysis + + except Exception as e: + + try: + screenshot_path.unlink() + except Exception: + pass + return f"Screen analysis failed: {e}" + + +def code_helper( + parameters: dict, + response=None, + player=None, + session_memory=None, + speak=None +) -> str: + """ + Called from main.py. + + parameters: + action : write | edit | explain | run | build | screen_debug | optimize | auto + description : What the code should do / what change to make / what problem to analyze + language : Programming language (default: python) + output_path : Where to save — user specifies full path or filename + file_path : Path to existing file (edit / explain / run / build / optimize) + code : Raw code string (explain/optimize without a file) + args : CLI argument list for run/build + timeout : Execution timeout in seconds (default: 30) + """ + p = parameters or {} + action = p.get("action", "auto").lower().strip() + description = p.get("description", "").strip() + language = p.get("language", "python").strip() + output_path = p.get("output_path", "").strip() + file_path = p.get("file_path", "").strip() + code = p.get("code", "").strip() + args = p.get("args", []) + timeout = int(p.get("timeout", 30)) + + if action == "auto": + action = _detect_intent(description, file_path, code) + print(f"[Code] 🤖 Auto-detected: {action}") + + if action == "write": + return _write_action(description, language, output_path, player) + + elif action == "edit": + return _edit_action( + file_path, + description or p.get("instruction", ""), + player + ) + + elif action == "explain": + return _explain_action(file_path, code, player) + + elif action == "run": + return _run_action(file_path, args, timeout, player) + + elif action == "build": + return _build(description, language, output_path, args, timeout, speak, player) + + elif action == "optimize": + return _optimize_action(file_path, code, language, output_path, player) + + elif action == "screen_debug": + return _screen_debug_action(description, file_path, player, speak) + + else: + return f"Unknown action: '{action}'. Use write, edit, explain, run, build, optimize, or screen_debug." \ No newline at end of file diff --git a/computer_settings.py b/computer_settings.py new file mode 100644 index 0000000..c8403c9 --- /dev/null +++ b/computer_settings.py @@ -0,0 +1,478 @@ +#computer_control.py +import io +import json +import re +import string +import subprocess +import sys +import time +import random +from pathlib import Path + +try: + import pyautogui + pyautogui.FAILSAFE = True + pyautogui.PAUSE = 0.05 + _PYAUTOGUI = True +except ImportError: + _PYAUTOGUI = False + +try: + import pyperclip + _PYPERCLIP = True +except ImportError: + _PYPERCLIP = False + +def _base_dir() -> Path: + if getattr(sys, "frozen", False): + return Path(sys.executable).parent + return Path(__file__).resolve().parent.parent + + +_BASE = _base_dir() +_CONFIG_PATH = _BASE / "config" / "api_keys.json" +_MEMORY_PATH = _BASE / "memory" / "long_term.json" + +def _load_config() -> dict: + try: + return json.loads(_CONFIG_PATH.read_text(encoding="utf-8")) + except Exception: + return {} + +def _get_os() -> str: + return _load_config().get("os_system", "windows").lower() + +_SAFE_SCREENSHOT_ROOTS = ( + Path.home(), +) + +def _safe_screenshot_path(requested: str | None) -> Path: + fallback = Path.home() / "Desktop" / "jarvis_screenshot.png" + if not requested: + return fallback + try: + p = Path(requested).expanduser().resolve() + for root in _SAFE_SCREENSHOT_ROOTS: + if p.is_relative_to(root.resolve()): + p.parent.mkdir(parents=True, exist_ok=True) + return p + except Exception: + pass + return fallback + +def _require_pyautogui(): + if not _PYAUTOGUI: + raise RuntimeError("PyAutoGUI not installed. Run: pip install pyautogui") + +_FIRST_NAMES = [ + "Alex", "Jordan", "Taylor", "Morgan", "Casey", "Riley", "Drew", "Quinn", + "Avery", "Blake", "Cameron", "Dakota", "Emerson", "Finley", "Harper", +] +_LAST_NAMES = [ + "Smith", "Johnson", "Williams", "Brown", "Jones", "Garcia", "Miller", + "Davis", "Wilson", "Moore", "Taylor", "Anderson", "Thomas", "Jackson", +] +_DOMAINS = ["gmail.com", "yahoo.com", "outlook.com", "proton.me", "mail.com"] + + +def _random_data(data_type: str) -> str: + dt = data_type.lower().strip() + + if dt == "first_name": + return random.choice(_FIRST_NAMES) + + if dt == "last_name": + return random.choice(_LAST_NAMES) + + if dt == "name": + return f"{random.choice(_FIRST_NAMES)} {random.choice(_LAST_NAMES)}" + + if dt == "email": + first = random.choice(_FIRST_NAMES).lower() + last = random.choice(_LAST_NAMES).lower() + num = random.randint(10, 999) + return f"{first}.{last}{num}@{random.choice(_DOMAINS)}" + + if dt == "username": + return f"{random.choice(_FIRST_NAMES).lower()}{random.randint(100, 9999)}" + + if dt == "password": + chars = string.ascii_letters + string.digits + "!@#$%" + raw = ( + random.choice(string.ascii_uppercase) + + random.choice(string.digits) + + random.choice("!@#$%") + + "".join(random.choices(chars, k=9)) + ) + return "".join(random.sample(raw, len(raw))) + + if dt == "phone": + return f"+1{random.randint(200,999)}{random.randint(1_000_000, 9_999_999)}" + + if dt == "birthday": + y = random.randint(1980, 2000) + m = random.randint(1, 12) + d = random.randint(1, 28) + return f"{m:02d}/{d:02d}/{y}" + + if dt == "address": + num = random.randint(100, 9999) + street = random.choice(["Main St", "Oak Ave", "Park Blvd", "Elm St", "Cedar Ln"]) + return f"{num} {street}" + + if dt == "zip_code": + return str(random.randint(10000, 99999)) + + if dt == "city": + return random.choice(["New York", "Los Angeles", "Chicago", "Houston", "Phoenix"]) + + return f"random_{data_type}_{random.randint(1000, 9999)}" + +def _user_profile() -> dict: + """Read identity fields from long-term memory.""" + try: + if _MEMORY_PATH.exists(): + data = json.loads(_MEMORY_PATH.read_text(encoding="utf-8")) + identity = data.get("identity", {}) + return {k: v.get("value", "") for k, v in identity.items()} + except Exception: + pass + return {} + +def _type(text: str, interval: float = 0.03) -> str: + _require_pyautogui() + time.sleep(0.3) + pyautogui.typewrite(text, interval=interval) + return f"Typed: {text[:60]}{'…' if len(text) > 60 else ''}" + + +def _smart_type(text: str, clear_first: bool = True) -> str: + _require_pyautogui() + if clear_first: + _clear_field() + time.sleep(0.1) + + if len(text) > 20 and _PYPERCLIP: + pyperclip.copy(text) + time.sleep(0.1) + pyautogui.hotkey("ctrl", "v") + return f"Smart-typed (clipboard): {text[:60]}{'…' if len(text) > 60 else ''}" + + pyautogui.typewrite(text, interval=0.04) + return f"Smart-typed: {text[:60]}{'…' if len(text) > 60 else ''}" + + +def _click(x=None, y=None, button: str = "left", clicks: int = 1) -> str: + _require_pyautogui() + if x is not None and y is not None: + pyautogui.click(x, y, button=button, clicks=clicks) + return f"{'Double-c' if clicks == 2 else 'C'}licked ({x}, {y}) [{button}]" + pyautogui.click(button=button, clicks=clicks) + return f"Clicked at current position [{button}]" + + +def _hotkey(*keys) -> str: + _require_pyautogui() + pyautogui.hotkey(*keys) + return f"Hotkey: {'+'.join(keys)}" + + +def _press(key: str) -> str: + _require_pyautogui() + pyautogui.press(key) + return f"Pressed: {key}" + + +def _scroll(direction: str = "down", amount: int = 3) -> str: + _require_pyautogui() + vertical = direction in ("up", "down") + clicks = amount if direction in ("up", "right") else -amount + pyautogui.scroll(clicks) if vertical else pyautogui.hscroll(clicks) + return f"Scrolled {direction} ×{amount}" + + +def _move(x: int, y: int, duration: float = 0.3) -> str: + _require_pyautogui() + pyautogui.moveTo(x, y, duration=duration) + return f"Mouse → ({x}, {y})" + + +def _drag(x1: int, y1: int, x2: int, y2: int, duration: float = 0.5) -> str: + _require_pyautogui() + pyautogui.moveTo(x1, y1, duration=0.2) + pyautogui.dragTo(x2, y2, duration=duration, button="left") + return f"Dragged ({x1},{y1}) → ({x2},{y2})" + + +def _clipboard_get() -> str: + if _PYPERCLIP: + return pyperclip.paste() + _hotkey("ctrl", "c") + time.sleep(0.2) + return "(copied — pyperclip unavailable for read)" + + +def _clipboard_paste(text: str) -> str: + if _PYPERCLIP: + pyperclip.copy(text) + time.sleep(0.1) + _require_pyautogui() + pyautogui.hotkey("ctrl", "v") + return f"Pasted: {text[:60]}{'…' if len(text) > 60 else ''}" + return "pyperclip not available" + + +def _screenshot(save_path: str | None = None) -> str: + _require_pyautogui() + path = _safe_screenshot_path(save_path) + img = pyautogui.screenshot() + img.save(str(path)) + return f"Screenshot saved: {path}" + + +def _clear_field() -> str: + _require_pyautogui() + pyautogui.hotkey("ctrl", "a") + time.sleep(0.1) + pyautogui.press("delete") + return "Field cleared" + +def _focus_window(title: str) -> str: + os_name = _get_os() + + if os_name == "windows": + try: + script = f'(New-Object -ComObject WScript.Shell).AppActivate("{title}")' + subprocess.run( + ["powershell", "-NoProfile", "-NonInteractive", "-Command", script], + capture_output=True, timeout=5, + ) + time.sleep(0.3) + return f"Focused window: {title}" + except Exception as e: + return f"focus_window (Windows) failed: {e}" + + if os_name == "mac": + script = ( + f'tell application "System Events" to ' + f'set frontmost of (first process whose name contains "{title}") to true' + ) + try: + subprocess.run( + ["osascript", "-e", script], + capture_output=True, timeout=5, + ) + time.sleep(0.3) + return f"Focused window: {title}" + except Exception as e: + return f"focus_window (macOS) failed: {e}" + + if os_name == "linux": + try: + result = subprocess.run( + ["wmctrl", "-a", title], + capture_output=True, timeout=5, + ) + if result.returncode == 0: + time.sleep(0.3) + return f"Focused window: {title}" + except FileNotFoundError: + pass + try: + result = subprocess.run( + ["xdotool", "search", "--name", title, "windowactivate"], + capture_output=True, timeout=5, + ) + time.sleep(0.3) + return f"Focused window: {title}" + except FileNotFoundError: + return "focus_window (Linux) requires wmctrl or xdotool" + except Exception as e: + return f"focus_window (Linux) failed: {e}" + + return f"focus_window: unknown OS '{os_name}'" +def _screen_find(description: str) -> tuple[int, int] | None: + try: + import base64 + from or_client import client + + _require_pyautogui() + w, h = pyautogui.size() + img = pyautogui.screenshot() + buf = io.BytesIO() + img.save(buf, format="PNG") + b64 = base64.b64encode(buf.getvalue()).decode() + + text = client.vision( + f"This is a screenshot of a {w}×{h} pixel screen. " + f"Locate the UI element: '{description}'. " + f"Reply ONLY with center coordinates as: x,y — or NOT_FOUND", + image_b64=b64, + mime="image/png", + ) + + if "NOT_FOUND" in text.upper(): + return None + match = re.search(r"(\d+)\s*,\s*(\d+)", text) + if match: + return int(match.group(1)), int(match.group(2)) + except Exception as e: + print(f"[ComputerControl] ⚠️ screen_find failed: {e}") + return None + +def computer_control( + parameters: dict, + response=None, + player=None, + session_memory=None, +) -> str: + """ + Dispatch table for all computer control actions. + + parameters keys (all optional unless noted): + action : (required) one of the actions listed below + text : text to type or paste + x, y : screen coordinates + button : 'left' | 'right' (default: left) + keys : hotkey string, e.g. 'ctrl+c' + key : single key name, e.g. 'enter' + direction : 'up' | 'down' | 'left' | 'right' + amount : scroll amount (default: 3) + seconds : wait duration + title : window title fragment for focus_window + description : natural-language element description for screen_find/click + type : data type for random_data + field : memory field name for user_data + clear_first : bool, clear field before typing (default: true) + path : save path for screenshot (must be inside home dir) + + Actions: + type — type text at cursor + smart_type — clear field + type (clipboard-backed) + click — left click + double_click — double left click + right_click — right click + move — move mouse + drag — click-drag between two points + hotkey — key combination + press — single key + scroll — scroll the wheel + copy — read clipboard + paste — write + paste clipboard + screenshot — capture screen (safe path only) + wait — sleep N seconds + clear_field — select-all + delete + focus_window — bring window to foreground + screen_find — AI element finder (returns x,y) + screen_click — AI element finder + click + random_data — generate fake form data + user_data — pull real data from memory + """ + params = parameters or {} + action = params.get("action", "").lower().strip() + + if not action: + return "No action specified for computer_control." + + if player: + player.write_log(f"[Computer] {action}") + + print(f"[ComputerControl] ▶ {action} {params}") + + try: + + if action == "type": + return _type(params.get("text", "")) + + if action == "smart_type": + return _smart_type( + params.get("text", ""), + clear_first=params.get("clear_first", True), + ) + + if action in ("click", "left_click"): + return _click(params.get("x"), params.get("y"), "left", 1) + + if action == "double_click": + return _click(params.get("x"), params.get("y"), "left", 2) + + if action == "right_click": + return _click(params.get("x"), params.get("y"), "right", 1) + + if action == "move": + return _move(int(params.get("x", 0)), int(params.get("y", 0))) + + if action == "drag": + return _drag( + int(params.get("x1", 0)), int(params.get("y1", 0)), + int(params.get("x2", 0)), int(params.get("y2", 0)), + ) + + if action == "hotkey": + raw = params.get("keys", "") + keys = [k.strip() for k in raw.split("+")] if isinstance(raw, str) else raw + return _hotkey(*keys) + + if action == "press": + return _press(params.get("key", "enter")) + + if action == "scroll": + return _scroll( + direction=params.get("direction", "down"), + amount=int(params.get("amount", 3)), + ) + + if action == "copy": + return _clipboard_get() + + if action == "paste": + return _clipboard_paste(params.get("text", "")) + + if action == "screenshot": + return _screenshot(params.get("path")) + + if action == "screen_find": + coords = _screen_find(params.get("description", "")) + return f"{coords[0]},{coords[1]}" if coords else "NOT_FOUND" + + if action == "screen_click": + desc = params.get("description", "") + coords = _screen_find(desc) + if coords: + time.sleep(0.2) + _click(x=coords[0], y=coords[1]) + return f"Clicked '{desc}' at {coords}" + return f"Element not found on screen: '{desc}'" + + if action == "wait": + secs = float(params.get("seconds", 1.0)) + secs = min(secs, 30.0) + time.sleep(secs) + return f"Waited {secs}s" + + if action == "clear_field": + return _clear_field() + + if action == "focus_window": + return _focus_window(params.get("title", "")) + + if action == "random_data": + dt = params.get("type", "name") + result = _random_data(dt) + print(f"[ComputerControl] 🎲 random {dt} → {result}") + return result + + if action == "user_data": + field = params.get("field", "name") + profile = _user_profile() + value = profile.get(field, "") + if not value: + value = _random_data(field) + print(f"[ComputerControl] ⚠️ No '{field}' in memory, using random: {value}") + return value + + return f"Unknown action: '{action}'" + + except Exception as e: + print(f"[ComputerControl] ❌ {action}: {e}") + return f"computer_control '{action}' failed: {e}" \ No newline at end of file