Add history learning module

shellgenius/history.py

@@ -0,0 +1,371 @@
+"""Command history learning system for ShellGenius."""
+
+import json
+import os
+from dataclasses import dataclass, field
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+import yaml
+
+from shellgenius.config import get_config
+
+
+@dataclass
+class HistoryEntry:
+    """A command history entry."""
+
+    id: str
+    timestamp: str
+    description: str
+    commands: List[str]
+    shell_type: str
+    usage_count: int = 1
+    tags: List[str] = field(default_factory=list)
+    success: bool = True
+
+
+class HistoryStorage:
+    """Storage management for command history."""
+
+    def __init__(self, storage_path: Optional[str] = None):
+        """Initialize history storage.
+
+        Args:
+            storage_path: Path to history file
+        """
+        config = get_config()
+        self.storage_path = storage_path or config.get(
+            "history.storage_path",
+            "~/.config/shellgenius/history.yaml",
+        )
+        self.storage_path = os.path.expanduser(self.storage_path)
+        self._ensure_storage_exists()
+
+    def _ensure_storage_exists(self) -> None:
+        """Ensure storage directory exists."""
+        Path(self.storage_path).parent.mkdir(parents=True, exist_ok=True)
+        if not Path(self.storage_path).exists():
+            self._save({"entries": [], "metadata": {"version": "1.0"}})
+
+    def _load(self) -> Dict[str, Any]:
+        """Load history from file.
+
+        Returns:
+            History data dictionary
+        """
+        try:
+            with open(self.storage_path, "r") as f:
+                return yaml.safe_load(f) or {"entries": [], "metadata": {}}
+        except Exception:
+            return {"entries": [], "metadata": {}}
+
+    def _save(self, data: Dict[str, Any]) -> None:
+        """Save history to file.
+
+        Args:
+            data: History data to save
+        """
+        with open(self.storage_path, "w") as f:
+            yaml.dump(data, f, default_flow_style=False, allow_unicode=True)
+
+    def add_entry(self, entry: HistoryEntry) -> None:
+        """Add a new history entry.
+
+        Args:
+            entry: HistoryEntry to add
+        """
+        data = self._load()
+        data["entries"].append(
+            {
+                "id": entry.id,
+                "timestamp": entry.timestamp,
+                "description": entry.description,
+                "commands": entry.commands,
+                "shell_type": entry.shell_type,
+                "usage_count": entry.usage_count,
+                "tags": entry.tags,
+                "success": entry.success,
+            }
+        )
+        self._save(data)
+
+    def get_entries(
+        self, limit: Optional[int] = None, offset: int = 0
+    ) -> List[HistoryEntry]:
+        """Get history entries.
+
+        Args:
+            limit: Maximum entries to return
+            offset: Starting offset
+
+        Returns:
+            List of HistoryEntry objects
+        """
+        data = self._load()
+        entries = data.get("entries", [])
+
+        if offset > 0:
+            entries = entries[offset:]
+        if limit:
+            entries = entries[:limit]
+
+        return [
+            HistoryEntry(
+                id=e["id"],
+                timestamp=e["timestamp"],
+                description=e["description"],
+                commands=e["commands"],
+                shell_type=e["shell_type"],
+                usage_count=e.get("usage_count", 1),
+                tags=e.get("tags", []),
+                success=e.get("success", True),
+            )
+            for e in entries
+        ]
+
+    def search(
+        self, query: str, limit: int = 10
+    ) -> List[HistoryEntry]:
+        """Search history by query.
+
+        Args:
+            query: Search query
+            limit: Maximum results
+
+        Returns:
+            List of matching HistoryEntry objects
+        """
+        data = self._load()
+        results = []
+        query_lower = query.lower()
+
+        for entry in data.get("entries", []):
+            if (
+                query_lower in entry["description"].lower()
+                or any(query_lower in cmd.lower() for cmd in entry["commands"])
+                or any(query_lower in tag.lower() for tag in entry.get("tags", []))
+            ):
+                results.append(
+                    HistoryEntry(
+                        id=entry["id"],
+                        timestamp=entry["timestamp"],
+                        description=entry["description"],
+                        commands=entry["commands"],
+                        shell_type=entry["shell_type"],
+                        usage_count=e.get("usage_count", 1),
+                        tags=e.get("tags", []),
+                        success=e.get("success", True),
+                    )
+                )
+                if len(results) >= limit:
+                    break
+
+        return results
+
+    def get_popular(self, limit: int = 10) -> List[HistoryEntry]:
+        """Get most used entries.
+
+        Args:
+            limit: Maximum entries to return
+
+        Returns:
+            List of popular HistoryEntry objects
+        """
+        data = self._load()
+        entries = [
+            HistoryEntry(
+                id=e["id"],
+                timestamp=e["timestamp"],
+                description=e["description"],
+                commands=e["commands"],
+                shell_type=e["shell_type"],
+                usage_count=e.get("usage_count", 1),
+                tags=e.get("tags", []),
+                success=e.get("success", True),
+            )
+            for e in data.get("entries", [])
+        ]
+        entries.sort(key=lambda x: x.usage_count, reverse=True)
+        return entries[:limit]
+
+    def update_usage(self, entry_id: str) -> bool:
+        """Increment usage count for an entry.
+
+        Args:
+            entry_id: Entry ID to update
+
+        Returns:
+            True if updated successfully
+        """
+        data = self._load()
+        for entry in data.get("entries", []):
+            if entry["id"] == entry_id:
+                entry["usage_count"] = entry.get("usage_count", 0) + 1
+                self._save(data)
+                return True
+        return False
+
+    def delete_entry(self, entry_id: str) -> bool:
+        """Delete a history entry.
+
+        Args:
+            entry_id: Entry ID to delete
+
+        Returns:
+            True if deleted successfully
+        """
+        data = self._load()
+        original_len = len(data.get("entries", []))
+        data["entries"] = [e for e in data.get("entries", []) if e["id"] != entry_id]
+        if len(data["entries"]) < original_len:
+            self._save(data)
+            return True
+        return False
+
+    def clear(self) -> None:
+        """Clear all history."""
+        self._save({"entries": [], "metadata": {"version": "1.0"}})
+
+
+class HistoryLearner:
+    """Learning from command history for personalized suggestions."""
+
+    def __init__(self):
+        """Initialize history learner."""
+        self.storage = HistoryStorage()
+        self.config = get_config()
+
+    def learn(
+        self,
+        description: str,
+        commands: List[str],
+        shell_type: str = "bash",
+        tags: Optional[List[str]] = None,
+    ) -> HistoryEntry:
+        """Learn from a generated command.
+
+        Args:
+            description: Description of what was generated
+            commands: List of commands generated
+            shell_type: Shell type used
+            tags: Optional tags for categorization
+
+        Returns:
+            Created HistoryEntry
+        """
+        import uuid
+
+        entry = HistoryEntry(
+            id=str(uuid.uuid4()),
+            timestamp=datetime.utcnow().isoformat(),
+            description=description,
+            commands=commands,
+            shell_type=shell_type,
+            tags=tags or [],
+        )
+
+        self.storage.add_entry(entry)
+        return entry
+
+    def find_similar(self, query: str, limit: int = 5) -> List[HistoryEntry]:
+        """Find similar commands from history.
+
+        Args:
+            query: Query to match against
+            limit: Maximum results
+
+        Returns:
+            List of similar HistoryEntry objects
+        """
+        return self.storage.search(query, limit)
+
+    def suggest(
+        self, description: str, limit: int = 3
+    ) -> List[HistoryEntry]:
+        """Get personalized suggestions based on history.
+
+        Args:
+            description: Current task description
+            limit: Maximum suggestions
+
+        Returns:
+            List of suggested HistoryEntry objects
+        """
+        similar = self.find_similar(description, limit)
+        popular = self.storage.get_popular(limit)
+
+        suggestions = []
+        seen = set()
+
+        for entry in similar + popular:
+            if entry.id not in seen:
+                suggestions.append(entry)
+                seen.add(entry.id)
+            if len(suggestions) >= limit:
+                break
+
+        return suggestions
+
+    def get_frequent_patterns(self) -> Dict[str, Any]:
+        """Analyze frequent command patterns.
+
+        Returns:
+            Dictionary with pattern statistics
+        """
+        entries = self.storage.get_entries(limit=100)
+
+        patterns = {
+            "shell_types": {},
+            "common_commands": {},
+            "frequent_tags": {},
+        }
+
+        for entry in entries:
+            patterns["shell_types"][entry.shell_type] = (
+                patterns["shell_types"].get(entry.shell_type, 0) + 1
+            )
+
+            for cmd in entry.commands:
+                first_word = cmd.strip().split()[0] if cmd.strip() else ""
+                if first_word:
+                    patterns["common_commands"][first_word] = (
+                        patterns["common_commands"].get(first_word, 0) + 1
+                    )
+
+            for tag in entry.tags:
+                patterns["frequent_tags"][tag] = (
+                    patterns["frequent_tags"].get(tag, 0) + 1
+                )
+
+        return patterns
+
+    def update_from_feedback(
+        self, entry_id: str, was_useful: bool, tags: Optional[List[str]] = None
+    ) -> bool:
+        """Update entry based on user feedback.
+
+        Args:
+            entry_id: Entry ID
+            was_useful: Whether the suggestion was useful
+            tags: Optional tags to add
+
+        Returns:
+            True if updated successfully
+        """
+        if was_useful:
+            return self.storage.update_usage(entry_id)
+        return False
+
+
+def get_history_storage(storage_path: Optional[str] = None) -> HistoryStorage:
+    """Get history storage instance.
+
+    Args:
+        storage_path: Optional storage path
+
+    Returns:
+        HistoryStorage instance
+    """
+    return HistoryStorage(storage_path)