Initial upload: shell-speak CLI tool with natural language to shell command conversion

shell_speak/history.py

@@ -0,0 +1,138 @@
+"""History management module."""
+
+import json
+from datetime import datetime
+from pathlib import Path
+from typing import List, Optional
+
+from shell_speak.config import get_history_file, ensure_data_dir
+from shell_speak.models import HistoryEntry
+
+
+class HistoryManager:
+    """Manages command history storage and retrieval."""
+
+    def __init__(self):
+        self._entries: List[HistoryEntry] = []
+        self._loaded = False
+
+    def load(self) -> None:
+        """Load history from file."""
+        history_file = get_history_file()
+        if not history_file.exists():
+            self._entries = []
+            self._loaded = True
+            return
+
+        try:
+            with open(history_file, 'r') as f:
+                data = json.load(f)
+                self._entries = []
+                for item in data.get("entries", []):
+                    entry = HistoryEntry(
+                        query=item.get("query", ""),
+                        command=item.get("command", ""),
+                        tool=item.get("tool", ""),
+                        timestamp=datetime.fromisoformat(item.get("timestamp", datetime.now().isoformat())),
+                        explanation=item.get("explanation", ""),
+                    )
+                    self._entries.append(entry)
+        except Exception:
+            self._entries = []
+
+        self._loaded = True
+
+    def save(self) -> None:
+        """Save history to file."""
+        ensure_data_dir()
+        history_file = get_history_file()
+
+        data = {
+            "version": "1.0",
+            "entries": [
+                {
+                    "query": entry.query,
+                    "command": entry.command,
+                    "tool": entry.tool,
+                    "timestamp": entry.timestamp.isoformat(),
+                    "explanation": entry.explanation,
+                }
+                for entry in self._entries
+            ],
+        }
+
+        with open(history_file, 'w') as f:
+            json.dump(data, f, indent=2)
+
+    def add(self, query: str, command: str, tool: str, explanation: str = "") -> None:
+        """Add a new entry to history."""
+        if not self._loaded:
+            self.load()
+
+        entry = HistoryEntry(
+            query=query,
+            command=command,
+            tool=tool,
+            timestamp=datetime.now(),
+            explanation=explanation,
+        )
+        self._entries.append(entry)
+
+        if len(self._entries) > 1000:
+            self._entries = self._entries[-1000:]
+
+        self.save()
+
+    def get_all(self) -> List[HistoryEntry]:
+        """Get all history entries."""
+        if not self._loaded:
+            self.load()
+        return self._entries.copy()
+
+    def get_recent(self, limit: int = 20) -> List[HistoryEntry]:
+        """Get recent history entries."""
+        if not self._loaded:
+            self.load()
+        return self._entries[-limit:]
+
+    def search(self, query: str, tool: Optional[str] = None) -> List[HistoryEntry]:
+        """Search history entries."""
+        if not self._loaded:
+            self.load()
+
+        results = []
+        query_lower = query.lower()
+
+        for entry in self._entries:
+            if query_lower in entry.query.lower() or query_lower in entry.command.lower():
+                if tool is None or entry.tool == tool:
+                    results.append(entry)
+
+        return results
+
+    def get_last_command(self, tool: Optional[str] = None) -> Optional[HistoryEntry]:
+        """Get the last command from history."""
+        if not self._loaded:
+            self.load()
+
+        for entry in reversed(self._entries):
+            if tool is None or entry.tool == tool:
+                return entry
+
+        return None
+
+    def clear(self) -> None:
+        """Clear all history."""
+        self._entries = []
+        self.save()
+
+
+_history_manager: Optional[HistoryManager] = None
+
+
+def get_history_manager() -> HistoryManager:
+    """Get the global history manager."""
+    global _history_manager
+    if _history_manager is None:
+        _history_manager = HistoryManager()
+    return _history_manager