♻️ refactor(clipboard, system): improve system plugin query performance and clipboard handling, invalidate cache on clipboard change, add command parsing to system plugin query

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 "<meta http-equiv" not in line]
+            if stdout_str.strip():
+                lines = stdout_str.strip().split("\n")
+                items = [line for line in lines if line and "<meta http-equiv" not in line]
+            else:
+                items = []
 
-            # Update cache
+            # Check if data has changed
             with self.cache_lock:
+                data_changed = (
+                    not self.clipboard_items_cache or
+                    len(items) != len(self.clipboard_items_cache) or
+                    items != self.clipboard_items_cache
+                )
+
+                # If cache is still valid and data hasn't changed, return cached data
+                if (not data_changed and
+                    self.clipboard_items_cache and
+                    current_time - self.cache_timestamp < self.cache_ttl):
+                    return self.clipboard_items_cache.copy()
+
+                # Update cache with fresh data
                 self.clipboard_items_cache = items
                 self.cache_timestamp = current_time
 
+                # If data changed significantly, clear image cache too
+                if data_changed:
+                    self.image_cache.clear()
+                    self.image_cache_timestamps.clear()
+
             return items
         except subprocess.CalledProcessError as e:
             print(f"Error loading clipboard history: {e}", file=sys.stderr)
@@ -300,6 +365,10 @@ class ClipboardPlugin(PluginBase):
                     check=True,
                     timeout=3
                 )
+
+            # Invalidate cache since clipboard content has changed
+            self.invalidate_cache()
+
         except subprocess.SubprocessError as e:
             print(f"Error copying to clipboard: {e}", file=sys.stderr)
         except subprocess.TimeoutExpired as e:

modules/launcher/plugins/system.py

@@ -5,7 +5,8 @@ Provides system commands and actions.
 
 import os
 import time
-from typing import List
+import shlex
+from typing import List, Union
 
 from fabric.utils import exec_shell_command_async
 import utils.icons as icons
@@ -55,27 +56,49 @@ class SystemPlugin(PluginBase):
 
     def query(self, query_string: str) -> 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()
+