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