From 6c010b49214b90958f99ca08241ad8cdc47921ab Mon Sep 17 00:00:00 2001 From: 7000pctAUTO Date: Sat, 31 Jan 2026 10:23:33 +0000 Subject: [PATCH] feat: add models, context, and search modules --- cli_memory/search.py | 111 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 111 insertions(+) create mode 100644 cli_memory/search.py diff --git a/cli_memory/search.py b/cli_memory/search.py new file mode 100644 index 0000000..4fc7e35 --- /dev/null +++ b/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, + }