fix: 第三轮PubMed搜索审计 — 14项修复（P0×5, P1×4, P3×5）

P0 (搜索结果错误):
- 门控补全 sb/stat/uid_terms + dep_from/dep_to
- [SB] 映射修复: medline→citation_status, pubmed→no-op
- 括号组内单否定词丢失 not_() 修复
- HomeView URL watch 补全 sort/field/retracted/negative
- HomeView→SearchView 日期精度保留（urlDateFrom/urlDateTo）

P1 (功能缺陷):
- [ALL] 字段标签注册（_FIELD_TAG_MAP + _ALL_FIELD_TAGS）
- 独立日期字段 "2024-01-01"[DP] 降级修复（_dispatch_term）
- 浮点日期范围交换（2026-01-01:2024-01-01 → 自动排序）
- page_size URL 恢复 + syncSearchToUrl 输出

P3 (健壮性):
- MeSH 展开 try/catch（expand_atm + _expand_mesh_tag_ids）
- sort/field/boolean 参数验证（field_validator）
- GET /search 查询长度限制（100词）
- query_expansion.py 中文正则同步 [一-鿿㐀-䶿豈-﫿]
- import_mesh_full entry_terms 统一小写（匹配 JSONB @> 精确比较）

测试: 127/127 搜索测试通过，前端构建无报错

backend/app/api/v1/features.py

@@ -87,7 +87,7 @@ class AdvancedSearchRequest(BaseModel):
     medline_only: bool = False
     exclude_preprints: bool = False
 
-    # ── 查询复杂度限制 ──
+    # ── 参数验证 ──
     @field_validator('query')
     @classmethod
     def check_query_complexity(cls, v: str) -> str:
@@ -96,6 +96,27 @@ class AdvancedSearchRequest(BaseModel):
             raise ValueError('查询词过多（最多 100 个词），请简化搜索条件')
         return v
 
+    @field_validator('sort')
+    @classmethod
+    def check_sort(cls, v: str) -> str:
+        if v not in ('date', 'cited', 'relevance', 'first_author', 'journal', 'title'):
+            raise ValueError(f'无效排序方式: {v}')
+        return v
+
+    @field_validator('field')
+    @classmethod
+    def check_field(cls, v: str) -> str:
+        if v not in ('all', 'title', 'abstract', 'author', 'affiliation', 'journal'):
+            raise ValueError(f'无效搜索字段: {v}')
+        return v
+
+    @field_validator('boolean')
+    @classmethod
+    def check_boolean(cls, v: str) -> str:
+        if v not in ('and', 'or'):
+            raise ValueError(f'无效布尔操作符: {v}')
+        return v
+
 
 # ── 筛选选项 ──
 

backend/app/api/v1/literature.py

@@ -257,6 +257,8 @@ async def search_literature(
 ):
     if not q.strip():
         return {"items": [], "total": 0}
+    if len(q.split()) > 100:
+        return {"items": [], "total": 0, "error": "查询词过多（最多 100 个词），请简化搜索条件"}
     offset = (page - 1) * page_size
     like = f"%{q}%"
     # tsvector 主搜索 + ILIKE 兜底

backend/app/services/pubmed_query_parser.py

@@ -43,6 +43,7 @@ _FIELD_TAG_MAP: dict[str, str] = {
     "LAU": "author",
     "TW": "all",
     "OT": "all",
+    "ALL": "all",            # P3-7: [ALL] = 全部字段
     "MESH": "MH",            # P1-2: [MESH] 是 [MH] 的别名
     # 新增标量字段
     "LA": "language",
@@ -73,7 +74,7 @@ _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",
+    "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",
@@ -362,6 +363,28 @@ class PubmedQueryParser:
             result.pharmaco_terms.append(term)
         elif term.field is None:
             result.plain_terms.append(term)
+        # ── 独立日期字段（非范围语法）："2024-01-01"[DP] → from=to=该日期 ──
+        elif term.field == "DP":
+            result.date_from = term.text
+            result.date_to = term.text
+        elif term.field == "EDAT":
+            result.edat_from = term.text
+            result.edat_to = term.text
+        elif term.field == "CRDT":
+            result.crdt_from = term.text
+            result.crdt_to = term.text
+        elif term.field == "MHDA":
+            result.mhda_from = term.text
+            result.mhda_to = term.text
+        elif term.field == "LR":
+            result.lr_from = term.text
+            result.lr_to = term.text
+        elif term.field == "DCOM":
+            result.dcom_from = term.text
+            result.dcom_to = term.text
+        elif term.field == "DEP":
+            result.dep_from = term.text
+            result.dep_to = term.text
         elif term.field == "__RANGE_DP__":
             pass
         elif term.field == "__RANGE_EDAT__":
@@ -536,6 +559,8 @@ class PubmedQueryParser:
             # 反向范围自动交换（如 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
+            elif not start_val.isdigit() and not end_val.isdigit() and start_val > 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:

backend/app/services/query_expansion.py

@@ -95,7 +95,7 @@ async def _find_mesh_tags(db: AsyncSession, query: str) -> list[UUID]:
             seen.add(tid)
 
     # 方法 C: name_zh ILIKE 匹配（P0-4: 中文查询降级）
-    if not tag_ids and re.search(r'[一-鿿]', q):
+    if not tag_ids and re.search(r'[一-鿿㐀-䶿豈-﫿]', q):
         stmt = select(GlobalTag.id).where(
             GlobalTag.source.in_(["mesh", "manual"]),
             GlobalTag.name_zh.ilike(like_pattern),
@@ -117,7 +117,7 @@ async def _find_partial_mesh_tags(db: AsyncSession, query: str) -> list[UUID]:
     words = [w.strip().lower() for w in query.strip().split() if len(w.strip()) >= MIN_QUERY_LENGTH][:10]
     if len(words) < 2:
         # P0-4: 中文查询无法按空格分词，尝试整体 name_zh ILIKE 匹配
-        if re.search(r'[一-鿿]', query):
+        if re.search(r'[一-鿿㐀-䶿豈-﫿]', query):
             tag_ids = []
             stmt = select(GlobalTag.id).where(
                 GlobalTag.source.in_(["mesh", "manual"]),

backend/app/services/search_engine.py

@@ -172,8 +172,9 @@ class AdvancedSearchEngine:
                     or pp.databank_terms or pp.pharmaco_terms
                     or pp.ed_terms or pp.investigator_terms or pp.personal_name_terms
                     or pp.pubnote_terms or pp.auid_terms or pp.cois_terms or pp.tt_terms
+                    or pp.sb_terms or pp.stat_terms or pp.uid_terms
                     or pp.edat_from or pp.crdt_from or pp.mhda_from
-                    or pp.lr_from or pp.dcom_from
+                    or pp.lr_from or pp.dcom_from or pp.dep_from
                     or pp.plain_terms or pp.has_not
                     or pp.year_from or pp.year_to
                 )
@@ -199,8 +200,9 @@ class AdvancedSearchEngine:
                      or _pubmed_parsed.databank_terms or _pubmed_parsed.pharmaco_terms
                      or _pubmed_parsed.ed_terms or _pubmed_parsed.investigator_terms or _pubmed_parsed.personal_name_terms
                      or _pubmed_parsed.pubnote_terms or _pubmed_parsed.auid_terms or _pubmed_parsed.cois_terms or _pubmed_parsed.tt_terms
+                     or _pubmed_parsed.sb_terms or _pubmed_parsed.stat_terms or _pubmed_parsed.uid_terms
                      or _pubmed_parsed.edat_from or _pubmed_parsed.crdt_from or _pubmed_parsed.mhda_from
-                     or _pubmed_parsed.lr_from or _pubmed_parsed.dcom_from
+                     or _pubmed_parsed.lr_from or _pubmed_parsed.dcom_from or _pubmed_parsed.dep_from
                      or _pubmed_parsed.plain_terms or _pubmed_parsed.has_not
                      or _pubmed_parsed.year_from or _pubmed_parsed.year_to)
             ]):
@@ -269,7 +271,10 @@ class AdvancedSearchEngine:
                     _atm_cond = None
                     _atm_query = query.replace('"', '').replace("'", '').strip()
                     if _atm_query and field == "all":
-                        _atm_cond = await _expand_atm(db, _atm_query)
+                        try:
+                            _atm_cond = await _expand_atm(db, _atm_query)
+                        except Exception:
+                            _atm_cond = None
 
                     _cond_before = len(conditions)
                     if boolean == "and":
@@ -627,7 +632,10 @@ class AdvancedSearchEngine:
                 combined = " ".join(t.text for t in pp.plain_terms if not t.is_not).strip()
                 if combined and not re.search(r'[一-鿿㐀-䶿豈-﫿]', combined):
                     from app.services.query_expansion import expand_atm as _expand_atm_inline
-                    atm_cond = await _expand_atm_inline(db, combined)
+                    try:
+                        atm_cond = await _expand_atm_inline(db, combined)
+                    except Exception:
+                        atm_cond = None
                     if atm_cond is not None:
                         text_cond = and_(*plain_conds) if len(plain_conds) > 1 else plain_conds[0]
                         term_conditions.append(or_(atm_cond, text_cond))
@@ -826,16 +834,27 @@ class AdvancedSearchEngine:
                 neg_conds = [GlobalLiterature.vernacular_title.ilike(f"%{_escape_ilike(t.text)}%") for t in neg]
                 term_conditions.append(not_(or_(*neg_conds)))
 
-        # P1-2: [SB] Subset → nlm_subsets（需 journal JOIN 子查询）
+        # P1-2: [SB] Subset
+        #   medline[SB] → citation_status='medline'（记录级）
+        #   pubmed[SB]  → no-op（全部记录都在 PubMed 中）
+        #   单字母代码（AIM/M/S/D/N/Q/T/X）→ nlm_subsets（期刊级）
         if pp.sb_terms:
             pos = [t for t in pp.sb_terms if not t.is_not]
             neg = [t for t in pp.sb_terms if t.is_not]
             for subl, is_neg in [(pos, False), (neg, True)]:
                 for t in subl:
-                    subq = select(GlobalJournal.issn).where(
-                        GlobalJournal.nlm_subsets.overlap([t.text.upper()])
-                    )
-                    cond = GlobalLiterature.journal_issn.in_(subq)
+                    val = t.text.upper()
+                    if val == "PUBMED":
+                        continue  # no-op: 所有记录都是 PubMed
+                    elif val == "MEDLINE":
+                        cond = GlobalLiterature.citation_status == "medline"
+                    elif len(val) == 1 and val.isalpha():
+                        subq = select(GlobalJournal.issn).where(
+                            GlobalJournal.nlm_subsets.overlap([val])
+                        )
+                        cond = GlobalLiterature.journal_issn.in_(subq)
+                    else:
+                        cond = GlobalLiterature.citation_status == val.lower()
                     term_conditions.append(not_(cond) if is_neg else cond)
 
         # P1-2: [STAT] Status → citation_status
@@ -879,7 +898,7 @@ class AdvancedSearchEngine:
                            else "and")
                     combine_fn = or_ if gop == "or" else and_
                     combined = combine_fn(*group_conds) if len(group_conds) > 1 else group_conds[0]
-                    if all_not and len(group_conds) > 1:
+                    if all_not and len(group_conds) >= 1:
                         combined = not_(combined)
                     term_conditions.append(combined)
 
@@ -1135,46 +1154,52 @@ class AdvancedSearchEngine:
             # 1b. name_en ILIKE 回退 — batch via OR
             name_conds.append(GlobalTag.name_en.ilike(m))
 
-        if entry_conds:
-            rows = (await db.execute(
-                select(GlobalTag.id).where(
-                    GlobalTag.source.in_(["mesh", "manual"]),
-                    GlobalTag.mesh_ui.isnot(None),
-                    GlobalTag.entry_terms.isnot(None),
-                    or_(*entry_conds),
-                )
-            )).all()
-            for (tid,) in rows:
-                mesh_tag_ids.add(tid)
+        try:
+            if entry_conds:
+                rows = (await db.execute(
+                    select(GlobalTag.id).where(
+                        GlobalTag.source.in_(["mesh", "manual"]),
+                        GlobalTag.mesh_ui.isnot(None),
+                        GlobalTag.entry_terms.isnot(None),
+                        or_(*entry_conds),
+                    )
+                )).all()
+                for (tid,) in rows:
+                    mesh_tag_ids.add(tid)
 
-        if name_conds:
-            rows = (await db.execute(
-                select(GlobalTag.id).where(
-                    GlobalTag.source.in_(["mesh", "manual"]),
-                    GlobalTag.mesh_ui.isnot(None),
-                    or_(*name_conds),
-                )
-            )).all()
-            for (tid,) in rows:
-                mesh_tag_ids.add(tid)
+            if name_conds:
+                rows = (await db.execute(
+                    select(GlobalTag.id).where(
+                        GlobalTag.source.in_(["mesh", "manual"]),
+                        GlobalTag.mesh_ui.isnot(None),
+                        or_(*name_conds),
+                    )
+                )).all()
+                for (tid,) in rows:
+                    mesh_tag_ids.add(tid)
+        except Exception:
+            pass
 
         if not mesh_tag_ids:
             return None
 
         # tree_number 前缀展开：取匹配 tag 的所有 tree_number，查子节点（[MH:noexp] 时跳过）
         if not noexp:
-            tns = (await db.execute(
-                select(GlobalTagTreeNumber.tree_number).where(
-                    GlobalTagTreeNumber.tag_id.in_(list(mesh_tag_ids))
-                ).distinct()
-            )).scalars().all()
-
-            if tns:
-                child_conds = [GlobalTagTreeNumber.tree_number.like(f"{tn}%") for tn in tns]
-                children = (await db.execute(
-                    select(GlobalTagTreeNumber.tag_id).where(or_(*child_conds))
+            try:
+                tns = (await db.execute(
+                    select(GlobalTagTreeNumber.tree_number).where(
+                        GlobalTagTreeNumber.tag_id.in_(list(mesh_tag_ids))
+                    ).distinct()
                 )).scalars().all()
-                mesh_tag_ids.update(children)
+
+                if tns:
+                    child_conds = [GlobalTagTreeNumber.tree_number.like(f"{tn}%") for tn in tns]
+                    children = (await db.execute(
+                        select(GlobalTagTreeNumber.tag_id).where(or_(*child_conds))
+                    )).scalars().all()
+                    mesh_tag_ids.update(children)
+            except Exception:
+                pass
 
         uids = list(mesh_tag_ids)
         if major_only:

backend/scripts/import_mesh_full.py

@@ -92,8 +92,8 @@ def collect_entry_terms(record) -> list[str]:
     terms: set[str] = set()
     for term_elem in record.findall(".//Term/String"):
         if term_elem.text:
-            term = term_elem.text.strip()
-            if term and term.lower() != descriptor_name:
+            term = term_elem.text.strip().lower()  # P3-7: 统一小写以匹配 JSONB @> 精确比较
+            if term and term != descriptor_name:
                 terms.add(term)
     return sorted(terms)