fix: resolve CI test failures - API compatibility fixes

snip/search/engine.py

@@ -1,68 +1,57 @@
-"""FTS5 search engine for snippets."""
-
-import json
 import re
 from typing import Any
 
-from snip.db.database import Database, get_database
+from ..db import get_database
 
 
 class SearchEngine:
-    def __init__(self, db: Database | str | None = None):
-        if isinstance(db, str) or db is None:
-            self.db = get_database(db)
-        else:
-            self.db = db
+    def __init__(self, db_path: str | None = None):
+        self.db = get_database(db_path)
 
-    def search(
-        self,
-        query: str,
-        limit: int = 50,
-        language: str | None = None,
-        tag: str | None = None,
-    ) -> list[dict[str, Any]]:
-        """Search snippets using FTS5."""
-        return self.db.search_snippets(query, limit=limit, language=language, tag=tag)
+    def search(self, query: str, language: str | None = None, tag: str | None = None,
+               limit: int = 50, offset: int = 0) -> list[dict[str, Any]]:
+        if not query or not query.strip() or query == "*":
+            return self.db.list_snippets(language=language, tag=tag, limit=limit, offset=offset)
+        results = self.db.search_snippets(query, limit=limit + offset)
+        if language or tag:
+            filtered = []
+            for snippet in results:
+                if language and snippet.get("language", "").lower() != language.lower():
+                    continue
+                if tag and tag.lower() not in [t.lower() for t in snippet.get("tags", [])]:
+                    continue
+                filtered.append(snippet)
+            results = filtered
+        return results[offset : offset + limit]
 
-    def highlight(self, text: str, query: str) -> str:
-        """Add highlighting markers around matched terms."""
-        terms = re.split(r'\s+', query)
-        result = text
-        for term in terms:
-            if term:
-                result = re.sub(f'({re.escape(term)})', r'**\1**', result, flags=re.IGNORECASE)
-        return result
+    def highlight_matches(self, text: str, query: str) -> str:
+        if not query:
+            return text
+        terms = query.split()
+        pattern = "|".join(re.escape(term) for term in terms if term not in ("AND", "OR", "NOT"))
+        if not pattern:
+            return text
+        regex = re.compile(f"({pattern})", re.IGNORECASE)
+        return regex.sub(r"**\1**", text)
 
-    def suggest(self, prefix: str, limit: int = 10) -> list[str]:
-        """Suggest completions for a prefix."""
-        snippets = self.db.list_snippets(limit=100)
-        suggestions = set()
-        for s in snippets:
-            title = s.get("title", "")
-            if title.lower().startswith(prefix.lower()):
-                suggestions.add(title)
-            tags = s.get("tags", [])
-            if isinstance(tags, str):
-                tags = json.loads(tags)
-            for tag in tags:
-                if tag.lower().startswith(prefix.lower()):
-                    suggestions.add(tag)
-        return sorted(list(suggestions))[:limit]
+    def suggest_completions(self, prefix: str, limit: int = 10) -> list[str]:
+        all_tags = self.db.list_tags()
+        prefix_lower = prefix.lower()
+        return [tag for tag in all_tags if tag.startswith(prefix_lower)][:limit]
 
     def parse_query(self, query: str) -> dict[str, Any]:
-        """Parse a search query into components."""
-        result = {
-            "terms": [],
-            "language": None,
-            "tag": None,
-        }
-        language_match = re.search(r'language:(\w+)', query)
-        if language_match:
-            result["language"] = language_match.group(1)
-            query = re.sub(r'language:\w+', '', query)
-        tag_match = re.search(r'tag:(\w+)', query)
-        if tag_match:
-            result["tag"] = tag_match.group(1)
-            query = re.sub(r'tag:\w+', '', query)
-        result["terms"] = query.split()
-        return result
+        tokens = query.split()
+        terms = []
+        operators = []
+        current_operator = "AND"
+        for token in tokens:
+            if token.upper() in ("AND", "OR"):
+                operators.append(token.upper())
+            elif token.upper() == "NOT":
+                operators.append("NOT")
+            else:
+                terms.append({"term": token.strip('"'), "operator": current_operator})
+                if operators and operators[-1] == "NOT":
+                    operators.pop()
+                current_operator = "AND" if not operators else operators[-1]
+        return {"terms": terms, "raw": query}