feat: migrate fuzzy search to textdistance (#36)

Author: julienld

AGENTS.md

@@ -385,7 +385,7 @@ Home Assistant MCP Server - Current Structure
 │   ├── manager.py             # MCP prompt templates
 │   └── enhanced.py            # Enhanced prompts
 └── Utils Layer (/src/ha_mcp/utils/)
-    ├── fuzzy_search.py        # Fuzzy matching engine with fuzzywuzzy
+    ├── fuzzy_search.py        # Fuzzy matching engine with textdistance
     ├── domain_handlers.py     # Home Assistant domain-specific logic
     ├── operation_manager.py   # Async operation management
     └── usage_logger.py        # Tool usage logging
@@ -416,7 +416,7 @@ Home Assistant MCP Server - Current Structure
 - **Connection Management**: Auto-reconnect with exponential backoff
 
 #### Smart Search Engine
-- **Fuzzy Matching**: Uses `fuzzywuzzy` with `python-levenshtein` for performance
+- **Fuzzy Matching**: Uses `textdistance[extras]` for high-performance similarity scoring
 - **Multi-language**: Supports French/English entity naming conventions
 - **Area-Based Search**: Groups entities by Home Assistant areas/rooms
 - **AI Optimization**: Provides system overviews optimized for AI understanding

pyproject.toml

@@ -28,11 +28,10 @@ classifiers = [
 dependencies = [
     "fastmcp>=2.11.0",
     "httpx>=0.27.0",
-    "fuzzywuzzy>=0.18.0",
+    "textdistance[extras]>=4.6.0",
     "pydantic>=2.5.0",
     "python-dotenv>=1.0.0",
     "websockets>=12.0",
-    "python-levenshtein>=0.25.0", # For faster fuzzy matching
 ]
 
 [project.optional-dependencies]

src/ha_mcp/tools/smart_search.py

@@ -7,7 +7,7 @@
 
 from ..client.rest_client import HomeAssistantClient
 from ..config import get_global_settings
-from ..utils.fuzzy_search import create_fuzzy_searcher
+from ..utils.fuzzy_search import calculate_partial_ratio, create_fuzzy_searcher
 
 logger = logging.getLogger(__name__)
 
@@ -637,14 +637,12 @@ def _search_in_dict(self, data: dict[str, Any] | list[Any] | Any, query: str) ->
 
         Returns a fuzzy match score based on how well the query matches values in the data.
         """
-        from fuzzywuzzy import fuzz
-
         max_score = 0
 
         if isinstance(data, dict):
             for key, value in data.items():
                 # Score the key itself
-                key_score = fuzz.partial_ratio(query, str(key).lower())
+                key_score = calculate_partial_ratio(query, str(key).lower())
                 max_score = max(max_score, key_score)
 
                 # Recursively score the value
@@ -658,11 +656,14 @@ def _search_in_dict(self, data: dict[str, Any] | list[Any] | Any, query: str) ->
 
         elif isinstance(data, str):
             # Direct fuzzy match on string values
-            max_score = max(max_score, fuzz.partial_ratio(query, data.lower()))
+            max_score = max(max_score, calculate_partial_ratio(query, data.lower()))
 
         elif data is not None:
             # Convert to string and match
-            max_score = max(max_score, fuzz.partial_ratio(query, str(data).lower()))
+            max_score = max(
+                max_score,
+                calculate_partial_ratio(query, str(data).lower()),
+            )
 
         return max_score
 

src/ha_mcp/utils/fuzzy_search.py

@@ -3,9 +3,12 @@
 """
 
 import logging
-from typing import Any
+from typing import Any, Iterable
 
-from fuzzywuzzy import fuzz, process
+import textdistance
+
+
+_LEVENSHTEIN = textdistance.Levenshtein()
 
 logger = logging.getLogger(__name__)
 
@@ -89,17 +92,17 @@ def _calculate_entity_score(
             score += 80
 
         # Fuzzy matching scores
-        entity_id_ratio = fuzz.ratio(query, entity_id.lower())
-        friendly_ratio = fuzz.ratio(query, friendly_name.lower())
-        domain_ratio = fuzz.ratio(query, domain.lower())
+        entity_id_ratio = calculate_ratio(query, entity_id.lower())
+        friendly_ratio = calculate_ratio(query, friendly_name.lower())
+        domain_ratio = calculate_ratio(query, domain.lower())
 
         # Partial ratio for substring matching
-        entity_partial = fuzz.partial_ratio(query, entity_id.lower())
-        friendly_partial = fuzz.partial_ratio(query, friendly_name.lower())
+        entity_partial = calculate_partial_ratio(query, entity_id.lower())
+        friendly_partial = calculate_partial_ratio(query, friendly_name.lower())
 
         # Token sort ratio for word order independence
-        entity_token = fuzz.token_sort_ratio(query, entity_id.lower())
-        friendly_token = fuzz.token_sort_ratio(query, friendly_name.lower())
+        entity_token = calculate_token_sort_ratio(query, entity_id.lower())
+        friendly_token = calculate_token_sort_ratio(query, friendly_name.lower())
 
         # Weight the scores
         score += max(entity_id_ratio, entity_partial, entity_token) * 0.7
@@ -185,7 +188,7 @@ def search_by_area(
                     continue
 
             # Fuzzy match on friendly name for room inference
-            area_score = fuzz.partial_ratio(area_lower, friendly_name.lower())
+            area_score = calculate_partial_ratio(area_lower, friendly_name.lower())
             if area_score >= self.threshold:
                 inferred_area = self._infer_area_from_name(friendly_name)
                 if inferred_area not in area_matches:
@@ -260,12 +263,12 @@ def get_smart_suggestions(
                 areas.add(inferred_area)
 
         # Fuzzy match against domains
-        domain_matches = process.extract(query, domains, limit=3, scorer=fuzz.ratio)
-        suggestions.extend([match[0] for match in domain_matches if match[1] >= 60])
+        domain_matches = extract_best_matches(query, domains, limit=3)
+        suggestions.extend([match for match, score in domain_matches if score >= 60])
 
         # Fuzzy match against areas
-        area_matches = process.extract(query, areas, limit=3, scorer=fuzz.ratio)
-        suggestions.extend([match[0] for match in area_matches if match[1] >= 60])
+        area_matches = extract_best_matches(query, areas, limit=3)
+        suggestions.extend([match for match, score in area_matches if score >= 60])
 
         # Add common search patterns
         if not suggestions:
@@ -290,3 +293,55 @@ def get_smart_suggestions(
 def create_fuzzy_searcher(threshold: int = 60) -> FuzzyEntitySearcher:
     """Create a new fuzzy entity searcher instance."""
     return FuzzyEntitySearcher(threshold)
+
+
+def calculate_ratio(query: str, value: str) -> int:
+    """Return the normalized Levenshtein similarity ratio (0-100)."""
+    if not query and not value:
+        return 100
+
+    max_len = max(len(query), len(value))
+    if max_len == 0:
+        return 0
+
+    distance = _LEVENSHTEIN.distance(query, value)
+    similarity = 1 - (distance / max_len)
+    return int(max(similarity, 0) * 100)
+
+
+def calculate_partial_ratio(query: str, value: str) -> int:
+    """Return the best similarity score for any substring match."""
+    if not query or not value:
+        return 0
+
+    shorter, longer = (query, value) if len(query) <= len(value) else (value, query)
+    window = len(shorter)
+    if window == 0:
+        return 0
+
+    best_score = 0
+    for start in range(len(longer) - window + 1):
+        substring = longer[start : start + window]
+        best_score = max(best_score, calculate_ratio(shorter, substring))
+        if best_score == 100:
+            break
+
+    return best_score
+
+
+def calculate_token_sort_ratio(query: str, value: str) -> int:
+    """Return similarity ratio after token sorting."""
+    query_sorted = " ".join(sorted(query.split()))
+    value_sorted = " ".join(sorted(value.split()))
+    return calculate_ratio(query_sorted, value_sorted)
+
+
+def extract_best_matches(
+    query: str, choices: Iterable[str], limit: int = 3
+) -> list[tuple[str, int]]:
+    """Return the highest scoring matches for a query among choices."""
+    scored_choices = [
+        (choice, calculate_ratio(query, choice)) for choice in choices if choice
+    ]
+    scored_choices.sort(key=lambda item: item[1], reverse=True)
+    return scored_choices[:limit]