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 } # Built-in power manager commands self.power_manager_commands = { "pm balanced": { "description": "Switch to balanced power profile", "profile": "balanced", "icon": "power-balanced", }, "pm performance": { "description": "Switch to performance power profile", "profile": "performance", "icon": "power-performance", }, "pm saver": { "description": "Switch to power saver profile", "profile": "power-saver", "icon": "power-saving", }, } # 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 # Battery service reference self._battery_service = None def initialize(self): """Initialize the bash scripts plugin.""" self.set_triggers(["sh"]) self._load_scripts_cache() self._start_background_cache_update() self._get_battery_service() 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 _get_battery_service(self): """Get reference to the battery service.""" try: from services.battery import Battery self._battery_service = Battery() except Exception as e: print(f"BashScriptsPlugin: Error getting battery service: {e}") self._battery_service = None def _set_power_profile(self, profile: str) -> bool: """Set the power profile using the battery service.""" if not self._battery_service: print("BashScriptsPlugin: Battery service not available") return False try: # Check if profile proxy is available if ( not hasattr(self._battery_service, "_profile_proxy") or not self._battery_service._profile_proxy ): print("BashScriptsPlugin: Power profile proxy not available") return False # Get available profiles profiles = self._battery_service._profile_proxy.Profiles available_profiles = [] for p in profiles: if isinstance(p, dict) and "Profile" in p: available_profiles.append(p["Profile"]) elif hasattr(p, "Profile"): available_profiles.append(p.Profile) elif isinstance(p, str): available_profiles.append(p) # Map profile types to actual profile names profile_mapping = { "power-saver": ["power-saver", "powersave", "power_saver"], "balanced": ["balanced", "balance"], "performance": ["performance", "performance-mode"], } # Find the actual profile name actual_profile = None if profile in profile_mapping: for candidate in profile_mapping[profile]: if candidate in available_profiles: actual_profile = candidate break elif profile in available_profiles: actual_profile = profile if actual_profile: self._battery_service._profile_proxy.ActiveProfile = actual_profile print( f"BashScriptsPlugin: Successfully set power profile to { actual_profile }" ) return True else: print( f"BashScriptsPlugin: No matching profile found for '{ profile }' in available profiles: {available_profiles}" ) return False except Exception as e: print(f"BashScriptsPlugin: Error setting power profile: {e}") return False 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 = [] # Check for power manager commands first power_results = self._search_power_commands(query) if power_results: results.extend(power_results) # Handle special commands if not query: # Show all scripts when no query results.extend(self._list_all_scripts()) # Also show power manager commands when no query if not power_results: results.extend(self._list_all_power_commands()) 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 _search_power_commands(self, query: str) -> List[Result]: """Search for power manager commands matching the query.""" if not query: return [] results = [] query_lower = query.lower() for cmd, info in self.power_manager_commands.items(): cmd_lower = cmd.lower() description_lower = info["description"].lower() # Check for exact match or partial match if ( query_lower == cmd_lower or cmd_lower.startswith(query_lower) or query_lower in cmd_lower or query_lower in description_lower ): relevance = 1.0 if query_lower == cmd_lower else 0.9 result = Result( title=cmd, subtitle=info["description"], icon_name=info["icon"], action=self._create_power_action(info["profile"]), relevance=relevance, plugin_name=self.display_name, data={ "command": cmd, "profile": info["profile"], "type": "power_manager", }, ) results.append(result) return results def _list_all_power_commands(self) -> List[Result]: """List all power manager commands.""" results = [] for cmd, info in self.power_manager_commands.items(): result = Result( title=cmd, subtitle=info["description"], icon_name=info["icon"], action=self._create_power_action(info["profile"]), relevance=0.8, plugin_name=self.display_name, data={ "command": cmd, "profile": info["profile"], "type": "power_manager", }, ) results.append(result) return results def _create_power_action(self, profile: str): """Create an action function for setting power profile.""" def action(): success = self._set_power_profile(profile) if success: # Clear the search query after successful execution try: from gi.repository import GLib from fabric import Application app = Application.get_default() if app and hasattr(app, "launcher"): launcher = app.launcher if launcher and hasattr(launcher, "search_entry"): def clear_search(): launcher.search_entry.set_text("") return False # Use a small delay to ensure the action completes first GLib.timeout_add(50, clear_search) except Exception as e: print(f"BashScriptsPlugin: Could not clear search query: {e}") return action 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}")