34047007@qq.com
723c4fc5c9
Batch1 — 解析器 (pubmed_query_parser.py) - P0-1: 未知字段标签降级为 WORD token 而非 ParseError - P2-1: 增加未消费 token 检查 - P2-2: PRISMA 字段标签正则 [\w:+] → [\w:]+ - P2-3: Unicode NFKC 规格化输入 - P2-4: re.ASCII 防止 Unicode 数字匹配 - P2-9: 移除重复 dataclass 字段 Batch2 — 搜索引擎 (search_engine.py) - P1-1: 批量 PMID 查询替代 N+1 循环 - P1-2: isdigit() → isdecimal() - P1-5: 移除 precision_mode 参数 - P2-5: 移除死代码 - P2-6: 统一 _CHINESE_RE 正则 Batch3 — ATM 引擎 (query_expansion.py) - P0-4: name_zh ILIKE 中文回退 + _find_mesh_tags 中文降级 - P1-3: name_en ILIKE 加 LIMIT 100 Batch4 — API 层 (features.py) - P1-5: 移除 precision_mode 请求字段 - P1-11: 增加 logging - P2-8: NLM_SUBSET_LABELS f-string 安全注释 - P2-12: split 校验器近似性注释 Batch5 — SearchView.vue - P0-5a/b/c: 日期修复（UTC 方法、互斥逻辑、restoreFromQuery 合并） - P2-10: 筛选模态关闭时重搜 - P3-1: 搜索框 aria-label Batch6 — HomeView.vue - P1-10: URL date → date_from/date_to - P2-11: clearSearch 清空 feedItems 并重加载 Batch7 — LiteratureCard.vue - P1-6: 字段标签正则 [\w-]+ → [\w:-]+ - P1-7: terms 切片限制 20 项防 ReDoS 后修复: - query_expansion.py _find_partial_mesh_tags UnboundLocalError（单非中文词未初始化 tag_ids） - HomeView.vue handleAdvancedSearch precision_mode 残留引用
617 lines
25 KiB
Python
617 lines
25 KiB
Python
"""PubMed 查询语法解析器
|
||
|
||
将用户输入的 PubMed 风格查询（如 `"lung cancer"[TI] AND "pembrolizumab"[TIAB]`）
|
||
解析为结构化数据，供 AdvancedSearchEngine.search() 消费。
|
||
|
||
支持的语法：
|
||
- 字段标签 (case-insensitive): [TI] [AB] [TIAB] [AU] [TA] [MH] [MAJR] [PT] [DP] [PMID] [DOI]
|
||
- 布尔运算符: AND / OR / NOT（大小写不敏感，前后需词边界）
|
||
- 引号短语: "exact phrase"
|
||
- 括号分组: (a OR b) AND c
|
||
- 日期范围: 2024:2026[DP]
|
||
|
||
设计原则：
|
||
- 向后兼容：不含特殊语法的纯文本查询原样返回为 plain_terms
|
||
- 容错：语法错误时退化到 plain_terms，不抛异常
|
||
- 组合：解析结果映射到 AdvancedSearchEngine.search() 的已有参数
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
import unicodedata
|
||
from dataclasses import dataclass, field
|
||
from enum import Enum, auto
|
||
|
||
# ─── 查询复杂度限制 ───
|
||
MAX_TERMS = 100 # P4-1: 放宽到 100 词（原 50 词）
|
||
MAX_PAREN_DEPTH = 10 # 括号嵌套最深层数
|
||
|
||
|
||
# ─── 字段标签映射 ───
|
||
|
||
_FIELD_TAG_MAP: dict[str, str] = {
|
||
"TI": "title",
|
||
"AB": "abstract",
|
||
"TIAB": "all",
|
||
"AU": "author",
|
||
"TA": "journal",
|
||
"JT": "journal", # P1-2: Journal Title (同 TA)
|
||
"AD": "affiliation",
|
||
"CN": "author",
|
||
"FAU": "author",
|
||
"LAU": "author",
|
||
"TW": "all",
|
||
"OT": "all",
|
||
"MESH": "MH", # P1-2: [MESH] 是 [MH] 的别名
|
||
# 新增标量字段
|
||
"LA": "language",
|
||
"VI": "volume",
|
||
"IP": "issue",
|
||
"PG": "pages",
|
||
"LID": "lid",
|
||
}
|
||
|
||
# 需要特殊处理的字段（不直接映射到 field 参数）
|
||
_SPECIAL_FIELDS = {
|
||
"MH", "MAJR", "PT", "DP", "PMID", "DOI",
|
||
# 日期字段
|
||
"EDAT", "CRDT", "MHDA", "LR", "DCOM",
|
||
# P1-2: 电子出版日期
|
||
"DEP",
|
||
# JSONB 字段
|
||
"GR", "SH", "RN", "NM", "SI", "PA",
|
||
# 待补充抽取的字段（标记为特殊以便未来实现）
|
||
"AUID", "COIS", "ED", "IR", "PS", "PUBN", "TT",
|
||
# P1-2: 新字段
|
||
"SB", "STAT", "UID",
|
||
}
|
||
|
||
# 支持日期范围语法的字段
|
||
_DATE_RANGE_FIELDS = {"DP", "EDAT", "CRDT", "MHDA", "LR", "DCOM", "DEP"}
|
||
|
||
# 所有合法字段标签（PubMed 全量字段）
|
||
# P1-1: 移除了 BOOK/FILTER/ISBN（未实现，降级为 plain text 不如报错透明）
|
||
_ALL_FIELD_TAGS = {
|
||
"AB", "AD", "AU", "CN", "FAU", "AUID", "LAU", "COIS",
|
||
"DCOM", "CRDT", "EDAT", "MHDA", "LR", "DP", "DOI",
|
||
"DEP", # P1-2: Date of Electronic Publication
|
||
"RN", "ED", "GR", "IR", "IP",
|
||
"TA", "JT", "LA", "LID", "MAJR", "SH", "MH", "MH:NOEXP", "OT", "PG",
|
||
"PA", "PT", "PMID", "PUBN", "SI", "PS", "NM", "TW",
|
||
"SB", "STAT", "UID", # P1-2: Subset, Status, UID
|
||
"MESH", # P1-2: [MH] 别名
|
||
"TI", "TIAB", "TT", "VI",
|
||
}
|
||
|
||
|
||
# ─── Token Types ───
|
||
|
||
class TokenType(Enum):
|
||
QUOTED = auto() # "text in quotes"
|
||
FIELD = auto() # [TI], [MH], etc.
|
||
AND = auto() # AND operator
|
||
OR = auto() # OR operator
|
||
NOT = auto() # NOT operator
|
||
LPAREN = auto() # (
|
||
RPAREN = auto() # )
|
||
COLON = auto() # :
|
||
WORD = auto() # unquoted text
|
||
NUMBER = auto() # digits (for PMID, year)
|
||
DATE = auto() # YYYY-MM-DD (normalized from YYYY/MM/DD)
|
||
UNKNOWN_FIELD = auto() # [...] 但内容不是已知字段标签
|
||
EOF = auto() # end of input
|
||
|
||
|
||
@dataclass
|
||
class Token:
|
||
type: TokenType
|
||
value: str = ""
|
||
|
||
|
||
# ─── Tokeniser ───
|
||
|
||
_TOKEN_PATTERNS: list[tuple[TokenType, str]] = [
|
||
(TokenType.QUOTED, r'"(?:[^"\\]|\\.)*"'),
|
||
(TokenType.FIELD, r'\[(?:' + '|'.join(_ALL_FIELD_TAGS) + r')\]'),
|
||
(TokenType.UNKNOWN_FIELD, r'\[[^\]]*\]'), # 未识别的字段标签 → 触发降级
|
||
(TokenType.AND, r'\bAND\b'),
|
||
(TokenType.OR, r'\bOR\b'),
|
||
(TokenType.NOT, r'\bNOT\b'),
|
||
(TokenType.LPAREN, r'\('),
|
||
(TokenType.RPAREN, r'\)'),
|
||
(TokenType.COLON, r':'),
|
||
(TokenType.DATE, r'\d{4}-\d{2}-\d{2}'),
|
||
(TokenType.NUMBER, r'\d+'),
|
||
(TokenType.WORD, r'[^\s"\[\]():]+'),
|
||
]
|
||
|
||
_TOKEN_RE = re.compile(
|
||
'|'.join(f'(?P<{t.name}>{p})' for t, p in _TOKEN_PATTERNS),
|
||
re.IGNORECASE | re.ASCII,
|
||
)
|
||
|
||
|
||
def tokenise(query: str) -> list[Token]:
|
||
"""将 PubMed 查询字符串分片为 Token 列表。"""
|
||
tokens: list[Token] = []
|
||
for m in _TOKEN_RE.finditer(query):
|
||
for name, value in m.groupdict().items():
|
||
if value is not None:
|
||
ttype = TokenType[name]
|
||
if ttype == TokenType.UNKNOWN_FIELD:
|
||
fname = value.strip('[]').upper()
|
||
# P0-1: 不认识字段标签时降级为 WORD，不终止解析
|
||
tokens.append(Token(TokenType.WORD, value.strip('[]')))
|
||
if len(tokens) > MAX_TERMS:
|
||
raise ParseError(f"查询词过多（超过 {MAX_TERMS} 个），降级为简单文本搜索")
|
||
continue
|
||
tokens.append(Token(ttype, value))
|
||
if len(tokens) > MAX_TERMS:
|
||
raise ParseError(f"查询词过多（超过 {MAX_TERMS} 个），降级为简单文本搜索")
|
||
tokens.append(Token(TokenType.EOF))
|
||
return tokens
|
||
|
||
|
||
# ─── AST Nodes ───
|
||
|
||
@dataclass
|
||
class Term:
|
||
"""单个搜索词项。"""
|
||
text: str
|
||
exact: bool = False # True if quoted phrase
|
||
field: str | None = None # None = plain (no field tag); "title"/"abstract"/etc for mapped; "MH"/"PT"/etc for special
|
||
is_not: bool = False # True if preceded by NOT
|
||
group_id: int = -1 # -1 = top-level, >= 0 = parenthesized group index
|
||
_noexp: bool = False # P1-4: [MH:noexp] 抑制树展开
|
||
|
||
|
||
@dataclass
|
||
class ParsedPubmedQuery:
|
||
"""PubMed 查询解析后的结构化表示。"""
|
||
title_terms: list[Term] = field(default_factory=list) # [TI]
|
||
abstract_terms: list[Term] = field(default_factory=list) # [AB]
|
||
tiab_terms: list[Term] = field(default_factory=list) # [TIAB]
|
||
author_terms: list[Term] = field(default_factory=list) # [AU]
|
||
journal_terms: list[Term] = field(default_factory=list) # [TA]
|
||
affiliation_terms: list[Term] = field(default_factory=list) # [AD]
|
||
language_terms: list[Term] = field(default_factory=list) # [LA]
|
||
volume_terms: list[Term] = field(default_factory=list) # [VI]
|
||
issue_terms: list[Term] = field(default_factory=list) # [IP]
|
||
pages_terms: list[Term] = field(default_factory=list) # [PG]
|
||
lid_terms: list[Term] = field(default_factory=list) # [LID]
|
||
date_from: str | None = None # [DP] lower bound (YYYY-MM-DD)
|
||
date_to: str | None = None # [DP] upper bound (YYYY-MM-DD)
|
||
year_from: int | None = None # [DP] year lower
|
||
year_to: int | None = None # [DP] year upper
|
||
# 更多日期字段
|
||
edat_from: str | None = None # [EDAT]
|
||
edat_to: str | None = None
|
||
crdt_from: str | None = None # [CRDT]
|
||
crdt_to: str | None = None
|
||
mhda_from: str | None = None # [MHDA]
|
||
mhda_to: str | None = None
|
||
lr_from: str | None = None # [LR]
|
||
lr_to: str | None = None
|
||
dcom_from: str | None = None # [DCOM]
|
||
dcom_to: str | None = None
|
||
dep_from: str | None = None # P1-2: [DEP]
|
||
dep_to: str | None = None
|
||
# 特殊字段（存储 Term 以保留 is_not 标志）
|
||
mesh_terms: list[Term] = field(default_factory=list) # [MH] (was list[str])
|
||
majr_terms: list[Term] = field(default_factory=list) # [MAJR] (was list[str])
|
||
pub_types: list[Term] = field(default_factory=list) # [PT] (was list[str])
|
||
doi_terms: list[Term] = field(default_factory=list) # [DOI] (was list[str])
|
||
pmid_terms: list[Term] = field(default_factory=list) # [PMID] (was list[int])
|
||
# JSONB 字段
|
||
grant_terms: list[Term] = field(default_factory=list) # [GR]
|
||
subheading_terms: list[Term] = field(default_factory=list) # [SH]
|
||
registry_terms: list[Term] = field(default_factory=list) # [RN]
|
||
substance_terms: list[Term] = field(default_factory=list) # [NM]
|
||
databank_terms: list[Term] = field(default_factory=list) # [SI]
|
||
pharmaco_terms: list[Term] = field(default_factory=list) # [PA]
|
||
ed_terms: list[Term] = field(default_factory=list) # [ED]
|
||
investigator_terms: list[Term] = field(default_factory=list) # [IR]
|
||
personal_name_terms: list[Term] = field(default_factory=list) # [PS]
|
||
pubnote_terms: list[Term] = field(default_factory=list) # [PUBN]
|
||
auid_terms: list[Term] = field(default_factory=list) # [AUID]
|
||
cois_terms: list[Term] = field(default_factory=list) # [COIS]
|
||
tt_terms: list[Term] = field(default_factory=list) # [TT]
|
||
sb_terms: list[Term] = field(default_factory=list) # P1-2: [SB] Subset
|
||
stat_terms: list[Term] = field(default_factory=list) # P1-2: [STAT] Status
|
||
uid_terms: list[Term] = field(default_factory=list) # P1-2: [UID] PMID/DOI 统一
|
||
plain_terms: list[Term] = field(default_factory=list) # no field tag
|
||
boolean_operator: str = "and" # "and" | "or" | "mixed"
|
||
has_not: bool = False # contains NOT
|
||
not_terms: list[Term] = field(default_factory=list) # terms under NOT
|
||
groups: list[list[Term]] = field(default_factory=list) # parenthesized sub-groups
|
||
group_operators: list[str] = field(default_factory=list) # "and"/"or" per group (P2-2)
|
||
negated_date_ranges: set[str] = field(default_factory=set) # date fields negated by NOT
|
||
|
||
|
||
# ─── Parser ───
|
||
|
||
class ParseError(ValueError):
|
||
"""查询语法解析错误（调用方应退化到 plain text 搜索）。"""
|
||
|
||
|
||
class PubmedQueryParser:
|
||
"""Recursive descent parser for PubMed query syntax.
|
||
|
||
Grammar:
|
||
query → or_expr
|
||
or_expr → and_expr (OR and_expr)*
|
||
and_expr → not_expr (AND not_expr)*
|
||
not_expr → NOT not_expr | primary
|
||
primary → atom FIELD? | LPAREN query RPAREN
|
||
atom → QUOTED | WORD | number (COLON number)?
|
||
"""
|
||
|
||
def __init__(self, tokens: list[Token]):
|
||
self.tokens = tokens
|
||
self.pos = 0
|
||
|
||
def peek(self) -> Token:
|
||
return self.tokens[self.pos]
|
||
|
||
def peek_n(self, n: int) -> Token | None:
|
||
idx = self.pos + n
|
||
return self.tokens[idx] if idx < len(self.tokens) else None
|
||
|
||
def advance(self) -> Token:
|
||
t = self.tokens[self.pos]
|
||
self.pos += 1
|
||
return t
|
||
|
||
def expect(self, *types: TokenType) -> Token:
|
||
t = self.peek()
|
||
if t.type not in types:
|
||
raise ParseError(
|
||
f"Expected one of {[tt.name for tt in types]}, got {t.type.name}({t.value!r})"
|
||
)
|
||
return self.advance()
|
||
|
||
def parse(self) -> ParsedPubmedQuery:
|
||
"""入口：解析完整的查询字符串。"""
|
||
result = ParsedPubmedQuery()
|
||
self._depth = 0 # 括号嵌套深度计数器
|
||
terms = self._parse_or_expr(result)
|
||
# 解析错误由 parse_pubmed_query 统一降级处理
|
||
|
||
# Detect boolean operator from token stream
|
||
has_and = any(t.type == TokenType.AND for t in self.tokens)
|
||
has_or = any(t.type == TokenType.OR for t in self.tokens)
|
||
if has_and and has_or:
|
||
result.boolean_operator = "mixed"
|
||
elif has_or and not has_and:
|
||
result.boolean_operator = "or"
|
||
|
||
# 分组词不从 flat lists 走，避免括号内外的词被一起 AND/OR
|
||
# 同时 has_not/not_terms 也只考虑非分组词
|
||
_ungrouped = [t for t in terms if not getattr(t, '_is_range_end', False) and t.group_id < 0]
|
||
result.has_not = any(t.is_not for t in _ungrouped)
|
||
result.not_terms = [t for t in _ungrouped if t.is_not]
|
||
for t in _ungrouped:
|
||
self._dispatch_term(result, t)
|
||
|
||
# P2-1: Handle unconsumed tokens (e.g., orphan text after RPAREN)
|
||
if self.pos < len(self.tokens) - 1:
|
||
for t in self.tokens[self.pos:-1]: # exclude EOF token
|
||
if t.type in (TokenType.WORD, TokenType.QUOTED, TokenType.NUMBER):
|
||
text = t.value.strip('"') if t.type == TokenType.QUOTED else t.value
|
||
result.plain_terms.append(Term(text=text, exact=(t.type == TokenType.QUOTED)))
|
||
|
||
return result
|
||
|
||
def _dispatch_term(self, result: ParsedPubmedQuery, term: Term) -> None:
|
||
if term.field == "title":
|
||
result.title_terms.append(term)
|
||
elif term.field == "abstract":
|
||
result.abstract_terms.append(term)
|
||
elif term.field == "all": # TIAB
|
||
result.tiab_terms.append(term)
|
||
elif term.field == "author":
|
||
result.author_terms.append(term)
|
||
elif term.field == "affiliation":
|
||
result.affiliation_terms.append(term)
|
||
elif term.field == "journal":
|
||
result.journal_terms.append(term)
|
||
elif term.field == "language":
|
||
result.language_terms.append(term)
|
||
elif term.field == "volume":
|
||
result.volume_terms.append(term)
|
||
elif term.field == "issue":
|
||
result.issue_terms.append(term)
|
||
elif term.field == "pages":
|
||
result.pages_terms.append(term)
|
||
elif term.field == "lid":
|
||
result.lid_terms.append(term)
|
||
elif term.field == "MH":
|
||
result.mesh_terms.append(term)
|
||
elif term.field == "MAJR":
|
||
result.majr_terms.append(term)
|
||
elif term.field == "PT":
|
||
result.pub_types.append(term)
|
||
elif term.field == "PMID":
|
||
try:
|
||
int(term.text) # validate
|
||
except ValueError:
|
||
result.plain_terms.append(term)
|
||
return
|
||
result.pmid_terms.append(term)
|
||
elif term.field == "DOI":
|
||
result.doi_terms.append(term)
|
||
elif term.field == "GR":
|
||
result.grant_terms.append(term)
|
||
elif term.field == "SH":
|
||
result.subheading_terms.append(term)
|
||
elif term.field == "RN":
|
||
result.registry_terms.append(term)
|
||
elif term.field == "NM":
|
||
result.substance_terms.append(term)
|
||
elif term.field == "SI":
|
||
result.databank_terms.append(term)
|
||
elif term.field == "PA":
|
||
result.pharmaco_terms.append(term)
|
||
elif term.field is None:
|
||
result.plain_terms.append(term)
|
||
elif term.field == "__RANGE_DP__":
|
||
pass
|
||
elif term.field == "__RANGE_EDAT__":
|
||
pass
|
||
elif term.field == "__RANGE_CRDT__":
|
||
pass
|
||
elif term.field == "__RANGE_MHDA__":
|
||
pass
|
||
elif term.field == "__RANGE_LR__":
|
||
pass
|
||
elif term.field == "__RANGE_DCOM__":
|
||
pass
|
||
elif term.field == "__RANGE_DEP__":
|
||
pass
|
||
elif term.field == "ED":
|
||
result.ed_terms.append(term)
|
||
elif term.field == "IR":
|
||
result.investigator_terms.append(term)
|
||
elif term.field == "PS":
|
||
result.personal_name_terms.append(term)
|
||
elif term.field == "PUBN":
|
||
result.pubnote_terms.append(term)
|
||
elif term.field == "AUID":
|
||
result.auid_terms.append(term)
|
||
elif term.field == "COIS":
|
||
result.cois_terms.append(term)
|
||
elif term.field == "TT":
|
||
result.tt_terms.append(term)
|
||
elif term.field == "SB":
|
||
result.sb_terms.append(term)
|
||
elif term.field == "STAT":
|
||
result.stat_terms.append(term)
|
||
elif term.field == "UID":
|
||
result.uid_terms.append(term)
|
||
else:
|
||
result.plain_terms.append(term)
|
||
|
||
def _parse_or_expr(self, result: ParsedPubmedQuery) -> list[Term]:
|
||
"""or_expr → and_expr (OR and_expr)*"""
|
||
left = self._parse_and_expr(result)
|
||
while self.peek().type == TokenType.OR:
|
||
self.advance()
|
||
right = self._parse_and_expr(result)
|
||
left.extend(right)
|
||
return left
|
||
|
||
def _is_primary_start(self, token: Token) -> bool:
|
||
"""Check if token could start a primary expression."""
|
||
return token.type in (TokenType.WORD, TokenType.QUOTED, TokenType.NUMBER, TokenType.DATE,
|
||
TokenType.LPAREN, TokenType.NOT)
|
||
|
||
def _parse_and_expr(self, result: ParsedPubmedQuery) -> list[Term]:
|
||
"""and_expr → not_expr (AND? not_expr)*
|
||
|
||
AND is optional between consecutive primaries (implicit AND).
|
||
e.g. ``smith j[AU]`` → smith AND j[AU]
|
||
"""
|
||
left = self._parse_not_expr(result)
|
||
while True:
|
||
tok = self.peek()
|
||
if tok.type == TokenType.AND:
|
||
self.advance()
|
||
elif self._is_primary_start(tok):
|
||
pass # implicit AND — continue without consuming
|
||
else:
|
||
break
|
||
right = self._parse_not_expr(result)
|
||
left.extend(right)
|
||
return left
|
||
|
||
def _parse_not_expr(self, result: ParsedPubmedQuery) -> list[Term]:
|
||
"""not_expr → NOT not_expr | primary"""
|
||
if self.peek().type == TokenType.NOT:
|
||
self.advance()
|
||
return self._parse_primary(result, negated=True)
|
||
return self._parse_primary(result, negated=False)
|
||
|
||
def _parse_primary(self, result: ParsedPubmedQuery, negated: bool = False) -> list[Term]:
|
||
"""primary → atom FIELD? | LPAREN query RPAREN"""
|
||
if self.peek().type == TokenType.LPAREN:
|
||
self._depth += 1
|
||
if self._depth > MAX_PAREN_DEPTH:
|
||
raise ParseError(f"括号嵌套过深（超过 {MAX_PAREN_DEPTH} 层），降级为简单文本搜索")
|
||
self.advance()
|
||
start_pos = self.pos # P2-2: 记录组起始标记位置
|
||
terms = self._parse_or_expr(result)
|
||
end_pos = self.pos # P2-2: 记录组结束标记位置
|
||
self.expect(TokenType.RPAREN)
|
||
self._depth -= 1
|
||
# 标记为子组，不放入 flat lists，保留括号分组结构
|
||
group_id = len(result.groups)
|
||
for t in terms:
|
||
t.group_id = group_id
|
||
result.groups.append(terms)
|
||
# P2-2: 检测组内是否有显式 OR
|
||
_has_or = any(t.type == TokenType.OR for t in self.tokens[start_pos:end_pos])
|
||
result.group_operators.append("or" if _has_or else "and")
|
||
if negated:
|
||
for t in terms:
|
||
t.is_not = True
|
||
return terms
|
||
|
||
return self._parse_atom(result, negated)
|
||
|
||
def _parse_atom(self, result: ParsedPubmedQuery, negated: bool = False) -> list[Term]:
|
||
"""Parse a single atom with optional field tag.
|
||
|
||
Handles:
|
||
- "quoted phrase"[FIELD]
|
||
- word[FIELD]
|
||
- NUMBER:NUMBER[DP] (year/date range)
|
||
- DATE:DATE[DP] (full date range like 2024-01-01:2024-12-31)
|
||
- NUMBER (bare number)
|
||
- word (bare word)
|
||
"""
|
||
# Look ahead for range pattern (NUMBER:NUMBER or DATE:DATE)
|
||
t0 = self.peek()
|
||
t1 = self.peek_n(1)
|
||
t2 = self.peek_n(2)
|
||
|
||
if (t1 is not None and t1.type == TokenType.COLON
|
||
and t2 is not None
|
||
and t0.type in (TokenType.NUMBER, TokenType.DATE, TokenType.WORD)
|
||
and t2.type in (TokenType.NUMBER, TokenType.DATE, TokenType.WORD)):
|
||
return self._parse_range(result, negated)
|
||
|
||
# Normal atom
|
||
token = self.advance()
|
||
text = token.value.strip('"') if token.type == TokenType.QUOTED else token.value
|
||
is_exact = (token.type == TokenType.QUOTED)
|
||
field = None
|
||
_noexp = False # P1-4
|
||
if self.peek().type == TokenType.FIELD:
|
||
ft = self.advance()
|
||
raw = ft.value[1:-1].upper()
|
||
# P1-4: [MH:noexp] → 抑制树展开
|
||
if raw == "MH:NOEXP":
|
||
field = "MH"
|
||
_noexp = True
|
||
else:
|
||
field = raw
|
||
if field in _FIELD_TAG_MAP:
|
||
field = _FIELD_TAG_MAP[field]
|
||
|
||
return [Term(text, exact=is_exact, field=field, is_not=negated, _noexp=_noexp)]
|
||
|
||
def _parse_range(self, result: ParsedPubmedQuery, negated: bool = False) -> list[Term]:
|
||
"""Parse NUMBER:NUMBER[FIELD] — handles date ranges specially."""
|
||
start_val = self.advance().value # NUMBER
|
||
self.advance() # COLON
|
||
end_val = self.advance().value # NUMBER
|
||
|
||
field = None
|
||
if self.peek().type == TokenType.FIELD:
|
||
ft = self.advance()
|
||
field = ft.value[1:-1].upper()
|
||
|
||
if field in _DATE_RANGE_FIELDS:
|
||
attr_map = {
|
||
"DP": ("date_from", "date_to", "year_from", "year_to", "__RANGE_DP__"),
|
||
"EDAT": ("edat_from", "edat_to", None, None, "__RANGE_EDAT__"),
|
||
"CRDT": ("crdt_from", "crdt_to", None, None, "__RANGE_CRDT__"),
|
||
"MHDA": ("mhda_from", "mhda_to", None, None, "__RANGE_MHDA__"),
|
||
"LR": ("lr_from", "lr_to", None, None, "__RANGE_LR__"),
|
||
"DCOM": ("dcom_from", "dcom_to", None, None, "__RANGE_DCOM__"),
|
||
"DEP": ("dep_from", "dep_to", None, None, "__RANGE_DEP__"), # P1-2
|
||
}
|
||
date_attr, date_attr_to, yr_from_attr, yr_to_attr, marker_field = attr_map[field]
|
||
# 反向范围自动交换（如 2026:2024[DP] → 2024:2026[DP]）
|
||
if start_val.isdigit() and end_val.isdigit() and int(start_val) > int(end_val):
|
||
start_val, end_val = end_val, start_val
|
||
# Year-only range (e.g., 2024:2026[EDAT])
|
||
if start_val.isdigit() and len(start_val) == 4:
|
||
if yr_from_attr:
|
||
setattr(result, yr_from_attr, int(start_val))
|
||
setattr(result, yr_to_attr, int(end_val))
|
||
else:
|
||
# For non-DP date fields: convert year to full date for consistency
|
||
setattr(result, date_attr, f"{start_val}-01-01")
|
||
setattr(result, date_attr_to, f"{end_val}-12-31")
|
||
else:
|
||
# Full date range (e.g., 2024-01-01:2024-12-31[EDAT])
|
||
setattr(result, date_attr, start_val)
|
||
setattr(result, date_attr_to, end_val)
|
||
marker = Term(f"{start_val}:{end_val}", field=marker_field, is_not=negated)
|
||
marker._is_range_end = True
|
||
if negated:
|
||
result.negated_date_ranges.add(field)
|
||
return [marker]
|
||
|
||
# Non-date range or no field → plain text
|
||
txt = f"{start_val}:{end_val}"
|
||
if field:
|
||
txt = f"{txt}[{field}]"
|
||
return [Term(txt, field=field, is_not=negated)]
|
||
|
||
|
||
# ─── Public API ───
|
||
|
||
def is_pubmed_syntax(query: str) -> bool:
|
||
"""快速检测查询串是否包含 PubMed 语法特征。
|
||
|
||
检测条件（任意满足其一）：
|
||
1. 包含 `[...]`（字段标签）
|
||
2. 包含布尔运算符 AND/OR/NOT（大写，词边界）
|
||
"""
|
||
if not query or not query.strip():
|
||
return False
|
||
if re.search(r'\[(' + '|'.join(_ALL_FIELD_TAGS) + r')\]', query, re.IGNORECASE):
|
||
return True
|
||
if re.search(r'\b(AND|OR|NOT)\b', query, re.IGNORECASE):
|
||
return True
|
||
return False
|
||
|
||
|
||
def parse_pubmed_query(query: str) -> ParsedPubmedQuery:
|
||
"""解析 PubMed 查询字符串。
|
||
|
||
如果解析失败（非法语法），返回全部内容为 plain_terms 的空结构。
|
||
"""
|
||
if not query or not query.strip():
|
||
return ParsedPubmedQuery()
|
||
|
||
try:
|
||
# P2-3: Unicode normalization — strip zero-width chars, normalize fullwidth digits
|
||
query = unicodedata.normalize('NFKC', query)
|
||
# 将 YYYY/MM/DD 格式的日期分隔符统一为 YYYY-MM-DD，使 tokeniser 正确识别为 DATE
|
||
import re as _re
|
||
query = _re.sub(r'(\d{4})/(\d{2})/(\d{2})', r'\1-\2-\3', query)
|
||
tokens = tokenise(query)
|
||
parser = PubmedQueryParser(tokens)
|
||
return parser.parse()
|
||
except (ParseError, IndexError, ValueError):
|
||
# P0-1: 降级时返回原始查询作为 plain_terms，不丢失用户输入
|
||
degraded = ParsedPubmedQuery()
|
||
for t in query.strip().split():
|
||
degraded.plain_terms.append(Term(text=t))
|
||
return degraded
|
||
|
||
|
||
def extract_pubmed_query_for_prisma(query: str) -> tuple[str, list[str]]:
|
||
"""提取 PubMed 查询和所用 MeSH 词，供 PRISMA 导出。
|
||
|
||
Returns:
|
||
(normalized_query, mesh_terms_used)
|
||
"""
|
||
parsed = parse_pubmed_query(query)
|
||
mesh_used = list(set(
|
||
[t.text for t in parsed.mesh_terms] + [t.text for t in parsed.majr_terms]
|
||
))
|
||
mesh_used.sort()
|
||
|
||
# 标准化：统一字段大写
|
||
normalized = re.sub(
|
||
r'\[([\w:]+)\]',
|
||
lambda m: f'[{m.group(1).upper()}]',
|
||
query,
|
||
)
|
||
return normalized, mesh_used
|