112 lines
3.5 KiB
Python
112 lines
3.5 KiB
Python
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,
|
|
}
|