"""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)