diff --git a/backend/app/services/search_engine.py b/backend/app/services/search_engine.py index f9d9661..d7a0252 100644 --- a/backend/app/services/search_engine.py +++ b/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 时分页) - cursor_date: str | None = None, + # Generic keyset cursor + 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 唯一标识翻页位置 - "cd": cursor_date, + # keyset cursor 唯一标识翻页位置（不含时为第 1 页） + "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 组合唯一标识翻页位置） - use_cursor = (cursor_date is not None and cursor_id is not None and sort == "date") + # 向后兼容：cursor_date → cursor_val（旧前端发 cursor_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)) - conditions.append(GlobalLiterature.id.in_(subq)) + from sqlalchemy.dialects.postgresql import array as _pg_array + 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 not _has_any_filter: + if year_counts: + 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") - 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 分页（所有列式排序模式统一，代替 OFFSET） ── + _keyset_cond = AdvancedSearchEngine._keyset_condition(sort, cursor_val, cursor_id) + if _keyset_cond is not None: + conditions.append(_keyset_cond) - # 重新构建查询（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: - # 总数 + # ── LIMIT page_size+1 探测下一页 + COUNT 第 1 页缓存 ── + result = await db.execute(q.limit(page_size + 1)) + items = result.scalars().all() + has_more = len(items) > page_size + items = items[:page_size] + + # COUNT 只在第 1 页计算，缓存到 facet key 供后续页复用 + total = 0 + if is_first_page: 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() - has_more = (offset + page_size) < total and len(items) == page_size + # 构建游标供翻页 + next_cursor_val = None + 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_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_ + # 从全局缓存加载期刊 tier/name（避免每页 IN 查询） + tier_map, name_map = await AdvancedSearchEngine._get_journal_map(db) + 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: - await _cache.set(_search_cache_key, result, ttl=AdvancedSearchEngine.SEARCH_CACHE_TTL) - - return result + 从 DB 按 ID 查询文献、从 ②③ 缓存加载 tags/journal，避免重算复杂 WHERE。 + """ + import uuid as _uuid + 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 = [] - 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))) + # 计算缓存键 + items = sorted(m.strip().lower() for m in mesh_names if m.strip()) + if not items: + return None + raw_key = f"{'|'.join(items)}:major={major_only}:noexp={noexp}" + cache_key = f"atm:{_hashlib.md5(raw_key.encode()).hexdigest()}" - 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) + # 查缓存 + cached = await _cache.get(cache_key) + if cached is not None: + mesh_tag_ids = {_uuid.UUID(uid) for uid in cached["tag_ids"]} + else: + mesh_tag_ids: set[_uuid.UUID] = set() - 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]) + # 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 + entry_conds.append(GlobalTag.entry_terms.contains([q])) + 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 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