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