backend/backend/app/services/search_engine.py

"""高级搜索服务：布尔运算 + 字段限定 + PubMed 查询语法"""

import logging
import re
from datetime import datetime

from sqlalchemy import String, and_, case, cast, func, literal_column, not_, or_, select, text

logger = logging.getLogger(__name__)
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.cache import cache as _cache
from app.models.literature import GlobalJournal, GlobalLiterature, GlobalLiteratureTag, GlobalTag, GlobalTagTreeNumber
from app.schemas.literature import cap_pub_date
from app.services.pubmed_query_parser import is_pubmed_syntax, parse_pubmed_query
from app.services.query_expansion import expand_atm as _expand_atm
from app.services.tag_loader import load_tags_for_literature


def _escape_ilike(s: str) -> str:
    """转义 ILIKE 模式中的通配符 _ 和 %，防止用户输入 'EGFR_mutation' 误匹配 'EGFR mutation'"""
    if not s:
        return s
    return s.replace('\\', '\\\\').replace('%', '\\%').replace('_', '\\_')


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,
        tag_ids: list[str] | None,
        retracted: str, negative_result: str,
        is_oa: bool | None, language: str | None, languages: list[str] | None, nlm_subsets: list[str] | None,
        page: int, page_size: int, sort: str,
        # PubMed filter params
        has_abstract: bool | None = None,
        is_free_full_text: bool | None = None,
        has_full_text: bool | None = None,
        has_associated_data: bool | None = None,
        species: list[str] | None = None,
        sex: list[str] | None = None,
        age: list[str] | None = None,
        medline_only: bool = False,
        exclude_preprints: bool = False,
        # Keyset cursor (sort=date 时分页)
        cursor_date: str | None = None,
        cursor_id: str | None = None,
    ) -> 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 [],
            "tid": sorted(tag_ids) if tag_ids else [],
            "r": retracted, "nr": negative_result,
            "oa": is_oa, "lang": language, "lgs": sorted(languages) if languages else [],
            "ns": sorted(nlm_subsets) if nlm_subsets else [],
            "p": page, "ps": page_size, "s": sort,
            "ha": has_abstract,
            "fft": is_free_full_text,
            "hft": has_full_text,
            "had": has_associated_data,
            "sp": sorted(species) if species else [],
            "sx": sorted(sex) if sex else [],
            "ag": sorted(age) if age else [],
            "mo": medline_only,
            "epr": exclude_preprints,
            # keyset cursor 唯一标识翻页位置
            "cd": cursor_date,
            "ci": cursor_id,
        }
        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,
        tag_ids: list[str] | None = None,
        retracted: str = "",             # "no", "only"
        negative_result: str = "",       # "yes", "no", "only"
        is_oa: bool | None = None,       # True = 仅开放获取
        language: str | None = None,      # 语言代码（en/zh/fr 等，向后兼容）
        languages: list[str] | None = None,  # 语言代码列表（多选）
        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）
        # ── PubMed 筛选器参数 ──
        has_abstract: bool | None = None,
        is_free_full_text: bool | None = None,
        has_full_text: bool | None = None,
        has_associated_data: bool | None = None,
        species: list[str] | None = None,
        sex: list[str] | None = None,
        age: list[str] | None = None,
        medline_only: bool = False,
        exclude_preprints: bool = False,
    ) -> dict:
        """执行高级搜索"""
        from sqlalchemy.dialects.postgresql import JSONB
        conditions = []

        # 60s 缓存（keyset cursor 页也参与缓存，cursor_date+cursor_id 组合唯一标识翻页位置）
        use_cursor = (cursor_date is not None and cursor_id is not None and sort == "date")
        _search_cache_key = AdvancedSearchEngine._search_cache_key(
            query, field, boolean, exact_phrase,
            year_from, year_to, date_from, date_to,
            journal_tiers, pub_types, tag_ids,
            retracted, negative_result,
            is_oa, language, languages, nlm_subsets,
            page, page_size, sort,
            has_abstract=has_abstract,
            is_free_full_text=is_free_full_text,
            has_full_text=has_full_text,
            has_associated_data=has_associated_data,
            species=species, sex=sex, age=age,
            medline_only=medline_only,
            exclude_preprints=exclude_preprints,
            cursor_date=cursor_date,
            cursor_id=cursor_id,
        )
        cached = await _cache.get(_search_cache_key)
        if cached is not None:
            return cached

        # 30 秒查询超时（放在缓存检查之后，缓存命中不执行）
        await db.execute(text("SET LOCAL statement_timeout = '30s'"))

        # ─── PubMed 语法检测与解析 ───
        _pubmed_parsed = None
        _is_flat_text = True  # 是否走传统 tsvector + ILIKE 路径
        if query.strip():

            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.affiliation_terms
                    or pp.language_terms or pp.volume_terms or pp.issue_terms
                    or pp.pages_terms or pp.lid_terms
                    or pp.grant_terms or pp.subheading_terms
                    or pp.registry_terms or pp.substance_terms
                    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.ot_terms or pp.gene_terms or pp.pmc_terms
                    or pp.edat_from or pp.crdt_from or pp.mhda_from
                    or pp.lr_from or pp.dcom_from or pp.dep_from
                    or pp.edat_to or pp.crdt_to or pp.mhda_to
                    or pp.lr_to or pp.dcom_to or pp.dep_to
                    or pp.date_from or pp.date_to
                    or pp.negated_date_ranges
                    or pp.groups
                    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():
            # 如果 PubMed 语法检测成功但解析后无有效词（如语法错误），剥离括号和操作符再搜索
            if _pubmed_parsed is not None and not any([
                bool(_pubmed_parsed.title_terms or _pubmed_parsed.abstract_terms or _pubmed_parsed.tiab_terms
                     or _pubmed_parsed.author_terms or _pubmed_parsed.journal_terms
                     or _pubmed_parsed.mesh_terms or _pubmed_parsed.majr_terms
                     or _pubmed_parsed.pub_types or _pubmed_parsed.doi_terms or _pubmed_parsed.pmid_terms
                     or _pubmed_parsed.affiliation_terms
                     or _pubmed_parsed.language_terms or _pubmed_parsed.volume_terms or _pubmed_parsed.issue_terms
                     or _pubmed_parsed.pages_terms or _pubmed_parsed.lid_terms
                     or _pubmed_parsed.grant_terms or _pubmed_parsed.subheading_terms
                     or _pubmed_parsed.registry_terms or _pubmed_parsed.substance_terms
                     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.ot_terms or _pubmed_parsed.gene_terms or _pubmed_parsed.pmc_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.dep_from
                     or _pubmed_parsed.edat_to or _pubmed_parsed.crdt_to or _pubmed_parsed.mhda_to
                     or _pubmed_parsed.lr_to or _pubmed_parsed.dcom_to or _pubmed_parsed.dep_to
                     or _pubmed_parsed.date_from or _pubmed_parsed.date_to
                     or _pubmed_parsed.negated_date_ranges
                     or _pubmed_parsed.groups
                     or _pubmed_parsed.plain_terms or _pubmed_parsed.has_not
                     or _pubmed_parsed.year_from or _pubmed_parsed.year_to)
            ]):
                # 解析失败但检测到 PubMed 语法 — 擦除 [field] 标签、布尔符、引号
                query = re.sub(r'\[[\w/: -]+\]', '', query)  # P5: [\w/: -] 覆盖 [Title/Abstract] 和 [MH:noexp]
                query = re.sub(r'\b(AND|OR|NOT)\b', '', query)
                query = query.replace('"', '').replace('(', '').replace(')', '')
                query = ' '.join(query.split())
            # 中文搜索：自动匹配 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'%{_escape_ilike(query.strip())}%'))
                    .limit(100)  # P5: limit to prevent oversized subquery
                )).scalars().all()
                if tag_matches:
                    existing = set(tag_ids or [])
                    tag_ids = list(existing | {str(t) for t in tag_matches})
                    # 不置空 query，保留 ILIKE 对中文标题/摘要的搜索能力
            if query.strip():
                # 提取引号短语作为完整词，避免 "lung cancer" 被拆散
                import re as _phrase_re
                _phrase_pat = _phrase_re.compile(r'"([^"]*)"')
                _phrases = _phrase_pat.findall(query)
                _query_no_quotes = _phrase_pat.sub(' ', query)
                _rest = [t.strip().strip('"').strip("'") for t in _query_no_quotes.split() if t.strip()]
                terms = [p for p in _phrases if p.strip()] + [t for t in _rest if t not in _phrases]
                # 单数字词：优先 PMID 精确匹配（unique index 5ms 返回）
                # 不是 PMID 时才回退到 ILIKE 兜底（DOI 片段等），不做 tsquery 避免 seq scan
                numeric_terms = [t for t in terms if re.match(r'^\d{1,15}$', t)]
                text_terms = [t for t in terms if not re.match(r'^\d{1,15}$', t)]
                if numeric_terms:
                    num_conds = []
                    if exact_phrase:
                        # 精确短语模式：跳过 PMID 快速路径，走 ILIKE
                        for t in numeric_terms:
                            num_conds.append(or_(
                                GlobalLiterature.title.ilike(t),
                                GlobalLiterature.doi.ilike(t),
                            ))
                    else:
                        # P1-1: 批量检查所有 PMID（单次 IN 查询替代 N+1 次 SELECT）
                        numeric_ids = [int(t) for t in numeric_terms]
                        db_pmids: set[int] = set()
                        rows = await db.execute(
                            select(GlobalLiterature.pmid).where(GlobalLiterature.pmid.in_(numeric_ids))
                        )
                        for (pmid,) in rows:
                            db_pmids.add(pmid)
                        for t in numeric_terms:
                            p = int(t)
                            if p in db_pmids:
                                num_conds.append(GlobalLiterature.pmid == p)
                            else:
                                p_like = f"%{_escape_ilike(t)}%"
                                num_conds.append(or_(
                                    GlobalLiterature.title.ilike(p_like),
                                    GlobalLiterature.doi.ilike(p_like),
                                ))
                    if boolean == "or" and len(num_conds) > 1:
                        conditions.append(or_(*num_conds))
                    else:
                        conditions.extend(num_conds)
                if text_terms:
                    # ATM 展开（仅 field="all" 时，字段搜索不应自动扩到 MeSH）
                    _atm_cond = None
                    _atm_query = query.replace('"', '').replace("'", '').replace('(', '').replace(')', '').strip()
                    if _atm_query and field == "all":
                        try:
                            _atm_cond = await _expand_atm(db, _atm_query)
                        except Exception:
                            logger.exception("ATM expansion failed (flat text): %s", _atm_query[:100])
                            _atm_cond = None

                    _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 is not None:
            conditions.append(GlobalLiterature.pub_year >= year_from)
        if year_to is not None:
            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 not issns:
                logger.warning("journal_tiers filter matched zero journals: %s", journal_tiers)
            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:
            type_conds = [GlobalLiterature.pub_types.cast(JSONB).contains([pt]) for pt in pub_types]
            conditions.append(or_(*type_conds))

        # 撤稿过滤
        if retracted == "no":
            conditions.append(GlobalLiterature.retracted == False)
        elif retracted in ("only", "yes"):
            conditions.append(GlobalLiterature.retracted == True)

        # 阴性结果过滤
        if negative_result == "no":
            conditions.append(GlobalLiterature.is_negative_result == False)
        elif negative_result in ("only", "yes"):
            conditions.append(GlobalLiterature.is_negative_result == True)

        # 开放获取
        if is_oa is not None:
            conditions.append(GlobalLiterature.is_oa == is_oa)

        # 语言（多选优先，向后兼容单语言）
        if languages:
            conditions.append(GlobalLiterature.language.in_(languages))
        elif 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 not issns:
                logger.warning("nlm_subsets filter matched zero journals: %s", nlm_subsets)
            conditions.append(GlobalLiterature.journal_issn.in_(issns))

        # ── PubMed 筛选器 ──

        # Text Availability
        if has_abstract:
            conditions.append(and_(
                GlobalLiterature.abstract.isnot(None),
                GlobalLiterature.abstract != '',
            ))
        if is_free_full_text:
            conditions.append(GlobalLiterature.is_oa == True)
        if has_full_text:
            conditions.append(GlobalLiterature.pmc_id.isnot(None))

        # Article Attribute: Associated data
        if has_associated_data:
            conditions.append(GlobalLiterature.databank_list.cast(JSONB) != '[]')

        # Species / Sex / Age — mesh_headings JSONB @> UI match
        if species:
            conditions.append(or_(*[
                GlobalLiterature.mesh_headings.contains([{"ui": ui}])
                for ui in species
            ]))
        if sex:
            conditions.append(or_(*[
                GlobalLiterature.mesh_headings.contains([{"ui": ui}])
                for ui in sex
            ]))
        if age:
            from app.api.v1.features import AGE_GROUP_UI_MAP
            age_uis = []
            for val in age:
                if val in AGE_GROUP_UI_MAP:
                    age_uis.extend(AGE_GROUP_UI_MAP[val])
                else:
                    age_uis.append(val)  # 直接传 UI 的情况（向后兼容）
            conditions.append(or_(*[
                GlobalLiterature.mesh_headings.contains([{"ui": ui}])
                for ui in set(age_uis)
            ]))

        # MEDLINE only
        if medline_only:
            conditions.append(GlobalLiterature.citation_status == 'MEDLINE')

        # Exclude Preprints
        if exclude_preprints:
            conditions.append(GlobalLiterature.is_preprint == False)

        # ── 按年份统计（Results by year，使用完整筛选条件） ──
        _yr_before = len(conditions)
        year_counts = []

        # 判断是否有任何筛选器/文本查询活跃（与旧 gate 逻辑一致）
        _has_any_filter = (query.strip() or year_from or year_to or date_from or date_to
                           or journal_tiers or pub_types or tag_ids
                           or retracted or negative_result or is_oa is not None
                           or language or languages or nlm_subsets
                           or has_abstract or is_free_full_text or has_full_text
                           or has_associated_data or species or sex or age
                           or medline_only or exclude_preprints)

        # ── 无文本无筛选 → 走 Redis 预计算缓存（每天 pipeline 后 cron 刷新） ──
        if not _has_any_filter:
            _cached = await _cache.get("search:year_counts:all")
            if _cached is not None:
                year_counts = _cached
            else:
                # 缓存未命中 → GROUP BY 兜底（首次部署、Redis 不可用时，conditions 可能为空）
                try:
                    yr_conds = conditions[:_yr_before]
                    yr_subq = select(GlobalLiterature.pub_year).where(
                        and_(*yr_conds) if yr_conds else text("TRUE")
                    ).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:
                    logger.exception("Year counts query failed")
                    year_counts = []

        # ── 有筛选条件 → 精确 GROUP BY（已移除旧 30 年限制，pub_year 低基数因此 GROUP BY 高效） ──
        elif conditions and _has_any_filter:
            try:
                yr_conds = conditions[:_yr_before]
                yr_subq = select(GlobalLiterature.pub_year).where(
                    and_(*yr_conds)
                ).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:
                logger.exception("Year counts query failed")
                year_counts = []

        # 排序（PubMed 查询时跳过 ts_rank，避免语法标签噪音）
        _relevance_query = query
        if _pubmed_parsed and sort in ("relevance", "best_match"):
            # 用纯文本词做相关性排序，去掉 [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 ""
        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))
        q = q.order_by(*AdvancedSearchEngine._apply_order_by(sort, _relevance_query))

        # 分页
        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

            result = await db.execute(q.offset(offset).limit(page_size))
            items = result.scalars().all()
            has_more = (offset + page_size) < total and len(items) == page_size

        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 做其他过滤）
        conditions = list(existing_conditions)
        # term_conditions 收集与查询词相关的条件，用于 OR 模式下统一包裹
        term_conditions: list = []

        # 1. 字段级搜索 [TI] [AB] [TIAB] [AU] [TA] [LA] [VI] [IP] [PG] [LID]
        field_combine = or_ if pp.boolean_operator == "or" else and_
        field_map = {
            "title": pp.title_terms,
            "abstract": pp.abstract_terms,
            "all": pp.tiab_terms,
            "author": pp.author_terms,
            "journal": pp.journal_terms,
            "affiliation": pp.affiliation_terms,
            "language": pp.language_terms,
            "volume": pp.volume_terms,
            "issue": pp.issue_terms,
            "pages": pp.pages_terms,
            "lid": pp.lid_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)
            term_conditions.append(field_combine(*field_conds) if len(field_conds) > 1 else field_conds[0])

        # 2. 纯文本词（无字段标签）— P0-2: 对无标签词补充 ATM MeSH 展开
        if pp.plain_terms:
            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:
                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
                    try:
                        atm_cond = await _expand_atm_inline(db, combined)
                    except Exception:
                        logger.exception("ATM expansion failed (pubmed plain_terms): %s", combined[:100])
                        atm_cond = None
                    if atm_cond is not None:
                        text_cond = field_combine(*plain_conds) if len(plain_conds) > 1 else plain_conds[0]
                        term_conditions.append(or_(atm_cond, text_cond))
                    else:
                        term_conditions.append(field_combine(*plain_conds) if len(plain_conds) > 1 else plain_conds[0])
                else:
                    term_conditions.append(field_combine(*plain_conds) if len(plain_conds) > 1 else plain_conds[0])

        # 3. [MH] → tree_number 展开，支持 is_not 和 _noexp
        if pp.mesh_terms:
            for is_neg in (False, True):
                subset = [t for t in pp.mesh_terms if t.is_not == is_neg]
                if not subset:
                    continue
                noexp_names = [t.text for t in subset if t._noexp]
                exp_names = [t.text for t in subset if not t._noexp]
                if exp_names:
                    cond = await AdvancedSearchEngine._expand_mesh_tag_ids(db, exp_names, major_only=False)
                    if cond is not None:
                        term_conditions.append(not_(cond) if is_neg else cond)
                if noexp_names:
                    cond = await AdvancedSearchEngine._expand_mesh_tag_ids(db, noexp_names, major_only=False, noexp=True)
                    if cond is not None:
                        term_conditions.append(not_(cond) if is_neg else cond)

        # 4. [MAJR] → tree_number 展开 + is_major=True，支持 is_not
        if pp.majr_terms:
            pos_names = [t.text for t in pp.majr_terms if not t.is_not]
            neg_names = [t.text for t in pp.majr_terms if t.is_not]
            if pos_names:
                cond = await AdvancedSearchEngine._expand_mesh_tag_ids(db, pos_names, major_only=True)
                if cond is not None:
                    term_conditions.append(cond)
            if neg_names:
                cond = await AdvancedSearchEngine._expand_mesh_tag_ids(db, neg_names, major_only=True)
                if cond is not None:
                    term_conditions.append(not_(cond))

        # 5. [PT] → pub_types JSONB contains，支持 is_not
        if pp.pub_types:
            pos = [t for t in pp.pub_types if not t.is_not]
            neg = [t for t in pp.pub_types if t.is_not]
            if pos:
                pos_conds = [GlobalLiterature.pub_types.cast(JSONB).contains([t.text]) for t in pos]
                term_conditions.append(or_(*pos_conds))
            if neg:
                neg_conds = [GlobalLiterature.pub_types.cast(JSONB).contains([t.text]) for t in neg]
                term_conditions.append(not_(or_(*neg_conds)))

        # 5b. [GR] [SH] [RN] [NM] [SI] [PA] → JSONB contains，支持 is_not
        if pp.grant_terms:
            pos = [t for t in pp.grant_terms if not t.is_not]
            neg = [t for t in pp.grant_terms if t.is_not]
            if pos:
                term_conditions.append(or_(*[
                    GlobalLiterature.grants.cast(JSONB).contains([{"grant_id": t.text}])
                    for t in pos
                ]))
            if neg:
                neg_conds = [GlobalLiterature.grants.cast(JSONB).contains([{"grant_id": t.text}]) for t in neg]
                term_conditions.append(not_(or_(*neg_conds)))
        if pp.subheading_terms:
            pos = [t for t in pp.subheading_terms if not t.is_not]
            neg = [t for t in pp.subheading_terms if t.is_not]
            if pos:
                term_conditions.append(or_(*[
                    GlobalLiterature.mesh_headings.cast(JSONB).contains([{"qualifiers": [t.text]}])
                    for t in pos
                ]))
            if neg:
                neg_conds = [GlobalLiterature.mesh_headings.cast(JSONB).contains([{"qualifiers": [t.text]}]) for t in neg]
                term_conditions.append(not_(or_(*neg_conds)))
        if pp.registry_terms:
            pos = [t for t in pp.registry_terms if not t.is_not]
            neg = [t for t in pp.registry_terms if t.is_not]
            if pos:
                term_conditions.append(or_(*[
                    GlobalLiterature.chemical_list.cast(JSONB).contains([{"registry_number": t.text}])
                    for t in pos
                ]))
            if neg:
                neg_conds = [GlobalLiterature.chemical_list.cast(JSONB).contains([{"registry_number": t.text}]) for t in neg]
                term_conditions.append(not_(or_(*neg_conds)))
        if pp.substance_terms:
            pos = [t for t in pp.substance_terms if not t.is_not]
            neg = [t for t in pp.substance_terms if t.is_not]
            if pos:
                term_conditions.append(or_(*[
                    GlobalLiterature.chemical_list.cast(JSONB).contains([{"name": t.text}])
                    for t in pos
                ]))
            if neg:
                neg_conds = [GlobalLiterature.chemical_list.cast(JSONB).contains([{"name": t.text}]) for t in neg]
                term_conditions.append(not_(or_(*neg_conds)))
        if pp.databank_terms:
            pos = [t for t in pp.databank_terms if not t.is_not]
            neg = [t for t in pp.databank_terms if t.is_not]
            if pos:
                term_conditions.append(or_(*[
                    GlobalLiterature.databank_list.cast(JSONB).contains([{"accession_numbers": [t.text]}])
                    for t in pos
                ]))
            if neg:
                neg_conds = [GlobalLiterature.databank_list.cast(JSONB).contains([{"accession_numbers": [t.text]}]) for t in neg]
                term_conditions.append(not_(or_(*neg_conds)))

        # 5c. [PA] → pharmacological_actions JSONB contains (by name or ui)，支持 is_not
        if pp.pharmaco_terms:
            pos = [t for t in pp.pharmaco_terms if not t.is_not]
            neg = [t for t in pp.pharmaco_terms if t.is_not]
            if pos:
                term_conditions.append(or_(*[
                    GlobalLiterature.pharmacological_actions.cast(JSONB).contains([{"name": t.text}])
                    for t in pos
                ]))
            if neg:
                neg_conds = [GlobalLiterature.pharmacological_actions.cast(JSONB).contains([{"name": t.text}]) for t in neg]
                term_conditions.append(not_(or_(*neg_conds)))

        # P4: [OT] → keywords JSONB contains（不再映射到 all）
        if pp.ot_terms:
            pos = [t for t in pp.ot_terms if not t.is_not]
            neg = [t for t in pp.ot_terms if t.is_not]
            if pos:
                term_conditions.append(or_(*[
                    GlobalLiterature.keywords.cast(JSONB).contains([t.text])
                    for t in pos
                ]))
            if neg:
                neg_conds = [GlobalLiterature.keywords.cast(JSONB).contains([t.text]) for t in neg]
                term_conditions.append(not_(or_(*neg_conds)))

        # P4: [GEN] → gene_symbols JSONB contains
        if pp.gene_terms:
            pos = [t for t in pp.gene_terms if not t.is_not]
            neg = [t for t in pp.gene_terms if t.is_not]
            if pos:
                term_conditions.append(or_(*[
                    GlobalLiterature.gene_symbols.cast(JSONB).contains([t.text])
                    for t in pos
                ]))
            if neg:
                neg_conds = [GlobalLiterature.gene_symbols.cast(JSONB).contains([t.text]) for t in neg]
                term_conditions.append(not_(or_(*neg_conds)))

        # 5d. [ED] [IR] [PS] [PUBN] [AUID] [COIS] [TT] → 新增字段搜索，支持 is_not
        if pp.ed_terms:
            pos = [t for t in pp.ed_terms if not t.is_not]
            neg = [t for t in pp.ed_terms if t.is_not]
            if pos:
                term_conditions.append(or_(*[
                    GlobalLiterature.authors.cast(JSONB).contains([{"type": "editor", "family": t.text}])
                    for t in pos
                ]))
            if neg:
                neg_conds = [GlobalLiterature.authors.cast(JSONB).contains([{"type": "editor", "family": t.text}]) for t in neg]
                term_conditions.append(not_(or_(*neg_conds)))
        if pp.investigator_terms:
            pos = [t for t in pp.investigator_terms if not t.is_not]
            neg = [t for t in pp.investigator_terms if t.is_not]
            if pos:
                term_conditions.append(or_(*[
                    GlobalLiterature.investigators.cast(JSONB).contains([{"family": t.text}])
                    for t in pos
                ]))
            if neg:
                neg_conds = [GlobalLiterature.investigators.cast(JSONB).contains([{"family": t.text}]) for t in neg]
                term_conditions.append(not_(or_(*neg_conds)))
        if pp.personal_name_terms:
            pos = [t for t in pp.personal_name_terms if not t.is_not]
            neg = [t for t in pp.personal_name_terms if t.is_not]
            if pos:
                term_conditions.append(or_(*[
                    GlobalLiterature.personal_name_subjects.cast(JSONB).contains([{"family": t.text}])
                    for t in pos
                ]))
            if neg:
                neg_conds = [GlobalLiterature.personal_name_subjects.cast(JSONB).contains([{"family": t.text}]) for t in neg]
                term_conditions.append(not_(or_(*neg_conds)))
        if pp.pubnote_terms:
            pos = [t for t in pp.pubnote_terms if not t.is_not]
            neg = [t for t in pp.pubnote_terms if t.is_not]
            if pos:
                term_conditions.append(or_(*[
                    cast(GlobalLiterature.publication_notes, String).ilike(f"%{_escape_ilike(t.text)}%")
                    for t in pos
                ]))
            if neg:
                neg_conds = [cast(GlobalLiterature.publication_notes, String).ilike(f"%{_escape_ilike(t.text)}%") for t in neg]
                term_conditions.append(not_(or_(*neg_conds)))
        if pp.auid_terms:
            pos = [t for t in pp.auid_terms if not t.is_not]
            neg = [t for t in pp.auid_terms if t.is_not]
            if pos:
                term_conditions.append(or_(*[
                    GlobalLiterature.auid_data.cast(JSONB).contains([{"value": t.text}])
                    for t in pos
                ]))
            if neg:
                neg_conds = [GlobalLiterature.auid_data.cast(JSONB).contains([{"value": t.text}]) for t in neg]
                term_conditions.append(not_(or_(*neg_conds)))
        if pp.cois_terms:
            pos = [t for t in pp.cois_terms if not t.is_not]
            neg = [t for t in pp.cois_terms if t.is_not]
            if pos:
                term_conditions.append(or_(*[
                    GlobalLiterature.cois_statement.ilike(f"%{_escape_ilike(t.text)}%")
                    for t in pos
                ]))
            if neg:
                neg_conds = [GlobalLiterature.cois_statement.ilike(f"%{_escape_ilike(t.text)}%") for t in neg]
                term_conditions.append(not_(or_(*neg_conds)))
        if pp.tt_terms:
            pos = [t for t in pp.tt_terms if not t.is_not]
            neg = [t for t in pp.tt_terms if t.is_not]
            if pos:
                term_conditions.append(or_(*[
                    GlobalLiterature.vernacular_title.ilike(f"%{_escape_ilike(t.text)}%")
                    for t in pos
                ]))
            if neg:
                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
        #   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:
                    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
        if pp.stat_terms:
            pos = [t for t in pp.stat_terms if not t.is_not]
            neg = [t for t in pp.stat_terms if t.is_not]
            if pos:
                term_conditions.append(or_(*[
                    GlobalLiterature.citation_status == t.text.lower()
                    for t in pos
                ]))
            if neg:
                neg_conds = [GlobalLiterature.citation_status == t.text.lower() for t in neg]
                term_conditions.append(not_(or_(*neg_conds)))

        # P1-2: [UID] → PMID 优先，兜底 DOI
        if pp.uid_terms:
            for t in pp.uid_terms:
                cond = None
                try:
                    cond = GlobalLiterature.pmid == int(t.text)
                except ValueError:
                    cond = GlobalLiterature.doi.ilike(f"%{_escape_ilike(t.text)}%")
                if cond is not None:
                    term_conditions.append(not_(cond) if t.is_not else cond)

        # 处理括号分组的词（保留 OR/AND 嵌套结构，P2-2）
        if pp.groups:
            for idx, group in enumerate(pp.groups):
                group_conds = []
                all_not = all(t.is_not for t in group)
                for t in group:
                    cond = await AdvancedSearchEngine._single_term_condition(db, t)
                    if cond is not None:
                        if not all_not and t.is_not:
                            cond = not_(cond)
                        group_conds.append(cond)
                if group_conds:
                    gop = (pp.group_operators[idx]
                           if idx < len(pp.group_operators)
                           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:
                        combined = not_(combined)
                    term_conditions.append(combined)

        # 将 term_conditions 加入 conditions
        if term_conditions:
            if pp.boolean_operator == "or":
                from sqlalchemy.sql.elements import UnaryExpression
                from sqlalchemy.sql import operators as _sa_ops
                # OR 模式下 NOT 项应独立 AND（PubMed: A OR B NOT C = (A OR B) AND NOT C）
                pos_conds = [c for c in term_conditions
                             if not (isinstance(c, UnaryExpression) and c.modifier == _sa_ops.inv)]
                neg_conds = [c for c in term_conditions
                             if isinstance(c, UnaryExpression) and c.modifier == _sa_ops.inv]
                if neg_conds:
                    if pos_conds:
                        conditions.append(or_(*pos_conds))
                    conditions.extend(neg_conds)
                else:
                    conditions.append(or_(*term_conditions))
            else:
                conditions.extend(term_conditions)

        # 以下为非词条件（日期、PMID、DOI），始终 AND
        # 6. [DP] → 年份/日期范围
        dp_negated = "DP" in getattr(pp, 'negated_date_ranges', set())
        dp_conds = []
        if pp.year_from is not None:
            dp_conds.append(GlobalLiterature.pub_year >= pp.year_from)
        if pp.year_to is not None:
            dp_conds.append(GlobalLiterature.pub_year <= pp.year_to)
        if pp.date_from:
            from datetime import date as _dt_date
            try:
                dp_conds.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:
                dp_conds.append(GlobalLiterature.pub_date <= _dt_date.fromisoformat(pp.date_to))
            except ValueError:
                pass
        if dp_conds:
            cond = and_(*dp_conds) if len(dp_conds) > 1 else dp_conds[0]
            conditions.append(not_(cond) if dp_negated else cond)
        elif dp_negated:
            # negated_date_ranges includes DP but no conditions built — edge case guard
            pass

        # 6b. [EDAT] [CRDT] [MHDA] [LR] [DCOM] [DEP] → 日期字段范围
        DATE_FIELD_COLS = {
            "edat": (GlobalLiterature.entrez_date, "EDAT"),
            "crdt": (GlobalLiterature.create_date, "CRDT"),
            "mhda": (GlobalLiterature.meshed_date, "MHDA"),
            "lr": (GlobalLiterature.pubmed_revised, "LR"),
            "dcom": (GlobalLiterature.date_completed, "DCOM"),
            "dep": (GlobalLiterature.pub_date, "DEP"),  # P1-2: [DEP] → pub_date
        }
        for prefix, (col, field_tag) in DATE_FIELD_COLS.items():
            _from = getattr(pp, f"{prefix}_from", None)
            _to = getattr(pp, f"{prefix}_to", None)
            field_conds = []
            if _from:
                from datetime import date as _dt_date
                try:
                    field_conds.append(col >= _dt_date.fromisoformat(_from))
                except ValueError:
                    pass
            if _to:
                from datetime import date as _dt_date
                try:
                    field_conds.append(col <= _dt_date.fromisoformat(_to))
                except ValueError:
                    pass
            if field_conds:
                cond = and_(*field_conds) if len(field_conds) > 1 else field_conds[0]
                negated = field_tag in getattr(pp, 'negated_date_ranges', set())
                conditions.append(not_(cond) if negated else cond)

        # 7. [PMID] → 精确匹配，支持 is_not
        for term in pp.pmid_terms:
            cond = GlobalLiterature.pmid == int(term.text)
            if term.is_not:
                cond = not_(cond)
            conditions.append(cond)

        # 8. [DOI] → ILIKE，支持 is_not
        for term in pp.doi_terms:
            cond = GlobalLiterature.doi.ilike(f"%{_escape_ilike(term.text)}%")
            if term.is_not:
                cond = not_(cond)
            conditions.append(cond)

        # P4: [PMC] → pmc_id 精确匹配，支持 is_not
        for term in pp.pmc_terms:
            cond = GlobalLiterature.pmc_id == term.text
            if term.is_not:
                cond = not_(cond)
            conditions.append(cond)

        return conditions

    @staticmethod
    async def _single_term_condition(db: AsyncSession, term) -> object | None:
        """将单个解析后的 term 转为 SQLAlchemy condition，处理所有字段类型。

        P0-3: 确保括号组内的特殊字段（MH/PT/GR/RN 等）不被降级到 "all"，走正确的条件生成。
        """
        from sqlalchemy.dialects.postgresql import JSONB

        field = term.field
        # 简单 ILIKE 字段
        if field in ("title", "abstract", "all", "author", "journal", "affiliation",
                     "language", "volume", "issue", "pages", "lid") or field is None:
            return AdvancedSearchEngine._field_condition(field or "all", term.text, term.exact)

        # 特殊字段 — 与 _pubmed_conditions 中 top-level dispatch 一致
        if field == "MH":
            return await AdvancedSearchEngine._expand_mesh_tag_ids(db, [term.text], major_only=False, noexp=term._noexp)
        if field == "MAJR":
            return await AdvancedSearchEngine._expand_mesh_tag_ids(db, [term.text], major_only=True)
        if field == "PT":
            return GlobalLiterature.pub_types.cast(JSONB).contains([term.text])
        if field == "PMID":
            try:
                return GlobalLiterature.pmid == int(term.text)
            except ValueError:
                return None
        if field == "DOI":
            return GlobalLiterature.doi.ilike(f"%{_escape_ilike(term.text)}%")
        if field == "GR":
            return GlobalLiterature.grants.cast(JSONB).contains([{"grant_id": term.text}])
        if field == "SH":
            return GlobalLiterature.mesh_headings.cast(JSONB).contains([{"qualifiers": [term.text]}])
        if field == "RN":
            return GlobalLiterature.chemical_list.cast(JSONB).contains([{"registry_number": term.text}])
        if field == "NM":
            return GlobalLiterature.chemical_list.cast(JSONB).contains([{"name": term.text}])
        if field == "SI":
            return GlobalLiterature.databank_list.cast(JSONB).contains([{"accession_numbers": [term.text]}])
        if field == "PA":
            return GlobalLiterature.pharmacological_actions.cast(JSONB).contains([{"name": term.text}])
        if field == "ED":
            return GlobalLiterature.authors.cast(JSONB).contains([{"type": "editor", "family": term.text}])
        if field == "IR":
            return GlobalLiterature.investigators.cast(JSONB).contains([{"family": term.text}])
        if field == "PS":
            return GlobalLiterature.personal_name_subjects.cast(JSONB).contains([{"family": term.text}])
        if field == "PUBN":
            return cast(GlobalLiterature.publication_notes, String).ilike(f"%{_escape_ilike(term.text)}%")
        if field == "AUID":
            return GlobalLiterature.auid_data.cast(JSONB).contains([{"value": term.text}])
        if field == "COIS":
            return GlobalLiterature.cois_statement.ilike(f"%{_escape_ilike(term.text)}%")
        if field == "TT":
            return GlobalLiterature.vernacular_title.ilike(f"%{_escape_ilike(term.text)}%")
        if field == "SB":
            val = term.text.upper()
            if val == "PUBMED":
                return text("TRUE")  # no-op: 所有记录都是 PubMed
            elif val == "MEDLINE":
                return GlobalLiterature.citation_status == "medline"
            elif len(val) == 1 and val.isalpha():
                subq = select(GlobalJournal.issn).where(
                    GlobalJournal.nlm_subsets.overlap([val])
                )
                return GlobalLiterature.journal_issn.in_(subq)
            else:
                return GlobalLiterature.citation_status == val.lower()
        if field == "STAT":
            return GlobalLiterature.citation_status == term.text.lower()
        if field == "UID":
            try:
                return GlobalLiterature.pmid == int(term.text)
            except ValueError:
                return GlobalLiterature.doi.ilike(f"%{_escape_ilike(term.text)}%")
        # P4: [OT] → keywords JSONB contains
        if field == "OT":
            return GlobalLiterature.keywords.cast(JSONB).contains([term.text])
        # P4: [GEN] → gene_symbols JSONB contains
        if field == "GEN":
            return GlobalLiterature.gene_symbols.cast(JSONB).contains([term.text])
        # P4: [PMC] → pmc_id 精确匹配
        if field == "PMC":
            return GlobalLiterature.pmc_id == term.text

        # 回退
        return AdvancedSearchEngine._field_condition("all", term.text, term.exact)

    @staticmethod
    def _field_condition(field: str, term: str, exact: bool) -> callable:
        _wildcard = term.endswith('*') and not exact
        _stem = term[:-1] if _wildcard else term
        _escaped = _escape_ilike(_stem)

        def _pt() -> str:
            """P2-1: wildcard → 右截断 ILIKE；其他 → 双百搭"""
            return f"{_escaped}%" if _wildcard else f"%{_escaped}%"

        if field == "title":
            return GlobalLiterature.title.ilike(_pt())
        elif field == "abstract":
            return GlobalLiterature.abstract.ilike(_pt())
        elif field == "author":
            return GlobalLiterature.author_names_text.ilike(_pt())
        elif field == "journal":
            pat = _pt()
            return or_(
                GlobalLiterature.journal.ilike(pat),
                GlobalLiterature.journal_iso.ilike(pat),
            )
        elif field == "affiliation":
            return cast(GlobalLiterature.authors, String).ilike(_pt())
        elif field == "language":
            return GlobalLiterature.language.ilike(_pt())
        elif field == "volume":
            return GlobalLiterature.volume.ilike(_pt())
        elif field == "issue":
            return GlobalLiterature.issue.ilike(_pt())
        elif field == "pages":
            return GlobalLiterature.pages.ilike(_pt())
        elif field == "lid":
            pat = _pt()
            return or_(
                GlobalLiterature.doi.ilike(pat),
                GlobalLiterature.pmc_id.ilike(pat),
            )
        else:  # "all" default
            pat = _pt()
            if exact and not _wildcard:
                # P4: 精确短语 → phraseto_tsquery（利用 GIN 索引，保留词序）
                return GlobalLiterature.search_tsv.op("@@")(func.phraseto_tsquery("english", term))
            if _wildcard:
                # wildcard → ILIKE 右截断（tsvector 不支持 *），多字段覆盖
                return or_(
                    GlobalLiterature.title.ilike(pat),
                    GlobalLiterature.abstract.ilike(pat),
                    cast(GlobalLiterature.pmid, String).ilike(pat),
                    GlobalLiterature.doi.ilike(pat),
                )
            like_val = f"%{_escaped}%"
            if "/" in term:
                if term.startswith("10."):
                    return or_(
                        GlobalLiterature.doi.ilike(_escape_ilike(term)),
                        GlobalLiterature.doi.ilike(like_val),
                    )
                return or_(
                    GlobalLiterature.title.ilike(like_val),
                    cast(GlobalLiterature.pmid, String).ilike(like_val),
                    GlobalLiterature.doi.ilike(like_val),
                )
            # P7-D2: Chinese → ILIKE fallback (tsvector is English-only)
            if re.search(r'[一-鿿㐀-䶿豈-﫿]', term):
                return or_(
                    GlobalLiterature.title.ilike(like_val),
                    GlobalLiterature.abstract.ilike(like_val),
                )
            return GlobalLiterature.search_tsv.op("@@")(func.plainto_tsquery("english", term))

    @staticmethod
    async def _expand_mesh_tag_ids(
        db: AsyncSession,
        mesh_names: list[str],
        major_only: bool = False,
        noexp: bool = False,  # P1-4: [MH:noexp] 抑制树展开
    ) -> object | None:
        """用 GlobalTagTreeNumber 展开 [MH]/[MAJR]

        1. 先精确入口词匹配（entry_terms JSONB @>），再退到 name_en ILIKE
        2. 用 tree_number 前缀展开子节点（C04.588 → 所有子树号）
        3. [MAJR] 额外 AND is_major=True
        返回 SQLAlchemy condition 或 None（无匹配时）。

        noexp=True 时跳过第 2 步（树展开），只搜精确词。
        """
        import uuid as _uuid
        mesh_tag_ids: set[_uuid.UUID] = set()

        # Batch all mesh name lookups — 2 queries instead of 2N
        entry_conds = []
        name_conds = []
        for m in mesh_names:
            q = m.strip().lower()
            if not q:
                continue
            # 1a. 精确入口词匹配（P1-3）— batch via OR
            entry_conds.append(GlobalTag.entry_terms.contains([q]))
            # 1b. name_en ILIKE 回退 — batch via OR
            name_conds.append(GlobalTag.name_en.ilike(_escape_ilike(m)))

        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)
        except Exception:
            logger.exception("MeSH tag lookup failed for mesh_names=%s", mesh_names[:5])

        if not mesh_tag_ids:
            return None

        # tree_number 前缀展开：取匹配 tag 的所有 tree_number，查子节点（[MH:noexp] 时跳过）
        if not noexp:
            try:
                tns = (await db.execute(
                    select(GlobalTagTreeNumber.tree_number).where(
                        GlobalTagTreeNumber.tag_id.in_(list(mesh_tag_ids))
                    ).distinct()
                )).scalars().all()

                if tns:
                    child_conds = [or_(
                        GlobalTagTreeNumber.tree_number == tn,
                        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:
                logger.exception("Tree number expansion failed for mesh_names=%s", mesh_names[:5])

        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),
            )
        return GlobalLiterature.id.in_(subq)

    @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()

    @staticmethod
    def _apply_order_by(sort: str, relevance_query: str):
        """Return a list of order_by expressions for the given sort mode."""
        if sort == "date":
            return [GlobalLiterature.pub_date.desc().nullslast(), GlobalLiterature.id.desc()]
        elif sort == "cited":
            return [GlobalLiterature.cited_by_count.desc().nullslast(), GlobalLiterature.id.desc()]
        elif sort == "best_match" and relevance_query.strip():
            tsq = func.plainto_tsquery("english", relevance_query)
            return [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)
            return [rank.desc()]
        elif sort == "first_author":
            return [GlobalLiterature.authors[0]['family'].astext.asc().nullslast()]
        elif sort == "journal":
            return [GlobalLiterature.journal.asc().nullslast()]
        elif sort == "title":
            return [GlobalLiterature.title.asc().nullslast()]
        else:
            return [GlobalLiterature.pub_date.desc().nullslast(), GlobalLiterature.id.desc()]