`057ca7f`

refactor stable unthorough

Authored by

espadonne 11 months ago

SHA: 057ca7f6921a703ea7844dc73c405f9f08b25b17
Parents: 9309653
Tree: 6c70249

7 changed files

Status	File	+	-
A	`settings.sample.toml`	25	0
M	`src/shtick/cli.py`	32	0
M	`src/shtick/commands.py`	138	0
M	`src/shtick/config.py`	58	8
M	`src/shtick/generator.py`	22	2
A	`src/shtick/settings.py`	200	0
M	`src/shtick/shtick.py`	113	7

settings.sample.tomladded

 +[generation]
 +# Shells to generate files for. Empty list = auto-detect based on current shell
 +shells = ["bash", "dash", "zsh", "nushell", "powershell", "yash", "fish", "oil", "elvish"]
 +# Enable parallel generation (not implemented yet)
 +parallel = false
 +# Consolidate all items into single files per shell
 +consolidate_files = true
++
 +[behavior]
 +# Prompt to source changes after modifications
 +auto_source_prompt = true
 +# Check for conflicts when adding items
 +check_conflicts = true
 +# Create backups before saving config
 +backup_on_save = false
 +# Enable interactive prompts
 +interactive_mode = true
++
 +[performance]
 +# Cache time-to-live in seconds
 +cache_ttl = 300
 +# Enable lazy loading (not implemented yet)
 +lazy_load = false
 +# Enable batch operations for better performance
 +batch_operations = true

src/shtick/cli.pymodified

          "--shell", help="Specify shell type (auto-detected if not provided)"
+     )
 +    # Settings command
 +    settings_parser = subparsers.add_parser("settings", help="Manage shtick settings")
 +    settings_subparsers = settings_parser.add_subparsers(
 +        dest="settings_command", help="Settings commands"
 +    )
++
 +    # Settings subcommands
 +    settings_init_parser = settings_subparsers.add_parser(
 +        "init", help="Create default settings file"
 +    )
++
 +    settings_show_parser = settings_subparsers.add_parser(
 +        "show", help="Show current settings"
 +    )
++
 +    settings_set_parser = settings_subparsers.add_parser(
 +        "set", help="Set a setting value"
 +    )
 +    settings_set_parser.add_argument(
 +        "key", help="Setting key (e.g., generation.shells)"
 +    )
 +    settings_set_parser.add_argument("value", help="Setting value")
++
      args = parser.parse_args()
      # Set up logging first
              display.shells(args.long)
          elif args.command == "source":
              commands.source_command(args.shell)
 +        elif args.command == "settings":
 +            if args.settings_command == "init":
 +                commands.settings_init()
 +            elif args.settings_command == "show":
 +                commands.settings_show()
 +            elif args.settings_command == "set":
 +                commands.settings_set(args.key, args.value)
 +            else:
 +                settings_parser.print_help()
      except KeyboardInterrupt:
          logger.debug("Operation cancelled by user")

src/shtick/commands.pymodified

      def offer_auto_source(self):
          """Offer to source shtick in current shell session"""
 +        # Check settings first
 +        from shtick.settings import Settings
++
 +        settings = Settings()
 +        if not settings.behavior.auto_source_prompt:
 +            return
++
          current_shell = self.get_current_shell()
          if not current_shell or current_shell not in ["bash", "zsh", "fish"]:
              return
          # Output the source command that can be eval'd
          print(f"source {loader_path}")
++
 +    # Settings commands
 +    def settings_init(self):
 +        """Initialize settings file with defaults"""
 +        from shtick.settings import Settings
++
 +        settings = Settings()
++
 +        if os.path.exists(settings._settings_path):
 +            try:
 +                response = (
 +                    input("Settings file already exists. Overwrite? [y/N]: ")
 +                    .strip()
 +                    .lower()
 +                )
 +                if response not in ["y", "yes"]:
 +                    print("Cancelled")
 +                    return
 +            except (KeyboardInterrupt, EOFError):
 +                print("\nCancelled")
 +                return
++
 +        settings.create_default_settings_file()
 +        print(f"✓ Created settings file at {settings._settings_path}")
 +        print("\nYou can now customize your shtick behavior by editing this file.")
++
 +    def settings_show(self):
 +        """Show current settings"""
 +        from shtick.settings import Settings
++
 +        settings = Settings()
++
 +        print("Shtick Settings")
 +        print("=" * 50)
++
 +        print("\n[generation]")
 +        print(f"  shells = {settings.generation.shells or '[] (auto-detect)'}")
 +        print(f"  parallel = {settings.generation.parallel}")
 +        print(f"  consolidate_files = {settings.generation.consolidate_files}")
++
 +        print("\n[behavior]")
 +        print(f"  auto_source_prompt = {settings.behavior.auto_source_prompt}")
 +        print(f"  check_conflicts = {settings.behavior.check_conflicts}")
 +        print(f"  backup_on_save = {settings.behavior.backup_on_save}")
 +        print(f"  interactive_mode = {settings.behavior.interactive_mode}")
++
 +        print("\n[performance]")
 +        print(f"  cache_ttl = {settings.performance.cache_ttl}")
 +        print(f"  lazy_load = {settings.performance.lazy_load}")
 +        print(f"  batch_operations = {settings.performance.batch_operations}")
++
 +        print(f"\nSettings file: {settings._settings_path}")
 +        if not os.path.exists(settings._settings_path):
 +            print("(No settings file found - using defaults)")
 +            print("Run 'shtick settings init' to create one")
++
 +    def settings_set(self, key: str, value: str):
 +        """Set a specific setting value"""
 +        from shtick.settings import Settings
++
 +        settings = Settings()
++
 +        # Parse the key (e.g., "generation.shells")
 +        parts = key.split(".")
 +        if len(parts) != 2:
 +            print(
 +                f"Error: Invalid key format. Use 'section.key' (e.g., 'generation.shells')"
 +            )
 +            sys.exit(1)
++
 +        section, setting_key = parts
++
 +        # Validate section
 +        if section not in ["generation", "behavior", "performance"]:
 +            print(
 +                f"Error: Invalid section '{section}'. Must be one of: generation, behavior, performance"
 +            )
 +            sys.exit(1)
++
 +        # Get the section object
 +        section_obj = getattr(settings, section)
++
 +        # Check if key exists
 +        if not hasattr(section_obj, setting_key):
 +            print(f"Error: Invalid key '{setting_key}' for section '{section}'")
 +            print(f"Valid keys: {', '.join(vars(section_obj).keys())}")
 +            sys.exit(1)
++
 +        # Parse the value based on type
 +        current_value = getattr(section_obj, setting_key)
 +        try:
 +            if isinstance(current_value, bool):
 +                # Parse boolean
 +                if value.lower() in ["true", "1", "yes", "on"]:
 +                    parsed_value = True
 +                elif value.lower() in ["false", "0", "no", "off"]:
 +                    parsed_value = False
 +                else:
 +                    raise ValueError(f"Invalid boolean value: {value}")
 +            elif isinstance(current_value, int):
 +                # Parse integer
 +                parsed_value = int(value)
 +            elif isinstance(current_value, list):
 +                # Parse list (simple eval for now - could be improved)
 +                if value == "[]":
 +                    parsed_value = []
 +                elif value.startswith("[") and value.endswith("]"):
 +                    # Simple parsing - just split by comma
 +                    parsed_value = [
 +                        s.strip().strip("\"'")
 +                        for s in value[1:-1].split(",")
 +                        if s.strip()
 +                    ]
 +                else:
 +                    # Single value becomes a list
 +                    parsed_value = [value]
 +            else:
 +                # String value
 +                parsed_value = value
 +        except Exception as e:
 +            print(f"Error parsing value: {e}")
 +            sys.exit(1)
++
 +        # Set the value
 +        setattr(section_obj, setting_key, parsed_value)
++
 +        # Save settings
 +        settings.save()
++
 +        print(f"✓ Set {key} = {parsed_value}")
 +        print(f"Settings saved to {settings._settings_path}")

src/shtick/config.pymodified

  class Config:
      """Main configuration handler"""
 -    # Class variable for shell detection caching
 +    # Class variables for caching
      _detected_shell = None
 +    _active_groups_cache = None
 +    _active_groups_mtime = None
 +    _active_groups_file_path = None
      def __init__(self, config_path: Optional[str] = None):
          self.config_path = config_path or self.get_default_config_path()
          """Clear the cached shell detection (useful for testing)"""
          cls._detected_shell = None
 +    @classmethod
 +    def clear_all_caches(cls):
 +        """Clear all caches (useful for testing or forced refresh)"""
 +        cls._detected_shell = None
 +        cls._active_groups_cache = None
 +        cls._active_groups_mtime = None
 +        cls._active_groups_file_path = None
++
      def load_active_groups(self) -> List[str]:
 -        """Load list of currently active groups"""
 +        """Load list of currently active groups with caching"""
          active_file = self.get_active_groups_file()
 -        if not os.path.exists(active_file):
 -            return []
 -        with open(active_file, "r") as f:
 -            return [line.strip() for line in f if line.strip()]
 +        # Check if we need to reload from disk
 +        if os.path.exists(active_file):
 +            current_mtime = os.path.getmtime(active_file)
 +            if (
 +                self._active_groups_cache is None
 +                or self._active_groups_file_path != active_file
 +                or self._active_groups_mtime != current_mtime
 +            ):
++
 +                logger.debug(f"Reloading active groups from {active_file}")
 +                with open(active_file, "r") as f:
 +                    self._active_groups_cache = [
 +                        line.strip() for line in f if line.strip()
 +                    ]
 +                self._active_groups_mtime = current_mtime
 +                self._active_groups_file_path = active_file
 +            else:
 +                logger.debug("Using cached active groups")
 +        else:
 +            # File doesn't exist, return empty list
 +            self._active_groups_cache = []
 +            self._active_groups_mtime = None
++
 +        return (
 +            self._active_groups_cache.copy()
 +        )  # Return a copy to prevent external modification
      def save_active_groups(self, active_groups: List[str]) -> None:
          """Save list of active groups to state file"""
              for group in active_groups:
                  f.write(f"{group}\n")
 +        # Invalidate cache by updating mtime
 +        self._active_groups_cache = active_groups.copy()
 +        self._active_groups_mtime = os.path.getmtime(active_file)
 +        self._active_groups_file_path = active_file
++
      def activate_group(self, group_name: str) -> bool:
          """Activate a group. Returns True if successful."""
          # Check if group exists
      def get_all_shells_to_generate(self) -> List[str]:
          """Get list of shells to generate files for based on user settings"""
 -        # For now, just detect current shell and common ones
 -        # Later this can read from settings.toml
 +        from .settings import Settings
++
 +        settings = Settings()
++
 +        # If shells are explicitly set in settings, use those
 +        if settings.generation.shells:
 +            logger.debug(f"Using shells from settings: {settings.generation.shells}")
 +            return settings.generation.shells
++
 +        # Otherwise auto-detect based on current shell
          current_shell = self.get_current_shell()
          if not current_shell:
 +            logger.debug("No current shell detected, using defaults")
              return ["bash", "zsh", "fish"]  # Common defaults
          # Include current shell and close relatives
+         }
          shells = shell_families.get(current_shell, [current_shell])
 +        logger.debug(
 +            f"Auto-detected shells based on current shell '{current_shell}': {shells}"
 +        )
          return list(set(shells))  # Remove duplicates

src/shtick/generator.pymodified

          self.output_base_dir = output_base_dir or Config.get_output_dir()
          # Get shells to generate for once
          self._shells_to_generate = None
 +        self._config_for_shells = None
      @property
      def shells_to_generate(self) -> List[str]:
          """Get list of shells to generate files for"""
          if self._shells_to_generate is None:
 -            config = Config()
 -            self._shells_to_generate = config.get_all_shells_to_generate()
 +            # If we have a config instance stored, use it
 +            if self._config_for_shells:
 +                self._shells_to_generate = (
 +                    self._config_for_shells.get_all_shells_to_generate()
 +                )
 +            else:
 +                # Otherwise create a temporary one just for getting shells
 +                config = Config()
 +                self._shells_to_generate = config.get_all_shells_to_generate()
              logger.debug(f"Will generate files for shells: {self._shells_to_generate}")
          return self._shells_to_generate
 +    def set_config_for_shells(self, config: Config) -> None:
 +        """Set config instance to use for shell detection"""
 +        self._config_for_shells = config
 +        # Clear cached shells to force re-read
 +        self._shells_to_generate = None
++
      def ensure_output_dir(self, group_name: str) -> str:
          """Ensure output directory exists and return the path"""
          output_dir = os.path.join(self.output_base_dir, group_name)
              print("No groups found in configuration")
              return
 +        # Set config for shell detection FIRST
 +        self.set_config_for_shells(config)
++
          print(f"Generating shell files for {len(config.groups)} groups...")
          print(f"Target shells: {', '.join(self.shells_to_generate)}")
          """Generate dynamic loader files that source persistent + active groups"""
          logger.info("Generating dynamic loader files...")
 +        # Set config for shell detection
 +        self.set_config_for_shells(config)
++
          active_groups = config.load_active_groups()
          persistent_group = config.get_persistent_group()

src/shtick/settings.pyadded

 +"""
 +Settings management for shtick
 +"""
++
 +import os
 +import tomllib
 +import logging
 +from pathlib import Path
 +from typing import Dict, List, Optional, Any
 +from dataclasses import dataclass, field
++
 +logger = logging.getLogger("shtick")
++
++
 +@dataclass
 +class GenerationSettings:
 +    """Settings for file generation"""
++
 +    shells: List[str] = field(default_factory=list)  # Empty = auto-detect
 +    parallel: bool = False  # Threading not implemented yet
 +    consolidate_files: bool = True
++
++
 +@dataclass
 +class BehaviorSettings:
 +    """Settings for shtick behavior"""
++
 +    auto_source_prompt: bool = True
 +    check_conflicts: bool = True
 +    backup_on_save: bool = False
 +    interactive_mode: bool = True
++
++
 +@dataclass
 +class PerformanceSettings:
 +    """Settings for performance optimization"""
++
 +    cache_ttl: int = 300  # 5 minutes
 +    lazy_load: bool = False  # Not implemented yet
 +    batch_operations: bool = True
++
++
 +class Settings:
 +    """Manages shtick settings and preferences"""
++
 +    # Singleton instance
 +    _instance = None
 +    _loaded = False
++
 +    def __new__(cls):
 +        if cls._instance is None:
 +            cls._instance = super().__new__(cls)
 +        return cls._instance
++
 +    def __init__(self):
 +        if not self._loaded:
 +            self.generation = GenerationSettings()
 +            self.behavior = BehaviorSettings()
 +            self.performance = PerformanceSettings()
 +            self._settings_path = self.get_settings_path()
 +            self._load()
 +            Settings._loaded = True
++
 +    @staticmethod
 +    def get_settings_path() -> str:
 +        """Get the settings file path"""
 +        return os.path.expanduser("~/.config/shtick/settings.toml")
++
 +    def _load(self) -> None:
 +        """Load settings from file if it exists"""
 +        if not os.path.exists(self._settings_path):
 +            logger.debug("No settings file found, using defaults")
 +            return
++
 +        try:
 +            logger.debug(f"Loading settings from {self._settings_path}")
 +            with open(self._settings_path, "rb") as f:
 +                data = tomllib.load(f)
++
 +            # Load generation settings
 +            if "generation" in data:
 +                gen_data = data["generation"]
 +                self.generation.shells = gen_data.get("shells", [])
 +                self.generation.parallel = gen_data.get("parallel", False)
 +                self.generation.consolidate_files = gen_data.get(
 +                    "consolidate_files", True
 +                )
++
 +            # Load behavior settings
 +            if "behavior" in data:
 +                beh_data = data["behavior"]
 +                self.behavior.auto_source_prompt = beh_data.get(
 +                    "auto_source_prompt", True
 +                )
 +                self.behavior.check_conflicts = beh_data.get("check_conflicts", True)
 +                self.behavior.backup_on_save = beh_data.get("backup_on_save", False)
 +                self.behavior.interactive_mode = beh_data.get("interactive_mode", True)
++
 +            # Load performance settings
 +            if "performance" in data:
 +                perf_data = data["performance"]
 +                self.performance.cache_ttl = perf_data.get("cache_ttl", 300)
 +                self.performance.lazy_load = perf_data.get("lazy_load", False)
 +                self.performance.batch_operations = perf_data.get(
 +                    "batch_operations", True
 +                )
++
 +            logger.debug("Settings loaded successfully")
++
 +        except Exception as e:
 +            logger.warning(f"Failed to load settings: {e}, using defaults")
++
 +    def save(self) -> None:
 +        """Save current settings to file"""
 +        # Ensure directory exists
 +        os.makedirs(os.path.dirname(self._settings_path), exist_ok=True)
++
 +        # Build settings dictionary
 +        settings_dict = {
 +            "generation": {
 +                "shells": self.generation.shells,
 +                "parallel": self.generation.parallel,
 +                "consolidate_files": self.generation.consolidate_files,
 +            },
 +            "behavior": {
 +                "auto_source_prompt": self.behavior.auto_source_prompt,
 +                "check_conflicts": self.behavior.check_conflicts,
 +                "backup_on_save": self.behavior.backup_on_save,
 +                "interactive_mode": self.behavior.interactive_mode,
 +            },
 +            "performance": {
 +                "cache_ttl": self.performance.cache_ttl,
 +                "lazy_load": self.performance.lazy_load,
 +                "batch_operations": self.performance.batch_operations,
 +            },
 +        }
++
 +        # Write TOML file
 +        with open(self._settings_path, "w") as f:
 +            f.write("# Shtick settings file\n")
 +            f.write("# Generated automatically - edit as needed\n\n")
++
 +            for section, values in settings_dict.items():
 +                f.write(f"[{section}]\n")
 +                for key, value in values.items():
 +                    if isinstance(value, bool):
 +                        f.write(f"{key} = {str(value).lower()}\n")
 +                    elif isinstance(value, list):
 +                        if value:  # Non-empty list
 +                            f.write(f"{key} = {repr(value)}\n")
 +                        else:
 +                            f.write(f"{key} = []  # Empty = auto-detect\n")
 +                    else:
 +                        f.write(f"{key} = {value}\n")
 +                f.write("\n")
++
 +    def create_default_settings_file(self) -> None:
 +        """Create a default settings file with comments"""
 +        os.makedirs(os.path.dirname(self._settings_path), exist_ok=True)
++
 +        content = """# Shtick settings file
 +# This file controls various shtick behaviors and optimizations
++
 +[generation]
 +# Shells to generate files for. Empty list = auto-detect based on current shell
 +shells = []
 +# Enable parallel generation (not implemented yet)
 +parallel = false
 +# Consolidate all items into single files per shell
 +consolidate_files = true
++
 +[behavior]
 +# Prompt to source changes after modifications
 +auto_source_prompt = true
 +# Check for conflicts when adding items
 +check_conflicts = true
 +# Create backups before saving config
 +backup_on_save = false
 +# Enable interactive prompts
 +interactive_mode = true
++
 +[performance]
 +# Cache time-to-live in seconds
 +cache_ttl = 300
 +# Enable lazy loading (not implemented yet)
 +lazy_load = false
 +# Enable batch operations for better performance
 +batch_operations = true
 +"""
++
 +        with open(self._settings_path, "w") as f:
 +            f.write(content)
++
 +        logger.info(f"Created default settings file at {self._settings_path}")
++
 +    @classmethod
 +    def reset(cls):
 +        """Reset the singleton instance (useful for testing)"""
 +        cls._instance = None
 +        cls._loaded = False

src/shtick/shtick.pymodified

              debug: Enable debug output
          """
          self.config_path = config_path or Config.get_default_config_path()
 +        self.debug = debug
 -        # Set up logging
 -        if debug:
 -            logging.basicConfig(
 -                level=logging.DEBUG, format="%(name)s - %(levelname)s - %(message)s"
 -            )
 -        else:
 -            logging.basicConfig(level=logging.INFO, format="%(message)s")
 +        # Set up logging using centralized setup
 +        from .logger import setup_logging
++
 +        self.logger = setup_logging(debug=debug)
          self._config = None
          self._generator = Generator()
          try:
              config = self._get_config()
 +            # Use settings if check_conflicts not explicitly set
 +            if check_conflicts is None:
 +                from .settings import Settings
++
 +                settings = Settings()
 +                check_conflicts = settings.behavior.check_conflicts
++
              # Check for conflicts if requested
              if check_conflicts:
                  conflicts = self.check_conflicts(item_type, key, group_name)
              logger.error(f"Error removing {item_type}: {e}")
              return False
 +    # Batch operations
 +    def add_items_batch(
 +        self, items: List[Dict[str, Union[str, bool]]]
 +    ) -> Dict[str, List[str]]:
 +        """
 +        Add multiple items in one batch operation.
++
 +        Args:
 +            items: List of dicts with keys: type, group, key, value, check_conflicts (optional)
 +                Example: [
 +                    {'type': 'alias', 'group': 'dev', 'key': 'll', 'value': 'ls -la'},
 +                    {'type': 'env', 'group': 'dev', 'key': 'DEBUG', 'value': '1'},
 +                ]
++
 +        Returns:
 +            Dictionary with 'success' and 'failed' lists of item keys
 +        """
 +        results = {"success": [], "failed": []}
 +        config = self._get_config()
 +        affected_groups = set()
++
 +        for item in items:
 +            try:
 +                item_type = item["type"]
 +                group_name = item["group"]
 +                key = item["key"]
 +                value = item["value"]
 +                check_conflicts = item.get("check_conflicts", True)
++
 +                # Check for conflicts if requested
 +                if check_conflicts:
 +                    conflicts = self.check_conflicts(item_type, key, group_name)
 +                    if conflicts:
 +                        logger.warning(
 +                            f"Item '{key}' exists in groups: {[c[0] for c in conflicts]}"
 +                        )
++
 +                # Add the item
 +                config.add_item(item_type, group_name, key, value)
 +                results["success"].append(key)
++
 +                # Track affected groups
 +                if group_name == "persistent" or config.is_group_active(group_name):
 +                    affected_groups.add(group_name)
++
 +            except Exception as e:
 +                logger.error(f"Failed to add item '{item.get('key', 'unknown')}': {e}")
 +                results["failed"].append(item.get("key", "unknown"))
++
 +        # Save and regenerate once for all affected groups
 +        if results["success"]:
 +            self._save_and_regenerate(list(affected_groups))
++
 +        return results
++
 +    def remove_items_batch(self, items: List[Dict[str, str]]) -> Dict[str, List[str]]:
 +        """
 +        Remove multiple items in one batch operation.
++
 +        Args:
 +            items: List of dicts with keys: type, group, key
 +                Example: [
 +                    {'type': 'alias', 'group': 'dev', 'key': 'll'},
 +                    {'type': 'env', 'group': 'dev', 'key': 'DEBUG'},
 +                ]
++
 +        Returns:
 +            Dictionary with 'success' and 'failed' lists of item keys
 +        """
 +        results = {"success": [], "failed": []}
 +        config = self._get_config()
 +        affected_groups = set()
++
 +        for item in items:
 +            try:
 +                item_type = item["type"]
 +                group_name = item["group"]
 +                key = item["key"]
++
 +                # Remove the item
 +                if config.remove_item(item_type, group_name, key):
 +                    results["success"].append(key)
++
 +                    # Track affected groups
 +                    if group_name == "persistent" or config.is_group_active(group_name):
 +                        affected_groups.add(group_name)
 +                else:
 +                    results["failed"].append(key)
++
 +            except Exception as e:
 +                logger.error(
 +                    f"Failed to remove item '{item.get('key', 'unknown')}': {e}"
 +                )
 +                results["failed"].append(item.get("key", "unknown"))
++
 +        # Save and regenerate once for all affected groups
 +        if results["success"]:
 +            self._save_and_regenerate(list(affected_groups))
++
 +        return results
++
      # Group management
      def activate_group(self, group: str) -> bool:
          """