nextcloud-mcp-server/nextcloud_mcp_server/search/fuzzy.py

"""Fuzzy search algorithm using character overlap matching on Qdrant payload."""

import logging
from typing import Any

from qdrant_client.models import FieldCondition, Filter, MatchValue

from nextcloud_mcp_server.config import get_settings
from nextcloud_mcp_server.search.algorithms import SearchAlgorithm, SearchResult
from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client

logger = logging.getLogger(__name__)


class FuzzySearchAlgorithm(SearchAlgorithm):
    """Fuzzy search using simple character-based similarity.

    Implements character overlap matching with configurable threshold:
    - Compares character sets between query and text
    - Requires configurable % character overlap to match (default: 70%)
    - Tolerant to typos and minor variations
    """

    def __init__(self, threshold: float = 0.7):
        """Initialize fuzzy search algorithm.

        Args:
            threshold: Minimum character overlap ratio (0-1, default: 0.7)
        """
        if not 0.0 <= threshold <= 1.0:
            raise ValueError(f"Threshold must be between 0.0 and 1.0, got {threshold}")
        self.threshold = threshold

    @property
    def name(self) -> str:
        return "fuzzy"

    async def search(
        self,
        query: str,
        user_id: str,
        limit: int = 10,
        doc_type: str | None = None,
        **kwargs: Any,
    ) -> list[SearchResult]:
        """Execute fuzzy search using character overlap on Qdrant payload.

        Queries Qdrant for all indexed documents, then scores based on character
        overlap in title and excerpt fields. Returns unverified results - access
        verification should be performed separately at the final output stage.

        Args:
            query: Search query
            user_id: User ID for filtering
            limit: Maximum results to return
            doc_type: Optional document type filter (None = all types)
            **kwargs: Additional parameters (threshold override)

        Returns:
            List of unverified SearchResult objects ranked by character overlap score
        """
        settings = get_settings()
        threshold = kwargs.get("threshold", self.threshold)

        logger.info(
            f"Fuzzy search: query='{query}', user={user_id}, "
            f"limit={limit}, threshold={threshold}, doc_type={doc_type}"
        )

        # Build Qdrant filter
        filter_conditions = [
            FieldCondition(key="user_id", match=MatchValue(value=user_id))
        ]
        if doc_type:
            filter_conditions.append(
                FieldCondition(key="doc_type", match=MatchValue(value=doc_type))
            )

        # Scroll through Qdrant to get all matching documents
        qdrant_client = await get_qdrant_client()
        collection = settings.get_collection_name()

        all_points = []
        offset = None

        # Scroll through all points matching filter
        while True:
            scroll_result, next_offset = await qdrant_client.scroll(
                collection_name=collection,
                scroll_filter=Filter(must=filter_conditions),
                limit=100,  # Batch size
                offset=offset,
                with_payload=["doc_id", "doc_type", "title", "excerpt", "chunk_index"],
                with_vectors=False,  # Don't need vectors
            )

            all_points.extend(scroll_result)

            if next_offset is None:
                break
            offset = next_offset

        logger.debug(f"Retrieved {len(all_points)} points from Qdrant for fuzzy search")

        # Deduplicate by (doc_id, doc_type) - keep first chunk
        seen_docs = {}
        for point in all_points:
            doc_id = int(point.payload["doc_id"])
            dtype = point.payload.get("doc_type", "note")
            doc_key = (doc_id, dtype)

            chunk_idx = point.payload.get("chunk_index", 0)
            if doc_key not in seen_docs or chunk_idx == 0:
                seen_docs[doc_key] = point

        logger.debug(f"Deduplicated to {len(seen_docs)} unique documents")

        # Score each document based on fuzzy matches
        scored_results = []
        query_lower = query.lower()

        for doc_key, point in seen_docs.items():
            doc_id, dtype = doc_key
            title = point.payload.get("title", "")
            excerpt = point.payload.get("excerpt", "")

            # Check title match
            title_score = self._calculate_char_overlap(query_lower, title.lower())

            # Check excerpt match
            excerpt_score = self._calculate_char_overlap(query_lower, excerpt.lower())

            # Use best score
            best_score = max(title_score, excerpt_score)

            if best_score >= threshold:
                match_location = "title" if title_score >= excerpt_score else "excerpt"
                scored_results.append(
                    {
                        "doc_id": doc_id,
                        "doc_type": dtype,
                        "title": title,
                        "excerpt": excerpt
                        if excerpt_score >= title_score
                        else f"Title match: {title}",
                        "score": best_score,
                        "match_location": match_location,
                    }
                )

        # Sort by score (descending) and limit
        scored_results.sort(key=lambda x: x["score"], reverse=True)
        top_results = scored_results[:limit]

        # Return unverified results (verification happens at output stage)
        final_results = []
        for result in top_results:
            final_results.append(
                SearchResult(
                    id=result["doc_id"],
                    doc_type=result["doc_type"],
                    title=result["title"],
                    excerpt=result["excerpt"],
                    score=result["score"],
                    metadata={"match_location": result["match_location"]},
                )
            )

        logger.info(f"Fuzzy search returned {len(final_results)} unverified results")
        if final_results:
            result_details = [
                f"{r.doc_type}_{r.id} (score={r.score:.3f}, title='{r.title}')"
                for r in final_results[:5]
            ]
            logger.debug(f"Top fuzzy results: {', '.join(result_details)}")

        return final_results

    def _calculate_char_overlap(self, query: str, text: str) -> float:
        """Calculate character overlap ratio between query and text.

        Args:
            query: Query string (normalized)
            text: Text to compare (normalized)

        Returns:
            Overlap ratio (0.0-1.0)
        """
        if not query or not text:
            return 0.0

        # Convert to character sets
        query_chars = set(query)
        text_chars = set(text)

        # Calculate overlap
        overlap = query_chars & text_chars
        overlap_ratio = len(overlap) / len(query_chars)

        return overlap_ratio

    def _extract_excerpt(self, content: str, max_length: int = 200) -> str:
        """Extract excerpt from content.

        Args:
            content: Full document content
            max_length: Maximum excerpt length

        Returns:
            Excerpt string
        """
        if not content:
            return ""

        excerpt = content[:max_length].strip()
        if len(content) > max_length:
            excerpt += "..."

        return excerpt