cli-command-memory/cli_memory/config.py

import os
from pathlib import Path
from typing import Any, Dict, Optional

import yaml
from dotenv import load_dotenv


class Config:
    def __init__(self, config_path: Optional[str] = None):
        self.config_path = config_path or self._find_config_file()
        self._config: Dict[str, Any] = {}
        self._load_config()

    def _find_config_file(self) -> str:
        env_path = os.environ.get("CLI_MEMORY_CONFIG")
        if env_path and os.path.exists(env_path):
            return env_path
        home_config = os.path.expanduser("~/.cli_memory/config.yaml")
        if os.path.exists(home_config):
            return home_config
        local_config = os.path.join(os.getcwd(), "config.yaml")
        if os.path.exists(local_config):
            return local_config
        return home_config

    def _load_config(self) -> None:
        load_dotenv()
        self._config = {
            "database": self._load_database_config(),
            "recording": self._load_recording_config(),
            "project": self._load_project_config(),
            "search": self._load_search_config(),
            "suggestions": self._load_suggestions_config(),
            "patterns": self._load_patterns_config(),
            "script": self._load_script_config(),
            "playback": self._load_playback_config(),
            "shell": self._load_shell_config(),
            "logging": self._load_logging_config(),
        }
        self._apply_env_overrides()

    def _load_database_config(self) -> Dict[str, Any]:
        return {
            "path": os.environ.get("DATABASE_PATH", "~/.cli_memory/database.db"),
            "wal_mode": True,
            "timeout": 30.0,
        }

    def _load_recording_config(self) -> Dict[str, Any]:
        return {
            "max_commands_per_workflow": int(os.environ.get("MAX_WORKFLOW_COMMANDS", 100)),
            "min_commands_for_workflow": 3,
            "auto_save_interval": 10,
            "capture_exit_code": True,
            "capture_duration": True,
        }

    def _load_project_config(self) -> Dict[str, Any]:
        return {
            "auto_detect_git": True,
            "cache_ttl": 3600,
            "excluded_dirs": ["node_modules", "__pycache__", ".git", "venv", ".venv"],
        }

    def _load_search_config(self) -> Dict[str, Any]:
        return {
            "max_results": 50,
            "default_limit": 20,
            "enable_fuzzy": True,
            "fuzzy_threshold": 0.6,
        }

    def _load_suggestions_config(self) -> Dict[str, Any]:
        return {
            "max_suggestions": int(os.environ.get("MAX_SUGGESTIONS", 10)),
            "min_confidence": 0.3,
            "decay_factor": 0.95,
            "recency_weight": 0.3,
            "frequency_weight": 0.4,
            "context_weight": 0.3,
        }

    def _load_patterns_config(self) -> Dict[str, Any]:
        return {
            "min_sequence_length": 3,
            "min_occurrences": 2,
            "max_pattern_length": 10,
            "similarity_threshold": 0.8,
            "timeout": 30.0,
        }

    def _load_script_config(self) -> Dict[str, Any]:
        return {
            "output_dir": os.path.expanduser("~/.cli_memory/scripts"),
            "include_error_handling": True,
            "include_logging": True,
            "dry_run_default": False,
        }

    def _load_playback_config(self) -> Dict[str, Any]:
        return {
            "default_speed": 1.0,
            "confirm_each": False,
            "show_progress": True,
            "pause_on_error": True,
        }

    def _load_shell_config(self) -> Dict[str, Any]:
        return {
            "enable_autocomplete": os.environ.get("ENABLE_AUTOCOMPLETE", "true").lower() == "true",
            "prompt_command": "cm-prompt",
            "history_file": os.path.expanduser("~/.cli_memory/shell_history"),
        }

    def _load_logging_config(self) -> Dict[str, Any]:
        return {
            "level": os.environ.get("LOG_LEVEL", "info"),
            "format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s",
            "file": os.path.expanduser("~/.cli_memory/logs/app.log"),
        }

    def _apply_env_overrides(self) -> None:
        if os.environ.get("CLI_MEMORY_HOME"):
            home = os.path.expanduser(os.environ["CLI_MEMORY_HOME"])
            self._config["database"]["path"] = os.path.join(home, "database.db")
            self._config["script"]["output_dir"] = os.path.join(home, "scripts")
            self._config["shell"]["history_file"] = os.path.join(home, "shell_history")
            self._config["logging"]["file"] = os.path.join(home, "logs", "app.log")

    def get(self, key: str, default: Any = None) -> Any:
        keys = key.split(".")
        value = self._config
        for k in keys:
            if isinstance(value, dict) and k in value:
                value = value[k]
            else:
                return default
        return value

    def set(self, key: str, value: Any) -> None:
        keys = key.split(".")
        config = self._config
        for k in keys[:-1]:
            if k not in config:
                config[k] = {}
            config = config[k]
        config[keys[-1]] = value

    def reload(self) -> None:
        self._load_config()

    def get_home_dir(self) -> str:
        home = os.environ.get("CLI_MEMORY_HOME", "~/.cli_memory")
        return os.path.expanduser(home)

    def ensure_directories(self) -> None:
        home = self.get_home_dir()
        for subdir in ["scripts", "logs"]:
            path = os.path.join(home, subdir)
            os.makedirs(path, exist_ok=True)