test: 更新 MeSH 打标测试适配 tag_service 重构

- _tag_article → tag_service.tag_article 导入路径
- 增加 name_en 查询 mock 返回值（3 次 execute 调用）

backend/alembic/versions/e0764f6d7c21_add_tag_ids_array_brin_indexes.py

@@ -0,0 +1,62 @@
+"""add_tag_ids_array_brin_indexes
+
+Revision ID: e0764f6d7c21
+Revises: g0h1i2j3k4l5
+Create Date: 2026-07-27 16:14:29.454432
+"""
+from typing import Sequence, Union
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+revision: str = 'e0764f6d7c21'
+down_revision: Union[str, None] = 'g0h1i2j3k4l5'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_index(op.f('ix_audit_action'), table_name='audit_logs')
+    op.drop_index(op.f('ix_audit_actor'), table_name='audit_logs')
+    op.drop_table('audit_logs')
+    op.add_column('global_literature', sa.Column('tag_ids', postgresql.ARRAY(sa.Uuid()), nullable=True))
+    op.drop_index(op.f('ix_gl_authors_text_trgm'), table_name='global_literature', postgresql_ops={'(authors::text)': 'gin_trgm_ops'}, postgresql_using='gin')
+    op.create_index('ix_gl_is_negative_true', 'global_literature', ['is_negative_result'], unique=False, postgresql_where=sa.text('is_negative_result = TRUE'))
+    op.create_index('ix_gl_is_oa_true', 'global_literature', ['is_oa'], unique=False, postgresql_where=sa.text('is_oa = TRUE'))
+    op.create_index('ix_gl_is_preprint_true', 'global_literature', ['is_preprint'], unique=False, postgresql_where=sa.text('is_preprint = TRUE'))
+    op.create_index('ix_gl_pub_date_brin', 'global_literature', ['pub_date'], unique=False, postgresql_using='brin')
+    op.create_index('ix_gl_pub_date_covering', 'global_literature', ['pub_date', 'id'], unique=False, postgresql_include={'citation_status', 'is_oa', 'cited_by_count', 'is_preprint', 'pub_year', 'doi', 'journal_issn', 'is_negative_result', 'article_date', 'retracted', 'journal', 'language', 'pmc_id'})
+    op.create_index('ix_gl_pub_year_brin', 'global_literature', ['pub_year'], unique=False, postgresql_using='brin')
+    op.create_index('ix_gl_retracted_true', 'global_literature', ['retracted'], unique=False, postgresql_where=sa.text('retracted = TRUE'))
+    op.create_index('ix_gl_tag_ids_gin', 'global_literature', ['tag_ids'], unique=False, postgresql_using='gin')
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_index('ix_gl_tag_ids_gin', table_name='global_literature', postgresql_using='gin')
+    op.drop_index('ix_gl_retracted_true', table_name='global_literature', postgresql_where=sa.text('retracted = TRUE'))
+    op.drop_index('ix_gl_pub_year_brin', table_name='global_literature', postgresql_using='brin')
+    op.drop_index('ix_gl_pub_date_covering', table_name='global_literature', postgresql_include={'citation_status', 'is_oa', 'cited_by_count', 'is_preprint', 'pub_year', 'doi', 'journal_issn', 'is_negative_result', 'article_date', 'retracted', 'journal', 'language', 'pmc_id'})
+    op.drop_index('ix_gl_pub_date_brin', table_name='global_literature', postgresql_using='brin')
+    op.drop_index('ix_gl_is_preprint_true', table_name='global_literature', postgresql_where=sa.text('is_preprint = TRUE'))
+    op.drop_index('ix_gl_is_oa_true', table_name='global_literature', postgresql_where=sa.text('is_oa = TRUE'))
+    op.drop_index('ix_gl_is_negative_true', table_name='global_literature', postgresql_where=sa.text('is_negative_result = TRUE'))
+    op.create_index(op.f('ix_gl_authors_text_trgm'), 'global_literature', [sa.literal_column('(authors::text)')], unique=False, postgresql_ops={'(authors::text)': 'gin_trgm_ops'}, postgresql_using='gin')
+    op.drop_column('global_literature', 'tag_ids')
+    op.create_table('audit_logs',
+    sa.Column('id', sa.UUID(), autoincrement=False, nullable=False),
+    sa.Column('actor_id', sa.UUID(), autoincrement=False, nullable=False),
+    sa.Column('tenant_id', sa.UUID(), autoincrement=False, nullable=True),
+    sa.Column('action', sa.VARCHAR(length=50), autoincrement=False, nullable=False),
+    sa.Column('target_type', sa.VARCHAR(length=50), autoincrement=False, nullable=True),
+    sa.Column('target_id', sa.VARCHAR(length=100), autoincrement=False, nullable=True),
+    sa.Column('detail', sa.TEXT(), autoincrement=False, nullable=True),
+    sa.Column('ip_address', sa.VARCHAR(length=45), autoincrement=False, nullable=True),
+    sa.Column('created_at', postgresql.TIMESTAMP(timezone=True), server_default=sa.text('now()'), autoincrement=False, nullable=False),
+    sa.PrimaryKeyConstraint('id', name=op.f('audit_logs_pkey'))
+    )
+    op.create_index(op.f('ix_audit_actor'), 'audit_logs', ['actor_id', 'created_at'], unique=False)
+    op.create_index(op.f('ix_audit_action'), 'audit_logs', ['action', 'created_at'], unique=False)
+    # ### end Alembic commands ###

backend/app/api/v1/admin.py

@@ -1468,7 +1468,7 @@ async def retag_tags(
     此接口仅用于回填管道上线前已存在的文献。
     """
     import logging
-    from app.services.pubmed_api import _tag_article
+    from app.services.tag_service import tag_article
 
     logger = logging.getLogger(__name__)
 
@@ -1494,7 +1494,7 @@ async def retag_tags(
         mh = lit.mesh_headings
         if not mh or (isinstance(mh, list) and len(mh) == 0):
             continue
-        n = await _tag_article(db, lit.id, mh)
+        n = await tag_article(db, lit.id, mh)
         if n:
             tagged_count += 1
             tags_added += n

backend/app/api/v1/features.py

@@ -68,8 +68,8 @@ class AdvancedSearchRequest(BaseModel):
     page: int = Field(1, ge=1)
     page_size: int = Field(20, ge=1, le=100)
     sort: str = "date"
-    # keyset 游标分页（设了 cursor 后 page 参数被忽略，不做 COUNT）
+    # keyset 游标分页（所有排序模式通用，设了 cursor 后 page 参数被忽略，不做 COUNT）
-    cursor_date: str | None = None   # 上一页最后一条的 pub_date（ISO 日期）
+    cursor_val: str | None = None    # 上一页最后一条的排序列值（字符串，服务端按 sort 模式解析）
     cursor_id: str | None = None     # 上一页最后一条的 id（UUID 字符串）
 
     # ── PubMed 筛选器参数 ──

backend/app/core/cache.py

@@ -66,6 +66,18 @@ class CacheService:
             self._store.popitem(last=False)
         return True
 
+    async def mget(self, keys: list[str]) -> list[dict | None]:
+        """批量获取，顺序对应 keys 列表。Redis 不可用时降级到内存模式。"""
+        r = await self._get_redis()
+        if r:
+            try:
+                vals = await r.mget(*keys)
+                return [json.loads(v) if v else None for v in vals]
+            except Exception:
+                logger.exception("Redis MGET failed")
+                return [None] * len(keys)
+        return [self._store.get(k) for k in keys]
+
     async def delete(self, key: str):
         r = await self._get_redis()
         if r:

backend/app/models/literature.py

@@ -3,7 +3,7 @@
 import uuid
 from datetime import date, datetime
 
-from sqlalchemy import Boolean, Date, DateTime, ForeignKey, Index, Integer, String, Text, Uuid, UniqueConstraint, func
+from sqlalchemy import Boolean, Date, DateTime, ForeignKey, Index, Integer, String, Text, Uuid, UniqueConstraint, func, text
 from sqlalchemy.dialects.postgresql import ARRAY, JSONB, TSVECTOR
 from sqlalchemy.orm import Mapped, mapped_column
 
@@ -78,11 +78,33 @@ class GlobalLiterature(Base):
     suppl_mesh_list: Mapped[dict] = mapped_column(JSONB, default=list)       # SupplMeshList [{descriptor, ui, type}]
     pharmacological_actions: Mapped[dict] = mapped_column(JSONB, default=list)  # PharmacologicalAction [{name, ui}]
 
+    tag_ids: Mapped[list | None] = mapped_column(ARRAY(Uuid))       # 标签 ID 数组 + GIN，免 JOIN global_literature_tags
+
     source: Mapped[str] = mapped_column(String(30), default="pubmed_ftp")
     created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
     updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
 
     __table_args__ = (
+        # 时序数据 BRIN 代替 B-tree：pub_date/pub_year 物理相关性高，BRIN 体积小两个数量级
+        Index("ix_gl_pub_date_brin", "pub_date", postgresql_using="brin"),
+        Index("ix_gl_pub_year_brin", "pub_year", postgresql_using="brin"),
+
+        # Covering index：date 排序 + 高频引用列（index-only scan，排除大 TEXT/JSONB）
+        Index("ix_gl_pub_date_covering", "pub_date", "id",
+              postgresql_include={"journal_issn", "cited_by_count", "is_oa", "retracted",
+                                  "is_negative_result", "is_preprint", "journal", "pub_year",
+                                  "article_date", "doi", "pmc_id", "language", "citation_status"}),
+
+        # 高频筛选 partial index
+        Index("ix_gl_retracted_true", "retracted", postgresql_where=text("retracted = TRUE")),
+        Index("ix_gl_is_oa_true", "is_oa", postgresql_where=text("is_oa = TRUE")),
+        Index("ix_gl_is_negative_true", "is_negative_result", postgresql_where=text("is_negative_result = TRUE")),
+        Index("ix_gl_is_preprint_true", "is_preprint", postgresql_where=text("is_preprint = TRUE")),
+
+        # 反范式 tag_ids GIN 索引：标签筛选免 JOIN global_literature_tags
+        Index("ix_gl_tag_ids_gin", "tag_ids", postgresql_using="gin"),
+
+        # 保留旧 B-tree 兼容（迁移后旧索引下线，IR 移除时机：所有实例完成 migration）
         Index("ix_gl_pub_date", "pub_date"),
         Index("ix_gl_pub_year", "pub_year"),
         Index("ix_gl_journal_issn", "journal_issn"),

backend/app/services/search_engine.py

@@ -38,7 +38,7 @@ class AdvancedSearchEngine:
         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,
+        page_size: int, sort: str,
         # PubMed filter params
         has_abstract: bool | None = None,
         is_free_full_text: bool | None = None,
@@ -49,8 +49,8 @@ class AdvancedSearchEngine:
         age: list[str] | None = None,
         medline_only: bool = False,
         exclude_preprints: bool = False,
-        # Keyset cursor (sort=date 时分页)
+        # Generic keyset cursor
-        cursor_date: str | None = None,
+        cursor_val: str | None = None,
         cursor_id: str | None = None,
     ) -> str:
         """归一化查询参数 → 确定性缓存 key（所有 list 排序后参与哈希）"""
@@ -66,7 +66,7 @@ class AdvancedSearchEngine:
             "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,
+            "ps": page_size, "s": sort,
             "ha": has_abstract,
             "fft": is_free_full_text,
             "hft": has_full_text,
@@ -76,13 +76,63 @@ class AdvancedSearchEngine:
             "ag": sorted(age) if age else [],
             "mo": medline_only,
             "epr": exclude_preprints,
-            # keyset cursor 唯一标识翻页位置
+            # keyset cursor 唯一标识翻页位置（不含时为第 1 页）
-            "cd": cursor_date,
+            "cv": cursor_val,
             "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
+    def _facet_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,
+        # PubMed filter params (same as _search_cache_key minus page/cursor/sort)
+        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,
+    ) -> str:
+        """归一化筛选条件 → year_counts/facet 缓存 key（不含 page/cursor/sort）
+
+        year_counts 只依赖筛选条件，分页/排序不影响分布。
+        与 _search_cache_key 的区别：不含 p/ps/s/cd/ci。
+        """
+        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 [],
+            "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,
+        }
+        raw = json.dumps(norm, sort_keys=True, ensure_ascii=False, default=str)
+        return f"search:year_counts:{hashlib.md5(raw.encode()).hexdigest()}"
+
     @staticmethod
     async def search(
         db: AsyncSession,
@@ -106,8 +156,9 @@ class AdvancedSearchEngine:
         page: int = 1,
         page_size: int = 20,
         sort: str = "date",
-        cursor_date: str | None = None,  # keyset 游标：上一页最后一条的 pub_date
+        cursor_val: str | None = None,   # 通用 keyset 游标值（sort 模式对应列的值）
         cursor_id: str | None = None,    # keyset 游标：上一页最后一条的 id（UUID）
+        cursor_date: str | None = None,  # 向后兼容：旧 date 游标，映射到 cursor_val
         # ── PubMed 筛选器参数 ──
         has_abstract: bool | None = None,
         is_free_full_text: bool | None = None,
@@ -123,15 +174,18 @@ class AdvancedSearchEngine:
         from sqlalchemy.dialects.postgresql import JSONB
         conditions = []
 
-        # 60s 缓存（keyset cursor 页也参与缓存，cursor_date+cursor_id 组合唯一标识翻页位置）
+        # 向后兼容：cursor_date → cursor_val（旧前端发 cursor_date）
-        use_cursor = (cursor_date is not None and cursor_id is not None and sort == "date")
+        if cursor_val is None and cursor_date is not None:
+            cursor_val = cursor_date
+
+        is_first_page = (cursor_val is None and cursor_id is None)
         _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,
+            page_size, sort,
             has_abstract=has_abstract,
             is_free_full_text=is_free_full_text,
             has_full_text=has_full_text,
@@ -139,12 +193,41 @@ class AdvancedSearchEngine:
             species=species, sex=sex, age=age,
             medline_only=medline_only,
             exclude_preprints=exclude_preprints,
-            cursor_date=cursor_date,
+            cursor_val=cursor_val,
             cursor_id=cursor_id,
         )
+        # year_counts 缓存键（不含 page/cursor/sort，所有页共享一份）
+        _facet_cache_key = AdvancedSearchEngine._facet_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,
+            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,
+        )
         cached = await _cache.get(_search_cache_key)
         if cached is not None:
-            return cached
+            # 谢缓存：只存了 lit_ids + 聚合数据，需从 ②③ 缓存回填 tags/journal
+            tier_map, name_map = await AdvancedSearchEngine._get_journal_map(db)
+            items = await AdvancedSearchEngine._hydrate_items(db, cached["lit_ids"], tier_map, name_map)
+            # year_counts 从独立 facet 缓存取
+            year_counts = await _cache.get(_facet_cache_key) or []
+            return {
+                "items": items,
+                "total": cached["total"],
+                "page": cached["page"],
+                "page_size": cached["page_size"],
+                "has_more": cached["has_more"],
+                "year_counts": year_counts,
+                "cursor_val": cached.get("cursor_val"),
+                "cursor_id": cached.get("cursor_id"),
+            }
 
         # 30 秒查询超时（放在缓存检查之后，缓存命中不执行）
         await db.execute(text("SET LOCAL statement_timeout = '30s'"))
@@ -353,8 +436,8 @@ class AdvancedSearchEngine:
                 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))
+            from sqlalchemy.dialects.postgresql import array as _pg_array
-            conditions.append(GlobalLiterature.id.in_(subq))
+            conditions.append(GlobalLiterature.tag_ids.overlap(_pg_array(uids)))
 
         # 发表类型（PG JSONB contains）
         if pub_types:
@@ -446,7 +529,12 @@ class AdvancedSearchEngine:
         _yr_before = len(conditions)
         year_counts = []
 
-        # 判断是否有任何筛选器/文本查询活跃（与旧 gate 逻辑一致）
+        # 先从 facet 缓存取（不含 page/cursor，所有页共享，TTL 更长）
+        yc_cached = await _cache.get(_facet_cache_key)
+        if yc_cached is not None:
+            year_counts = yc_cached if isinstance(yc_cached, list) else yc_cached.get("year_counts", [])
+
+        # 判断是否有任何筛选器/文本查询活跃
         _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
@@ -455,13 +543,13 @@ class AdvancedSearchEngine:
                            or has_associated_data or species or sex or age
                            or medline_only or exclude_preprints)
 
-        # ── 无文本无筛选 → 走 Redis 预计算缓存（每天 pipeline 后 cron 刷新） ──
+        if year_counts:
-        if not _has_any_filter:
+            pass  # facet 缓存命中
+        elif 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(
@@ -477,8 +565,8 @@ class AdvancedSearchEngine:
                 except Exception:
                     logger.exception("Year counts query failed")
                     year_counts = []
+                await _cache.set("search:year_counts:all", year_counts, ttl=1800)
 
-        # ── 有筛选条件 → 精确 GROUP BY（已移除旧 30 年限制，pub_year 低基数因此 GROUP BY 高效） ──
         elif conditions and _has_any_filter:
             try:
                 yr_conds = conditions[:_yr_before]
@@ -495,6 +583,7 @@ class AdvancedSearchEngine:
             except Exception:
                 logger.exception("Year counts query failed")
                 year_counts = []
+            await _cache.set(_facet_cache_key, year_counts, ttl=1800)
 
         # 排序（PubMed 查询时跳过 ts_rank，避免语法标签噪音）
         _relevance_query = query
@@ -504,43 +593,26 @@ class AdvancedSearchEngine:
             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")
+        # ── 通用 keyset 分页（所有列式排序模式统一，代替 OFFSET） ──
-        if use_keyset:
+        _keyset_cond = AdvancedSearchEngine._keyset_condition(sort, cursor_val, cursor_id)
-            from datetime import date as dt_date
+        if _keyset_cond is not None:
-            try:
+            conditions.append(_keyset_cond)
-                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))
 
-        # 分页
+        # ── LIMIT page_size+1 探测下一页 + COUNT 第 1 页缓存 ──
-        has_more = False
+        result = await db.execute(q.limit(page_size + 1))
-        if use_keyset:
+        items = result.scalars().all()
-            # cursor 模式：取 page_size+1 条判断是否有下一页，不做 COUNT
+        has_more = len(items) > page_size
-            result = await db.execute(q.limit(page_size + 1))
+        items = items[:page_size]
-            items = result.scalars().all()
+
-            has_more = len(items) > page_size
+        # COUNT 只在第 1 页计算，缓存到 facet key 供后续页复用
-            items = items[:page_size]
+        total = 0
-            total = 0
+        if is_first_page:
-        else:
-            # 总数
             count_q = select(func.count()).select_from(
                 select(literal_column("1"))
                 .select_from(GlobalLiterature)
@@ -548,36 +620,92 @@ class AdvancedSearchEngine:
                 .subquery()
             )
             total = (await db.execute(count_q)).scalar() or 0
-            offset = (page - 1) * page_size
+            await _cache.set(_facet_cache_key, {"year_counts": year_counts, "total": total}, ttl=1800)
+        else:
+            # 后续页从 facet 缓存读 total
+            facet_cached = await _cache.get(_facet_cache_key)
+            if facet_cached:
+                total = facet_cached.get("total", 0)
 
-            result = await db.execute(q.offset(offset).limit(page_size))
+        # 构建游标供翻页
-            items = result.scalars().all()
+        next_cursor_val = None
-            has_more = (offset + page_size) < total and len(items) == page_size
+        next_cursor_id = None
+        if has_more and items:
+            last = items[-1]
+            next_cursor_val = AdvancedSearchEngine._cursor_from_item(last, sort)
+            next_cursor_id = str(last.id)
 
         tm = await load_tags_for_literature(db, [str(lit.id) for lit in items])
 
-        # Batch load journal tiers + canonical names
+        # 从全局缓存加载期刊 tier/name（避免每页 IN 查询）
-        tier_map = {}
+        tier_map, name_map = await AdvancedSearchEngine._get_journal_map(db)
-        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 = AdvancedSearchEngine._build_item_dicts(items, tm, tier_map, name_map)
+
+        result = {
+            "items": results,
+            "total": total,
+            "page": page,
+            "page_size": page_size,
+            "has_more": has_more,
+            "year_counts": year_counts,
+            "cursor_val": next_cursor_val,
+            "cursor_id": next_cursor_id,
+        }
+
+        if _search_cache_key is not None:
+            # 只缓存 lit_ids + 聚合数据，tags/journal 从 ②③ 缓存取
+            slim = {
+                "lit_ids": [str(lit.id) for lit in items],
+                "total": total,
+                "page": page,
+                "page_size": page_size,
+                "has_more": has_more,
+                "cursor_val": next_cursor_val,
+                "cursor_id": next_cursor_id,
+            }
+            await _cache.set(_search_cache_key, slim, ttl=AdvancedSearchEngine.SEARCH_CACHE_TTL)
+
+        return result
+
+    @staticmethod
+    async def _get_journal_map(db: AsyncSession) -> tuple[dict[str, int | None], dict[str, str | None]]:
+        """返回全局 (tier_map, name_map)，以 journal_issn 为 key。
+
+        缓存在 Redis 中（journals:map），TTL 3600s。
+        期刊 tier/name 几乎从不变化，全表查询一次即可。
+        """
+        cached = await _cache.get("journals:map")
+        if cached is not None:
+            tier_map = cached.get("tier_map", {})
+            name_map = cached.get("name_map", {})
+        else:
+            rows = (await db.execute(
+                select(GlobalJournal.issn, GlobalJournal.tier, GlobalJournal.name)
+            )).all()
+            tier_map = {issn: tier for issn, tier, name_ in rows}
+            name_map = {issn: name_ for issn, tier, name_ in rows}
+            await _cache.set("journals:map", {"tier_map": tier_map, "name_map": name_map}, ttl=3600)
+        return tier_map, name_map
+
+    @staticmethod
+    def _build_item_dicts(
+        items: list,
+        tag_map: dict[str, list[dict]],
+        tier_map: dict[str, int | None],
+        name_map: dict[str, str | None],
+    ) -> list[dict]:
+        """将 GlobalLiterature 模型列表转为 API 响应所需的 dict 列表"""
         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),
+                "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), []),
+                "pub_year": lit.pub_year, "tags": tag_map.get(str(lit.id), []),
                 "abstract": lit.abstract[:300] if lit.abstract else None,
                 "doi": lit.doi,
                 "pmc_id": lit.pmc_id,
@@ -590,13 +718,29 @@ class AdvancedSearchEngine:
                 "pub_types": lit.pub_types,
                 "affiliation": authors[0].get("affiliation", "") if authors else "",
             })
+        return results
 
-        result = {"items": results, "total": total, "page": page, "page_size": page_size, "has_more": has_more, "year_counts": year_counts}
+    @staticmethod
+    async def _hydrate_items(
+        db: AsyncSession,
+        lit_ids: list[str],
+        tier_map: dict[str, int | None],
+        name_map: dict[str, str | None],
+    ) -> list[dict]:
+        """从 lit_ids 重新构建 items（用于缓存命中时的回填）
 
-        if _search_cache_key is not None:
+        从 DB 按 ID 查询文献、从 ②③ 缓存加载 tags/journal，避免重算复杂 WHERE。
-            await _cache.set(_search_cache_key, result, ttl=AdvancedSearchEngine.SEARCH_CACHE_TTL)
+        """
-
+        import uuid as _uuid
-        return result
+        uids = [_uuid.UUID(s) for s in lit_ids]
+        rows = (await db.execute(
+            select(GlobalLiterature).where(GlobalLiterature.id.in_(uids))
+        )).scalars().all()
+        # 保持传入顺序（PostgreSQL WHERE id IN 不保证排序）
+        id_order = {str(lit.id): lit for lit in rows}
+        ordered = [id_order[sid] for sid in lit_ids if sid in id_order]
+        tag_map = await load_tags_for_literature(db, lit_ids)
+        return AdvancedSearchEngine._build_item_dicts(ordered, tag_map, tier_map, name_map)
 
     @staticmethod
     async def _pubmed_conditions(
@@ -1219,72 +1363,91 @@ class AdvancedSearchEngine:
         返回 SQLAlchemy condition 或 None（无匹配时）。
 
         noexp=True 时跳过第 2 步（树展开），只搜精确词。
+
+        缓存策略：atm:{md5(归一化查询+参数)} → expanded_tag_ids list。
+        GlobalTag/TreeNumber 几乎不变，TTL 1 小时。
         """
+        import hashlib as _hashlib
         import uuid as _uuid
-        mesh_tag_ids: set[_uuid.UUID] = set()
 
-        # Batch all mesh name lookups — 2 queries instead of 2N
+        # 计算缓存键
-        entry_conds = []
+        items = sorted(m.strip().lower() for m in mesh_names if m.strip())
-        name_conds = []
+        if not items:
-        for m in mesh_names:
+            return None
-            q = m.strip().lower()
+        raw_key = f"{'|'.join(items)}:major={major_only}:noexp={noexp}"
-            if not q:
+        cache_key = f"atm:{_hashlib.md5(raw_key.encode()).hexdigest()}"
-                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:
+        cached = await _cache.get(cache_key)
-                rows = (await db.execute(
+        if cached is not None:
-                    select(GlobalTag.id).where(
+            mesh_tag_ids = {_uuid.UUID(uid) for uid in cached["tag_ids"]}
-                        GlobalTag.source.in_(["mesh", "manual"]),
+        else:
-                        GlobalTag.mesh_ui.isnot(None),
+            mesh_tag_ids: set[_uuid.UUID] = set()
-                        GlobalTag.entry_terms.isnot(None),
-                        or_(*entry_conds),
-                    )
-                )).all()
-                for (tid,) in rows:
-                    mesh_tag_ids.add(tid)
 
-            if name_conds:
+            # Batch all mesh name lookups — 2 queries instead of 2N
-                rows = (await db.execute(
+            entry_conds = []
-                    select(GlobalTag.id).where(
+            name_conds = []
-                        GlobalTag.source.in_(["mesh", "manual"]),
+            for m in mesh_names:
-                        GlobalTag.mesh_ui.isnot(None),
+                q = m.strip().lower()
-                        or_(*name_conds),
+                if not q:
-                    )
+                    continue
-                )).all()
+                entry_conds.append(GlobalTag.entry_terms.contains([q]))
-                for (tid,) in rows:
+                name_conds.append(GlobalTag.name_en.ilike(_escape_ilike(m)))
-                    mesh_tag_ids.add(tid)
+
-        except Exception:
+            try:
-            logger.exception("MeSH tag lookup failed for mesh_names=%s", mesh_names[:5])
+                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 mesh_tag_ids:
+                # tree_number 前缀展开
+                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])
+
+            # 写缓存（即使为空也缓存，避免重复查空）
+            await _cache.set(cache_key, {"tag_ids": [str(tid) for tid in mesh_tag_ids]}, ttl=3600)
 
         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(
@@ -1319,6 +1482,8 @@ class AdvancedSearchEngine:
         )
         return score.desc()
 
+    KEYSET_COLUMN_SORTS = {"date", "cited", "title", "journal", "first_author"}
+
     @staticmethod
     def _apply_order_by(sort: str, relevance_query: str):
         """Return a list of order_by expressions for the given sort mode."""
@@ -1341,3 +1506,73 @@ class AdvancedSearchEngine:
             return [GlobalLiterature.title.asc().nullslast()]
         else:
             return [GlobalLiterature.pub_date.desc().nullslast(), GlobalLiterature.id.desc()]
+
+    @staticmethod
+    def _keyset_condition(sort: str, cursor_val: str | None, cursor_id: str | None) -> object | None:
+        """构建 keyset WHERE 条件（适用所有列式排序模式）。
+
+        date/cited → DESC（col < val)
+        title/journal/first_author → ASC（col > val)
+        best_match/relevance 为计算表达式，降级到 OFFSET（返回 None）。
+        统一用 id 做 tiebreaker。
+        """
+        if sort not in AdvancedSearchEngine.KEYSET_COLUMN_SORTS:
+            return None
+        if cursor_val is None or cursor_id is None:
+            return None
+        import uuid as _uuid
+        try:
+            uid = _uuid.UUID(cursor_id)
+        except (ValueError, AttributeError):
+            return None
+
+        try:
+            if sort == "date":
+                from datetime import date as dt_date
+                val = dt_date.fromisoformat(cursor_val)
+                return or_(
+                    GlobalLiterature.pub_date < val,
+                    and_(GlobalLiterature.pub_date == val, GlobalLiterature.id < uid),
+                )
+            elif sort == "cited":
+                val = int(cursor_val)
+                return or_(
+                    GlobalLiterature.cited_by_count < val,
+                    and_(GlobalLiterature.cited_by_count == val, GlobalLiterature.id < uid),
+                )
+            elif sort == "title":
+                return or_(
+                    GlobalLiterature.title > cursor_val,
+                    and_(GlobalLiterature.title == cursor_val, GlobalLiterature.id > uid),
+                )
+            elif sort == "journal":
+                return or_(
+                    GlobalLiterature.journal > cursor_val,
+                    and_(GlobalLiterature.journal == cursor_val, GlobalLiterature.id > uid),
+                )
+            elif sort == "first_author":
+                family_col = GlobalLiterature.authors[0]['family'].astext
+                return or_(
+                    family_col > cursor_val,
+                    and_(family_col == cursor_val, GlobalLiterature.id > uid),
+                )
+        except (ValueError, TypeError):
+            return None
+        return None
+
+    @staticmethod
+    def _cursor_from_item(lit, sort: str) -> str | None:
+        """从末尾条目标提取 keyset 游标值。"""
+        if sort == "date":
+            val = str(lit.pub_date or lit.article_date or "")
+            return val if val else None
+        elif sort == "cited":
+            return str(lit.cited_by_count) if lit.cited_by_count is not None else None
+        elif sort == "title":
+            return lit.title or None
+        elif sort == "journal":
+            return lit.journal or None
+        elif sort == "first_author":
+            authors = lit.authors or []
+            return authors[0].get("family") if authors else None
+        return None

backend/app/services/tag_loader.py

@@ -1,40 +1,68 @@
-"""共享标签加载工具（带单次请求级内存缓存）"""
+"""共享标签加载工具（三级缓存：请求级内存 → Redis → DB）"""
 
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from app.core.cache import cache as _redis_cache
 from app.models.literature import GlobalLiteratureTag, GlobalTag
 
-# 请求级缓存：同一批 literature_ids 在短时间内重复查询不走 DB
+# 请求级缓存：同一批 literature_ids 在短时间内重复查询不走 DB/Redis
 _cache: dict[str, dict[str, list[dict]]] = {}
 _CACHE_MAX_KEYS = 5
+_TAGS_CACHE_TTL = 86400  # 24h，标签变更极低频
 
 
 async def load_tags_for_literature(db: AsyncSession, literature_ids: list[str]) -> dict[str, list[dict]]:
-    """批量加载文献标签，返回 {literature_id: [{name_zh, path, category, is_major}]}"""
+    """批量加载文献标签，返回 {literature_id: [{name_zh, path, category, is_major}]}
+
+    三级缓存：
+      L1 — 请求级 dict（同一批 ID 在本次请求中复用）
+      L2 — Redis tags:{lit_id}（24h TTL，跨请求/跨用户）
+      L3 — DB JOIN（miss 时回退）
+    """
     if not literature_ids:
         return {}
-    # 缓存 key 基于排序后的 ID 列表，确保相同集合命中
+
-    cache_key = ",".join(sorted(literature_ids))
+    # L1: 请求级缓存
+    import uuid as _uuid
+    str_ids = [str(x) if not isinstance(x, str) else x for x in literature_ids]
+    cache_key = ",".join(sorted(str_ids))
     if cache_key in _cache:
         return _cache[cache_key]
 
-    import uuid as _uuid
+    # L2: Redis 单文献缓存（批量查）
-    uids = [_uuid.UUID(x) if isinstance(x, str) else x for x in literature_ids]
+    redis_keys = [f"tags:{sid}" for sid in str_ids]
-    result = await db.execute(
+    cached_results = await _redis_cache.mget(redis_keys)
-        select(GlobalLiteratureTag.literature_id, GlobalTag.id, GlobalTag.name_zh, GlobalTag.name_en, GlobalTag.path,
-               GlobalTag.tag_category, GlobalLiteratureTag.is_major)
-        .join(GlobalTag, GlobalLiteratureTag.tag_id == GlobalTag.id)
-        .where(GlobalLiteratureTag.literature_id.in_(uids))
-    )
-    tags_map: dict[str, list[dict]] = {}
-    for lit_id, tid, nz, ne, p, cat, maj in result:
-        sid = str(lit_id)
-        if sid not in tags_map:
-            tags_map[sid] = []
-        tags_map[sid].append({"id": str(tid), "name_zh": nz, "name_en": ne, "path": p, "category": cat, "is_major": maj})
 
-    # 存入请求级缓存，限制大小
+    tags_map: dict[str, list[dict]] = {}
+    miss_ids: list[str] = []
+
+    for sid, cached in zip(str_ids, cached_results):
+        if cached is not None and "tags" in cached:
+            tags_map[sid] = cached["tags"]
+        else:
+            miss_ids.append(sid)
+
+    # L3: DB 回退（只查 miss 的文献）
+    if miss_ids:
+        miss_uids = [_uuid.UUID(sid) for sid in miss_ids]
+        result = await db.execute(
+            select(GlobalLiteratureTag.literature_id, GlobalTag.id, GlobalTag.name_zh, GlobalTag.name_en, GlobalTag.path,
+                   GlobalTag.tag_category, GlobalLiteratureTag.is_major)
+            .join(GlobalTag, GlobalLiteratureTag.tag_id == GlobalTag.id)
+            .where(GlobalLiteratureTag.literature_id.in_(miss_uids))
+        )
+        for lit_id, tid, nz, ne, p, cat, maj in result:
+            sid = str(lit_id)
+            if sid not in tags_map:
+                tags_map[sid] = []
+            tags_map[sid].append({"id": str(tid), "name_zh": nz, "name_en": ne, "path": p, "category": cat, "is_major": maj})
+
+        # 回填 Redis（空列表也缓存，避免空文献反复查 DB）
+        for sid in miss_ids:
+            await _redis_cache.set(f"tags:{sid}", {"tags": tags_map.get(sid, [])}, ttl=_TAGS_CACHE_TTL)
+
+    # L1: 存入请求级缓存
     if len(_cache) >= _CACHE_MAX_KEYS:
         _cache.pop(next(iter(_cache)), None)
     _cache[cache_key] = tags_map

backend/app/services/tag_service.py

@@ -6,9 +6,10 @@
 import uuid
 import logging
 
-from sqlalchemy import select
+from sqlalchemy import func, select, update as sql_update
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from app.core.cache import cache as _redis_cache
 from app.models.literature import GlobalTag, GlobalLiteratureTag
 
 logger = logging.getLogger(__name__)
@@ -78,6 +79,20 @@ async def tag_article(db: AsyncSession, lit_id: uuid.UUID,
         tag.article_count += 1
         count += 1
 
+    if count > 0:
+        # 维护反范式 tag_ids 数组
+        new_ids = [tag.id for tag in matched if tag.id not in existing_tag_ids]
+        if new_ids:
+            from sqlalchemy import update as sql_update
+            from app.models.literature import GlobalLiterature
+            stmt = (
+                sql_update(GlobalLiterature)
+                .where(GlobalLiterature.id == lit_id)
+                .values(tag_ids=func.array_cat(func.coalesce(GlobalLiterature.tag_ids, "{}"), new_ids))
+            )
+            await db.execute(stmt)
+        await _redis_cache.delete(f"tags:{lit_id}")
+
     return count
 
 

backend/app/tasks/pubmed_pipeline.py

@@ -16,6 +16,7 @@ from app.compat import UTC
 from app.db import async_session
 from app.models.literature import GlobalLiterature
 from app.services.journal_utils import ensure_journal_async
+from app.services.tag_service import tag_article
 
 # DOI 正则：从 PDF 全文提取
 DOI_PATTERN = re.compile(r'10\.\d{4,9}/[-._;()/:A-Za-z0-9]+')
@@ -185,6 +186,12 @@ async def _upsert_article_impl(article_data: dict, db: AsyncSession) -> bool:
             lit.doi = article_data["doi"]
             lit.mesh_headings = article_data.get("mesh_headings", lit.mesh_headings)
             lit.updated_at = datetime.now(UTC)
+        mh = article_data.get("mesh_headings", [])
+        if mh:
+            try:
+                await tag_article(db, lit.id, mh)
+            except Exception:
+                pass
         await db.commit()
         return False
     else:
@@ -194,5 +201,13 @@ async def _upsert_article_impl(article_data: dict, db: AsyncSession) -> bool:
             await ensure_journal_async(db, article_data.get("journal_issn"), article_data.get("journal"))
         except Exception:
             pass
+        # 需要 flush 才能拿到 lit.id
+        try:
+            await db.flush()
+            mh = article_data.get("mesh_headings", [])
+            if mh:
+                await tag_article(db, lit.id, mh)
+        except Exception:
+            pass
         await db.commit()
         return True

backend/tests/test_service_pubmed_api.py

@@ -155,9 +155,9 @@ async def test_tag_article_no_mesh():
     """Empty mesh_headings -> returns 0"""
     db = AsyncMock()
 
-    from app.services.pubmed_api import _tag_article
+    from app.services.tag_service import tag_article
 
-    result = await _tag_article(db, uuid.uuid4(), [])
+    result = await tag_article(db, uuid.uuid4(), [])
     assert result == 0
 
 
@@ -166,9 +166,9 @@ async def test_tag_article_no_mesh_ui():
     """Mesh headings without UI values -> returns 0"""
     db = AsyncMock()
 
-    from app.services.pubmed_api import _tag_article
+    from app.services.tag_service import tag_article
 
-    result = await _tag_article(
+    result = await tag_article(
         db, uuid.uuid4(),
         [{"descriptor": "Cancer", "ui": "", "major": False}],
     )
@@ -188,15 +188,17 @@ async def test_tag_article_matches_tags():
 
     mock_match = MagicMock()
     mock_match.scalars.return_value.all.return_value = [mock_tag]
-    mock_empty = MagicMock()
+    mock_no_name = MagicMock()
-    mock_empty.scalar.return_value = None
+    mock_no_name.scalars.return_value.all.return_value = []
-    db.execute.side_effect = [mock_match, mock_empty]
+    mock_no_existing = MagicMock()
+    mock_no_existing.all.return_value = []
+    db.execute.side_effect = [mock_match, mock_no_name, mock_no_existing]
 
     mesh_headings = [{"descriptor": "Cancer", "ui": "D000001", "major": True}]
 
-    from app.services.pubmed_api import _tag_article
+    from app.services.tag_service import tag_article
 
-    result = await _tag_article(db, lit_id, mesh_headings)
+    result = await tag_article(db, lit_id, mesh_headings)
     assert result == 1
     db.add.assert_called_once()
 
@@ -213,9 +215,9 @@ async def test_tag_article_no_match():
 
     mesh_headings = [{"descriptor": "RareDisease", "ui": "D999999", "major": False}]
 
-    from app.services.pubmed_api import _tag_article
+    from app.services.tag_service import tag_article
 
-    result = await _tag_article(db, lit_id, mesh_headings)
+    result = await tag_article(db, lit_id, mesh_headings)
     assert result == 0
     db.add.assert_not_called()
 

frontend/src/types/index.ts

@@ -330,7 +330,7 @@ export interface SearchRequestBody {
   page: number
   page_size: number
   sort?: string
-  cursor_date?: string
+  cursor_val?: string
   cursor_id?: string
   year_from?: number
   year_to?: number

frontend/src/views/app/SearchView.vue

@@ -63,9 +63,9 @@ const pageSize = ref(Number(localStorage.getItem('search:pageSize')) || 20)
 // 从 URL 恢复的页码（syncSearchToUrl 写入）
 const restoredPage = ref(1)
 
-// ── Keyset 游标分页（sort=date 时使用，跳过 COUNT） ──
+// ── Keyset 游标分页（所有排序模式通用，跳过 COUNT 和 OFFSET） ──
 const keysetPage = ref(1)
-const keysetCursors = ref<Array<{cursor_date: string, cursor_id: string} | null>>([null])
+const keysetCursors = ref<Array<{cursor_val: string, cursor_id: string} | null>>([null])
 const keysetHasMore = ref(false)
 
 const minYear = computed(() => yearCounts.value.length ? Math.min(...yearCounts.value.map(y => y.year)) : 2000)
@@ -279,21 +279,18 @@ const { page, total, goToPage } = usePagination({
       }
       if (field.value !== 'all') body.field = field.value
       // P3-6: precision_mode 不再发送（后端已忽略）
-      // ── Keyset 游标分页（sort=date 时跳过 COUNT，纯翻页模式） ──
+      // ── Keyset 游标分页（所有排序模式通用，跳过 COUNT + OFFSET） ──
-      const useKeyset = sort.value === 'date'
+      if (p === 1) {
-      if (useKeyset) {
+        delete (body as any).page
-        if (p === 1) {
+        keysetCursors.value = [null]; keysetPage.value = 1
+      } else {
+        const entry = keysetCursors.value[p]
+        if (entry) {
           delete (body as any).page
-          keysetCursors.value = [null]; keysetPage.value = 1
+          body.cursor_val = entry.cursor_val
-        } else {
+          body.cursor_id = entry.cursor_id
-          const entry = keysetCursors.value[p]
-          if (entry) {
-            delete (body as any).page
-            body.cursor_date = entry.cursor_date
-            body.cursor_id = entry.cursor_id
-          }
-          // cursor 不存在时保留 body.page，回退到 offset 分页
         }
+        // cursor 不存在时保留 body.page，回退到 offset 分页
       }
       // 年份 / 日期（datePreset 与 year_* 互斥）
       const yf = Number(yearFromStr.value)
@@ -343,26 +340,19 @@ const { page, total, goToPage } = usePagination({
       if (negativeResult.value) body.negative_result = negativeResult.value
       const { data } = await api.post('/features/search/advanced', body, { signal })
       results.value = data.items || []
-      if (useKeyset) {
+      // 游标分页：首页走 COUNT 存 total，后续页沿用；cursor 和数据一起返回
-        // 第一页走 COUNT，存下总数；后面沿用第一页的数值
+      if (p === 1) total.value = data.total || 0
-        if (p === 1) total.value = data.total || 0
+      keysetHasMore.value = data.has_more || false
-        keysetHasMore.value = data.has_more || false
+      keysetPage.value = p
-        keysetPage.value = p
+      // 从响应保存游标，供下一页使用
-        // 从当前页最后一条保存游标，供下一页使用
+      const items = data.items || []
-        const items = data.items || []
+      if (items.length > 0 && data.has_more && data.cursor_val) {
-        if (items.length > 0) {
+        keysetCursors.value[p + 1] = {
-          const last = items[items.length - 1]
+          cursor_val: data.cursor_val,
-          keysetCursors.value[p + 1] = {
+          cursor_id: data.cursor_id || items[items.length - 1].id,
-            cursor_date: last.pub_date || last.article_date || '',
-            cursor_id: last.id,
-          }
-          // 当两个日期都为空时，不设游标（避免空串 fromisoformat 失败）
-          if (!last.pub_date && !last.article_date) {
-            delete keysetCursors.value[p + 1]
-          }
         }
       } else {
-        total.value = data.total || 0
+        delete keysetCursors.value[p + 1]
       }
       yearCounts.value = data.year_counts || []
     } catch (e: any) {

frontend/src/views/public/HomeView.vue

@@ -149,7 +149,7 @@ async function fetchData(resetPage = true) {
     if (searchParams.value.negative_result) body.negative_result = searchParams.value.negative_result
     // keyset 游标
     if (cursorDate.value && cursorId.value) {
-      body.cursor_date = cursorDate.value
+      body.cursor_val = cursorDate.value
       body.cursor_id = cursorId.value
     }
     const { data } = await api.post('/features/search/advanced', body)
@@ -160,12 +160,18 @@ async function fetchData(resetPage = true) {
       feedItems.value.push(...(data.items || []))
     }
     hasMoreItems.value = data.has_more ?? false
-    // 记录游标（取最后一条）
+    // keyset 游标（全部使用服务端返回的游标，通用所有排序模式）
-    const items = data.items || []
+    if (searchParams.value.sort !== 'date') {
-    if (items.length > 0) {
+      // 非 date 排序由服务端返回 cursor_val，用 response 字段
-      const last = items[items.length - 1]
+      if (data.cursor_val) {
-      cursorDate.value = (last.article_date || last.pub_date)?.slice(0, 10) || null
+        cursorDate.value = data.cursor_val
-      cursorId.value = last.id || null
+        cursorId.value = data.cursor_id
+      }
+    } else {
+      if (data.cursor_val && data.cursor_id) {
+        cursorDate.value = data.cursor_val
+        cursorId.value = data.cursor_id
+      }
     }
     searched.value = true
   } catch (e) { toast.apiError(e, '搜索文献失败，请重试') }