"""高级搜索服务：布尔运算 + 字段限定 + PubMed 查询语法""" from datetime import datetime from sqlalchemy import String, and_, case, cast, func, literal_column, not_, or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.models.literature import GlobalJournal, GlobalLiterature, GlobalLiteratureTag, GlobalTag, GlobalTagTreeNumber from app.schemas.literature import cap_pub_date class AdvancedSearchEngine: """PG 高级搜索（ES 就绪后切换 search_service.py）""" SEARCH_CACHE_TTL = 300 # 秒（夜间流水线更新，5分钟缓存很安全） @staticmethod def _search_cache_key( query: str, field: str, boolean: str, exact_phrase: bool, year_from: int | None, year_to: int | None, date_from: str | None, date_to: str | None, journal_tiers: list[str] | None, pub_types: list[str] | None, study_design: str | None, tag_ids: list[str] | None, retracted: str, negative_result: str, is_oa: bool | None, language: str | None, nlm_subsets: list[str] | None, page: int, page_size: int, sort: str, ) -> str: """归一化查询参数 → 确定性缓存 key（所有 list 排序后参与哈希）""" import hashlib, json norm = { "q": query.strip().lower(), "f": field, "b": boolean, "ep": exact_phrase, "yf": year_from, "yt": year_to, "df": date_from, "dt": date_to, "jt": sorted(journal_tiers) if journal_tiers else [], "pt": sorted(pub_types) if pub_types else [], "sd": study_design, "tid": sorted(tag_ids) if tag_ids else [], "r": retracted, "nr": negative_result, "oa": is_oa, "lang": language, "ns": sorted(nlm_subsets) if nlm_subsets else [], "p": page, "ps": page_size, "s": sort, } raw = json.dumps(norm, sort_keys=True, ensure_ascii=False, default=str) return f"search:advanced:{hashlib.md5(raw.encode()).hexdigest()}" @staticmethod async def search( db: AsyncSession, query: str = "", field: str = "all", boolean: str = "and", exact_phrase: bool = False, year_from: int | None = None, year_to: int | None = None, date_from: str | None = None, # YYYY-MM-DD date_to: str | None = None, # YYYY-MM-DD journal_tiers: list[str] | None = None, pub_types: list[str] | None = None, study_design: str | None = None, # primary category: interventional / observational / synthesis / etc tag_ids: list[str] | None = None, retracted: str = "", # "yes", "no", "only" negative_result: str = "", # "yes", "no", "only" is_oa: bool | None = None, # True = 仅开放获取 language: str | None = None, # 语言代码（en/zh/fr 等） nlm_subsets: list[str] | None = None, # NLM 期刊子集（AIM/M/S 等） page: int = 1, page_size: int = 20, sort: str = "date", cursor_date: str | None = None, # keyset 游标：上一页最后一条的 pub_date cursor_id: str | None = None, # keyset 游标：上一页最后一条的 id（UUID） ) -> dict: """执行高级搜索""" conditions = [] # 60s 缓存（仅 page 模式，cursor 游标不重复） use_cursor = (cursor_date is not None and cursor_id is not None and sort == "date") _cache = None _search_cache_key = None if not use_cursor: from app.core.cache import cache as _cache _search_cache_key = AdvancedSearchEngine._search_cache_key( query, field, boolean, exact_phrase, year_from, year_to, date_from, date_to, journal_tiers, pub_types, study_design, tag_ids, retracted, negative_result, is_oa, language, nlm_subsets, page, page_size, sort, ) cached = await _cache.get(_search_cache_key) if cached is not None: return cached # ─── PubMed 语法检测与解析 ─── _pubmed_parsed = None _is_flat_text = True # 是否走传统 tsvector + ILIKE 路径 if query.strip(): from app.services.pubmed_query_parser import is_pubmed_syntax, parse_pubmed_query if is_pubmed_syntax(query): pp = parse_pubmed_query(query) _pubmed_parsed = pp # 只要解析器产出了有效结构，就走 PubMed 路径 has_pubmed_terms = bool( pp.title_terms or pp.abstract_terms or pp.tiab_terms or pp.author_terms or pp.journal_terms or pp.mesh_terms or pp.majr_terms or pp.pub_types or pp.doi_terms or pp.pmid_terms or pp.plain_terms or pp.has_not or pp.year_from or pp.year_to ) if has_pubmed_terms: _is_flat_text = False conditions = await AdvancedSearchEngine._pubmed_conditions( db, pp, conditions, ) # ─── 传统搜索路径（纯文本 / PubMed 退化） ─── if _is_flat_text and query.strip(): # 中文搜索：自动匹配 GlobalTag.name_zh → 注入 tag_ids，跳过 ILIKE import re as _re _CHINESE_RE = _re.compile(r'[一-鿿]') if _CHINESE_RE.search(query): tag_matches = (await db.execute( select(GlobalTag.id).where(GlobalTag.name_zh.ilike(f'%{query.strip()}%')) )).scalars().all() if tag_matches: existing = set(tag_ids or []) tag_ids = list(existing | {str(t.id) for t in tag_matches}) query = "" # ILIKE 对中文标题/摘要无效，跳过 if query.strip(): terms = [t.strip() for t in query.split() if t.strip()] # 单数字词：优先 PMID 精确匹配（unique index 5ms 返回） # 不是 PMID 时才回退到 ILIKE 兜底（DOI 片段等），不做 tsquery 避免 seq scan numeric_terms = [t for t in terms if t.isdigit() and len(t) <= 15] text_terms = [t for t in terms if not (t.isdigit() and len(t) <= 15)] for t in numeric_terms: p = int(t) exists = (await db.execute( select(GlobalLiterature.id).where(GlobalLiterature.pmid == p).limit(1) )).scalar_one_or_none() if exists: conditions.append(GlobalLiterature.pmid == p) else: # 不是 PMID → ILIKE 搜索（DOI 等），不做 tsquery 避免 seq scan p_like = t if exact_phrase else f"%{t}%" conditions.append(or_( GlobalLiterature.title.ilike(p_like), GlobalLiterature.doi.ilike(p_like), )) if text_terms: # ATM 展开（仅非中文查询） _atm_cond = None if not _CHINESE_RE.search(query): from app.services.query_expansion import expand_atm as _expand_atm _atm_cond = await _expand_atm(db, query.strip()) _cond_before = len(conditions) if boolean == "and": for term in text_terms: conditions.append(AdvancedSearchEngine._field_condition(field, term, exact_phrase)) else: or_conds = [AdvancedSearchEngine._field_condition(field, t, exact_phrase) for t in text_terms] conditions.append(or_(*or_conds)) # 将文本条件与 ATM 条件 OR 组合 if _atm_cond is not None: _text_conds = conditions[_cond_before:] del conditions[_cond_before:] if _text_conds: if len(_text_conds) == 1: conditions.append(or_(_atm_cond, _text_conds[0])) else: conditions.append(or_(_atm_cond, and_(*_text_conds))) else: conditions.append(_atm_cond) # 年份范围 if year_from: conditions.append(GlobalLiterature.pub_year >= year_from) if year_to: conditions.append(GlobalLiterature.pub_year <= year_to) # 具体日期范围（按天搜索） from datetime import date as dt_date if date_from: try: df = dt_date.fromisoformat(date_from) conditions.append(GlobalLiterature.pub_date >= df) except ValueError: raise ValueError(f"Invalid date_from format: {date_from}") if date_to: try: dt = dt_date.fromisoformat(date_to) conditions.append(GlobalLiterature.pub_date <= dt) except ValueError: raise ValueError(f"Invalid date_to format: {date_to}") # 期刊等级 if journal_tiers: subq = select(GlobalJournal.issn).where(GlobalJournal.tier.in_(journal_tiers)) result = await db.execute(subq) issns = [r for (r,) in result.all()] if issns: conditions.append(GlobalLiterature.journal_issn.in_(issns)) # 标签筛选（含子标签递归） if tag_ids: import uuid as _uuid tag_uuids = [_uuid.UUID(tid) if isinstance(tid, str) else tid for tid in tag_ids] all_tags = (await db.execute(select(GlobalTag).where(GlobalTag.id.in_(tag_uuids)))).scalars().all() all_tag_ids = set(t.id for t in all_tags if t) # Batch child tag lookup (1 query instead of N) paths = [t.path + "::" for t in all_tags if t] if paths: child_conds = [GlobalTag.path.like(f"{p}%") for p in paths] children = (await db.execute(select(GlobalTag.id).where(or_(*child_conds)))).scalars().all() all_tag_ids.update(children) uids = list(all_tag_ids) subq = select(func.distinct(GlobalLiteratureTag.literature_id)).where(GlobalLiteratureTag.tag_id.in_(uids)) conditions.append(GlobalLiterature.id.in_(subq)) # 发表类型（PG JSONB contains） if pub_types: from sqlalchemy.dialects.postgresql import JSONB type_conds = [GlobalLiterature.pub_types.cast(JSONB).contains([pt]) for pt in pub_types] conditions.append(or_(*type_conds)) # 研究设计分类过滤（PG JSON path access） if study_design: conditions.append( GlobalLiterature.study_design['primary'].astext == study_design ) # 撤稿过滤 if retracted == "yes": conditions.append(GlobalLiterature.retracted == True) elif retracted == "no": conditions.append(GlobalLiterature.retracted == False) elif retracted == "only": conditions.append(GlobalLiterature.retracted == True) # 阴性结果过滤 if negative_result == "yes": conditions.append(GlobalLiterature.is_negative_result == True) elif negative_result == "no": conditions.append(GlobalLiterature.is_negative_result == False) elif negative_result == "only": conditions.append(GlobalLiterature.is_negative_result == True) # 开放获取 if is_oa is not None: conditions.append(GlobalLiterature.is_oa == is_oa) # 语言 if language: conditions.append(GlobalLiterature.language == language) # NLM 期刊子集（如 AIM / Core Clinical Journals） if nlm_subsets: subq = select(GlobalJournal.issn).where(GlobalJournal.nlm_subsets.overlap(nlm_subsets)) result = await db.execute(subq) issns = [r for (r,) in result.all()] if issns: conditions.append(GlobalLiterature.journal_issn.in_(issns)) # ── 按年份统计（Results by year，与筛选条件一致） ── year_counts = [] if conditions and (query.strip() or year_from or year_to or date_from or date_to or journal_tiers or pub_types or study_design or tag_ids or retracted or negative_result or is_oa is not None or language or nlm_subsets): try: yr_subq = select(GlobalLiterature.pub_year).where( and_(*conditions) ).subquery() year_count_q = select( yr_subq.c.pub_year, func.count().label("cnt") ).group_by(yr_subq.c.pub_year).order_by(yr_subq.c.pub_year.desc()) year_rows = await db.execute(year_count_q) year_counts = [ {"year": y, "count": c} for y, c in year_rows if y is not None ] except Exception: year_counts = [] # 构建查询 q = select(GlobalLiterature) if conditions: q = q.where(and_(*conditions)) # 排序（PubMed 查询时跳过 ts_rank，避免语法标签噪音） _relevance_query = query if _pubmed_parsed and sort == "relevance": # 用纯文本词做相关性排序，去掉 [field] 标签 plain_parts = [t.text for t in _pubmed_parsed.plain_terms] plain_parts += [t.text for t in _pubmed_parsed.title_terms] plain_parts += [t.text for t in _pubmed_parsed.tiab_terms] _relevance_query = " ".join(plain_parts) if plain_parts else "" if sort == "date": q = q.order_by(GlobalLiterature.pub_date.desc().nullslast()) elif sort == "cited": q = q.order_by(GlobalLiterature.cited_by_count.desc().nullslast()) elif sort == "best_match" and _relevance_query.strip(): tsq = func.plainto_tsquery("english", _relevance_query) q = q.order_by(AdvancedSearchEngine._best_match_order(tsq)) elif sort == "relevance" and _relevance_query.strip(): tsq = func.plainto_tsquery("english", _relevance_query) rank = func.ts_rank(GlobalLiterature.search_tsv, tsq) q = q.order_by(rank.desc()) else: q = q.order_by(GlobalLiterature.pub_date.desc().nullslast()) # keyset 游标分页（仅 date 排序支持，page 参数被忽略，不做 COUNT） use_keyset = (cursor_date is not None and cursor_id is not None and sort == "date") if use_keyset: from datetime import date as dt_date try: cursor_dt = dt_date.fromisoformat(cursor_date) except (ValueError, TypeError): use_keyset = False if use_keyset: import uuid as _cuuid try: cursor_uuid = _cuuid.UUID(cursor_id) conditions.append( or_( GlobalLiterature.pub_date < cursor_dt, and_(GlobalLiterature.pub_date == cursor_dt, GlobalLiterature.id < cursor_uuid), ) ) except (ValueError, AttributeError): use_keyset = False # 重新构建查询（keyset 条件可能已追加） q = select(GlobalLiterature) if conditions: q = q.where(and_(*conditions)) if sort == "date": q = q.order_by(GlobalLiterature.pub_date.desc().nullslast()) elif sort == "cited": q = q.order_by(GlobalLiterature.cited_by_count.desc().nullslast()) elif sort == "best_match" and _relevance_query.strip(): tsq = func.plainto_tsquery("english", _relevance_query) q = q.order_by(AdvancedSearchEngine._best_match_order(tsq)) elif sort == "relevance" and _relevance_query.strip(): tsq = func.plainto_tsquery("english", _relevance_query) rank = func.ts_rank(GlobalLiterature.search_tsv, tsq) q = q.order_by(rank.desc()) else: q = q.order_by(GlobalLiterature.pub_date.desc().nullslast()) # 分页 has_more = False if use_keyset: # cursor 模式：取 page_size+1 条判断是否有下一页，不做 COUNT result = await db.execute(q.limit(page_size + 1)) items = result.scalars().all() has_more = len(items) > page_size items = items[:page_size] total = 0 else: # 总数 count_q = select(func.count()).select_from( select(literal_column("1")) .select_from(GlobalLiterature) .where(and_(*conditions) if conditions else True) .subquery() ) total = (await db.execute(count_q)).scalar() or 0 offset = (page - 1) * page_size # ── date 排序优化 ── # PG 对常见词（如 "cancer" >80% 匹配率）会选 Parallel Seq Scan（~22s）。 # 嵌入子查询先取最近 N 条 ID（走 ix_gl_pub_date 索引），迫使 PG # 使用索引而非全表扫描。 # 条件：仅当无可搜索查询且非历史日期筛选时启用，避免丢失早期文献 _has_meaningful_query = bool(query.strip()) _has_historical_filter = year_from is not None and year_from < datetime.now().year - 5 if sort == "date" and conditions and not _has_meaningful_query and not _has_historical_filter: recent_n = max(offset + page_size * 5, 2000) recent_subq = select(GlobalLiterature.id).where( GlobalLiterature.pub_date.isnot(None) ).order_by(GlobalLiterature.pub_date.desc()).limit(recent_n) data_q = select(GlobalLiterature).where( GlobalLiterature.id.in_(recent_subq), and_(*conditions) ).order_by(GlobalLiterature.pub_date.desc().nullslast()) else: data_q = q result = await db.execute(data_q.offset(offset).limit(page_size)) items = result.scalars().all() has_more = (offset + page_size) < total and len(items) == page_size from app.services.tag_loader import load_tags_for_literature tm = await load_tags_for_literature(db, [str(lit.id) for lit in items]) # Batch load journal tiers + canonical names tier_map = {} name_map = {} if items: issns = list(set(lit.journal_issn for lit in items if lit.journal_issn)) if issns: jr = await db.execute( select(GlobalJournal.issn, GlobalJournal.tier, GlobalJournal.name).where(GlobalJournal.issn.in_(issns)) ) for issn_, tier_, name_ in jr.all(): tier_map[issn_] = tier_ name_map[issn_] = name_ results = [] for lit in items: authors = lit.authors or [] results.append({ "id": str(lit.id), "pmid": lit.pmid, "title": lit.title, "first_author": authors[0].get("family", "") if authors else "", "journal": name_map.get(lit.journal_issn) or lit.journal, "pub_date": cap_pub_date(lit.pub_date), "article_date": lit.article_date.isoformat() if lit.article_date else None, "pub_year": lit.pub_year, "tags": tm.get(str(lit.id), []), "abstract": lit.abstract[:300] if lit.abstract else None, "doi": lit.doi, "pmc_id": lit.pmc_id, "is_oa": lit.is_oa, "cited_by_count": lit.cited_by_count, "created_at": lit.created_at.isoformat() if lit.created_at else None, "updated_at": lit.updated_at.isoformat() if lit.updated_at else None, "journal_issn": lit.journal_issn, "journal_tier": tier_map.get(lit.journal_issn), "pub_types": lit.pub_types, "affiliation": authors[0].get("affiliation", "") if authors else "", }) result = {"items": results, "total": total, "page": page, "page_size": page_size, "has_more": has_more, "year_counts": year_counts} if _search_cache_key is not None: await _cache.set(_search_cache_key, result, ttl=AdvancedSearchEngine.SEARCH_CACHE_TTL) return result @staticmethod async def _pubmed_conditions( db: AsyncSession, pp, existing_conditions: list, ) -> list: """将解析后的 PubMed 查询转换为 SQLAlchemy 条件列表。""" from sqlalchemy.dialects.postgresql import JSONB conditions = list(existing_conditions) # 1. 字段级搜索 [TI] [AB] [TIAB] [AU] [TA] field_map = { "title": pp.title_terms, "abstract": pp.abstract_terms, "all": pp.tiab_terms, "author": pp.author_terms, "journal": pp.journal_terms, } for fld, terms in field_map.items(): if not terms: continue field_conds = [] for term in terms: cond = AdvancedSearchEngine._field_condition(fld, term.text, term.exact) if term.is_not: cond = not_(cond) field_conds.append(cond) # 同 field 内多个 term 按 boolean_operator 组合 if pp.boolean_operator == "or": conditions.append(or_(*field_conds)) else: conditions.extend(field_conds) # and_() 由外部统一组合 # 2. 纯文本词（无字段标签） if pp.plain_terms: if pp.boolean_operator == "or": plain_conds = [] for term in pp.plain_terms: cond = AdvancedSearchEngine._field_condition("all", term.text, term.exact) if term.is_not: cond = not_(cond) plain_conds.append(cond) if plain_conds: conditions.append(or_(*plain_conds)) else: for term in pp.plain_terms: cond = AdvancedSearchEngine._field_condition("all", term.text, term.exact) if term.is_not: cond = not_(cond) conditions.append(cond) # 3. [MH] → tree_number 展开（逐 term 独立 subq，AND 组合） if pp.mesh_terms: if pp.boolean_operator == "or": conditions = await AdvancedSearchEngine._expand_mesh_tag_ids( db, pp.mesh_terms, conditions) else: for m in pp.mesh_terms: conditions = await AdvancedSearchEngine._expand_mesh_tag_ids( db, [m], conditions) # 4. [MAJR] → tree_number 展开 + is_major=True（逐 term 独立 subq） if pp.majr_terms: if pp.boolean_operator == "or": conditions = await AdvancedSearchEngine._expand_mesh_tag_ids( db, pp.majr_terms, conditions, major_only=True) else: for m in pp.majr_terms: conditions = await AdvancedSearchEngine._expand_mesh_tag_ids( db, [m], conditions, major_only=True) # 5. [PT] → pub_types JSONB contains if pp.pub_types: type_conds = [ GlobalLiterature.pub_types.cast(JSONB).contains([pt]) for pt in pp.pub_types ] conditions.append(or_(*type_conds)) # 5. [DP] → 年份/日期范围 if pp.year_from: conditions.append(GlobalLiterature.pub_year >= pp.year_from) if pp.year_to: conditions.append(GlobalLiterature.pub_year <= pp.year_to) if pp.date_from: from datetime import date as _dt_date try: conditions.append(GlobalLiterature.pub_date >= _dt_date.fromisoformat(pp.date_from)) except ValueError: pass if pp.date_to: from datetime import date as _dt_date try: conditions.append(GlobalLiterature.pub_date <= _dt_date.fromisoformat(pp.date_to)) except ValueError: pass # 6. [PMID] → 精确匹配 for pmid_val in pp.pmid_terms: conditions.append(GlobalLiterature.pmid == pmid_val) # 7. [DOI] → ILIKE for doi_term in pp.doi_terms: conditions.append(GlobalLiterature.doi.ilike(f"%{doi_term}%")) return conditions @staticmethod def _field_condition(field: str, term: str, exact: bool) -> callable: if field == "title": if exact: return GlobalLiterature.title.ilike(term if exact else f"%{term}%") # 只用 tsquery（GIN 索引），不掺 ILIKE——OR 会让 PG 放弃 GIN 索引走 seq scan return GlobalLiterature.search_tsv.op("@@")(func.plainto_tsquery("english", term)) elif field == "abstract": if exact: return GlobalLiterature.abstract.ilike(term if exact else f"%{term}%") return GlobalLiterature.search_tsv.op("@@")(func.plainto_tsquery("english", term)) elif field == "author": return GlobalLiterature.search_tsv.op("@@")(func.plainto_tsquery("english", term)) elif field == "journal": return or_( GlobalLiterature.journal.ilike(f"%{term}%"), GlobalLiterature.journal_iso.ilike(f"%{term}%"), ) elif field == "affiliation": return GlobalLiterature.search_tsv.op("@@")(func.plainto_tsquery("english", term)) else: # "all" default like_val = term if exact else f"%{term}%" if exact: return or_( GlobalLiterature.title.ilike(like_val), cast(GlobalLiterature.pmid, String).ilike(f"%{term}%"), GlobalLiterature.doi.ilike(like_val), ) # 含 "/" → 跳过 tsquery if "/" in term: if term.startswith("10."): return or_( GlobalLiterature.doi == term, GlobalLiterature.doi.ilike(like_val), ) return or_( GlobalLiterature.title.ilike(like_val), cast(GlobalLiterature.pmid, String).ilike(f"%{term}%"), GlobalLiterature.doi.ilike(like_val), ) # 只用 tsquery（GIN 索引），不掺 ILIKE——OR 会让 PG 放弃 GIN 索引走 seq scan return GlobalLiterature.search_tsv.op("@@")(func.plainto_tsquery("english", term)) @staticmethod async def _expand_mesh_tag_ids( db: AsyncSession, mesh_names: list[str], conditions: list, major_only: bool = False, ) -> list: """用 GlobalTagTreeNumber 展开 [MH]/[MAJR] 1. ILIKE 匹配 mesh_names → tag_ids 2. 用 tree_number 前缀展开子节点（C04.588 → 所有子树号） 3. [MAJR] 额外 AND is_major=True """ import uuid as _uuid mesh_conds = [GlobalTag.name_en.ilike(f"%{m}%") for m in mesh_names] tag_rows = (await db.execute( select(GlobalTag.id).where(or_(*mesh_conds)) )).all() if not tag_rows: return conditions mesh_tag_ids: set[_uuid.UUID] = set(tid for (tid,) in tag_rows) # tree_number 前缀展开：取匹配 tag 的所有 tree_number，查子节点 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)) )).scalars().all() mesh_tag_ids.update(children) uids = list(mesh_tag_ids) if major_only: subq = select(func.distinct(GlobalLiteratureTag.literature_id)).where( GlobalLiteratureTag.tag_id.in_(uids), GlobalLiteratureTag.is_major == True, ) else: subq = select(func.distinct(GlobalLiteratureTag.literature_id)).where( GlobalLiteratureTag.tag_id.in_(uids), ) conditions.append(GlobalLiterature.id.in_(subq)) return conditions @staticmethod def _best_match_order(tsq): """构建 best_match 排序表达式：ts_rank + 引用数对数 + 近期度梯度加分 近期度加分通过 EXTRACT(YEAR FROM NOW()) 动态计算，分三档: - 当年: +10 - 去年: +7 - 前年: +3 """ _cy = func.extract("year", func.now()) score = ( func.ts_rank(GlobalLiterature.search_tsv, tsq) * 0.3 + func.ln(func.coalesce(GlobalLiterature.cited_by_count, 0) + 1) * 2 + case( (GlobalLiterature.pub_year >= _cy, 10), (GlobalLiterature.pub_year >= _cy - 1, 7), (GlobalLiterature.pub_year >= _cy - 2, 3), else_=0, ) ) return score.desc()