webmail-client/backend/app/search/queryparser.py

"""Operator query grammar (SPEC §8.2), engine-independent.

Tokenization: whitespace split respecting ``"quoted phrases"``. A ``key:value``
(or ``-key:value``) token with a known key becomes a filter; everything else is
free text. ``-`` negates. ``OR`` (uppercase) groups adjacent clauses into a
disjunction; the default combinator is AND. Unknown ``key:`` tokens are treated
as free text (forward compatibility). The parser never raises.
"""

from __future__ import annotations

from dataclasses import dataclass
from uuid import UUID

from app.search.base import SearchQuery

# Known operator keys (SPEC §8.2). Unknown keys fall through to free text.
KNOWN_KEYS: frozenset[str] = frozenset(
    {
        "from",
        "to",
        "cc",
        "subject",
        "account",
        "in",
        "is",
        "has",
        "filename",
        "before",
        "after",
        "newer_than",
        "older_than",
        "larger",
        "smaller",
    }
)


@dataclass(frozen=True)
class Clause:
    """One parsed unit: a free-text term or a ``key:value`` filter."""

    type: str  # "text" | "filter"
    key: str | None  # operator key for filters, else None
    value: str
    negate: bool
    phrase: bool  # value was a quoted phrase

    def as_dict(self) -> dict[str, object]:
        return {
            "type": self.type,
            "key": self.key,
            "value": self.value,
            "negate": self.negate,
            "phrase": self.phrase,
        }


# A parsed query is an AND of groups; each group is an OR of clauses.
ParsedQuery = list[list[Clause]]


@dataclass(frozen=True)
class _Token:
    text: str
    negate: bool
    starts_with_quote: bool  # a bare quoted phrase -> never a key:value filter


def _tokenize(query: str) -> list[_Token]:
    """Split into tokens honouring quotes; never raises on unbalanced quotes.

    Tracks a leading ``-`` (negation) and whether the token began with a quote
    (a bare phrase), so ``key:"quoted value"`` is still recognised as a filter
    while ``"key: value"`` stays free text.
    """
    tokens: list[_Token] = []
    i, n = 0, len(query)
    while i < n:
        while i < n and query[i].isspace():
            i += 1
        if i >= n:
            break
        negate = False
        if query[i] == "-" and i + 1 < n and not query[i + 1].isspace():
            negate = True
            i += 1
        buf: list[str] = []
        starts_with_quote = i < n and query[i] == '"'
        while i < n and not query[i].isspace():
            if query[i] == '"':
                i += 1
                while i < n and query[i] != '"':
                    buf.append(query[i])
                    i += 1
                if i < n and query[i] == '"':
                    i += 1  # consume closing quote
            else:
                buf.append(query[i])
                i += 1
        tokens.append(_Token(text="".join(buf), negate=negate, starts_with_quote=starts_with_quote))
    return tokens


def _to_clause(token: _Token) -> Clause | None:
    if not token.text and not token.starts_with_quote:
        return None
    if not token.starts_with_quote and ":" in token.text:
        key, _, value = token.text.partition(":")
        if key.lower() in KNOWN_KEYS and value:
            return Clause(
                type="filter",
                key=key.lower(),
                value=value,
                negate=token.negate,
                phrase=(" " in value),
            )
    return Clause(
        type="text",
        key=None,
        value=token.text,
        negate=token.negate,
        phrase=token.starts_with_quote or (" " in token.text),
    )


def parse(query: str) -> ParsedQuery:
    """Parse ``query`` into AND-of-OR groups of :class:`Clause`."""
    groups: ParsedQuery = []
    pending_or = False
    for token in _tokenize(query):
        if token.text == "OR" and not token.negate and not token.starts_with_quote:
            pending_or = True
            continue
        clause = _to_clause(token)
        if clause is None:
            continue
        if pending_or and groups:
            groups[-1].append(clause)
        else:
            groups.append([clause])
        pending_or = False
    return groups


def free_text(groups: ParsedQuery) -> str | None:
    """Return the positive free-text remainder (standalone, non-negated text terms)."""
    terms = [g[0].value for g in groups if len(g) == 1 and g[0].type == "text" and not g[0].negate]
    return " ".join(terms) if terms else None


def to_search_query(query: str, user_id: UUID) -> SearchQuery:
    """Parse ``query`` into a :class:`SearchQuery` (I-7: ``user_id`` always set)."""
    groups = parse(query)
    filters: dict[str, object] = {
        "groups": [[clause.as_dict() for clause in group] for group in groups]
    }
    return SearchQuery(text=free_text(groups), filters=filters, user_id=user_id)