Compare commits

...
6 Commits
Author SHA1 Message Date
34047007@qq.com 4a24f764f7 test: 更新 MeSH 打标测试适配 tag_service 重构
CI / backend (push) Canceled after 0s
CI / frontend (push) Canceled after 0s
- _tag_article → tag_service.tag_article 导入路径
- 增加 name_en 查询 mock 返回值(3 次 execute 调用)
2026-07-27 16:16:43 +08:00
34047007@qq.com 388cbaaa95 refactor: 前端 keyset 分页推广 + cursor_date→cursor_val 统一
- AdvancedSearchRequest.cursor_date → cursor_val(通用游标值字段)
- SearchView 移除 sort===date 守卫,所有排序模式统一使用 keyset
- HomeView 游标字段同步更新
- 类型定义同步
2026-07-27 16:16:35 +08:00
34047007@qq.com 52a0d01823 perf: 模型索引优化 — tag_ids 反范式 + BRIN/partial/covering 索引
- GlobalLiterature.tag_ids ARRAY(Uuid) 列 + GIN 索引(标签筛选免 JOIN)
- BRIN 索引:pub_date / pub_year(时序数据,体积缩小 100x)
- Covering 索引:(pub_date, id) INCLUDE 高频列(index-only scan)
- Partial 索引:retracted / is_oa / is_negative_result / is_preprint
2026-07-27 16:16:30 +08:00
34047007@qq.com 0895359b6f perf: 搜索引擎优化 — 五重缓存 + Keyset 全排序 + tag_ids 筛选
- ATM 展开缓存(atm:{md5}, 3600s TTL)
- 期刊 map 全表缓存(journals:map, 3600s TTL)
- 搜索结果精简缓存(search:{md5},不含 tags/journal 纯数据)
- 分面 year_counts 独立缓存(search:year_counts:{md5},首页用)
- 标签筛选改为 tag_ids && ARRAY[uids] GIN 索引扫描,免 JOIN
- Keyset 分页推广至 date/cited/title/journal/first_author 全部排序模式
- _keyset_condition() + _cursor_from_item() + _hydrate_items() 通用化
- COUNT 仅首页执行,结果缓存到分面键
2026-07-27 16:16:26 +08:00
34047007@qq.com 0b2bd87d61 perf: 标签二级缓存 + cache.mget 批量读取
- tag_loader 增加 Redis L2 缓存(tags:{lit_id}, 86400s TTL),批量 mget
- cache.py 新增 mget() 方法,同键 Redis 批量查询 + memory 降级
- tag_service.py 打标后自动清除对应缓存
2026-07-27 16:16:21 +08:00
34047007@qq.com c791c06096 fix: FTP pipeline MeSH 打标链 — 修复 _upsert_article_impl 缺少 tag_article() 调用
FTP 增量更新写入 mesh_headings JSONB 后未调用 tag_article(),导致
GlobalLiteratureTag 无关联记录,[MH]/[MAJR] 搜索返回空。新增 tag_ids
数组维护 + 标签缓存失效。
2026-07-27 16:16:17 +08:00
13 changed files with 595 additions and 208 deletions
@@ -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 ###
+2 -2
View File
@@ -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
+2 -2
View File
@@ -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)
cursor_date: str | None = None # 上一页最后一条的 pub_dateISO 日期
# keyset 游标分页(所有排序模式通用,设了 cursor 后 page 参数被忽略,不做 COUNT)
cursor_val: str | None = None # 上一页最后一条的排序列值(字符串,服务端按 sort 模式解析
cursor_id: str | None = None # 上一页最后一条的 id(UUID 字符串)
# ── PubMed 筛选器参数 ──
+12
View File
@@ -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:
+23 -1
View File
@@ -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-treepub_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 indexdate 排序 + 高频引用列(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"),
+309 -74
View File
@@ -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
# ── 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
else:
# 总数
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,8 +1363,25 @@ 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
# 计算缓存键
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()}"
# 查缓存
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()
# Batch all mesh name lookups — 2 queries instead of 2N
@@ -1230,9 +1391,7 @@ class AdvancedSearchEngine:
q = m.strip().lower()
if not q:
continue
# 1a. 精确入口词匹配(P1-3)— batch via OR
entry_conds.append(GlobalTag.entry_terms.contains([q]))
# 1b. name_en ILIKE 回退 — batch via OR
name_conds.append(GlobalTag.name_en.ilike(_escape_ilike(m)))
try:
@@ -1261,10 +1420,8 @@ class AdvancedSearchEngine:
except Exception:
logger.exception("MeSH tag lookup failed for mesh_names=%s", mesh_names[:5])
if not mesh_tag_ids:
return None
# tree_number 前缀展开:取匹配 tag 的所有 tree_number,查子节点([MH:noexp] 时跳过)
if mesh_tag_ids:
# tree_number 前缀展开
if not noexp:
try:
tns = (await db.execute(
@@ -1285,6 +1442,12 @@ class AdvancedSearchEngine:
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
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 → DESCcol < val)
title/journal/first_author → ASCcol > 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
+38 -10
View File
@@ -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 JOINmiss 时回退
"""
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
uids = [_uuid.UUID(x) if isinstance(x, str) else x for x in literature_ids]
# L2: Redis 单文献缓存(批量查)
redis_keys = [f"tags:{sid}" for sid in str_ids]
cached_results = await _redis_cache.mget(redis_keys)
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_(uids))
.where(GlobalLiteratureTag.literature_id.in_(miss_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})
# 存入请求级缓存,限制大小
# 回填 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
+16 -1
View File
@@ -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
+15
View File
@@ -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
+13 -11
View File
@@ -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_empty.scalar.return_value = None
db.execute.side_effect = [mock_match, mock_empty]
mock_no_name = MagicMock()
mock_no_name.scalars.return_value.all.return_value = []
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()
+1 -1
View File
@@ -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
+10 -20
View File
@@ -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,9 +279,7 @@ const { page, total, goToPage } = usePagination({
}
if (field.value !== 'all') body.field = field.value
// P3-6: precision_mode
// Keyset sort=date COUNT
const useKeyset = sort.value === 'date'
if (useKeyset) {
// Keyset COUNT + OFFSET
if (p === 1) {
delete (body as any).page
keysetCursors.value = [null]; keysetPage.value = 1
@@ -289,12 +287,11 @@ const { page, total, goToPage } = usePagination({
const entry = keysetCursors.value[p]
if (entry) {
delete (body as any).page
body.cursor_date = entry.cursor_date
body.cursor_val = entry.cursor_val
body.cursor_id = entry.cursor_id
}
// cursor body.page退 offset
}
}
// / datePreset year_*
const yf = Number(yearFromStr.value)
const yt = Number(yearToStr.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沿
// COUNT total沿cursor
if (p === 1) total.value = data.total || 0
keysetHasMore.value = data.has_more || false
keysetPage.value = p
// 使
// 使
const items = data.items || []
if (items.length > 0) {
const last = items[items.length - 1]
if (items.length > 0 && data.has_more && data.cursor_val) {
keysetCursors.value[p + 1] = {
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]
}
cursor_val: data.cursor_val,
cursor_id: data.cursor_id || items[items.length - 1].id,
}
} else {
total.value = data.total || 0
delete keysetCursors.value[p + 1]
}
yearCounts.value = data.year_counts || []
} catch (e: any) {
+13 -7
View File
@@ -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
//
const items = data.items || []
if (items.length > 0) {
const last = items[items.length - 1]
cursorDate.value = (last.article_date || last.pub_date)?.slice(0, 10) || null
cursorId.value = last.id || null
// keyset 使
if (searchParams.value.sort !== 'date') {
// date cursor_val response
if (data.cursor_val) {
cursorDate.value = data.cursor_val
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, '搜索文献失败,请重试') }