From 519f273d2012784c5606d8ddbe34d1965679972b Mon Sep 17 00:00:00 2001 From: S4NKALP Date: Sun, 22 Jun 2025 14:41:14 +0545 Subject: [PATCH] =?UTF-8?q?=E2=99=BB=EF=B8=8F=20refactor(clipboard,=20syst?= =?UTF-8?q?em):=20improve=20system=20plugin=20query=20performance=20and=20?= =?UTF-8?q?clipboard=20handling,=20invalidate=20cache=20on=20clipboard=20c?= =?UTF-8?q?hange,=20add=20command=20parsing=20to=20system=20plugin=20query?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- modules/launcher/plugins/clipboard.py | 95 +++++++++++++++++++++++---- modules/launcher/plugins/system.py | 61 ++++++++++++++--- 2 files changed, 133 insertions(+), 23 deletions(-) diff --git a/modules/launcher/plugins/clipboard.py b/modules/launcher/plugins/clipboard.py index f1c3858..36fed0a 100644 --- a/modules/launcher/plugins/clipboard.py +++ b/modules/launcher/plugins/clipboard.py @@ -10,7 +10,7 @@ import tempfile import threading import time from typing import List, Dict, Optional -from concurrent.futures import ThreadPoolExecutor, as_completed +from concurrent.futures import ThreadPoolExecutor from modules.launcher.plugin_base import PluginBase from modules.launcher.result import Result from gi.repository import GdkPixbuf, GLib @@ -25,7 +25,7 @@ class ClipboardPlugin(PluginBase): # Performance settings self.max_results = 20 - self.cache_ttl = 30 # Cache clipboard items for 30 seconds + self.cache_ttl = 5 # Cache clipboard items for 5 seconds (more responsive) self.image_cache_ttl = 300 # Cache images for 5 minutes # Initialize cache and temp directory @@ -74,30 +74,95 @@ class ClipboardPlugin(PluginBase): except Exception as e: print(f"Error cleaning up temporary files: {e}", file=sys.stderr) + def invalidate_cache(self): + """Force invalidation of the clipboard cache.""" + with self.cache_lock: + self.clipboard_items_cache.clear() + self.cache_timestamp = 0 + self.image_cache.clear() + self.image_cache_timestamps.clear() + + def _force_launcher_refresh(self): + """Force the launcher to refresh its results.""" + try: + from gi.repository import GLib + + def trigger_refresh(): + try: + # Try to access the launcher through the fabric Application + from fabric import Application + app = Application.get_default() + + if app and hasattr(app, 'launcher'): + launcher = app.launcher + if launcher and hasattr(launcher, 'search_entry') and hasattr(launcher, '_perform_search'): + # Get current search text to preserve the query + current_text = launcher.search_entry.get_text() + # Trigger the search to refresh results + launcher._perform_search(current_text) + return False + + # Fallback: try to find launcher instance through other means + import gc + for obj in gc.get_objects(): + if (hasattr(obj, '__class__') and obj.__class__.__name__ == 'Launcher'): + if hasattr(obj, 'search_entry') and hasattr(obj, '_perform_search'): + current_text = obj.search_entry.get_text() + obj._perform_search(current_text) + return False + + except Exception as e: + print(f"Error forcing launcher refresh: {e}") + + return False # Don't repeat + + # Use immediate refresh + GLib.timeout_add(10, trigger_refresh) + + except Exception as e: + print(f"Could not trigger refresh: {e}") + + + def _load_clipboard_items_cached(self) -> List[str]: - """Load clipboard items from cliphist with caching.""" + """Load clipboard items from cliphist with caching and change detection.""" current_time = time.time() - # Check if cache is still valid - with self.cache_lock: - if (self.clipboard_items_cache and - current_time - self.cache_timestamp < self.cache_ttl): - return self.clipboard_items_cache.copy() - - # Load fresh data + # Always load fresh data to check for changes try: result = subprocess.run( ["cliphist", "list"], capture_output=True, check=True, timeout=5 ) stdout_str = result.stdout.decode("utf-8", errors="replace") - lines = stdout_str.strip().split("\n") - items = [line for line in lines if line and " List[Result]: """Search for system commands matching the query.""" - query = query_string.lower().strip() + query = query_string.strip() if not query: return [] results = [] + # Parse the query to extract binary name and arguments + query_parts = query.split() + if not query_parts: + return [] + + binary_query = query_parts[0].lower() + full_command = query # Keep the original case and spacing + # Check system binaries self._update_bin_cache() for binary in self._bin_cache: - if query in binary.lower(): - relevance = self._calculate_binary_relevance(binary, query) + if binary_query in binary.lower(): + relevance = self._calculate_binary_relevance(binary, binary_query) if relevance > 0: + # If the query starts with this binary, use the full command + if binary.lower() == binary_query: + # Exact binary match - use full command with arguments + display_command = full_command + command_to_execute = full_command + elif binary.lower().startswith(binary_query): + # Binary starts with query - suggest the binary name only + display_command = binary + command_to_execute = binary + else: + # Partial match - suggest the binary name only + display_command = binary + command_to_execute = binary + result = Result( - title=binary, - subtitle=f"Execute {binary}", + title=display_command, + subtitle=f"Execute: {display_command}", icon_markup=icons.terminal, - action=lambda b=binary: self._execute_command([b]), + action=self._create_action(command_to_execute), relevance=relevance, plugin_name=self.display_name, - data={"command": [binary], "id": binary}, + data={"command": command_to_execute, "id": binary}, ) results.append(result) @@ -99,6 +122,24 @@ class SystemPlugin(PluginBase): return 0.0 - def _execute_command(self, command: List[str]): + def _create_action(self, command: Union[str, List[str]]): + """Create an action function for the given command.""" + def action(): + self._execute_command(command) + return action + + def _execute_command(self, command: Union[str, List[str]]): """Execute a system command.""" - exec_shell_command_async(command) + try: + if isinstance(command, str): + # Handle string commands with arguments - split into list for proper execution + command_list = shlex.split(command) + exec_shell_command_async(command_list) + else: + # Handle list commands (backward compatibility) + exec_shell_command_async(command) + except Exception as e: + print(f"SystemPlugin: Error executing command '{command}': {e}") + import traceback + traceback.print_exc() +