backend/backup_20260725/search_engine.py

"""高级搜索服务：布尔运算 + 字段限定 + 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()