From 079c58afbe426d266b2ce68d9f725dab4c5d3a2f Mon Sep 17 00:00:00 2001 From: 7000pctAUTO Date: Wed, 4 Feb 2026 11:01:48 +0000 Subject: [PATCH] Add history learning module --- shellgenius/history.py | 371 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 371 insertions(+) create mode 100644 shellgenius/history.py diff --git a/shellgenius/history.py b/shellgenius/history.py new file mode 100644 index 0000000..deace2f --- /dev/null +++ b/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)