feat: add models, context, and search modules
This commit is contained in:
111
cli_memory/search.py
Normal file
111
cli_memory/search.py
Normal file
@@ -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,
|
||||
}
|
||||
Reference in New Issue
Block a user