1339 lines
64 KiB
Python
1339 lines
64 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
|
||
|
||
# P11: partial date YYYY-MM pattern, expanded to full month range
|
||
_PARTIAL_DATE_RE = re.compile(r'^\d{4}-\d{2}$')
|
||
_LAST_DAY = {1:31, 2:29, 3:31, 4:30, 5:31, 6:30, 7:31, 8:31, 9:30, 10:31, 11:30, 12:31}
|
||
|
||
|
||
def _is_leap_year(year: int) -> bool:
|
||
"""Check if year is a leap year."""
|
||
return (year % 4 == 0 and year % 100 != 0) or (year % 400 == 0)
|
||
|
||
|
||
def _validate_date_str(s: str) -> bool:
|
||
"""Validate that a string is a valid calendar date: YYYY or YYYY-MM-DD."""
|
||
if s.isdigit() and len(s) == 4:
|
||
return True
|
||
if len(s) == 10 and s[4] == '-' and s[7] == '-' and s.replace('-', '').isdigit():
|
||
try:
|
||
y, m, d = int(s[:4]), int(s[5:7]), int(s[8:10])
|
||
if m < 1 or m > 12 or d < 1:
|
||
return False
|
||
last = _LAST_DAY.get(m, 31)
|
||
if m == 2 and last == 29 and not _is_leap_year(y):
|
||
last = 28
|
||
return d <= last
|
||
except (ValueError, IndexError):
|
||
return False
|
||
return False
|
||
|
||
|
||
def _expand_partial_date(text: str) -> tuple[str, str]:
|
||
"""Expand YYYY-MM to full month range (YYYY-MM-01 to YYYY-MM-last_day)."""
|
||
date_from = f"{text}-01"
|
||
y, m = text.split("-")
|
||
month = int(m)
|
||
if month < 1 or month > 12:
|
||
# P12: invalid month → return year-only range
|
||
return f"{y}-01-01", f"{y}-12-31"
|
||
last_day = _LAST_DAY.get(month, 31)
|
||
if month == 2 and last_day == 29 and not _is_leap_year(int(y)):
|
||
last_day = 28
|
||
date_to = f"{y}-{m}-{last_day}"
|
||
return date_from, date_to
|
||
|
||
|
||
def _normalize_field_label(raw: str) -> str | None:
|
||
"""Normalize raw PubMed field label to internal field name. (P12)"""
|
||
if raw in ("MH:NOEXP", "MESH:NOEXP"):
|
||
return "MH"
|
||
if raw in _FIELD_TAG_MAP:
|
||
return _FIELD_TAG_MAP[raw]
|
||
if raw in _SPECIAL_FIELDS:
|
||
return raw
|
||
return None
|
||
|
||
|
||
# ─── 查询复杂度限制 ───
|
||
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",
|
||
"ALL": "all", # P3-7: [ALL] = 全部字段
|
||
"MESH": "MH", # P1-2: [MESH] 是 [MH] 的别名
|
||
# 新增标量字段
|
||
"LA": "language",
|
||
"VI": "volume",
|
||
"IP": "issue",
|
||
"PG": "pages",
|
||
"LID": "lid",
|
||
# P4 新增字段标签
|
||
"Title/Abstract": "all", # [Title/Abstract] 长标签 → all
|
||
"TITLE/ABSTRACT": "all", # 解析器 .upper() 后的大写版本
|
||
"OAB": "all", # [OAB] Other Abstract → all
|
||
"WORD": "all", # [WORD] Word in text → all
|
||
"FI": "GR", # [FI] Funder Identifier → 同 GR（grant_id）
|
||
"SO": "journal", # [SO] Source → journal（近似）
|
||
"PL": "journal", # [PL] Place of Publication → journal（近似）
|
||
# P20: PubMed 长格式字段标签
|
||
"TITLE": "title",
|
||
"ABSTRACT": "abstract",
|
||
"ALL FIELDS": "all",
|
||
"MESH TERMS": "MH",
|
||
"MESH MAJOR TOPIC": "MAJR",
|
||
"TEXT WORD": "all",
|
||
"LANGUAGE": "language",
|
||
"AUTHOR": "author",
|
||
"JOURNAL": "journal",
|
||
"AFFILIATION": "affiliation",
|
||
"PUBLICATION DATE": "DP",
|
||
"SUBSTANCE NAME": "NM",
|
||
"GRANT NUMBER": "GR",
|
||
"PHARMACOLOGICAL ACTION": "PA",
|
||
"MESH SUBHEADING": "SH",
|
||
"PUBLICATION TYPE": "PT",
|
||
"DATE COMPLETED": "DCOM",
|
||
"DATE CREATED": "CRDT",
|
||
"DATE MESH CREATED": "MHDA",
|
||
"ENTRY DATE": "EDAT",
|
||
"LAST REVISED": "LR",
|
||
"DATE REVISED": "LR",
|
||
"DATE OF ELECTRONIC PUBLICATION": "DEP",
|
||
"SECONDARY SOURCE ID": "SI",
|
||
"SUBSET": "SB",
|
||
"STATUS": "STAT",
|
||
"TRANSLITERATED TITLE": "TT",
|
||
"VERNACULAR TITLE": "TT",
|
||
"OTHER TERM": "OT",
|
||
"GENE SYMBOL": "GEN",
|
||
"PMC ID": "PMC",
|
||
"VOLUME": "volume",
|
||
"ISSUE": "issue",
|
||
"PAGINATION": "pages",
|
||
"PERSONAL NAME AS SUBJECT": "PS",
|
||
"INVESTIGATOR": "IR",
|
||
"CONFLICT OF INTEREST STATEMENT": "COIS",
|
||
"AUTHOR IDENTIFIER": "AUID",
|
||
}
|
||
|
||
# 需要特殊处理的字段（不直接映射到 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",
|
||
# P4 新增：独立语义字段
|
||
"OT", # [OT] → keywords JSONB（不再映射到 all）
|
||
"GEN", # [GEN] → gene_symbols JSONB
|
||
"PMC", # [PMC] → pmc_id
|
||
}
|
||
|
||
# 支持日期范围语法的字段
|
||
_DATE_RANGE_FIELDS = {"DP", "EDAT", "CRDT", "MHDA", "LR", "DCOM", "DEP"}
|
||
|
||
# 所有合法字段标签（PubMed 全量字段）
|
||
# P1-1: 移除了 BOOK/FILTER/ISBN（未实现，降级为 plain text 不如报错透明）
|
||
_ALL_FIELD_TAGS = {
|
||
"AB", "AD", "ALL", "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", "MESH: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",
|
||
# P4 新增字段标签
|
||
"Title/Abstract", # [Title/Abstract] 长标签
|
||
"OAB", "WORD", # [OAB] Other Abstract, [WORD] Word in text
|
||
"FI", "GEN", "PMC", "SO", "PL", # [FI] Funder, [GEN] Gene, [PMC] PMCID, [SO] Source, [PL] Place
|
||
# P20: PubMed 长格式字段标签
|
||
"Title", "Abstract", "All Fields",
|
||
"MeSH Terms", "MeSH Major Topic",
|
||
"Text Word", "Language", "Author", "Journal", "Affiliation",
|
||
"Publication Date", "Substance Name", "Grant Number",
|
||
"Pharmacological Action", "MeSH Subheading", "Publication Type",
|
||
"Date Completed", "Date Created", "Date MeSH Created", "Entry Date",
|
||
"Last Revised", "Date Revised",
|
||
"Date of Electronic Publication",
|
||
"Secondary Source ID", "Subset", "Status",
|
||
"Transliterated Title", "Vernacular Title",
|
||
"Other Term", "Gene Symbol", "PMC ID",
|
||
"Volume", "Issue", "Pagination",
|
||
"Personal Name as Subject", "Investigator",
|
||
"Conflict of Interest Statement", "Author Identifier",
|
||
}
|
||
|
||
|
||
# ─── 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] = []
|
||
last_end = 0
|
||
for m in _TOKEN_RE.finditer(query):
|
||
# P5: 检测未匹配的字符（不在任何 token 模式中的字符被静默丢弃）
|
||
if m.start() > last_end:
|
||
gap = query[last_end:m.start()]
|
||
if gap.strip():
|
||
tokens.append(Token(TokenType.WORD, gap.strip()))
|
||
if len(tokens) > MAX_TERMS:
|
||
raise ParseError(f"查询词过多（超过 {MAX_TERMS} 个），降级为简单文本搜索")
|
||
last_end = m.end()
|
||
for name, value in m.groupdict().items():
|
||
if value is not None:
|
||
ttype = TokenType[name]
|
||
if ttype == TokenType.UNKNOWN_FIELD:
|
||
stripped = value.strip('[]')
|
||
# P20: skip empty brackets like `[]`
|
||
if not stripped:
|
||
continue
|
||
tokens.append(Token(TokenType.WORD, stripped))
|
||
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} 个），降级为简单文本搜索")
|
||
# R23-1: capture trailing characters after last token
|
||
if last_end < len(query):
|
||
gap = query[last_end:]
|
||
if gap.strip():
|
||
tokens.append(Token(TokenType.WORD, gap.strip()))
|
||
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 统一
|
||
# P4 新增独立语义字段
|
||
ot_terms: list[Term] = field(default_factory=list) # [OT] → keywords JSONB
|
||
gene_terms: list[Term] = field(default_factory=list) # [GEN] → gene_symbols JSONB
|
||
pmc_terms: list[Term] = field(default_factory=list) # [PMC] → pmc_id
|
||
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)
|
||
group_negated: list[bool] = field(default_factory=list) # P16: True if group was wrapped by NOT (external negation)
|
||
sub_group_refs: list[list[int]] = field(default_factory=list) # P20: parent_gid → [child_gid, ...] for AND sub-groups
|
||
negated_date_ranges: set[str] = field(default_factory=set) # date fields negated by NOT
|
||
_date_range_markers: list[Term] = field(default_factory=list, repr=False) # internal: date range Term collectors
|
||
_top_level_date_fields: set[str] = field(default_factory=set, repr=False) # date fields with ungrouped terms
|
||
# R26: separate negated date bounds (NOT year[DP], NOT yyyy:mm[DP]) that should
|
||
# produce independent NOT conditions instead of contaminating the positive range.
|
||
# Keyed by field tag ("DP", "EDAT", etc.), value is list of (from_str, to_str) tuples.
|
||
_neg_single_dates: dict[str, list[tuple[str | None, str | None]]] = field(default_factory=dict)
|
||
|
||
|
||
# ─── 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 top-level (depth=0) token stream only
|
||
# P11: 括号内的 AND/OR 不应影响顶层布尔操作符判定
|
||
depth = 0
|
||
has_and = has_or = False
|
||
for t in self.tokens:
|
||
if t.type == TokenType.LPAREN:
|
||
depth += 1
|
||
elif t.type == TokenType.RPAREN:
|
||
depth -= 1
|
||
elif depth == 0:
|
||
if t.type == TokenType.AND:
|
||
has_and = True
|
||
elif t.type == TokenType.OR:
|
||
has_or = True
|
||
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]
|
||
# P12: has_not 同时检查分组内 NOT（如 NOT (a OR b)）
|
||
# R23-3: 也检查 group_negated（NOT 包组时 is_not 被还原，group_negated 真正标记）
|
||
result.has_not = any(t.is_not for t in _ungrouped) or any(
|
||
t.is_not for g in result.groups for t in g
|
||
) or any(result.group_negated) or any(t.is_not for t in result._date_range_markers)
|
||
result.not_terms = [t for t in _ungrouped if t.is_not]
|
||
for t in _ungrouped:
|
||
self._dispatch_term(result, t)
|
||
# P25: track date fields with top-level (ungrouped) terms for De Morgan handling
|
||
if t.field in _DATE_RANGE_FIELDS:
|
||
result._top_level_date_fields.add(t.field)
|
||
|
||
# R27: also track top-level date range markers (excluded from _ungrouped by _is_range_end filter)
|
||
for t in terms:
|
||
if getattr(t, '_is_range_end', False) and t.group_id < 0:
|
||
_tag = t.field.replace("__RANGE_", "").replace("__", "")
|
||
if _tag in _DATE_RANGE_FIELDS:
|
||
result._top_level_date_fields.add(_tag)
|
||
|
||
# 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, TokenType.DATE):
|
||
text = t.value.strip('"') if t.type == TokenType.QUOTED else t.value
|
||
result.plain_terms.append(Term(text=text, exact=(t.type == TokenType.QUOTED)))
|
||
|
||
# Recompute negated_date_ranges from TOP-LEVEL marker terms only.
|
||
# R21: use |= not = to preserve negated_date_ranges added by _dispatch_term single-date NOTs
|
||
# R28: exclude group-scoped markers (t.group_id >= 0) — engine's group loop handles them.
|
||
result.negated_date_ranges |= {
|
||
t.field.replace("__RANGE_", "").replace("__", "")
|
||
for t in result._date_range_markers if t.is_not and t.group_id < 0
|
||
}
|
||
|
||
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)
|
||
# ── 独立日期字段（非范围语法）："2024-01-01"[DP] → from=to=该日期 ──
|
||
# 纯 4 位年份 "2024"[DP] 用 year_from/year_to，避免 fromisoformat 问题
|
||
# is_not 时加入 negated_date_ranges，引擎据此 NOT 条件
|
||
elif term.field == "DP":
|
||
if term.text.isdigit() and len(term.text) == 4:
|
||
y = int(term.text)
|
||
if term.is_not:
|
||
result._neg_single_dates.setdefault("DP", []).append((f"{y}-01-01", f"{y}-12-31"))
|
||
else:
|
||
result.year_from = max(result.year_from, y) if result.year_from is not None else y
|
||
result.year_to = min(result.year_to, y) if result.year_to is not None else y
|
||
elif _PARTIAL_DATE_RE.match(term.text):
|
||
df, dt = _expand_partial_date(term.text)
|
||
if term.is_not:
|
||
result._neg_single_dates.setdefault("DP", []).append((df, dt))
|
||
else:
|
||
result.date_from = max(result.date_from, df) if result.date_from is not None else df
|
||
result.date_to = min(result.date_to, dt) if result.date_to is not None else dt
|
||
else:
|
||
if _validate_date_str(term.text):
|
||
df = dt = term.text
|
||
if term.is_not:
|
||
result._neg_single_dates.setdefault("DP", []).append((df, dt))
|
||
else:
|
||
result.date_from = max(result.date_from, df) if result.date_from is not None else df
|
||
result.date_to = min(result.date_to, dt) if result.date_to is not None else dt
|
||
else:
|
||
result.plain_terms.append(term)
|
||
return
|
||
elif term.field == "EDAT":
|
||
if term.text.isdigit() and len(term.text) == 4:
|
||
y = int(term.text)
|
||
_f = f"{y}-01-01"
|
||
_t = f"{y}-12-31"
|
||
if term.is_not:
|
||
result._neg_single_dates.setdefault("EDAT", []).append((_f, _t))
|
||
else:
|
||
result.edat_from = max(result.edat_from, _f) if result.edat_from is not None else _f
|
||
result.edat_to = min(result.edat_to, _t) if result.edat_to is not None else _t
|
||
elif _PARTIAL_DATE_RE.match(term.text):
|
||
df, dt = _expand_partial_date(term.text)
|
||
if term.is_not:
|
||
result._neg_single_dates.setdefault("EDAT", []).append((df, dt))
|
||
else:
|
||
result.edat_from = max(result.edat_from, df) if result.edat_from is not None else df
|
||
result.edat_to = min(result.edat_to, dt) if result.edat_to is not None else dt
|
||
else:
|
||
if _validate_date_str(term.text):
|
||
if term.is_not:
|
||
result._neg_single_dates.setdefault("EDAT", []).append((term.text, term.text))
|
||
else:
|
||
result.edat_from = max(result.edat_from, term.text) if result.edat_from is not None else term.text
|
||
result.edat_to = min(result.edat_to, term.text) if result.edat_to is not None else term.text
|
||
else:
|
||
result.plain_terms.append(term)
|
||
return
|
||
elif term.field == "CRDT":
|
||
if term.text.isdigit() and len(term.text) == 4:
|
||
y = int(term.text)
|
||
_f = f"{y}-01-01"
|
||
_t = f"{y}-12-31"
|
||
if term.is_not:
|
||
result._neg_single_dates.setdefault("CRDT", []).append((_f, _t))
|
||
else:
|
||
result.crdt_from = max(result.crdt_from, _f) if result.crdt_from is not None else _f
|
||
result.crdt_to = min(result.crdt_to, _t) if result.crdt_to is not None else _t
|
||
elif _PARTIAL_DATE_RE.match(term.text):
|
||
df, dt = _expand_partial_date(term.text)
|
||
if term.is_not:
|
||
result._neg_single_dates.setdefault("CRDT", []).append((df, dt))
|
||
else:
|
||
result.crdt_from = max(result.crdt_from, df) if result.crdt_from is not None else df
|
||
result.crdt_to = min(result.crdt_to, dt) if result.crdt_to is not None else dt
|
||
else:
|
||
if _validate_date_str(term.text):
|
||
if term.is_not:
|
||
result._neg_single_dates.setdefault("CRDT", []).append((term.text, term.text))
|
||
else:
|
||
result.crdt_from = max(result.crdt_from, term.text) if result.crdt_from is not None else term.text
|
||
result.crdt_to = min(result.crdt_to, term.text) if result.crdt_to is not None else term.text
|
||
else:
|
||
result.plain_terms.append(term)
|
||
return
|
||
elif term.field == "MHDA":
|
||
if term.text.isdigit() and len(term.text) == 4:
|
||
y = int(term.text)
|
||
_f = f"{y}-01-01"
|
||
_t = f"{y}-12-31"
|
||
if term.is_not:
|
||
result._neg_single_dates.setdefault("MHDA", []).append((_f, _t))
|
||
else:
|
||
result.mhda_from = max(result.mhda_from, _f) if result.mhda_from is not None else _f
|
||
result.mhda_to = min(result.mhda_to, _t) if result.mhda_to is not None else _t
|
||
elif _PARTIAL_DATE_RE.match(term.text):
|
||
df, dt = _expand_partial_date(term.text)
|
||
if term.is_not:
|
||
result._neg_single_dates.setdefault("MHDA", []).append((df, dt))
|
||
else:
|
||
result.mhda_from = max(result.mhda_from, df) if result.mhda_from is not None else df
|
||
result.mhda_to = min(result.mhda_to, dt) if result.mhda_to is not None else dt
|
||
else:
|
||
if _validate_date_str(term.text):
|
||
if term.is_not:
|
||
result._neg_single_dates.setdefault("MHDA", []).append((term.text, term.text))
|
||
else:
|
||
result.mhda_from = max(result.mhda_from, term.text) if result.mhda_from is not None else term.text
|
||
result.mhda_to = min(result.mhda_to, term.text) if result.mhda_to is not None else term.text
|
||
else:
|
||
result.plain_terms.append(term)
|
||
return
|
||
elif term.field == "LR":
|
||
if term.text.isdigit() and len(term.text) == 4:
|
||
y = int(term.text)
|
||
_f = f"{y}-01-01"
|
||
_t = f"{y}-12-31"
|
||
if term.is_not:
|
||
result._neg_single_dates.setdefault("LR", []).append((_f, _t))
|
||
else:
|
||
result.lr_from = max(result.lr_from, _f) if result.lr_from is not None else _f
|
||
result.lr_to = min(result.lr_to, _t) if result.lr_to is not None else _t
|
||
elif _PARTIAL_DATE_RE.match(term.text):
|
||
df, dt = _expand_partial_date(term.text)
|
||
if term.is_not:
|
||
result._neg_single_dates.setdefault("LR", []).append((df, dt))
|
||
else:
|
||
result.lr_from = max(result.lr_from, df) if result.lr_from is not None else df
|
||
result.lr_to = min(result.lr_to, dt) if result.lr_to is not None else dt
|
||
else:
|
||
if _validate_date_str(term.text):
|
||
if term.is_not:
|
||
result._neg_single_dates.setdefault("LR", []).append((term.text, term.text))
|
||
else:
|
||
result.lr_from = max(result.lr_from, term.text) if result.lr_from is not None else term.text
|
||
result.lr_to = min(result.lr_to, term.text) if result.lr_to is not None else term.text
|
||
else:
|
||
result.plain_terms.append(term)
|
||
return
|
||
elif term.field == "DCOM":
|
||
if term.text.isdigit() and len(term.text) == 4:
|
||
y = int(term.text)
|
||
_f = f"{y}-01-01"
|
||
_t = f"{y}-12-31"
|
||
if term.is_not:
|
||
result._neg_single_dates.setdefault("DCOM", []).append((_f, _t))
|
||
else:
|
||
result.dcom_from = max(result.dcom_from, _f) if result.dcom_from is not None else _f
|
||
result.dcom_to = min(result.dcom_to, _t) if result.dcom_to is not None else _t
|
||
elif _PARTIAL_DATE_RE.match(term.text):
|
||
df, dt = _expand_partial_date(term.text)
|
||
if term.is_not:
|
||
result._neg_single_dates.setdefault("DCOM", []).append((df, dt))
|
||
else:
|
||
result.dcom_from = max(result.dcom_from, df) if result.dcom_from is not None else df
|
||
result.dcom_to = min(result.dcom_to, dt) if result.dcom_to is not None else dt
|
||
else:
|
||
if _validate_date_str(term.text):
|
||
if term.is_not:
|
||
result._neg_single_dates.setdefault("DCOM", []).append((term.text, term.text))
|
||
else:
|
||
result.dcom_from = max(result.dcom_from, term.text) if result.dcom_from is not None else term.text
|
||
result.dcom_to = min(result.dcom_to, term.text) if result.dcom_to is not None else term.text
|
||
else:
|
||
result.plain_terms.append(term)
|
||
return
|
||
elif term.field == "DEP":
|
||
if term.text.isdigit() and len(term.text) == 4:
|
||
y = int(term.text)
|
||
_f = f"{y}-01-01"
|
||
_t = f"{y}-12-31"
|
||
if term.is_not:
|
||
result._neg_single_dates.setdefault("DEP", []).append((_f, _t))
|
||
else:
|
||
result.dep_from = max(result.dep_from, _f) if result.dep_from is not None else _f
|
||
result.dep_to = min(result.dep_to, _t) if result.dep_to is not None else _t
|
||
elif _PARTIAL_DATE_RE.match(term.text):
|
||
df, dt = _expand_partial_date(term.text)
|
||
if term.is_not:
|
||
result._neg_single_dates.setdefault("DEP", []).append((df, dt))
|
||
else:
|
||
result.dep_from = max(result.dep_from, df) if result.dep_from is not None else df
|
||
result.dep_to = min(result.dep_to, dt) if result.dep_to is not None else dt
|
||
else:
|
||
if _validate_date_str(term.text):
|
||
if term.is_not:
|
||
result._neg_single_dates.setdefault("DEP", []).append((term.text, term.text))
|
||
else:
|
||
result.dep_from = max(result.dep_from, term.text) if result.dep_from is not None else term.text
|
||
result.dep_to = min(result.dep_to, term.text) if result.dep_to is not None else term.text
|
||
else:
|
||
result.plain_terms.append(term)
|
||
return
|
||
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)
|
||
# P4 新增独立语义字段
|
||
elif term.field == "OT":
|
||
result.ot_terms.append(term)
|
||
elif term.field == "GEN":
|
||
result.gene_terms.append(term)
|
||
elif term.field == "PMC":
|
||
result.pmc_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)*
|
||
|
||
P7-D3: Preserve AND clusters when OR is mixed (e.g. ``A OR B AND C``).
|
||
Collect each ``_parse_and_expr`` result as a separate cluster. When OR
|
||
is found AND a cluster has >1 term (implicit AND), wrap it in a group
|
||
so the engine produces ``A OR (B AND C)`` instead of ``A OR B OR C``.
|
||
"""
|
||
clusters = [self._parse_and_expr(result)]
|
||
had_or = False
|
||
while self.peek().type == TokenType.OR:
|
||
had_or = True
|
||
self.advance()
|
||
if self.peek().type == TokenType.EOF:
|
||
break # trailing OR, ignore silently
|
||
clusters.append(self._parse_and_expr(result))
|
||
|
||
if not had_or:
|
||
return clusters[0] if clusters else []
|
||
|
||
# OR present: group any AND-cluster with >1 term
|
||
all_terms: list[Term] = []
|
||
for cluster in clusters:
|
||
if len(cluster) > 1 and not any(t.group_id >= 0 for t in cluster):
|
||
gid = len(result.groups)
|
||
for t in cluster:
|
||
t.group_id = gid
|
||
result.groups.append(cluster)
|
||
result.group_operators.append("and")
|
||
result.group_negated.append(False) # P19: keep lengths aligned
|
||
result.sub_group_refs.append([]) # P20: keep lengths aligned
|
||
elif len(cluster) > 1 and any(t.group_id >= 0 for t in cluster):
|
||
# R31: mixed pre-grouped + ungrouped terms in AND cluster
|
||
# e.g. (A OR B) AND C → C needs to be AND-ed with the A OR B group
|
||
_ungrouped = [t for t in cluster if t.group_id < 0]
|
||
_child_gids = sorted(set(t.group_id for t in cluster if t.group_id >= 0))
|
||
if _ungrouped:
|
||
gid = len(result.groups)
|
||
for t in _ungrouped:
|
||
t.group_id = gid
|
||
result.groups.append(_ungrouped)
|
||
result.group_operators.append("and")
|
||
result.group_negated.append(False)
|
||
while len(result.sub_group_refs) < gid:
|
||
result.sub_group_refs.append([])
|
||
result.sub_group_refs.append(_child_gids if _child_gids else [])
|
||
all_terms.extend(cluster)
|
||
return all_terms
|
||
|
||
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)
|
||
left = list(left) # P12: copy to prevent mutation of shared group list
|
||
while True:
|
||
tok = self.peek()
|
||
if tok.type == TokenType.AND:
|
||
self.advance()
|
||
if self.peek().type == TokenType.EOF:
|
||
break
|
||
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, _not_depth: int = 0) -> list[Term]:
|
||
"""not_expr → NOT not_expr | primary"""
|
||
if _not_depth > MAX_PAREN_DEPTH:
|
||
raise ParseError(f"NOT 嵌套过深（超过 {MAX_PAREN_DEPTH} 层），降级为简单文本搜索")
|
||
if self.peek().type == TokenType.NOT:
|
||
self.advance()
|
||
# P5: trailing NOT at end of input → ignore silently (avoid IndexError peeking past EOF)
|
||
if self.peek().type == TokenType.EOF:
|
||
return []
|
||
inner = self._parse_not_expr(result, _not_depth + 1)
|
||
for t in inner:
|
||
t.is_not = not t.is_not
|
||
# P19: NOT (A OR B) should negate the group, not individual terms.
|
||
# When all inner terms belong to groups, revert per-term toggles
|
||
# and set group_negated[gid] = True instead.
|
||
if inner and all(t.group_id >= 0 for t in inner):
|
||
for t in inner:
|
||
t.is_not = not t.is_not # revert
|
||
seen = set()
|
||
for t in inner:
|
||
gid = t.group_id
|
||
if gid >= 0 and gid not in seen:
|
||
seen.add(gid)
|
||
if gid < len(result.group_negated):
|
||
result.group_negated[gid] = not result.group_negated[gid]
|
||
return inner
|
||
return self._parse_primary(result, negated=(_not_depth % 2 == 1))
|
||
|
||
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
|
||
# P2-F12: (a OR b)[TI] — 组后字段标签应用到组内所有词
|
||
if self.peek().type == TokenType.FIELD:
|
||
ft = self.advance()
|
||
_raw_field = ft.value[1:-1].upper()
|
||
_field = _normalize_field_label(_raw_field)
|
||
for t in terms:
|
||
if not getattr(t, '_is_range_end', False): # R28: skip date range markers
|
||
t.field = _field
|
||
# P17: (lung OR breast)[MH:NOEXP] — 将 _noexp 传播到组内词
|
||
if _raw_field in ("MH:NOEXP", "MESH:NOEXP"):
|
||
for t in terms:
|
||
t._noexp = True
|
||
# P15-PRIMARY: 如果 _parse_or_expr 已为 AND 集群（如 A OR B AND C → [B, C] sub-group）
|
||
# 或嵌套括号创建了子组，则这些子组已正确处理结构。
|
||
# 只对尚未分组的词创建外层组，避免 Term 被放入两个组导致 _pubmed_conditions 重复处理。
|
||
_parent_gid = len(result.groups)
|
||
_ungrouped = [t for t in terms if t.group_id < 0]
|
||
# P20: track sub-groups created by _parse_or_expr inside this paren
|
||
# R30: exclude gids already in existing sub_group_refs (transitive children)
|
||
# and gids that are themselves parent groups (have sub_group_refs)
|
||
_existing_children = set(g for refs in result.sub_group_refs for g in refs)
|
||
_child_candidates = sorted(set(t.group_id for t in terms if t.group_id >= 0))
|
||
_child_gids = [
|
||
gid for gid in _child_candidates
|
||
if gid not in _existing_children
|
||
and (gid >= len(result.sub_group_refs) or not result.sub_group_refs[gid])
|
||
]
|
||
for t in _ungrouped:
|
||
t.group_id = _parent_gid
|
||
if _ungrouped:
|
||
# P20: ensure sub_group_refs is aligned with groups
|
||
while len(result.sub_group_refs) < _parent_gid:
|
||
result.sub_group_refs.append([])
|
||
if _child_gids:
|
||
result.sub_group_refs.append(_child_gids)
|
||
else:
|
||
result.sub_group_refs.append([])
|
||
# R21: only write sub_group_refs when creating a parent group,
|
||
# preventing phantom entries when _ungrouped is empty (all terms already in child groups)
|
||
result.groups.append(_ungrouped)
|
||
# R30: track paren depth to avoid counting OR inside nested parens
|
||
_or_depth = 0
|
||
_has_or = False
|
||
for _t in self.tokens[start_pos:end_pos]:
|
||
if _t.type == TokenType.LPAREN:
|
||
_or_depth += 1
|
||
elif _t.type == TokenType.RPAREN:
|
||
_or_depth -= 1
|
||
elif _t.type == TokenType.OR and _or_depth == 0:
|
||
_has_or = True
|
||
break
|
||
result.group_operators.append("or" if _has_or else "and")
|
||
result.group_negated.append(False) # P16: P19 revert logic handles NOT group tracking
|
||
# 已分组的 Term（嵌套括号 OR 子组）不再重复加组
|
||
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)
|
||
|
||
# R29: open-ended start range :2024[DP] → COLON NUMBER/DATE FIELD
|
||
if t0.type == TokenType.COLON and t1 is not None and t1.type in (TokenType.NUMBER, TokenType.DATE):
|
||
self.advance() # consume COLON
|
||
e_val = self.advance().value # end value
|
||
return self._handle_date_range_edge(result, negated, start_val=None, end_val=e_val)
|
||
|
||
# R29: open-ended end range 2024:[DP] → NUMBER/DATE COLON FIELD
|
||
if (t0.type in (TokenType.NUMBER, TokenType.DATE) and t1 is not None and t1.type == TokenType.COLON
|
||
and t2 is not None and t2.type == TokenType.FIELD):
|
||
s_val = self.advance().value # start value
|
||
self.advance() # consume COLON
|
||
return self._handle_date_range_edge(result, negated, start_val=s_val, end_val=None)
|
||
|
||
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 — only consume if peek is a valid atomic token
|
||
t0 = self.peek()
|
||
if t0.type not in (TokenType.WORD, TokenType.QUOTED, TokenType.NUMBER, TokenType.DATE):
|
||
return []
|
||
token = self.advance()
|
||
text = token.value.strip('"') if token.type == TokenType.QUOTED else token.value
|
||
# R21: skip empty quoted text ""[TI]; R27: also skip whitespace-only
|
||
if not text or not text.strip():
|
||
return []
|
||
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] / [MESH:noexp] → 抑制树展开
|
||
if raw in ("MH:NOEXP", "MESH: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=False, _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()
|
||
# R23-2: normalize field label for non-date ranges too
|
||
_norm = _normalize_field_label(field)
|
||
if _norm is not None:
|
||
field = _norm
|
||
|
||
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]）
|
||
# P5: also handle mixed types (e.g. 2026:2024-01-01 → 2024-01-01:2026)
|
||
_start_is_digit = start_val.isdigit()
|
||
_end_is_digit = end_val.isdigit()
|
||
if _start_is_digit and _end_is_digit and int(start_val) > int(end_val):
|
||
start_val, end_val = end_val, start_val
|
||
elif not _start_is_digit and not _end_is_digit and start_val > end_val:
|
||
start_val, end_val = end_val, start_val
|
||
elif _start_is_digit and not _end_is_digit:
|
||
try:
|
||
if int(start_val) > int(end_val[:4]):
|
||
start_val, end_val = end_val, start_val
|
||
except (ValueError, TypeError):
|
||
pass
|
||
elif not _start_is_digit and _end_is_digit:
|
||
try:
|
||
if int(start_val[:4]) > int(end_val):
|
||
start_val, end_val = end_val, start_val
|
||
except (ValueError, TypeError):
|
||
pass
|
||
# P12: normalize compact YYYYMMDD dates → YYYY-MM-DD
|
||
if start_val.isdigit() and len(start_val) == 8:
|
||
try:
|
||
start_val = f"{start_val[:4]}-{start_val[4:6]}-{start_val[6:8]}"
|
||
except IndexError:
|
||
pass
|
||
if end_val.isdigit() and len(end_val) == 8:
|
||
try:
|
||
end_val = f"{end_val[:4]}-{end_val[4:6]}-{end_val[6:8]}"
|
||
except IndexError:
|
||
pass
|
||
# P12: validate date values — non-numeric garbage falls back to plain text
|
||
# R27: expand partial dates YYYY-MM in ranges before validation
|
||
if _PARTIAL_DATE_RE.match(start_val):
|
||
start_val, _ = _expand_partial_date(start_val)
|
||
if _PARTIAL_DATE_RE.match(end_val):
|
||
_, end_val = _expand_partial_date(end_val)
|
||
_valid_date = lambda s: _validate_date_str(s)
|
||
if not _valid_date(start_val) or not _valid_date(end_val):
|
||
txt = f"{start_val}:{end_val}[{field}]"
|
||
return [Term(txt, field=None, is_not=False)]
|
||
# 确定两端是否是 4 位年份
|
||
_start_is_year = start_val.isdigit() and len(start_val) == 4
|
||
_end_is_year = end_val.isdigit() and len(end_val) == 4
|
||
# Year-only range (e.g., 2024:2026[EDAT])
|
||
# R27: only set global attributes for top-level ranges (depth == 0)
|
||
# Ranges inside groups are handled via markers in the engine's group path.
|
||
if self._depth == 0:
|
||
if _start_is_year and _end_is_year:
|
||
if negated:
|
||
# R26: store in _neg_single_dates instead of main fields
|
||
result._neg_single_dates.setdefault(field, []).append(
|
||
(f"{start_val}-01-01", f"{end_val}-12-31")
|
||
)
|
||
else:
|
||
try:
|
||
if yr_from_attr:
|
||
curr_f = getattr(result, yr_from_attr)
|
||
new_f = int(start_val)
|
||
setattr(result, yr_from_attr, max(curr_f, new_f) if curr_f is not None else new_f)
|
||
curr_t = getattr(result, yr_to_attr)
|
||
new_t = int(end_val)
|
||
setattr(result, yr_to_attr, min(curr_t, new_t) if curr_t is not None else new_t)
|
||
else:
|
||
# For non-DP date fields: convert year to full date for consistency
|
||
curr_f = getattr(result, date_attr)
|
||
new_f = f"{start_val}-01-01"
|
||
setattr(result, date_attr, max(curr_f, new_f) if curr_f is not None else new_f)
|
||
curr_t = getattr(result, date_attr_to)
|
||
new_t = f"{end_val}-12-31"
|
||
setattr(result, date_attr_to, min(curr_t, new_t) if curr_t is not None else new_t)
|
||
except (ValueError, TypeError):
|
||
pass
|
||
elif _start_is_year and not _end_is_year:
|
||
# Mixed: start is year, end is full date (e.g., 2024:2024-12-01[EDAT])
|
||
if negated:
|
||
result._neg_single_dates.setdefault(field, []).append(
|
||
(f"{start_val}-01-01", end_val)
|
||
)
|
||
else:
|
||
if yr_from_attr:
|
||
try:
|
||
curr = getattr(result, yr_from_attr)
|
||
v = int(start_val)
|
||
setattr(result, yr_from_attr, max(curr, v) if curr is not None else v)
|
||
except (ValueError, TypeError):
|
||
pass
|
||
curr_f = getattr(result, date_attr)
|
||
new_f = f"{start_val}-01-01"
|
||
setattr(result, date_attr, max(curr_f, new_f) if curr_f is not None else new_f)
|
||
curr_t = getattr(result, date_attr_to)
|
||
setattr(result, date_attr_to, min(curr_t, end_val) if curr_t is not None else end_val)
|
||
elif not _start_is_year and _end_is_year:
|
||
# Mixed: start is full date, end is year (e.g., 2024-01-01:2026[EDAT])
|
||
if negated:
|
||
result._neg_single_dates.setdefault(field, []).append(
|
||
(start_val, f"{end_val}-12-31")
|
||
)
|
||
else:
|
||
curr_f = getattr(result, date_attr)
|
||
setattr(result, date_attr, max(curr_f, start_val) if curr_f is not None else start_val)
|
||
curr_t = getattr(result, date_attr_to)
|
||
v = f"{end_val}-12-31"
|
||
setattr(result, date_attr_to, min(curr_t, v) if curr_t is not None else v)
|
||
else:
|
||
# Full date range (e.g., 2024-01-01:2024-12-31[EDAT])
|
||
if negated:
|
||
result._neg_single_dates.setdefault(field, []).append((start_val, end_val))
|
||
else:
|
||
curr_f = getattr(result, date_attr)
|
||
setattr(result, date_attr, max(curr_f, start_val) if curr_f is not None else start_val)
|
||
curr_t = getattr(result, date_attr_to)
|
||
setattr(result, date_attr_to, min(curr_t, end_val) if curr_t is not None else end_val)
|
||
marker = Term(f"{start_val}:{end_val}", field=marker_field, is_not=False)
|
||
marker._is_range_end = True
|
||
result._date_range_markers.append(marker)
|
||
return [marker]
|
||
|
||
# Non-date range or no field → plain text
|
||
txt = f"{start_val}:{end_val}"
|
||
# R29: field tag is already stored in Term.field — do not append to text
|
||
return [Term(txt, field=field or "", is_not=False)]
|
||
|
||
# ─── R29: Open-ended date range helper ───
|
||
def _handle_date_range_edge(self, result: ParsedPubmedQuery, negated: bool = False,
|
||
start_val: str | None = None, end_val: str | None = None) -> list[Term]:
|
||
"""Handle open-ended ranges like :2024[DP] or 2024:[DP]."""
|
||
field = None
|
||
if self.peek().type == TokenType.FIELD:
|
||
ft = self.advance()
|
||
field = ft.value[1:-1].upper()
|
||
_norm = _normalize_field_label(field)
|
||
if _norm is not None:
|
||
field = _norm
|
||
|
||
if field not in _DATE_RANGE_FIELDS:
|
||
txt = f"{start_val or ''}:{end_val or ''}"
|
||
return [Term(txt, field=field or "", is_not=False)]
|
||
|
||
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__"),
|
||
}
|
||
date_attr, date_attr_to, yr_from_attr, yr_to_attr, marker_field = attr_map[field]
|
||
|
||
if self._depth == 0:
|
||
if start_val is not None and end_val is None:
|
||
# Open-ended end: 2024:[DP] → from start_val onwards
|
||
if start_val.isdigit() and len(start_val) == 4:
|
||
if yr_from_attr is not None:
|
||
if negated:
|
||
result._neg_single_dates.setdefault(field, []).append(
|
||
(f"{start_val}-01-01", None))
|
||
else:
|
||
curr = getattr(result, yr_from_attr)
|
||
v = int(start_val)
|
||
setattr(result, yr_from_attr, max(curr, v) if curr is not None else v)
|
||
else:
|
||
if negated:
|
||
result._neg_single_dates.setdefault(field, []).append(
|
||
(f"{start_val}-01-01", None))
|
||
else:
|
||
curr = getattr(result, date_attr)
|
||
v = f"{start_val}-01-01"
|
||
setattr(result, date_attr, max(curr, v) if curr is not None else v)
|
||
else:
|
||
if _PARTIAL_DATE_RE.match(start_val):
|
||
start_val, _ = _expand_partial_date(start_val)
|
||
if negated:
|
||
result._neg_single_dates.setdefault(field, []).append((start_val, None))
|
||
else:
|
||
curr = getattr(result, date_attr)
|
||
setattr(result, date_attr, max(curr, start_val) if curr is not None else start_val)
|
||
elif end_val is not None and start_val is None:
|
||
# Open-ended start: :2024[DP] → up to end_val
|
||
if end_val.isdigit() and len(end_val) == 4:
|
||
if yr_to_attr is not None:
|
||
if negated:
|
||
result._neg_single_dates.setdefault(field, []).append(
|
||
(None, f"{end_val}-12-31"))
|
||
else:
|
||
curr = getattr(result, yr_to_attr)
|
||
v = int(end_val)
|
||
setattr(result, yr_to_attr, min(curr, v) if curr is not None else v)
|
||
else:
|
||
if negated:
|
||
result._neg_single_dates.setdefault(field, []).append(
|
||
(None, f"{end_val}-12-31"))
|
||
else:
|
||
curr = getattr(result, date_attr_to)
|
||
v = f"{end_val}-12-31"
|
||
setattr(result, date_attr_to, min(curr, v) if curr is not None else v)
|
||
else:
|
||
if _PARTIAL_DATE_RE.match(end_val):
|
||
_, end_val = _expand_partial_date(end_val)
|
||
if negated:
|
||
result._neg_single_dates.setdefault(field, []).append((None, end_val))
|
||
else:
|
||
curr = getattr(result, date_attr_to)
|
||
setattr(result, date_attr_to, min(curr, end_val) if curr is not None else end_val)
|
||
|
||
marker = Term(f"{start_val or ''}:{end_val or ''}", field=marker_field, is_not=False)
|
||
marker._is_range_end = True
|
||
result._date_range_markers.append(marker)
|
||
return [marker]
|
||
|
||
|
||
# ─── Public API ───
|
||
|
||
def is_pubmed_syntax(query: str) -> bool:
|
||
"""快速检测查询串是否包含 PubMed 语法特征。
|
||
|
||
检测条件（任意满足其一）：
|
||
1. 包含 `[...]`（字段标签）
|
||
2. 包含布尔运算符 AND/OR/NOT（大写，词边界）
|
||
"""
|
||
if not query or not query.strip():
|
||
return False
|
||
# P5: Normalize fullwidth characters before checking
|
||
query = unicodedata.normalize('NFKC', query)
|
||
if re.search(r'\[(' + '|'.join(_ALL_FIELD_TAGS) + r')\]', query, re.IGNORECASE | re.ASCII):
|
||
return True
|
||
if re.search(r'\b(AND|OR|NOT)\b', query, re.IGNORECASE | re.ASCII):
|
||
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/M/D 格式的日期分隔符统一为 YYYY-MM-DD，使 tokeniser 正确识别为 DATE
|
||
# Anchored with context boundaries to avoid over-matching inside URLs/paths
|
||
query = re.sub(
|
||
r'(^|[\[\s":])(\d{4})/(\d{1,2})/(\d{1,2})(?=\s|$|[\[\]":])',
|
||
lambda m: f'{m.group(1)}{m.group(2)}-{int(m.group(3)):02d}-{int(m.group(4)):02d}',
|
||
query,
|
||
)
|
||
# P5: Normalize single-digit month/day (2024-1-1 → 2024-01-01) to match DATE token pattern
|
||
# Anchored with context boundaries to avoid over-matching inside non-date text
|
||
query = re.sub(
|
||
r'(^|[\[\s":])(\d{4})-(\d{1,2})-(\d{1,2})(?=\s|$|[\[\]":])',
|
||
lambda m: f'{m.group(1)}{m.group(2)}-{int(m.group(3)):02d}-{int(m.group(4)):02d}',
|
||
query,
|
||
)
|
||
# P5: Normalize YYYY-MM (partial month) to YYYY-MM-01 when followed by date field tag
|
||
# R23-3: support single-digit month (2024-1[DP] → 2024-01-01)
|
||
# The \s*\[ lookahead prevents false match on YYYY-MM-DD sequences
|
||
query = re.sub(
|
||
r'(\b\d{4})-(\d{1,2})(?=\s*\[(?:DP|EDAT|DEP|CRDT|MHDA|LR|DCOM)\])',
|
||
lambda m: f'{m.group(1)}-{int(m.group(2)):02d}',
|
||
query,
|
||
)
|
||
tokens = tokenise(query)
|
||
parser = PubmedQueryParser(tokens)
|
||
return parser.parse()
|
||
except (ParseError, IndexError, ValueError):
|
||
# P0-1: 降级时清理查询中的 [field] 标签、布尔符、引号和括号
|
||
degraded = ParsedPubmedQuery()
|
||
import re as _degrade_re
|
||
_clean = _degrade_re.sub(r'\[[\w/: -]+\]', '', query)
|
||
_clean = _degrade_re.sub(r'\b(AND|OR|NOT)\b', '', _clean)
|
||
_clean = _clean.replace('"', '').replace('(', '').replace(')', '')
|
||
for t in _clean.split():
|
||
if t.strip():
|
||
degraded.plain_terms.append(Term(text=t.strip()))
|
||
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()
|
||
|
||
# P5: Handle field tags with '/' (e.g. Title/Article) or special chars
|
||
normalized = re.sub(
|
||
r'\[([\w/: -]+)\]',
|
||
lambda m: f'[{m.group(1).upper()}]',
|
||
query,
|
||
)
|
||
return normalized, mesh_used
|