Add recorder, project, context, search, and history modules

.cli_memory/search.py

@@ -0,0 +1,111 @@
+import re
+import logging
+from typing import Optional, List, Dict, Any
+from difflib import SequenceMatcher
+
+from .config import Config
+from .models import Command, CommandType
+
+logger = logging.getLogger(__name__)
+
+
+class SearchEngine:
+    def __init__(self, config: Optional[Config] = None):
+        self.config = config or Config()
+
+    def search_commands(
+        self,
+        commands: List[Command],
+        query: str,
+        project_id: Optional[int] = None,
+        command_type: Optional[str] = None,
+        start_time: Optional[Any] = None,
+        end_time: Optional[Any] = None,
+        limit: int = 50,
+        fuzzy: bool = False,
+    ) -> List[Command]:
+        results = []
+
+        for cmd in commands:
+            if project_id is not None and cmd.project_id != project_id:
+                continue
+            if command_type is not None and cmd.command_type.value != command_type:
+                continue
+
+            if self._matches_query(cmd.command, query, fuzzy):
+                results.append(cmd)
+
+            if len(results) >= limit:
+                break
+
+        return results
+
+    def _matches_query(self, command: str, query: str, fuzzy: bool = False) -> bool:
+        if fuzzy:
+            ratio = SequenceMatcher(None, command.lower(), query.lower()).ratio()
+            return ratio >= self.config.get("search.fuzzy_threshold", 0.6)
+        else:
+            return query.lower() in command.lower()
+
+    def search_by_project(
+        self, commands: List[Command], project_id: int
+    ) -> List[Command]:
+        return [cmd for cmd in commands if cmd.project_id == project_id]
+
+    def search_by_technology(
+        self, commands: List[Command], technology: str
+    ) -> List[Command]:
+        tech_commands = {
+            "git": CommandType.GIT,
+            "docker": CommandType.DOCKER,
+            "python": CommandType.OTHER,
+            "node": CommandType.OTHER,
+        }
+        cmd_type = tech_commands.get(technology.lower())
+        if cmd_type:
+            return [cmd for cmd in commands if cmd.command_type == cmd_type]
+        return []
+
+    def search_recent(
+        self, commands: List[Command], hours: int = 24
+    ) -> List[Command]:
+        from datetime import datetime, timedelta
+
+        cutoff = datetime.utcnow() - timedelta(hours=hours)
+        return [cmd for cmd in commands if cmd.timestamp >= cutoff]
+
+    def get_command_statistics(self, commands: List[Command]) -> Dict[str, Any]:
+        if not commands:
+            return {
+                "total_commands": 0,
+                "by_type": {},
+                "avg_duration_ms": 0,
+                "most_common": [],
+            }
+
+        type_counts = {}
+        duration_sum = 0
+        duration_count = 0
+        command_counts = {}
+
+        for cmd in commands:
+            type_str = cmd.command_type.value
+            type_counts[type_str] = type_counts.get(type_str, 0) + 1
+
+            if cmd.duration_ms is not None:
+                duration_sum += cmd.duration_ms
+                duration_count += 1
+
+            cmd_key = cmd.command.split()[0] if cmd.command else ""
+            command_counts[cmd_key] = command_counts.get(cmd_key, 0) + 1
+
+        most_common = sorted(
+            command_counts.items(), key=lambda x: x[1], reverse=True
+        )[:10]
+
+        return {
+            "total_commands": len(commands),
+            "by_type": type_counts,
+            "avg_duration_ms": duration_sum // duration_count if duration_count > 0 else 0,
+            "most_common": most_common,
+        }