Blame for metadata_search/boolean_query.py

"""Google Scholar-style boolean query parsing (+required, -exclude, \"phrases\")."""

from __future__ import annotations

import re
from dataclasses import dataclass

from metadata_search.corpus import tokenize

_QUOTED = re.compile(r'"([^"]+)"')
_DISCOVERY_PREFIX = re.compile(
    r"^(?:please\s+)?(?:find|search|show|list|recommend|retrieve|get)\s+"
    r"(?:me\s+)?(?:some\s+)?(?:papers?\s+)?(?:on|about|for|related to)\s+",
    re.I,
)
_BUT_NOT = re.compile(r"\s+but\s+not\s+", re.I)
_BUT_EXCLUDE = re.compile(r"\s+but\s+(?=-)", re.I)
_AND_NOT = re.compile(r"\s+and\s+not\s+", re.I)
_TRAILING_BUT = re.compile(r"\s+but\s*$", re.I)
_NL_NEGATION = re.compile(
    r"(?:,\s*|\s+)(?:but\s+not|not|excluding|without|except)\s+(.+)$",
    re.I,
)
_EXCLUDED_FILLER_PREFIX = re.compile(
    r"^(?:anything|something|any|studies|papers|research|work|articles?)\s+"
    r"(?:on|about|for|regarding|related to)\s+",
    re.I,
)
_EXCLUDED_LEADING_PREP = re.compile(r"^(?:on|about|for|regarding|related to)\s+", re.I)
_TRAILING_FILLER = re.compile(
    r"(?:,\s*)?\b(?:though|tho|however|please|thanks|thank you|just|really|actually|anyway|right|ok|okay|hmm|um|uh|lol)\b[\s,.!?]*$",
    re.I,
)
_LEADING_FILLER = re.compile(
    r"^[\s,.!?]*\b(?:please|just|really|actually|maybe|kind of|sort of)\b\s+",
    re.I,
)
_FILLER_WORDS = frozenset(
    {
        "though",
        "tho",
        "however",
        "please",
        "thanks",
        "just",
        "really",
        "actually",
        "anyway",
        "right",
        "ok",
        "okay",
        "hmm",
        "um",
        "uh",
    }
)


def _strip_conversational_fillers(text: str) -> str:
    """Remove trailing/leading chat filler ('… though', 'please …')."""
    cleaned = text.strip()
    if not cleaned:
        return ""
    for _ in range(5):
        updated = _TRAILING_FILLER.sub("", cleaned).strip(" ,.")
        updated = _LEADING_FILLER.sub("", updated).strip(" ,.")
        if updated == cleaned:
            break
        cleaned = updated
    return cleaned


def _clean_excluded_phrase(phrase: str) -> str:
    """Drop chat filler from NL exclusions: 'anything on drinking water though' → 'drinking water'."""
    text = _strip_conversational_fillers(phrase.strip())
    if not text:
        return ""
    for _ in range(3):
        updated = _EXCLUDED_FILLER_PREFIX.sub("", text).strip()
        updated = _EXCLUDED_LEADING_PREP.sub("", updated).strip()
        updated = _strip_conversational_fillers(updated)
        if updated == text:
            break
        text = updated
    return text


def normalize_search_query(query: str) -> str:
    """Strip chat phrasing (e.g. 'Find papers on …') before boolean parsing."""
    text = (query or "").strip()
    if not text:
        return ""
    text = _DISCOVERY_PREFIX.sub("", text).strip(" .")
    text = _BUT_NOT.sub(" not ", text)
    text = _AND_NOT.sub(" not ", text)
    text = _BUT_EXCLUDE.sub(" ", text)
    text = re.sub(r"\s+", " ", text).strip()
    return text


def has_natural_language_negation(query: str) -> bool:
    text = (query or "").strip()
    return bool(text and _NL_NEGATION.search(text))


def split_natural_language_negation(query: str) -> tuple[str, tuple[str, ...]]:
    """Split trailing ', not X' / ' not X' style constraints from the positive query."""
    text = (query or "").strip()
    if not text:
        return "", ()
    match = _NL_NEGATION.search(text)
    if not match:
        return text, ()
    positive = text[: match.start()].strip(" ,.")
    positive = _TRAILING_BUT.sub("", positive).strip(" ,.")
    excluded = _clean_excluded_phrase(match.group(1).strip(" ,."))
    if not positive:
        positive = text
        excluded = ""
    return positive, (excluded,) if excluded else ()


def has_boolean_syntax(query: str) -> bool:
    text = (query or "").strip()
    if not text:
        return False
    if has_natural_language_negation(text):
        return True
    if _QUOTED.search(text):
        return True
    return bool(re.search(r"(?:^|\s)[+-]", text))


@dataclass(frozen=True)
class BooleanQuery:
    raw: str
    ranking_text: str
    required_phrases: tuple[str, ...] = ()
    required_terms: tuple[str, ...] = ()
    excluded_phrases: tuple[str, ...] = ()
    excluded_terms: tuple[str, ...] = ()
    optional_terms: tuple[str, ...] = ()


def _format_exclusion(phrase: str) -> str:
    """Multi-word exclusions must be quoted or only the first token is excluded."""
    text = phrase.strip()
    if not text:
        return ""
    if " " in text:
        return f'-"{text}"'
    return f"-{text}"


def _format_required(term: str) -> str:
    text = term.strip()
    if not text:
        return ""
    if " " in text:
        return f'"{text}"'
    return f"+{text}"


def _build_effective_raw(

    *,

    required_phrases: list[str],

    required_terms: list[str],

    optional_terms: list[str],

    excluded_phrases: list[str],

    excluded_terms: list[str],

    fallback: str,

) -> str:
    parts: list[str] = []
    for phrase in required_phrases:
        parts.append(f'"{phrase}"')
    for term in required_terms:
        parts.append(_format_required(term))
    for term in optional_terms:
        parts.append(term)
    for phrase in excluded_phrases:
        parts.append(_format_exclusion(phrase))
    for term in excluded_terms:
        parts.append(_format_exclusion(term))
    return " ".join(part for part in parts if part).strip() or fallback


def parse_boolean_query(query: str) -> BooleanQuery:
    """Parse +term, -term, \"phrase\", and natural-language \"not X\" constraints."""
    raw = normalize_search_query(query)
    if not raw:
        return BooleanQuery(raw="", ranking_text="")

    positive_text, nl_excluded_phrases = split_natural_language_negation(raw)

    required_phrases: list[str] = []
    excluded_phrases: list[str] = list(nl_excluded_phrases)
    required_terms: list[str] = []
    excluded_terms: list[str] = []
    optional_terms: list[str] = []
    ranking_parts: list[str] = []

    def _add_term(bucket: list[str], token: str) -> None:
        normalized = token.strip().lower()
        if normalized:
            bucket.append(normalized)

    remaining = positive_text
    for match in _QUOTED.finditer(positive_text):
        phrase = match.group(1).strip()
        if not phrase:
            continue
        start = match.start()
        prefix = positive_text[max(0, start - 1) : start]
        if prefix == "-":
            excluded_phrases.append(_clean_excluded_phrase(phrase))
        else:
            required_phrases.append(phrase)
            ranking_parts.append(phrase)
        remaining = remaining.replace(match.group(0), " ", 1)

    tokens = remaining.split()
    index = 0
    while index < len(tokens):
        token = tokens[index].strip()
        if not token:
            index += 1
            continue
        if token.startswith("+"):
            term = token[1:].strip("+-")
            _add_term(required_terms, term)
            if term:
                ranking_parts.append(term)
            index += 1
            continue
        if token.startswith("-"):
            term = token[1:].strip("+-")
            phrase_parts: list[str] = [term] if term else []
            next_index = index + 1
            while next_index < len(tokens):
                follower = tokens[next_index].strip()
                if not follower or follower.startswith(("+", "-")):
                    break
                phrase_parts.append(follower)
                next_index += 1
            phrase = " ".join(part for part in phrase_parts if part).strip()
            phrase = _clean_excluded_phrase(phrase)
            if " " in phrase:
                excluded_phrases.append(phrase)
            elif phrase:
                _add_term(excluded_terms, phrase)
            index = next_index
            continue
        _add_term(optional_terms, token)
        if token.lower() not in _FILLER_WORDS:
            ranking_parts.append(token)
        index += 1

    excluded_phrases = [_clean_excluded_phrase(phrase) for phrase in excluded_phrases]
    excluded_phrases = [phrase for phrase in excluded_phrases if phrase]
    optional_terms = [term for term in optional_terms if term.lower() not in _FILLER_WORDS]
    ranking_parts = [part for part in ranking_parts if part.lower() not in _FILLER_WORDS]

    ranking_text = " ".join(part for part in ranking_parts if part).strip() or positive_text or raw
    effective_raw = _build_effective_raw(
        required_phrases=required_phrases,
        required_terms=required_terms,
        optional_terms=optional_terms,
        excluded_phrases=excluded_phrases,
        excluded_terms=excluded_terms,
        fallback=raw,
    )
    return BooleanQuery(
        raw=effective_raw,
        ranking_text=ranking_text,
        required_phrases=tuple(required_phrases),
        required_terms=tuple(required_terms),
        excluded_phrases=tuple(excluded_phrases),
        excluded_terms=tuple(excluded_terms),
        optional_terms=tuple(optional_terms),
    )


def _haystack(row: dict[str, object]) -> str:
    return " ".join(
        [
            str(row.get("title") or row.get("TI") or ""),
            str(row.get("abstract") or row.get("full_abstract") or row.get("AB") or ""),
            str(row.get("keywords") or row.get("DE") or ""),
        ]
    ).lower()


def _contains_term(haystack: str, term: str) -> bool:
    normalized = term.lower().strip()
    if not normalized:
        return True
    if " " in normalized:
        return normalized in haystack
    return bool(re.search(rf"\b{re.escape(normalized)}\b", haystack)) or normalized in haystack


def matches_boolean_query(row: dict[str, object], parsed: BooleanQuery) -> bool:
    haystack = _haystack(row)
    for phrase in parsed.required_phrases:
        if phrase.lower() not in haystack:
            return False
    for term in parsed.required_terms:
        if not _contains_term(haystack, term):
            return False
    for phrase in parsed.excluded_phrases:
        if phrase.lower() in haystack:
            return False
    for term in parsed.excluded_terms:
        if _contains_term(haystack, term):
            return False
    return True


def focus_tokens(text: str, *, limit: int = 12) -> list[str]:
    """Tokens for re-ranking agent paper picks (no hardcoded stopword list)."""
    tokens = tokenize(text or "")
    seen: set[str] = set()
    focused: list[str] = []
    for token in tokens:
        if token.startswith(("lemma:", "phrase:", "concept:")):
            continue
        if token in seen or len(token) < 3:
            continue
        seen.add(token)
        focused.append(token)
        if len(focused) >= limit:
            break
    return focused


def plan_summary(parsed: BooleanQuery) -> dict[str, object]:
    return {
        "raw": parsed.raw,
        "ranking_text": parsed.ranking_text,
        "required_phrases": list(parsed.required_phrases),
        "required_terms": list(parsed.required_terms),
        "excluded_terms": list(parsed.excluded_terms),
        "excluded_phrases": list(parsed.excluded_phrases),
        "optional_terms": list(parsed.optional_terms),
    }