import json import os import threading import time from typing import Dict, List import config.data as data from fabric.utils import exec_shell_command_async from modules.launcher.plugin_base import PluginBase from modules.launcher.result import Result class BashScriptsPlugin(PluginBase): """ Plugin for managing and executing bash scripts. """ def __init__(self): super().__init__() self.display_name = "Bash Scripts" self.description = "Manage and execute bash scripts" # Configuration self.scripts_cache_file = os.path.join(data.CACHE_DIR, "bash_scripts.json") # Default script directory to scan (only Modus scripts) self.modus_scripts_dir = os.path.expanduser("~/.config/Modus/scripts") # Scripts to exclude from discovery self.excluded_scripts = { "screen-capture.sh" # Exclude screen-capture.sh as it's handled by screencapture plugin } # In-memory cache self._scripts_cache: Dict[str, Dict] = {} self._last_cache_update = 0 self._cache_update_interval = 300 # 5 minutes # Background cache building self._cache_building = False self._cache_thread = None def initialize(self): """Initialize the bash scripts plugin.""" self.set_triggers(["sh"]) self._load_scripts_cache() self._start_background_cache_update() def cleanup(self): """Cleanup the bash scripts plugin.""" self._scripts_cache.clear() if self._cache_thread and self._cache_thread.is_alive(): # Note: We don't join the thread to avoid blocking cleanup pass def _load_scripts_cache(self): """Load scripts cache from JSON file.""" try: if os.path.exists(self.scripts_cache_file): with open(self.scripts_cache_file, "r", encoding="utf-8") as f: cache_data = json.load(f) self._scripts_cache = cache_data.get("scripts", {}) self._last_cache_update = cache_data.get("last_update", 0) else: print( "BashScriptsPlugin: No cache file found, will build cache in background" ) except Exception as e: print(f"BashScriptsPlugin: Error loading scripts cache: {e}") self._scripts_cache = {} self._last_cache_update = 0 def _save_scripts_cache(self): """Save scripts cache to JSON file.""" try: os.makedirs(data.CACHE_DIR, exist_ok=True) cache_data = { "scripts": self._scripts_cache, "last_update": self._last_cache_update, } with open(self.scripts_cache_file, "w", encoding="utf-8") as f: json.dump(cache_data, f, indent=2) except Exception as e: print(f"BashScriptsPlugin: Error saving scripts cache: {e}") def _start_background_cache_update(self): """Start background thread to update scripts cache.""" current_time = time.time() # Check if cache needs updating if ( current_time - self._last_cache_update > self._cache_update_interval or not self._scripts_cache ): if not self._cache_building: self._cache_building = True self._cache_thread = threading.Thread( target=self._build_scripts_cache_background, daemon=True ) self._cache_thread.start() def _build_scripts_cache_background(self): """Build scripts cache in background thread.""" try: new_cache = {} # Scan Modus scripts directory for discovered scripts if os.path.exists(self.modus_scripts_dir) and os.path.isdir( self.modus_scripts_dir ): try: self._scan_directory_for_scripts(self.modus_scripts_dir, new_cache) except (PermissionError, FileNotFoundError, OSError) as e: print( f"BashScriptsPlugin: Error scanning Modus scripts directory: { e }" ) # Update cache atomically self._scripts_cache = new_cache self._last_cache_update = time.time() # Save to disk self._save_scripts_cache() except Exception as e: print(f"BashScriptsPlugin: Error building scripts cache: {e}") finally: self._cache_building = False def _scan_directory_for_scripts(self, directory: str, cache: Dict): """Scan a directory for bash scripts and add them to cache.""" try: with os.scandir(directory) as entries: for entry in entries: if entry.is_file(follow_symlinks=False): script_path = entry.path script_name = entry.name # Skip excluded scripts if script_name in self.excluded_scripts: continue # Check if it's a script file if self._is_script_file(script_path): cache[script_name] = { "path": script_path, "name": script_name, "description": self._get_script_description( script_path ), "type": "discovered", "executable": os.access(script_path, os.X_OK), "args": [], "category": os.path.basename(directory), } except (PermissionError, FileNotFoundError, OSError) as e: print(f"BashScriptsPlugin: Error scanning directory {directory}: {e}") except Exception as e: print(f"BashScriptsPlugin: Unexpected error scanning {directory}: {e}") def _is_script_file(self, file_path: str) -> bool: """Check if a file is a bash script.""" try: # Check file extension first (most common case) if file_path.endswith((".sh", ".bash")): return True # For files without extension, check shebang try: with open(file_path, "rb") as f: first_line = f.readline(100).decode("utf-8", errors="ignore") if first_line.startswith("#!") and ( "bash" in first_line or "sh" in first_line ): return True except (PermissionError, FileNotFoundError, UnicodeDecodeError): pass return False except Exception: return False def _get_script_description(self, script_path: str) -> str: """Extract description from script comments.""" try: with open(script_path, "r", encoding="utf-8", errors="ignore") as f: lines = f.readlines() # Look for description in first few comment lines for line in lines[:10]: line = line.strip() if line.startswith("#") and not line.startswith("#!"): # Remove leading # and whitespace desc = line[1:].strip() if desc and len(desc) > 5: # Meaningful description return desc return f"Script: {os.path.basename(script_path)}" except (PermissionError, FileNotFoundError, UnicodeDecodeError): return f"Script: {os.path.basename(script_path)}" def query(self, query_string: str) -> List[Result]: """Search for bash scripts matching the query.""" query = query_string.strip() # Start background update if needed (non-blocking) if not self._scripts_cache or ( time.time() - self._last_cache_update > self._cache_update_interval ): self._start_background_cache_update() results = [] # Handle special commands if not query: # Show all scripts when no query results.extend(self._list_all_scripts()) else: # Search for scripts results.extend(self._search_scripts(query)) return results def _list_all_scripts(self) -> List[Result]: """List all available scripts.""" results = [] max_results = 20 # Sort scripts by name for consistent ordering sorted_scripts = sorted( self._scripts_cache.items(), key=lambda x: x[1].get("name", "") ) # Add scripts (limit to max_results) for script_name, script_info in sorted_scripts: script_results = self._create_script_results_with_args( script_name, script_info, 0.8 ) for script_result in script_results: if len(results) < max_results: results.append(script_result) else: break if len(results) >= max_results: break return results def _search_scripts(self, query: str) -> List[Result]: """Search for scripts matching the query.""" results = [] query_lower = query.lower() max_results = 15 # Categorize matches for better sorting exact_matches = [] prefix_matches = [] partial_matches = [] description_matches = [] for script_name, script_info in self._scripts_cache.items(): script_name_lower = script_name.lower() description_lower = script_info.get("description", "").lower() # Skip if no match at all if ( query_lower not in script_name_lower and query_lower not in description_lower ): continue # Categorize matches if script_name_lower == query_lower: exact_matches.append((script_name, script_info, 1.0)) elif script_name_lower.startswith(query_lower): prefix_matches.append((script_name, script_info, 0.9)) elif query_lower in script_name_lower: partial_matches.append((script_name, script_info, 0.7)) elif query_lower in description_lower: description_matches.append((script_name, script_info, 0.5)) # Combine results in priority order all_matches = ( exact_matches + prefix_matches + partial_matches + description_matches ) # Convert to Result objects for script_name, script_info, relevance in all_matches: script_results = self._create_script_results_with_args( script_name, script_info, relevance ) for script_result in script_results: if len(results) < max_results: results.append(script_result) else: break if len(results) >= max_results: break return results def _create_script_result( self, script_name: str, script_info: Dict, relevance: float ) -> Result: """Create a Result object for a script.""" script_path = script_info.get("path", "") description = script_info.get("description", "") script_type = script_info.get("type", "discovered") executable = script_info.get("executable", False) category = script_info.get("category", "") # Create subtitle with additional info subtitle_parts = [] if description: subtitle_parts.append(description) if category: subtitle_parts.append(f"[{category}]") if not executable: subtitle_parts.append("(not executable)") subtitle = ( " | ".join(subtitle_parts) if subtitle_parts else f"Execute: {script_name}" ) # Choose icon based on script type and status if not executable: icon_name = "gtk-file" elif script_type == "custom": icon_name = "folder-script-symbolic" else: icon_name = "terminalc" return Result( title=script_name, subtitle=subtitle, icon_name=icon_name, action=self._create_script_action(script_name, script_info), relevance=relevance, plugin_name=self.display_name, data={ "script_name": script_name, "script_path": script_path, "type": script_type, }, ) def _create_script_results_with_args( self, script_name: str, script_info: Dict, relevance: float ) -> List[Result]: """Create multiple Result objects for scripts that support arguments.""" results = [] # Check for special scripts that need argument variants if script_name == "hyprpicker.sh": # For hyprpicker, only show the argument variants (skip basic version) variants = [ ("-rgb", "Pick RGB color"), ("-hex", "Pick HEX color"), ("-hsv", "Pick HSV color"), ] for arg, desc in variants: variant_result = Result( title=f"{script_name} {arg}", subtitle=f"{desc} | [scripts]", icon_name="terminal-symbolic", action=self._create_script_action_with_args( script_name, script_info, [arg] ), relevance=relevance + 0.1, # Slightly higher relevance for specific variants plugin_name=self.display_name, data={ "script_name": script_name, "script_path": script_info.get("path", ""), "type": script_info.get("type", "discovered"), "args": [arg], }, ) results.append(variant_result) else: # For other scripts, create the basic result basic_result = self._create_script_result( script_name, script_info, relevance ) results.append(basic_result) return results def _create_script_action(self, script_name: str, script_info: Dict): """Create an action function for executing a script.""" def action(): self._execute_script(script_name, script_info) return action def _create_script_action_with_args( self, script_name: str, script_info: Dict, args: List[str] ): """Create an action function for executing a script with specific arguments.""" def action(): # Create a modified script_info with the specific arguments modified_script_info = script_info.copy() modified_script_info["args"] = args self._execute_script(script_name, modified_script_info) return action def _execute_script(self, script_name: str, script_info: Dict): """Execute a bash script.""" try: script_path = script_info.get("path", "") script_args = script_info.get("args", []) if not os.path.exists(script_path): return if not script_info.get("executable", False): return # Build command command = [script_path] + script_args exec_shell_command_async(command) except Exception as e: print(f"BashScriptsPlugin: Error executing script '{script_name}': {e}")