chore: batch commit remaining changes

Includes search engine improvements, Alembic migrations, new services (pubmed_daily_update, query_expansion), frontend updates, and documentation sync.
2026-07-27 08:35:12 +08:00
parent 35b0a5c565
commit 62ca8fa6b8
82 changed files with 314248 additions and 1519 deletions
@@ -994,14 +994,31 @@ async def pipeline_runs(
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
 ):
+    """管道运行历史"""
    offset = (page - 1) * page_size
    total = (await db.execute(select(func.count(PipelineRun.id)))).scalar() or 0
    r = await db.execute(select(PipelineRun).order_by(PipelineRun.created_at.desc()).offset(offset).limit(page_size))
    runs = r.scalars().all()
-    return {"total": total, "items": [{"id": str(p.id), "type": p.run_type, "status": p.status, "new": p.articles_new,
-                                       "filtered": p.articles_filtered, "feeds": p.feeds_generated, "error": p.error_log,
-                                       "started": p.started_at.strftime("%Y-%m-%dT%H:%M:%SZ") if p.started_at else None,
-                                       "completed": p.completed_at.strftime("%Y-%m-%dT%H:%M:%SZ") if p.completed_at else None} for p in runs]}
+    return {
+        "total": total,
+        "items": [{
+            "id": str(p.id),
+            "type": p.run_type,
+            "status": p.status,
+            "files_processed": p.files_processed,
+            "total_articles": p.articles_total,
+            "articles_new": p.articles_new,
+            "articles_updated": p.articles_updated,
+            "articles_deleted": p.articles_deleted,
+            "articles_filtered": p.articles_filtered,
+            "feeds_generated": p.feeds_generated,
+            "error": p.error_log,
+            "processed_date": str(p.processed_date) if p.processed_date else None,
+            "metadata": p.run_metadata,
+            "started": p.started_at.strftime("%Y-%m-%dT%H:%M:%SZ") if p.started_at else None,
+            "completed": p.completed_at.strftime("%Y-%m-%dT%H:%M:%SZ") if p.completed_at else None,
+        } for p in runs]
+    }


@router.post("/pipeline/refresh-citations", response_model=dict, summary="刷新引用次数")
@@ -1492,50 +1509,19 @@ async def retag_tags(
@router.post("/pipeline/run", response_model=dict, summary="运行 PubMed 检索流水线")
 async def run_pipeline(
    db: AsyncSession = Depends(get_db),
-    mode: str = Query("full", pattern="^(full|majr|broad)$", description="full=全量, majr=仅精搜, broad=仅宽搜"),
+    mode: str = Query("ftp", pattern="^(full|majr|broad|ftp)$", description="full=全量, majr=仅精搜, broad=仅宽搜, ftp=FTP每日增量"),
    max_per_query: int = Query(30, ge=5, le=5000, description="每查询最多结果数（Europe PMC 支持到 5000）"),
    precision_mode: str = Query("majr", pattern="^(majr|mesh)$", description="majr=Major Topic优先, mesh=MeSH更高召回"),
    source: str = Query("ncbi", pattern="^(ncbi|europe_pmc)$", description="ncbi=NCBI E-utilities, europe_pmc=Europe PMC API"),
    _op: dict = Depends(require_platform_operator),
 ):
-    """运行肿瘤学文献检索流水线，支持精度模式、搜索范围控制和数据源切换。"""
-    from app.services.pubmed_api import _run_pipeline
+    """运行肿瘤学文献检索流水线。

-    use_majr = mode in ("full", "majr")
-    use_broad = mode in ("full", "broad")
-    stats = await _run_pipeline(
-        db, max_per_query,
-        use_majr=use_majr, use_broad=use_broad,
-        precision_mode=precision_mode,
-        use_europe_pmc=(source == "europe_pmc"),
-    )
-
-    # 记录运行日志
-    from datetime import datetime
-    from app.compat import UTC
-    run_type = f"manual_{mode}"
-    if source == "europe_pmc":
-        run_type += "_epmc"
-    run_log = PipelineRun(
-        id=uuid.uuid4(),
-        run_type=run_type,
-        status="success",
-        articles_total=stats["fetched"],
-        articles_new=stats["new"],
-        articles_filtered=stats["tagged"],
-        feeds_generated=stats["feeds"],
-        started_at=datetime.now(UTC),
-        completed_at=datetime.now(UTC),
-    )
-    db.add(run_log)
-    await db.commit()
-
-    # 刷新 stats 缓存
-    try:
-        from app.api.v1.public import refresh_stats_cache
-        await refresh_stats_cache()
-    except Exception:
-        pass
+    默认模式为 ftp（FTP每日增量更新），取代旧的 full/majr/broad 模式。
+    """
+    if mode == "ftp":
+        return await _run_ftp_pipeline(db)
+    return await _run_eutils_pipeline(db, mode, max_per_query, precision_mode, source)

    return {"status": "ok", "mode": mode, "precision": precision_mode, **stats}

@@ -1567,6 +1553,57 @@ async def trigger_ai_summary(
    return {"status": "ok", "processed": done, "total": len(articles)}


+async def _run_ftp_pipeline(db: AsyncSession) -> dict:
+    """运行 FTP 每日增量更新（新默认管道）"""
+    from app.services.pubmed_daily_update import run_daily_ftp_update
+
+    stats = await run_daily_ftp_update()
+    logger.info("FTP pipeline done: %s", stats)
+
+    try:
+        from app.api.v1.public import refresh_stats_cache
+        await refresh_stats_cache()
+    except Exception:
+        pass
+    return stats
+
+
+async def _run_eutils_pipeline(db: AsyncSession, mode: str, max_per_query: int,
+                                precision_mode: str, source: str) -> dict:
+    """运行 E-utilities 旧管道（降级/兼容用）"""
+    from app.services.pubmed_api import _run_pipeline
+
+    use_majr = mode in ("full", "majr")
+    use_broad = mode in ("full", "broad")
+    stats = await _run_pipeline(
+        db, max_per_query,
+        use_majr=use_majr, use_broad=use_broad,
+        precision_mode=precision_mode,
+        use_europe_pmc=(source == "europe_pmc"),
+    )
+
+    run_type = f"manual_{mode}"
+    if source == "europe_pmc":
+        run_type += "_epmc"
+    from datetime import datetime
+    from app.compat import UTC
+    run_log = PipelineRun(
+        id=uuid.uuid4(), run_type=run_type, status="success",
+        articles_total=stats["fetched"], articles_new=stats["new"],
+        articles_filtered=stats["tagged"], feeds_generated=stats["feeds"],
+        started_at=datetime.now(UTC), completed_at=datetime.now(UTC),
+    )
+    db.add(run_log)
+    await db.commit()
+
+    try:
+        from app.api.v1.public import refresh_stats_cache
+        await refresh_stats_cache()
+    except Exception:
+        pass
+    return stats
+
+
 # ─── 网站访问记录 ───

@router.get("/page-views", response_model=dict, summary="页面访问记录")
@@ -3,14 +3,15 @@
 import uuid

 from fastapi import APIRouter, Depends, HTTPException, Query
-from pydantic import BaseModel
-from sqlalchemy import select
+from pydantic import BaseModel, Field, field_validator
+from sqlalchemy import func, select, text
 from sqlalchemy.ext.asyncio import AsyncSession

+from app.core.cache import cache
 from app.core.permissions import get_current_user, require_platform_operator
 from app.db import get_db
-from app.models.literature import GlobalLiterature
-from app.models.user import User
+from app.models.literature import GlobalJournal, GlobalLiterature, GlobalTag
+from app.models.user import User, UserSavedFilter
 from app.services.daily_digest import send_daily_digest_to_user
 from app.services.rule_engine import PRESET_RULES, RuleEngine
 from app.services.search_engine import AdvancedSearchEngine
@@ -57,20 +58,338 @@ class AdvancedSearchRequest(BaseModel):
    tag_ids: list[str] | None = None
    retracted: str = ""
    negative_result: str = ""
-    page: int = 1
-    page_size: int = 20
+    is_oa: bool | None = None
+    language: str | None = None
+    languages: list[str] | None = None
+    precision_mode: str = "majr"
+    nlm_subsets: list[str] | None = None
+    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_date（ISO 日期）
    cursor_id: str | None = None     # 上一页最后一条的 id（UUID 字符串）

+    # ── PubMed 筛选器参数 ──
+    # Text Availability
+    has_abstract: bool | None = None
+    is_free_full_text: bool | None = None
+    has_full_text: bool | None = None
+    # Article Attribute
+    has_associated_data: bool | None = None
+    # Species / Sex / Age（传 mesh_ui 列表）
+    species: list[str] | None = None
+    sex: list[str] | None = None
+    age: list[str] | None = None
+    # Journal Categories
+    medline_only: bool = False
+    exclude_preprints: bool = False
+
+    # ── 查询复杂度限制 ──
+    @field_validator('query')
+    @classmethod
+    def check_query_complexity(cls, v: str) -> str:
+        if len(v.split()) > 50:
+            raise ValueError('查询词过多（最多 50 个词），请简化搜索条件')
+        return v
+
+
+# ── 筛选选项 ──
+
+SPECIAL_MESH_UIS = {
+    "species": ["D006801", "D000818"],
+    "sex": ["D005260", "D008297"],
+    "age": [
+        "D007231",  # Infant, Newborn (birth-1mo)
+        "D007223",  # Infant (1-23mo)
+        "D002675",  # Preschool Child (2-5yr)
+        "D002648",  # Child (6-12yr)
+        "D000293",  # Adolescent (13-18yr)
+        "D000328",  # Adult (19-44yr)
+        "D055815",  # Young Adult (19-24yr)
+        "D008875",  # Middle Aged (45-64yr)
+        "D000368",  # Aged (65-79yr)
+        "D000369",  # 80 and over
+    ],
+}
+
+# AGE 筛选层级定义：(key, label, mesh_uis, indent)
+AGE_GROUPS = [
+    ("child_0_18",    "Child: birth-18 years",         ["D007231", "D007223", "D002675", "D002648", "D000293"], 0),
+    ("newborn",       "Newborn: birth-1 month",        ["D007231"], 1),
+    ("infant_0_23",   "Infant: birth-23 months",       ["D007231", "D007223"], 1),
+    ("infant_1_23",   "Infant: 1-23 months",           ["D007223"], 1),
+    ("preschool",     "Preschool Child: 2-5 years",    ["D002675"], 1),
+    ("child_6_12",    "Child: 6-12 years",             ["D002648"], 1),
+    ("adolescent",    "Adolescent: 13-18 years",       ["D000293"], 1),
+    ("adult_19_plus", "Adult: 19+ years",              ["D000328", "D008875", "D000368", "D000369"], 0),
+    ("young_adult",   "Young Adult: 19-24 years",      ["D055815"], 1),
+    ("adult_19_44",   "Adult: 19-44 years",            ["D000328"], 1),
+    ("middle_aged_plus","Middle Aged + Aged: 45+ years", ["D008875", "D000368", "D000369"], 1),
+    ("middle_aged",   "Middle Aged: 45-64 years",      ["D008875"], 1),
+    ("aged",          "Aged: 65+ years",               ["D000368", "D000369"], 1),
+    ("aged_80_plus",  "80 and over: 80+ years",       ["D000369"], 1),
+]
+
+# 预构建 key → UIs 查找表（搜索时用）
+AGE_GROUP_UI_MAP: dict[str, list[str]] = {k: u for k, _, u, _ in AGE_GROUPS}
+
+NLM_SUBSET_LABELS = {
+    "AIM": "Core clinical journals",
+    "M": "MEDLINE",
+    "S": "PubMed Central",
+    "D": "Dental journals",
+    "N": "Nursing journals",
+    "Q": "History of Medicine",
+    "T": "Health Technology Assessment",
+    "X": "AIDS/HIV",
+}
+
+
+@router.get("/search/filter-options", summary="获取搜索筛选选项")
+async def get_filter_options(db: AsyncSession = Depends(get_db)):
+    """返回搜索筛选面板所需的所有选项列表（缓存 1 小时）"""
+    result = await cache.get_or_set("filter-options", lambda: _load_filter_options(db), ttl=3600)
+    return result
+
+
+async def _load_filter_options(db: AsyncSession) -> dict:
+    """从数据库加载筛选选项（被 get_filter_options 缓存调用）"""
+    # 1. pub_types（按频率降序）
+    pub_types_rows = await db.execute(text("""
+        SELECT pt, COUNT(*) as cnt
+        FROM global_literature, jsonb_array_elements_text(pub_types) pt
+        GROUP BY pt ORDER BY cnt DESC
+    """))
+    pub_types = [{"name": r[0], "count": r[1]} for r in pub_types_rows if r[0]]
+
+    # 2. languages（按频率降序）
+    lang_rows = await db.execute(text("""
+        SELECT language, COUNT(*) as cnt
+        FROM global_literature WHERE language IS NOT NULL AND language != ''
+        GROUP BY language ORDER BY cnt DESC
+    """))
+    languages = [{"code": r[0], "count": r[1]} for r in lang_rows]
+
+    # 3. nlm_subsets（单次 JOIN + COUNT FILTER，代替 8 次独立查询）
+    subset_filter_cols = ", ".join(
+        f'COUNT(*) FILTER (WHERE j.nlm_subsets @> ARRAY[\'{code}\']) AS "{code}"'
+        for code in NLM_SUBSET_LABELS
+    )
+    subset_row = (await db.execute(text(f"""
+        SELECT {subset_filter_cols}
+        FROM global_literature l
+        JOIN global_journals j ON l.journal_issn = j.issn
+    """))).one()
+    nlm_subsets = []
+    for code, label in NLM_SUBSET_LABELS.items():
+        cnt = getattr(subset_row, code, None)
+        if cnt is not None and cnt > 0:
+            nlm_subsets.append({"code": code, "label": label, "count": cnt or 0})
+    nlm_subsets.sort(key=lambda x: -x["count"])
+
+    # 4. special_tags（Species/Sex/Age）
+    special_tags: dict[str, list[dict]] = {}
+    for group_name, mesh_uis in SPECIAL_MESH_UIS.items():
+        if group_name == "age":
+            # AGE 按层级结构返回
+            age_options = []
+            for key, label, group_uis, indent in AGE_GROUPS:
+                age_options.append({
+                    "key": key,
+                    "label": label,
+                    "mesh_uis": group_uis,
+                    "indent": indent,
+                })
+            special_tags["age"] = age_options
+        else:
+            rows = await db.execute(
+                select(GlobalTag.id, GlobalTag.mesh_ui, GlobalTag.name_en, GlobalTag.name_zh)
+                .where(GlobalTag.mesh_ui.in_(mesh_uis))
+            )
+            tags = [
+                {"id": str(r[0]), "mesh_ui": r[1], "name_en": r[2] or "", "name_zh": r[3] or ""}
+                for r in rows
+            ]
+            if tags:
+                special_tags[group_name] = tags
+
+    # 5. text_availability counts（单次扫描，4 个 COUNT FILTER）
+    count_row = (await db.execute(text("""
+        SELECT
+          COUNT(*) AS total,
+          COUNT(*) FILTER (WHERE abstract IS NOT NULL AND abstract != '') AS has_abstract,
+          COUNT(*) FILTER (WHERE is_oa = TRUE) AS free_full_text,
+          COUNT(*) FILTER (WHERE pmc_id IS NOT NULL) AS has_full_text
+        FROM global_literature
+    """))).one()
+    total_count = count_row.total or 0
+    has_abstract_count = count_row.has_abstract or 0
+    free_full_text_count = count_row.free_full_text or 0
+    has_full_text_count = count_row.has_full_text or 0
+
+    return {
+        "pub_types": pub_types,
+        "languages": languages,
+        "nlm_subsets": nlm_subsets,
+        "special_tags": special_tags,
+        "text_availability": {
+            "has_abstract": has_abstract_count,
+            "is_free_full_text": free_full_text_count,
+            "has_full_text": has_full_text_count,
+            "total": total_count,
+        },
+    }
+

@router.post("/search/advanced", summary="高级搜索")
 async def advanced_search(req: AdvancedSearchRequest, db: AsyncSession = Depends(get_db)):
    try:
        return await AdvancedSearchEngine.search(db, **req.model_dump())
+    except ValueError:
+        raise HTTPException(status_code=400, detail="搜索参数错误，请检查输入") from None
    except Exception as e:
-        raise HTTPException(status_code=400, detail=str(e)) from e
+        raise HTTPException(status_code=500, detail="搜索服务内部错误") from e
+
+
+# ─── 自定义筛选器（My Custom Filters）───
+
+class SavedFilterCreate(BaseModel):
+    name: str
+    query_string: str
+
+
+class SavedFilterUpdate(BaseModel):
+    name: str | None = None
+    query_string: str | None = None
+
+
+class SavedFilterReorder(BaseModel):
+    id: str
+    sort_order: int
+
+
+@router.get("/search/saved-filters", summary="获取当前用户的自定义筛选器列表")
+async def list_saved_filters(
+    db: AsyncSession = Depends(get_db),
+    user: dict = Depends(get_current_user),
+):
+    user_id = uuid.UUID(user["sub"])
+    result = await db.execute(
+        select(UserSavedFilter)
+        .where(UserSavedFilter.user_id == user_id)
+        .order_by(UserSavedFilter.sort_order, UserSavedFilter.created_at)
+    )
+    filters = result.scalars().all()
+    return {
+        "filters": [
+            {"id": str(f.id), "name": f.name, "query_string": f.query_string, "sort_order": f.sort_order}
+            for f in filters
+        ]
+    }
+
+
+@router.post("/search/saved-filters", summary="新建自定义筛选器")
+async def create_saved_filter(
+    req: SavedFilterCreate,
+    db: AsyncSession = Depends(get_db),
+    user: dict = Depends(get_current_user),
+):
+    user_id = uuid.UUID(user["sub"])
+    # 获取当前最大 sort_order
+    result = await db.execute(
+        select(UserSavedFilter.sort_order)
+        .where(UserSavedFilter.user_id == user_id)
+        .order_by(UserSavedFilter.sort_order.desc())
+        .limit(1)
+    )
+    max_order = result.scalar() or 0
+    sf = UserSavedFilter(
+        user_id=user_id,
+        name=req.name,
+        query_string=req.query_string,
+        sort_order=max_order + 1,
+    )
+    db.add(sf)
+    await db.commit()
+    await db.refresh(sf)
+    return {"id": str(sf.id), "name": sf.name, "query_string": sf.query_string, "sort_order": sf.sort_order}
+
+
+@router.put("/search/saved-filters/{filter_id}", summary="编辑自定义筛选器")
+async def update_saved_filter(
+    filter_id: str,
+    req: SavedFilterUpdate,
+    db: AsyncSession = Depends(get_db),
+    user: dict = Depends(get_current_user),
+):
+    try:
+        fid = uuid.UUID(filter_id)
+    except ValueError:
+        raise HTTPException(400, detail="无效的筛选器 ID")
+    user_id = uuid.UUID(user["sub"])
+    result = await db.execute(
+        select(UserSavedFilter).where(
+            UserSavedFilter.id == fid,
+            UserSavedFilter.user_id == user_id,
+        )
+    )
+    sf = result.scalar()
+    if not sf:
+        raise HTTPException(404, detail="筛选器不存在")
+    if req.name is not None:
+        sf.name = req.name
+    if req.query_string is not None:
+        sf.query_string = req.query_string
+    await db.commit()
+    await db.refresh(sf)
+    return {"id": str(sf.id), "name": sf.name, "query_string": sf.query_string, "sort_order": sf.sort_order}
+
+
+@router.delete("/search/saved-filters/{filter_id}", summary="删除自定义筛选器")
+async def delete_saved_filter(
+    filter_id: str,
+    db: AsyncSession = Depends(get_db),
+    user: dict = Depends(get_current_user),
+):
+    try:
+        fid = uuid.UUID(filter_id)
+    except ValueError:
+        raise HTTPException(400, detail="无效的筛选器 ID")
+    user_id = uuid.UUID(user["sub"])
+    result = await db.execute(
+        select(UserSavedFilter).where(
+            UserSavedFilter.id == fid,
+            UserSavedFilter.user_id == user_id,
+        )
+    )
+    sf = result.scalar()
+    if not sf:
+        raise HTTPException(404, detail="筛选器不存在")
+    await db.delete(sf)
+    await db.commit()
+    return {"status": "ok"}
+
+
+@router.put("/search/saved-filters/reorder", summary="批量排序自定义筛选器")
+async def reorder_saved_filters(
+    req: list[SavedFilterReorder],
+    db: AsyncSession = Depends(get_db),
+    user: dict = Depends(get_current_user),
+):
+    user_id = uuid.UUID(user["sub"])
+    for item in req:
+        result = await db.execute(
+            select(UserSavedFilter).where(
+                UserSavedFilter.id == uuid.UUID(item.id),
+                UserSavedFilter.user_id == user_id,
+            )
+        )
+        sf = result.scalar()
+        if sf:
+            sf.sort_order = item.sort_order
+    await db.commit()
+    return {"status": "ok"}


 # ─── 每日摘要 ───
@@ -16,6 +16,7 @@ from app.models.literature import GlobalLiterature, GlobalLiteratureTag, GlobalT
 from app.schemas.literature import FeedResponse, LiteratureCard, LiteratureDetail, cap_pub_date
 from app.core.cache import cache
 from app.services.cos_client import cached_get_full_text
+from app.services.search_engine import AdvancedSearchEngine
 from app.services.tag_loader import load_tags_for_literature

 router = APIRouter()
@@ -30,24 +31,31 @@ async def _record_dismissed_tags(db: AsyncSession, user_id: uuid.UUID, feed_item
    """提取 feed_item 的 matched_tags 并记录为负偏好信号（upsert 累计计数）"""
    if not feed_item.matched_tags:
        return
+    tag_uuids = []
    for tag_info in (feed_item.matched_tags or []):
        tag_id_str = tag_info.get("tag_id")
        if not tag_id_str:
            continue
        try:
-            tag_uuid = uuid.UUID(tag_id_str)
+            tag_uuids.append(uuid.UUID(tag_id_str))
        except (ValueError, TypeError):
            continue
-        existing = await db.execute(
-            select(UserDismissedTag).where(
-                UserDismissedTag.user_id == user_id,
-                UserDismissedTag.tag_id == tag_uuid,
-            )
+    if not tag_uuids:
+        return
+    # 批量查询已有记录
+    existing = await db.execute(
+        select(UserDismissedTag).where(
+            UserDismissedTag.user_id == user_id,
+            UserDismissedTag.tag_id.in_(tag_uuids),
        )
-        dt = existing.scalar()
-        if dt:
+    )
+    existing_map = {dt.tag_id: dt for dt in existing.scalars().all()}
+    now = datetime.now(timezone.utc)
+    for tag_uuid in tag_uuids:
+        if tag_uuid in existing_map:
+            dt = existing_map[tag_uuid]
            dt.dismiss_count += 1
-            dt.last_dismissed_at = datetime.now(timezone.utc)
+            dt.last_dismissed_at = now
        else:
            db.add(UserDismissedTag(
                user_id=user_id, tag_id=tag_uuid, dismiss_count=1,
@@ -115,10 +123,11 @@ async def personal_feed(
        for a in authors:
            aff = a.get("affiliation","") or ""
            if aff.strip(): affiliation = aff.split(",")[0].strip()[:30]; break
-        tlist = tm.get(str(lit.id), [])
-        for t in tlist:
+        tlist = []
+        for t in (tm.get(str(lit.id), [])):
            if '::' in t.get('path', ''):
-                t['name_zh'] = t['path'].split('::')[-1]
+                t = {**t, 'name_zh': t['path'].split('::')[-1]}
+            tlist.append(t)
        items.append(LiteratureCard(
            id=str(lit.id), pmid=lit.pmid, title=lit.title,
            first_author=authors[0].get("family", "") if authors else "",
@@ -170,39 +179,54 @@ class FolderMoveBody(BaseModel):
 async def batch_save_feed(req: BatchFeedPmids, db: AsyncSession = Depends(get_db), user: dict = Depends(get_current_user)):
    """批量收藏 Feed 中的文献。已收藏的自动跳过。"""
    user_id = uid(user)
-    saved = 0
-    for pmid in req.pmids[:200]:
-        lit = (await db.execute(select(GlobalLiterature.id).where(GlobalLiterature.pmid == pmid))).scalar()
-        if not lit:
-            continue
-        exists = (await db.execute(
-            select(UserLiterature).where(UserLiterature.user_id == user_id, UserLiterature.literature_id == lit)
-        )).scalar()
-        if not exists:
-            db.add(UserLiterature(user_id=user_id, literature_id=lit))
-            saved += 1
+    pmids = req.pmids[:200]
+    if not pmids:
+        return {"saved": 0, "total": 0}
+    lit_ids = (await db.execute(
+        select(GlobalLiterature.id).where(GlobalLiterature.pmid.in_(pmids))
+    )).scalars().all()
+    if not lit_ids:
+        return {"saved": 0, "total": len(pmids)}
+    existing = (await db.execute(
+        select(UserLiterature.literature_id).where(
+            UserLiterature.user_id == user_id,
+            UserLiterature.literature_id.in_(lit_ids),
+        )
+    )).scalars().all()
+    existing_set = set(existing)
+    new_entries = [
+        UserLiterature(user_id=user_id, literature_id=lid)
+        for lid in lit_ids if lid not in existing_set
+    ]
+    db.add_all(new_entries)
    await db.commit()
-    return {"saved": saved, "total": len(req.pmids)}
+    return {"saved": len(new_entries), "total": len(pmids)}


@router.post("/feed/batch-dismiss", summary="批量忽略")
 async def batch_dismiss_feed(req: BatchFeedPmids, db: AsyncSession = Depends(get_db), user: dict = Depends(get_current_user)):
    """批量忽略 Feed 中的文献推荐。"""
    user_id = uid(user)
-    dismissed = 0
-    for pmid in req.pmids[:200]:
-        lit = (await db.execute(select(GlobalLiterature.id).where(GlobalLiterature.pmid == pmid))).scalar()
-        if not lit:
-            continue
-        feed_item = (await db.execute(
-            select(UserFeed).where(UserFeed.user_id == user_id, UserFeed.literature_id == lit)
-        )).scalar()
-        if feed_item and not feed_item.is_dismissed:
-            feed_item.is_dismissed = True
-            dismissed += 1
-            await _record_dismissed_tags(db, user_id, feed_item)
+    pmids = req.pmids[:200]
+    if not pmids:
+        return {"dismissed": 0, "total": 0}
+    lit_ids = (await db.execute(
+        select(GlobalLiterature.id).where(GlobalLiterature.pmid.in_(pmids))
+    )).scalars().all()
+    if not lit_ids:
+        return {"dismissed": 0, "total": len(pmids)}
+    feed_items = (await db.execute(
+        select(UserFeed).where(
+            UserFeed.user_id == user_id,
+            UserFeed.literature_id.in_(lit_ids),
+            UserFeed.is_dismissed == False,
+        )
+    )).scalars().all()
+    for feed_item in feed_items:
+        feed_item.is_dismissed = True
+        await _record_dismissed_tags(db, user_id, feed_item)
    await db.commit()
-    return {"dismissed": dismissed, "total": len(req.pmids)}
+    return {"dismissed": len(feed_items), "total": len(pmids)}


@router.post("/feed/{pmid}/read", summary="标记已读")
@@ -244,14 +268,10 @@ async def search_literature(
    count_q = select(func.count(GlobalLiterature.id)).where(search_cond)
    total = (await db.execute(count_q)).scalar() or 0
    tsq = func.plainto_tsquery("english", q)
-    best_match_rank = (
-        func.ts_rank(GlobalLiterature.search_tsv, tsq) * 0.3
-        + func.ln(func.coalesce(GlobalLiterature.cited_by_count, 0) + 1) * 2
-        + case((GlobalLiterature.pub_year >= 2020, 5), else_=0)
-    )
+    best_match_rank = AdvancedSearchEngine._best_match_order(tsq)
    result = await db.execute(
        select(GlobalLiterature).where(search_cond)
-        .order_by(best_match_rank.desc()).offset(offset).limit(page_size)
+        .order_by(best_match_rank).offset(offset).limit(page_size)
    )
    lit_list = result.scalars().all()
    tm = await load_tags_for_literature(db, [str(lit.id) for lit in lit_list])
@@ -400,6 +420,9 @@ async def literature_detail(pmid: int, db: AsyncSession = Depends(get_db), user:

    authors = lit.authors or []

+    # 先获取全文（可能失败），再提交状态修改
+    full_text_sections = await cached_get_full_text(lit.pmid, lit)
+
    # Auto-mark feed item as read
    feed_item = await db.execute(
        select(UserFeed).where(
@@ -450,7 +473,7 @@ async def literature_detail(pmid: int, db: AsyncSession = Depends(get_db), user:
        reading_status=reading_status,
        license_info=lit.license_info,
        pre_extracted_data=lit.pre_extracted_data,
-        full_text_sections=await cached_get_full_text(lit.pmid, lit),
+        full_text_sections=full_text_sections,
        chemical_list=lit.chemical_list or [],
        gene_symbols=lit.gene_symbols or [],
        num_refs=lit.num_refs,
@@ -585,7 +608,7 @@ async def export_single_endpoint(pmid: int, fmt: str, db: AsyncSession = Depends


@router.post("/export/batch/{fmt}", response_class=PlainTextResponse, summary="批量导出")
-async def export_batch_endpoint(fmt: str, pmids: list[int] = Query(default=[]), db: AsyncSession = Depends(get_db), user: dict = Depends(get_current_user)):
+async def export_batch_endpoint(fmt: str, pmids: list[int] = Query(default_factory=list), db: AsyncSession = Depends(get_db), user: dict = Depends(get_current_user)):
    """批量导出"""
    if fmt not in ("bibtex", "ris", "endnote", "csv", "mla"):
        raise HTTPException(status_code=400, detail="Unsupported format")
@@ -2,15 +2,18 @@

 import json
 import logging
+from collections import OrderedDict

 logger = logging.getLogger(__name__)


 class CacheService:
-    """内存缓存实现（生产环境可替换为 Redis）"""
+    """内存缓存实现（生产环境可替换为 Redis），内存模式最多保留 1000 条"""
+
+    MAX_MEMORY_ITEMS = 1000

    def __init__(self):
-        self._store: dict[str, dict] = {}
+        self._store: OrderedDict[str, dict] = OrderedDict()
        self._redis_failed = False
        self._redis = None

@@ -42,7 +45,10 @@ class CacheService:
            except Exception:
                logger.exception("Redis GET failed: %s", key)
                return None
-        return self._store.get(key)
+        val = self._store.get(key)
+        if val is not None:
+            self._store.move_to_end(key)
+        return val

    async def set(self, key: str, value: dict, ttl: int = 300) -> bool:
        r = await self._get_redis()
@@ -53,7 +59,11 @@ class CacheService:
            except Exception:
                logger.exception("Redis SET failed: %s", key)
                return False
+        # 内存模式：自动淘汰最旧条目，避免无限增长
+        # 注：OrderedDict.__setitem__ 在 Python 3.9+ 已自动移到末尾，无需 move_to_end
        self._store[key] = value
+        if len(self._store) > self.MAX_MEMORY_ITEMS:
+            self._store.popitem(last=False)
        return True

    async def delete(self, key: str):
@@ -79,14 +89,59 @@ class CacheService:
        return self._store.pop(key, None)

    async def get_or_set(self, key: str, loader, ttl: int = 300) -> dict | None:
-        """原子化 get-then-set-or-fallback：先查缓存，命中即返回；未命中调用 loader() 加载、写缓存后返回。"""
+        """原子化 get-then-set-or-fallback：先查缓存，命中即返回；未命中调用 loader() 加载、写缓存后返回。
+
+        防缓存雪崩（cache stampede）：Redis 模式用 SETNX 互斥锁，内存模式用 asyncio.Lock。
+        未抢到锁的请求会短暂自旋等待，避免重复计算。
+        """
        cached = await self.get(key)
        if cached is not None:
            return cached
-        value = await loader()
-        if value is not None:
-            await self.set(key, value, ttl=ttl)
-        return value
+
+        lock_key = f"lock:{key}"
+        if r := await self._get_redis():
+            # Redis 模式：SETNX 互斥锁
+            try:
+                locked = await r.set(lock_key, "1", nx=True, ex=10)
+            except Exception:
+                locked = True  # 锁操作失败时直接执行 loader，不做等待
+            if locked:
+                try:
+                    value = await loader()
+                    if value is not None:
+                        await r.set(key, json.dumps(value, default=str), ex=ttl)
+                    return value
+                finally:
+                    await r.delete(lock_key)
+            # 未抢到锁：自旋等待（最多 3 次，每次 100ms）
+            import asyncio
+            for _ in range(3):
+                await asyncio.sleep(0.1)
+                cached = await self.get(key)
+                if cached is not None:
+                    return cached
+            # 超时仍未获缓存：直接执行（保险）
+            value = await loader()
+            if value is not None:
+                await self.set(key, value, ttl=ttl)
+            return value
+        else:
+            # 内存模式：asyncio.Lock 互斥
+            import asyncio
+            if not hasattr(self, '_mem_locks'):
+                self._mem_locks: dict[str, asyncio.Lock] = {}
+            if key not in self._mem_locks:
+                self._mem_locks[key] = asyncio.Lock()
+            async with self._mem_locks[key]:
+                # 双检锁（double-check）：获取锁后可能已被其他协程写入
+                double_check = await self.get(key)
+                if double_check is not None:
+                    return double_check
+                value = await loader()
+                if value is not None:
+                    self._store[key] = value
+                    self._store.move_to_end(key)
+                return value

    async def invalidate_user(self, user_id: str):
        await self.delete(f"user:{user_id}:profile")
@@ -25,7 +25,7 @@ from app.models.operations import (
 )
 from app.models.review import ReviewLiterature, SystematicReview
 from app.models.team import Invitation, JournalClubAttendance, JournalClubQueue, SharedFolder, Team, TeamMember
-from app.models.user import LoginLog, Tenant, User, UserTenant
+from app.models.user import LoginLog, Tenant, User, UserSavedFilter, UserTenant

 __all__ = [
    "Tenant", "User", "UserTenant", "LoginLog",
@@ -33,6 +33,7 @@ __all__ = [
    "UserJournalSubscription", "UserSubscription", "UserDismissedTag", "UserFeed", "UserLiterature", "UserFolder",
    "AnonymousPageView", "UserNote", "UserPdfHighlight", "UserFeedback", "UserActivityLog",
    "Team", "TeamMember", "Invitation", "SharedFolder", "JournalClubAttendance", "JournalClubQueue",
+    "UserSavedFilter",
    "DrugApproval", "GuidelineVersion", "GuidelineEvidence", "ApprovalWorkflow", "ApprovalStep",
    "PipelineRun", "TenantSubscription", "SubscriptionEvent",
    "ExportJob", "ApiKey", "ApiUsageLog", "SystemNotification",
@@ -4,7 +4,7 @@ import uuid
 from datetime import date, datetime

 from sqlalchemy import Boolean, Date, DateTime, ForeignKey, Index, Integer, String, Text, Uuid, UniqueConstraint, func
-from sqlalchemy.dialects.postgresql import JSONB, TSVECTOR
+from sqlalchemy.dialects.postgresql import ARRAY, JSONB, TSVECTOR
 from sqlalchemy.orm import Mapped, mapped_column

 from app.db import Base, new_uuid
@@ -37,7 +37,7 @@ class GlobalLiterature(Base):
    pmc_id: Mapped[str | None] = mapped_column(String(20))           # PMC1234567
    is_oa: Mapped[bool] = mapped_column(default=False)               # 开放存取
    keywords: Mapped[dict] = mapped_column(JSONB, default=list)       # 作者关键词
-    pubmed_revised: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))  # PubMed 修订日期
+    pubmed_revised: Mapped[date | None] = mapped_column(Date)  # PubMed 修订日期（纯日期，NLM 仅提供 Y/M/D）
    grants: Mapped[dict] = mapped_column(JSONB, default=list)         # 基金资助
    cited_by_count: Mapped[int] = mapped_column(Integer, default=0)   # 引用次数
    full_text_sections: Mapped[dict | None] = mapped_column(JSONB)     # PMC OA 全文结构化段落（旧，不断新增存 COS）
@@ -46,7 +46,8 @@ class GlobalLiterature(Base):
    study_design: Mapped[dict | None] = mapped_column(JSONB)            # 研究设计分类 {primary, sub, design}
    pico: Mapped[dict | None] = mapped_column(JSONB)                    # PICO {population, intervention, comparator, outcome, sample_size, effect_size}
    citation_status: Mapped[str | None] = mapped_column(String(20))    # MedlineCitation Status: publisher/in-process/medline/...
-    date_completed: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))  # MeSH 标引完成日期
+    date_completed: Mapped[date | None] = mapped_column(Date)  # NLM 整条编目记录完成日（含 MeSH 标引在内的全部处理完成）
+    meshed_date: Mapped[date | None] = mapped_column(Date)  # MeSH 数据版本日期。通常=date_completed（MeSH 是编目最后一步，两者同天）。年更刷新后=新的 date_completed。NULL=无MeSH(in-process/publisher)
    retracted: Mapped[bool] = mapped_column(Boolean, default=False)    # 是否被撤稿
    retraction_details: Mapped[dict | None] = mapped_column(JSONB)       # 撤稿详情 {ref_type, ref_pmid, ref_source}
    license_info: Mapped[dict | None] = mapped_column(JSONB)              # 许可信息 {href, type, text}
@@ -54,19 +55,28 @@ class GlobalLiterature(Base):
    is_negative_result: Mapped[bool] = mapped_column(Boolean, default=False)  # 阴性结果
    rct_detection: Mapped[dict | None] = mapped_column(JSONB)              # RCT 检测结果 {is_rct, confidence, source, evidence}
    search_tsv: Mapped[str | None] = mapped_column(TSVECTOR)               # 全文检索向量（PG tsvector）
+    author_names_text: Mapped[str | None] = mapped_column(Text)             # 从 authors JSONB 提取的 family 文本，由触发器维护，供 trgm 索引
    negative_result_details: Mapped[dict | None] = mapped_column(JSONB)   # 阴性结果详情
    ai_summary: Mapped[dict | None] = mapped_column(JSONB)                  # AI 摘要 {one_liner, structured, implication}
+    is_preprint: Mapped[bool] = mapped_column(Boolean, default=False)       # 是否为预印本
+    auid_data: Mapped[dict | None] = mapped_column(JSONB)                     # Author Identifiers [{"type": "ORCID", "value": "0000-...", "author_index": 0}]
+    cois_statement: Mapped[str | None] = mapped_column(Text)                   # Conflict of Interest Statement（来自 <CoiStatement>）
+    vernacular_title: Mapped[str | None] = mapped_column(Text)                 # Transliterated/Vernacular Title（来自 <VernacularTitle>）

    # ═══ PubMed 补充解析字段 ═══
    chemical_list: Mapped[dict] = mapped_column(JSONB, default=list)        # ChemicalList [{name, registry_number, mesh_ui}]
+    investigators: Mapped[dict] = mapped_column(JSONB, default=list)        # InvestigatorList [{family, given, affiliation, identifiers}]
+    personal_name_subjects: Mapped[dict] = mapped_column(JSONB, default=list)  # PersonalNameSubjectList [{family, given}]
+    publication_notes: Mapped[dict] = mapped_column(JSONB, default=list)    # PubmedData/PublicationNote ["note1", "note2"]
    gene_symbols: Mapped[dict] = mapped_column(JSONB, default=list)          # GeneSymbolList
    num_refs: Mapped[int | None] = mapped_column(Integer)                   # NumberOfReferences
    publication_status: Mapped[str | None] = mapped_column(String(30))      # PublicationStatus (epublish/ppublish/aheadofprint)
    article_date: Mapped[date | None] = mapped_column(Date)                 # 电子出版日期（在线先发，早于纸质版）【PubMed: ArticleDate】
-    create_date: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))  # PubMed 记录创建日期【PubMed: CRDT】
-    entrez_date: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))  # PubMed 收录日期【PubMed: EDAT】
+    create_date: Mapped[date | None] = mapped_column(Date)  # PubMed 记录创建日期（纯日期）【PubMed: CRDT】
+    entrez_date: Mapped[date | None] = mapped_column(Date)  # PubMed 收录日期（纯日期）【PubMed: EDAT】
    databank_list: Mapped[dict] = mapped_column(JSONB, default=list)         # DataBankList [{databank_name, accession_numbers}]
    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}]

    source: Mapped[str] = mapped_column(String(30), default="pubmed_ftp")
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
@@ -79,6 +89,31 @@ class GlobalLiterature(Base):
        Index("ix_gl_created", "created_at"),
        Index("ix_gl_pmc_id", "pmc_id"),
        Index("ix_gl_search_tsv", "search_tsv", postgresql_using="gin"),
+        # 筛选性能：布尔/枚举列在纯筛选（无文本搜索）时避免顺序扫描
+        Index("ix_gl_retracted", "retracted"),
+        Index("ix_gl_is_oa", "is_oa"),
+        Index("ix_gl_is_negative", "is_negative_result"),
+        Index("ix_gl_is_preprint", "is_preprint"),
+        Index("ix_gl_citation_status", "citation_status"),
+        # JSONB 筛选索引：Species/Sex/Age 通过 mesh_headings @> 查询
+        Index("ix_gl_mesh_headings", "mesh_headings", postgresql_using="gin"),
+        # PubMed [TA] 搜索：journal ILIKE + journal_iso ILIKE 兜底
+        Index("ix_global_literature_journal_iso_trgm", "journal_iso", postgresql_using="gin", postgresql_ops={"journal_iso": "gin_trgm_ops"}),
+        # ILIKE 性能索引（pg_trgm）：title/abstract/journal 用于字段搜索和全文兜底
+        Index("ix_global_literature_title_trgm", "title", postgresql_using="gin", postgresql_ops={"title": "gin_trgm_ops"}),
+        Index("ix_global_literature_abstract_trgm", "abstract", postgresql_using="gin", postgresql_ops={"abstract": "gin_trgm_ops"}),
+        Index("ix_global_literature_journal_trgm", "journal", postgresql_using="gin", postgresql_ops={"journal": "gin_trgm_ops"}),
+        # Author 搜索：author_names_text 列 + trgm 索引
+        Index("ix_gl_author_names_trgm", "author_names_text", postgresql_using="gin", postgresql_ops={"author_names_text": "gin_trgm_ops"}),
+        # 常用筛选列 B-tree
+        Index("ix_gl_doi", "doi"),
+        Index("ix_gl_journal_iso", "journal_iso"),
+        Index("ix_gl_language", "language"),
+        # JSONB @> 查询索引：pub_type/grant/chemical/study_design 筛选
+        Index("ix_gl_pub_types_gin", "pub_types", postgresql_using="gin"),
+        Index("ix_gl_study_design_gin", "study_design", postgresql_using="gin"),
+        Index("ix_gl_grants_gin", "grants", postgresql_using="gin"),
+        Index("ix_gl_chemical_list_gin", "chemical_list", postgresql_using="gin"),
    )


@@ -166,6 +201,7 @@ class GlobalJournal(Base):
    priority_score: Mapped[float | None] = mapped_column()       # 综合优先级评分 0-100
    priority_source: Mapped[str | None] = mapped_column(String(20), default="auto")  # 'auto' | 'manual'
    specialty: Mapped[str | None] = mapped_column(String(50))    # 'oncology' | NULL
+    nlm_subsets: Mapped[list[str] | None] = mapped_column(ARRAY(String(30)), default=list)  # NLM 期刊子集（AIM/D/N/S 等）
    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())

@@ -1,9 +1,9 @@
 """运营与基础设施模型"""

 import uuid
-from datetime import datetime
+from datetime import date, datetime

-from sqlalchemy import JSON, Boolean, DateTime, ForeignKey, Index, Integer, String, Text, Uuid, func
+from sqlalchemy import Date, JSON, Boolean, DateTime, ForeignKey, Index, Integer, String, Text, Uuid, func
 from sqlalchemy.orm import Mapped, mapped_column

 from app.db import Base, new_uuid
@@ -24,6 +24,8 @@ class PipelineRun(Base):
    users_matched: Mapped[int] = mapped_column(Integer, default=0)
    feeds_generated: Mapped[int] = mapped_column(Integer, default=0)
    error_log: Mapped[str | None] = mapped_column(Text)
+    processed_date: Mapped[date | None] = mapped_column(Date)  # FTP 增量检查点：已处理的 EDAT 日期
+    run_metadata: Mapped[dict | None] = mapped_column("metadata", JSON)  # 扩展元数据（基线年份、文件序号范围等）
    started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
@@ -71,6 +71,22 @@ class UserTenant(Base):
    )


+class UserSavedFilter(Base):
+    __tablename__ = "user_saved_filters"
+
+    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=new_uuid)
+    user_id: Mapped[uuid.UUID] = mapped_column(Uuid, ForeignKey("users.id", ondelete="CASCADE"), nullable=False)
+    name: Mapped[str] = mapped_column(String(100), nullable=False)
+    query_string: Mapped[str] = mapped_column(String(500), nullable=False)
+    sort_order: Mapped[int] = mapped_column(Integer, default=0)
+    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__ = (
+        Index("ix_usf_user_sort", "user_id", "sort_order"),
+    )
+
+
 class LoginLog(Base):
    __tablename__ = "login_logs"

@@ -1,6 +1,7 @@
 """期刊工具：确保期刊在 global_journals 中存在（同步 + 异步版本）"""

-from sqlalchemy import func, insert as sa_insert, select
+from sqlalchemy import func, select
+from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import Session

@@ -68,7 +69,7 @@ def ensure_journal_sync(db: Session, issn: str | None, name: str | None) -> None
        if existing:
            return
        db.execute(
-            sa_insert(GlobalJournal)
+            pg_insert(GlobalJournal)
            .values(id=new_uuid(), name=(name or "Unknown")[:500], issn=issn,
                    tier="4", is_active=True)
            .on_conflict_do_nothing(index_elements=["issn"])
@@ -113,12 +114,16 @@ async def ensure_journal_async(db: AsyncSession, issn: str | None, name: str | N
        if existing.scalar():
            return
        stmt = (
-            sa_insert(GlobalJournal)
+            pg_insert(GlobalJournal)
            .values(id=new_uuid(), name=(name or "Unknown")[:500], issn=issn,
                    tier="4", is_active=True)
            .on_conflict_do_nothing(index_elements=["issn"])
        )
-        await db.execute(stmt)
+        try:
+            async with db.begin_nested():
+                await db.execute(stmt)
+        except Exception:
+            pass  # journal with same name already exists, skip

    elif name:
        existing = await db.execute(
@@ -141,9 +146,13 @@ async def ensure_journal_async(db: AsyncSession, issn: str | None, name: str | N
        except Exception:
            pass

-        stmt = (
-            sa_insert(GlobalJournal)
-            .values(id=new_uuid(), name=name[:500], tier="4", is_active=True)
-            .on_conflict_do_nothing(index_elements=["name"])
-        )
-        await db.execute(stmt)
+        try:
+            async with db.begin_nested():
+                stmt = (
+                    pg_insert(GlobalJournal)
+                    .values(id=new_uuid(), name=name[:500], tier="4", is_active=True)
+                    .on_conflict_do_nothing(index_elements=["name"])
+                )
+                await db.execute(stmt)
+        except Exception:
+            pass
@@ -8,7 +8,7 @@

 import asyncio
 import logging
-from datetime import datetime, timedelta
+from datetime import date, datetime, timedelta

 from sqlalchemy import select, and_, not_
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -82,8 +82,14 @@ async def recheck_batch(lits: list[GlobalLiterature]) -> int:
            # 有新 MeSH → 更新
            lit.mesh_headings = new_mesh
            lit.citation_status = article.get("citation_status", lit.citation_status)
+            # 同步更新 MeSH 数据版本时间：优先取 NLM date_completed（实际 MeSH 完成时间）
            if article.get("date_completed"):
                lit.date_completed = article["date_completed"]
+                lit.meshed_date = article["date_completed"]
+            else:
+                lit.meshed_date = date.today()  # 降级：NLM 未给日期则取我们的刷新日期
+
+            # 重新打标

            # 重新打标
            tagged = await tag_article(db, lit.id, new_mesh)
@@ -14,9 +14,14 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from app.compat import UTC
 from app.config import settings

-# 动态计算搜索年范围（覆盖最近两年，非硬编码）
-_CURRENT_YEAR: int = datetime.now().year
-_YEAR_FILTER: str = f"{_CURRENT_YEAR - 2}:{_CURRENT_YEAR}[dp]"
+# 预印本检测常量
+_PREPRINT_DOI_PREFIXES = ("10.1101/", "10.21203/", "10.22541/", "10.20944/", "10.26434/", "10.31219/", "10.2139/", "10.48550/")
+_PREPRINT_JOURNALS = frozenset({"medrxiv", "biorxiv", "arxiv", "research square", "preprints.org", "ssrn", "chemrxiv", "authorea"})
+
+# 动态计算搜索年范围（覆盖最近两年，非硬编码） — 每次调用时实时计算
+def _year_filter() -> str:
+    now = datetime.now()
+    return f"{now.year - 2}:{now.year}[dp]"

 from app.db import async_session
 from app.models.literature import GlobalLiterature
@@ -39,9 +44,6 @@ ELINK_URL = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils/elink.fcgi"

 # Europe PMC RESTful API（支持 cursor 分页，无 API Key 限制）
 EUROPE_PMC_SEARCH_URL = "https://www.ebi.ac.uk/europepmc/webservices/rest/search"
-# 内存缓存：{pmid: {"cited_by_count": N, "_fetched_at": timestamp}}
-# Elink 返回的引用数可能滞后，Europe PMC 的 citedByCount 更实时
-_europe_pmc_cache: dict[int, dict] = {}

 API_KEY = settings.PUBMED_API_KEY or None  # 空字符串视为 None
 TOOL_NAME = "scilit_oncology"
@@ -86,26 +88,27 @@ _PRECISION_MODE: str = "majr"
 def _build_oncology_queries(precision_mode: str = "majr") -> list[str]:
    """根据精度模式生成肿瘤学 MeSH 查询列表。"""
    tag = "MAJR" if precision_mode == "majr" else "MeSH"
+    year_f = _year_filter()
    queries = [
-        f'("Lung Neoplasms"[{tag}]) AND {_YEAR_FILTER}',
-        f'("Breast Neoplasms"[{tag}]) AND {_YEAR_FILTER}',
-        f'("Colorectal Neoplasms"[{tag}]) AND {_YEAR_FILTER}',
-        f'("Stomach Neoplasms"[{tag}]) AND {_YEAR_FILTER}',
-        f'("Liver Neoplasms"[{tag}]) AND {_YEAR_FILTER}',
-        f'("Prostatic Neoplasms"[{tag}]) AND {_YEAR_FILTER}',
-        f'("Leukemia"[{tag}]) AND {_YEAR_FILTER}',
-        f'("Lymphoma"[{tag}]) AND {_YEAR_FILTER}',
-        f'("Melanoma"[{tag}]) AND {_YEAR_FILTER}',
-        f'("Pancreatic Neoplasms"[{tag}]) AND {_YEAR_FILTER}',
-        f'("Ovarian Neoplasms"[{tag}]) AND {_YEAR_FILTER}',
-        f'("Esophageal Neoplasms"[{tag}]) AND {_YEAR_FILTER}',
-        f'("Glioma"[{tag}]) AND {_YEAR_FILTER}',
-        f'("Head and Neck Neoplasms"[{tag}]) AND {_YEAR_FILTER}',
-        f'("Sarcoma"[{tag}]) AND {_YEAR_FILTER}',
-        f'("Immunotherapy"[{tag}]) AND ("Neoplasms"[{tag}]) AND {_YEAR_FILTER}',
-        f'("Molecular Targeted Therapy"[{tag}]) AND ("Neoplasms"[{tag}]) AND {_YEAR_FILTER}',
+        f'("Lung Neoplasms"[{tag}]) AND {year_f}',
+        f'("Breast Neoplasms"[{tag}]) AND {year_f}',
+        f'("Colorectal Neoplasms"[{tag}]) AND {year_f}',
+        f'("Stomach Neoplasms"[{tag}]) AND {year_f}',
+        f'("Liver Neoplasms"[{tag}]) AND {year_f}',
+        f'("Prostatic Neoplasms"[{tag}]) AND {year_f}',
+        f'("Leukemia"[{tag}]) AND {year_f}',
+        f'("Lymphoma"[{tag}]) AND {year_f}',
+        f'("Melanoma"[{tag}]) AND {year_f}',
+        f'("Pancreatic Neoplasms"[{tag}]) AND {year_f}',
+        f'("Ovarian Neoplasms"[{tag}]) AND {year_f}',
+        f'("Esophageal Neoplasms"[{tag}]) AND {year_f}',
+        f'("Glioma"[{tag}]) AND {year_f}',
+        f'("Head and Neck Neoplasms"[{tag}]) AND {year_f}',
+        f'("Sarcoma"[{tag}]) AND {year_f}',
+        f'("Immunotherapy"[{tag}]) AND ("Neoplasms"[{tag}]) AND {year_f}',
+        f'("Molecular Targeted Therapy"[{tag}]) AND ("Neoplasms"[{tag}]) AND {year_f}',
        # 兜底：任何 Neoplasm 相关（覆盖甲状腺/肾/膀胱/宫颈/骨髓瘤等非白名单癌种）
-        f'("Neoplasms"[{tag}]) AND {_YEAR_FILTER}',
+        f'("Neoplasms"[{tag}]) AND {year_f}',
    ]
    return queries

@@ -114,25 +117,27 @@ def _build_oncology_queries(precision_mode: str = "majr") -> list[str]:
 ONCOLOGY_SEARCH_QUERIES = _build_oncology_queries("majr")

 # 2. 高召回：Title/Abstract 关键词，覆盖 in-process + publisher（尚未 MeSH-indexed）
-BROAD_ONCOLOGY_QUERIES = [
-    f'("lung cancer"[Title/Abstract] OR "lung carcinoma"[Title/Abstract] OR "lung adenocarcinoma"[Title/Abstract]) AND {_YEAR_FILTER}',
-    f'("breast cancer"[Title/Abstract] OR "breast carcinoma"[Title/Abstract]) AND {_YEAR_FILTER}',
-    f'("colorectal cancer"[Title/Abstract] OR "colon cancer"[Title/Abstract] OR "rectal cancer"[Title/Abstract]) AND {_YEAR_FILTER}',
-    f'("gastric cancer"[Title/Abstract] OR "stomach cancer"[Title/Abstract]) AND {_YEAR_FILTER}',
-    f'("hepatocellular"[Title/Abstract] OR "liver cancer"[Title/Abstract] OR "hepatic"[Title/Abstract]) AND {_YEAR_FILTER}',
-    f'("prostate cancer"[Title/Abstract] OR "prostatic cancer"[Title/Abstract]) AND {_YEAR_FILTER}',
-    f'("leukemia"[Title/Abstract] OR "leukaemia"[Title/Abstract]) AND {_YEAR_FILTER}',
-    f'("lymphoma"[Title/Abstract] OR "Hodgkin"[Title/Abstract] OR "non-Hodgkin"[Title/Abstract]) AND {_YEAR_FILTER}',
-    f'("melanoma"[Title/Abstract]) AND {_YEAR_FILTER}',
-    f'("pancreatic cancer"[Title/Abstract]) AND {_YEAR_FILTER}',
-    f'("ovarian cancer"[Title/Abstract] OR "ovarian carcinoma"[Title/Abstract]) AND {_YEAR_FILTER}',
-    f'("esophageal cancer"[Title/Abstract] OR "oesophageal"[Title/Abstract]) AND {_YEAR_FILTER}',
-    f'("glioma"[Title/Abstract] OR "glioblastoma"[Title/Abstract]) AND {_YEAR_FILTER}',
-    f'("head and neck cancer"[Title/Abstract] OR "oral cancer"[Title/Abstract] OR "squamous cell carcinoma"[Title/Abstract]) AND {_YEAR_FILTER}',
-    f'("sarcoma"[Title/Abstract] OR "osteosarcoma"[Title/Abstract]) AND {_YEAR_FILTER}',
-    # 兜底：非白名单癌种（甲状腺/肾/膀胱/宫颈/骨髓瘤等）
-    f'("cancer"[Title/Abstract] OR "tumor"[Title/Abstract] OR "tumour"[Title/Abstract] OR "neoplasm"[Title/Abstract] OR "malignan*"[Title/Abstract]) AND {_YEAR_FILTER}',
-]
+def _build_broad_queries() -> list[str]:
+    year_f = _year_filter()
+    return [
+        f'("lung cancer"[Title/Abstract] OR "lung carcinoma"[Title/Abstract] OR "lung adenocarcinoma"[Title/Abstract]) AND {year_f}',
+        f'("breast cancer"[Title/Abstract] OR "breast carcinoma"[Title/Abstract]) AND {year_f}',
+        f'("colorectal cancer"[Title/Abstract] OR "colon cancer"[Title/Abstract] OR "rectal cancer"[Title/Abstract]) AND {year_f}',
+        f'("gastric cancer"[Title/Abstract] OR "stomach cancer"[Title/Abstract]) AND {year_f}',
+        f'("hepatocellular"[Title/Abstract] OR "liver cancer"[Title/Abstract] OR "hepatic"[Title/Abstract]) AND {year_f}',
+        f'("prostate cancer"[Title/Abstract] OR "prostatic cancer"[Title/Abstract]) AND {year_f}',
+        f'("leukemia"[Title/Abstract] OR "leukaemia"[Title/Abstract]) AND {year_f}',
+        f'("lymphoma"[Title/Abstract] OR "Hodgkin"[Title/Abstract] OR "non-Hodgkin"[Title/Abstract]) AND {year_f}',
+        f'("melanoma"[Title/Abstract]) AND {year_f}',
+        f'("pancreatic cancer"[Title/Abstract]) AND {year_f}',
+        f'("ovarian cancer"[Title/Abstract] OR "ovarian carcinoma"[Title/Abstract]) AND {year_f}',
+        f'("esophageal cancer"[Title/Abstract] OR "oesophageal"[Title/Abstract]) AND {year_f}',
+        f'("glioma"[Title/Abstract] OR "glioblastoma"[Title/Abstract]) AND {year_f}',
+        f'("head and neck cancer"[Title/Abstract] OR "oral cancer"[Title/Abstract] OR "squamous cell carcinoma"[Title/Abstract]) AND {year_f}',
+        f'("sarcoma"[Title/Abstract] OR "osteosarcoma"[Title/Abstract]) AND {year_f}',
+        # 兜底：非白名单癌种（甲状腺/肾/膀胱/宫颈/骨髓瘤等）
+        f'("cancer"[Title/Abstract] OR "tumor"[Title/Abstract] OR "tumour"[Title/Abstract] OR "neoplasm"[Title/Abstract] OR "malignan*"[Title/Abstract]) AND {year_f}',
+    ]


 def _uid() -> uuid.UUID:
@@ -252,8 +257,6 @@ def _parse_europe_pmc_article(art: dict) -> dict | None:

    # Cache citedByCount for later use
    cbc = art.get("citedByCount", 0)
-    if isinstance(cbc, int) and cbc > 0:
-        _europe_pmc_cache[pmid] = {"cited_by_count": cbc, "_fetched_at": datetime.now(UTC)}

    return {
        "pmid": pmid,
@@ -261,12 +264,12 @@ def _parse_europe_pmc_article(art: dict) -> dict | None:
        "abstract": (art.get("abstractText") or "").strip() or None,
        "authors": authors,
        "doi": doi,
-        "pmc_id": (art.get("pmcid") or "").lstrip("PMC"),
+        "pmc_id": (art.get("pmcid") or "").removeprefix("PMC"),
        "is_oa": bool(art.get("pmcid")),
        "pubmed_revised": None,
        "citation_status": None,
        "date_completed": None,
-        "keywords": keywords,
+        "meshed_date": None,
        "grants": [],
        "journal": art.get("journalTitle", ""),
        "journal_issn": art.get("journalIssn", "") or art.get("journalISSN", ""),
@@ -298,6 +301,13 @@ def _parse_europe_pmc_article(art: dict) -> dict | None:
                             "ui": s.get("ui", s.get("supplMeshUI", "")),
                             "type": s.get("type", s.get("supplMeshType", ""))}
                            for s in (art.get("supplMeshList") or []) if isinstance(s, dict) and (s.get("name") or s.get("supplMeshName"))],
+        "pharmacological_actions": [],
+        "cois_statement": None,
+        "vernacular_title": None,
+        "is_preprint": (
+            (doi or "").lower().startswith(_PREPRINT_DOI_PREFIXES)
+            or (art.get("journalTitle") or "").lower() in _PREPRINT_JOURNALS
+        ),
        "source": "europe_pmc",
    }

@@ -415,7 +425,7 @@ def _parse_pubmed_xml(xml_text: str) -> list[dict]:
            abstract_parts = []
            for at in article.findall(".//Abstract/AbstractText"):
                label = at.get("Label", "")
-                text = "".join(at.itertext()) if at.text else ""
+                text = "".join(at.itertext()).strip()
                if label:
                    abstract_parts.append(f"{label}: {text}")
                else:
@@ -429,7 +439,38 @@ def _parse_pubmed_xml(xml_text: str) -> list[dict]:
                fore = au.findtext("ForeName", "")
                aff_elem = au.find(".//AffiliationInfo/Affiliation")
                aff = aff_elem.text if aff_elem is not None else ""
-                authors.append({"family": last, "given": fore, "affiliation": aff})
+                # Author Identifiers (ORCID etc.)
+                identifiers = []
+                for id_elem in au.findall("Identifier"):
+                    id_source = id_elem.get("Source", "")
+                    id_val = id_elem.text or ""
+                    if id_source and id_val:
+                        identifiers.append({"type": id_source, "value": id_val})
+                entry = {"family": last, "given": fore, "affiliation": aff}
+                if identifiers:
+                    entry["identifiers"] = identifiers
+                # Type="editor" in AuthorList
+                if au.get("Type") == "editor":
+                    entry["type"] = "editor"
+                authors.append(entry)
+
+            # InvestigatorList（多中心试验研究者）
+            investigators = []
+            for inv in article.findall(".//InvestigatorList/Investigator"):
+                last = inv.findtext("LastName", "")
+                fore = inv.findtext("ForeName", "")
+                aff_elem = inv.find(".//AffiliationInfo/Affiliation")
+                aff = aff_elem.text if aff_elem is not None else ""
+                identifiers = []
+                for id_elem in inv.findall("Identifier"):
+                    id_source = id_elem.get("Source", "")
+                    id_val = id_elem.text or ""
+                    if id_source and id_val:
+                        identifiers.append({"type": id_source, "value": id_val})
+                entry = {"family": last, "given": fore, "affiliation": aff}
+                if identifiers:
+                    entry["identifiers"] = identifiers
+                investigators.append(entry)

            # DOI
            doi = None
@@ -442,7 +483,7 @@ def _parse_pubmed_xml(xml_text: str) -> list[dict]:
            pmc_id = None
            for aid in article_elem.findall(".//PubmedData/ArticleIdList/ArticleId"):
                if aid.get("IdType") == "pmc":
-                    pmc_id = (aid.text or "").lstrip("PMC")
+                    pmc_id = (aid.text or "").removeprefix("PMC")
                    break

            # 临床试验注册号
@@ -464,7 +505,7 @@ def _parse_pubmed_xml(xml_text: str) -> list[dict]:
                    yr = int(dr_elem.findtext("Year", "0"))
                    mo = int(dr_elem.findtext("Month", "1"))
                    dy = int(dr_elem.findtext("Day", "1"))
-                    pubmed_revised = datetime(yr, mo, dy, tzinfo=UTC)
+                    pubmed_revised = date(yr, mo, dy)
                except (ValueError, TypeError):
                    pass

@@ -479,7 +520,7 @@ def _parse_pubmed_xml(xml_text: str) -> list[dict]:
                    yr = int(dc_elem.findtext("Year", "0"))
                    mo = int(dc_elem.findtext("Month", "1"))
                    dy = int(dc_elem.findtext("Day", "1"))
-                    date_completed = datetime(yr, mo, dy, tzinfo=UTC)
+                    date_completed = date(yr, mo, dy)
                except (ValueError, TypeError):
                    pass

@@ -609,6 +650,14 @@ def _parse_pubmed_xml(xml_text: str) -> list[dict]:
                    "mesh_ui": ui,
                })

+            # ─── PharmacologicalAction（药理作用）───
+            pharmacological_actions = []
+            for pa in medline.findall(".//PharmacologicalAction/NameOfSubstance"):
+                name = pa.text or ""
+                ui = pa.get("UI", "")
+                if name:
+                    pharmacological_actions.append({"name": name, "ui": ui})
+
            # ─── GeneSymbolList（基因符号）───
            gene_symbols = [gs.text for gs in article.findall(".//GeneSymbolList/GeneSymbol") if gs.text]

@@ -655,7 +704,7 @@ def _parse_pubmed_xml(xml_text: str) -> list[dict]:
                if h_year and h_year.isdigit():
                    h_mo = int(hist.findtext("Month", "1"))
                    h_dy = int(hist.findtext("Day", "1"))
-                    h_dt = datetime(int(h_year), h_mo, h_dy, tzinfo=timezone.utc)
+                    h_dt = date(int(h_year), h_mo, h_dy)
                    if status == "pubmed":
                        create_date = h_dt
                    elif status == "entrez":
@@ -684,6 +733,28 @@ def _parse_pubmed_xml(xml_text: str) -> list[dict]:
                    "type": stype,
                })

+            # ─── CoiStatement（利益冲突声明）───
+            cois_elem = article.find(".//CoiStatement")
+            cois_statement = cois_elem.text.strip() if cois_elem is not None and cois_elem.text else None
+
+            # ─── VernacularTitle（非英文标题的拉丁转写）───
+            vt_elem = article.find(".//VernacularTitle")
+            vernacular_title = vt_elem.text.strip() if vt_elem is not None and vt_elem.text else None
+
+            # ─── PersonalNameSubjectList（作为主题的人名）───
+            personal_name_subjects = []
+            for pns in medline.findall(".//PersonalNameSubjectList/PersonalNameSubject"):
+                last = pns.findtext("LastName", "")
+                fore = pns.findtext("ForeName", "")
+                if last or fore:
+                    personal_name_subjects.append({"family": last, "given": fore})
+
+            # ─── PubmedData/PublicationNote（出版注释）───
+            publication_notes = []
+            for pn in article_elem.findall(".//PubmedData/PublicationNote"):
+                if pn.text and pn.text.strip():
+                    publication_notes.append(pn.text.strip())
+
            articles.append({
                "pmid": pmid,
                "title": title.strip(),
@@ -694,7 +765,7 @@ def _parse_pubmed_xml(xml_text: str) -> list[dict]:
                "is_oa": pmc_id is not None,
                "pubmed_revised": pubmed_revised,
                "citation_status": citation_status,
-                "date_completed": date_completed,
+                "meshed_date": date_completed,  # 首次导入=date_completed，年更后刷新
                "keywords": keywords,
                "grants": grants,
                "journal": journal_title,
@@ -721,6 +792,16 @@ def _parse_pubmed_xml(xml_text: str) -> list[dict]:
                "entrez_date": entrez_date,
                "databank_list": databank_list,
                "suppl_mesh_list": suppl_mesh_list,
+                "pharmacological_actions": pharmacological_actions,
+                "cois_statement": cois_statement,
+                "vernacular_title": vernacular_title,
+                "is_preprint": (
+                    (doi or "").lower().startswith(_PREPRINT_DOI_PREFIXES)
+                    or (journal_title or "").lower() in _PREPRINT_JOURNALS
+                ),
+                "investigators": investigators,
+                "personal_name_subjects": personal_name_subjects,
+                "publication_notes": publication_notes,
                "source": "pubmed_api",
            })
        except Exception:
@@ -844,11 +925,15 @@ def _update_lit_from_article(lit: GlobalLiterature, article: dict):
        "title", "abstract", "authors", "doi", "journal", "journal_issn", "journal_iso",
        "volume", "issue", "pages", "pub_date", "pub_year", "pub_types",
        "mesh_headings", "language", "pmc_id", "keywords", "grants",
-        "pubmed_revised", "trial_reg", "citation_status", "date_completed",
+        "pubmed_revised", "trial_reg", "citation_status", "date_completed", "meshed_date",
        "retracted", "retraction_details",
        "chemical_list", "gene_symbols", "num_refs", "publication_status",
        "article_date", "print_date", "create_date", "entrez_date",
        "databank_list", "suppl_mesh_list",
+        "pharmacological_actions",
+        "cois_statement", "vernacular_title",
+        "investigators", "personal_name_subjects", "publication_notes",
+        "is_preprint",
    ]
    for f in fields:
        val = article.get(f)
@@ -866,6 +951,9 @@ def _update_lit_from_article(lit: GlobalLiterature, article: dict):
    # 阴性结果检测
    if article.get("abstract") or article.get("title"):
        from app.services.negative_detector import detect_negative_result
+        # 每次重新抓取后重置 is_negative_result，避免旧标记永不清零
+        lit.is_negative_result = False
+        lit.negative_result_details = None
        neg = detect_negative_result(article.get("abstract"), article.get("title", ""))
        if neg.get("is_negative"):
            lit.is_negative_result = True
@@ -886,7 +974,9 @@ async def _run_pipeline(
    当 use_europe_pmc=True 时使用 Europe PMC API（cursor 分页，适合大数据量）；
    False 时使用 NCBI E-utilities（默认，向后兼容）。
    """
-    stats = {"searched": 0, "fetched": 0, "new": 0, "updated": 0, "tagged": 0, "feeds": 0, "citations": 0, "oa_fetched": 0}
+    # 记录管线开始时间，供 PipelineRun duration 计算
+    _started_at = datetime.now(UTC)
+    stats = {"searched": 0, "fetched": 0, "new": 0, "updated": 0, "tagged": 0, "feeds": 0, "citations": 0, "oa_fetched": 0, "started_at": _started_at}

    all_new_pmids: list[int] = []
    oa_pending: dict[int, str] = {}  # pmid -> pmc_id（待抓全文的新 OA 文献）
@@ -953,6 +1043,7 @@ async def _run_pipeline(
                pubmed_revised=article.get("pubmed_revised"),
                citation_status=article.get("citation_status"),
                date_completed=article.get("date_completed"),
+                meshed_date=article.get("meshed_date"),
                grants=article.get("grants", []),
                trial_reg=article.get("trial_reg"),
                study_design=classify_study_design(article.get("pub_types", [])),
@@ -970,6 +1061,13 @@ async def _run_pipeline(
                entrez_date=article.get("entrez_date"),
                databank_list=article.get("databank_list", []),
                suppl_mesh_list=article.get("suppl_mesh_list", []),
+                pharmacological_actions=article.get("pharmacological_actions", []),
+                cois_statement=article.get("cois_statement"),
+                vernacular_title=article.get("vernacular_title"),
+                is_preprint=article.get("is_preprint", False),
+                investigators=article.get("investigators", []),
+                personal_name_subjects=article.get("personal_name_subjects", []),
+                publication_notes=article.get("publication_notes", []),
                source=article.get("source", "pubmed_api"),
                raw_xml_hash=hashlib.sha256(article["title"].encode()).hexdigest()[:16],
            )
@@ -1058,7 +1156,7 @@ async def _run_pipeline(
    if use_majr:
        await _process_queries(_build_oncology_queries(precision_mode))
    if use_broad:
-        await _process_queries(BROAD_ONCOLOGY_QUERIES)
+        await _process_queries(_build_broad_queries())

    # OA 全文抓取（新创建的 OA 文献）
    if oa_pending:
@@ -1152,7 +1250,7 @@ async def pull_daily_oncology(precision_mode: str = "majr"):
            id=_uid(), run_type="daily_update", status="success",
            articles_total=stats["fetched"], articles_new=stats["new"],
            articles_filtered=stats["tagged"], feeds_generated=stats["feeds"],
-            started_at=datetime.now(UTC),
+            started_at=stats["started_at"],
            completed_at=datetime.now(UTC),
        )
        db.add(run)
@@ -1175,7 +1273,7 @@ async def pull_broad_oncology():
            id=_uid(), run_type="broad_update", status="success",
            articles_total=stats["fetched"], articles_new=stats["new"],
            articles_filtered=stats["tagged"], feeds_generated=stats["feeds"],
-            started_at=datetime.now(UTC),
+            started_at=stats["started_at"],
            completed_at=datetime.now(UTC),
        )
        db.add(run)
@@ -1207,7 +1305,7 @@ async def pull_europe_pmc_oncology(max_results_per_query: int = 5000):
            id=_uid(), run_type="europe_pmc_full", status="success",
            articles_total=stats["fetched"], articles_new=stats["new"],
            articles_filtered=stats["tagged"], feeds_generated=stats["feeds"],
-            started_at=datetime.now(UTC),
+            started_at=stats["started_at"],
            completed_at=datetime.now(UTC),
        )
        db.add(run)
@@ -0,0 +1,457 @@
+"""PubMed FTP 每日增量更新管道
+
+取代旧的 E-utilities API 多路搜索（daily MAJR + weekly broad + retagger）。
+数据源：ftp://ftp.ncbi.nlm.nih.gov/pubmed/updatefiles/pubmed25updateNNNN.xml.gz
+
+流程：
+1. 读检查点 → 2. FTP 列表获取未处理文件 → 3. 下载 + 解析 + 过滤 + upsert + 删除标记 → 4. 更新检查点
+"""
+
+import asyncio
+import gzip
+import hashlib
+import io
+import logging
+import uuid
+from datetime import date, datetime, timezone
+from typing import Optional
+
+from lxml import etree
+from sqlalchemy import func, select
+
+from app.compat import UTC
+from app.db import async_session
+from app.models.literature import GlobalLiterature
+from app.models.operations import PipelineRun
+from app.services.feed_engine import generate_feeds_for_literature
+from app.services.journal_utils import ensure_journal_async
+from app.services.pubmed_api import _update_lit_from_article
+from app.services.tag_service import tag_article
+from scripts.pubmed_baseline import _extract_article, _is_oncology
+
+logger = logging.getLogger(__name__)
+
+FTP_HOST = "ftp.ncbi.nlm.nih.gov"
+FTP_PATH = "/pubmed/updatefiles/"
+FILE_SUFFIX = ".xml.gz"
+FTP_TIMEOUT = 30
+
+# 降级：FTP 不可用时回退 E-utilities API
+FALLBACK_ENABLED = True
+
+
+def _ftp_file_prefix(year: int) -> str:
+    """根据基线年份生成 FTP 文件前缀
+
+    >>> _ftp_file_prefix(2026)
+    'pubmed26n'
+    """
+    return f"pubmed{year % 100:02d}n"
+
+
+class FtpFileInfo:
+    """FTP 文件基本信息"""
+    __slots__ = ("name", "sequence", "size", "modified")
+
+    def __init__(self, name: str, sequence: int, size: int, modified: datetime):
+        self.name = name
+        self.sequence = sequence
+        self.size = size
+        self.modified = modified
+
+
+# ─── Checkpoint ───
+
+
+async def get_last_checkpoint() -> Optional[date]:
+    """读取最后一次成功运行的 processed_date（检查点）"""
+    async with async_session() as db:
+        result = await db.execute(
+            select(PipelineRun.processed_date)
+            .where(PipelineRun.run_type == "daily_ftp_update", PipelineRun.status == "success")
+            .order_by(PipelineRun.processed_date.desc().nulls_last())
+            .limit(1)
+        )
+        return result.scalar()
+
+
+# ─── 基线年份检测 ───
+
+
+async def get_baseline_year() -> int:
+    """获取基线导入的年份，用于确定 FTP 文件前缀。
+
+    读取 pipeline_runs 中 run_type='baseline_import' 的记录，
+    优先取 run_metadata['year']，fallback 到当前年份。
+
+    例如基线是 2026 年 → FTP 文件前缀为 pubmed26update*。
+    """
+    async with async_session() as db:
+        result = await db.execute(
+            select(PipelineRun.run_metadata)
+            .where(PipelineRun.run_type == "baseline_import", PipelineRun.status == "success")
+            .order_by(PipelineRun.created_at.desc())
+            .limit(1)
+        )
+        meta = result.scalar()
+    if meta and "year" in meta:
+        return int(meta["year"])
+    # 没有基线记录 → 用当前年份
+    return CURRENT_YEAR
+
+
+# ─── FTP 操作 ───
+
+
+def _list_files_sync(year: int) -> list[FtpFileInfo]:
+    """同步列出 FTP 目录（在 executor 中运行），只匹配指定年份的 update 文件"""
+    prefix = _ftp_file_prefix(year)
+    import ftplib
+
+    ftp = ftplib.FTP(FTP_HOST, timeout=FTP_TIMEOUT)
+    ftp.login()
+    ftp.cwd(FTP_PATH)
+
+    files: list[FtpFileInfo] = []
+    for name in ftp.nlst():
+        if not name.startswith(prefix) or not name.endswith(FILE_SUFFIX):
+            continue
+        basename = name.rsplit("/", 1)[-1]
+        try:
+            seq_str = basename[len(prefix):-len(FILE_SUFFIX)]
+            sequence = int(seq_str)
+        except (ValueError, IndexError):
+            continue
+
+        # MDTM → 发布时间
+        try:
+            raw = ftp.voidcmd(f"MDTM {basename}")
+            mod_time = datetime.strptime(raw[4:].strip(), "%Y%m%d%H%M%S").replace(tzinfo=timezone.utc)
+        except Exception:
+            mod_time = datetime.now(timezone.utc)
+
+        try:
+            size = ftp.size(basename) or 0
+        except Exception:
+            size = 0
+
+        files.append(FtpFileInfo(basename, sequence, size, mod_time))
+
+    ftp.quit()
+    return files
+
+
+async def list_update_files(year: int) -> list[FtpFileInfo]:
+    """异步获取 FTP 上指定年份的所有更新文件列表"""
+    return await asyncio.to_thread(_list_files_sync, year)
+
+
+def _download_file_sync(name: str) -> bytes:
+    """同步下载单个 FTP 文件（在 executor 中运行）"""
+    import ftplib
+
+    ftp = ftplib.FTP(FTP_HOST, timeout=FTP_TIMEOUT)
+    ftp.login()
+    ftp.cwd(FTP_PATH)
+    buf = io.BytesIO()
+    ftp.retrbinary(f"RETR {name}", buf.write)
+    ftp.quit()
+    return buf.getvalue()
+
+
+async def download_file(name: str) -> Optional[bytes]:
+    """异步下载单个 FTP 文件"""
+    try:
+        return await asyncio.to_thread(_download_file_sync, name)
+    except Exception as e:
+        logger.error("FTP download failed: %s — %s", name, e)
+        return None
+
+
+# ─── 文件处理 ───
+
+
+async def process_update_file(data: bytes, filename: str) -> dict:
+    """处理单个 update 文件：删除 + 解析 + 过滤 + upsert，整个文件一个事务"""
+    stats = {"articles_total": 0, "articles_new": 0, "articles_updated": 0,
+             "articles_deleted": 0, "articles_filtered": 0, "feeds_generated": 0}
+
+    tree = etree.parse(io.BytesIO(data))
+    root = tree.getroot()
+
+    # ── 1. DeleteCitation ──
+    delete_pmids = []
+    for dc in root.findall(".//DeleteCitation/PMID"):
+        if dc.text:
+            delete_pmids.append(int(dc.text))
+
+    # ── 2. 解析 & 过滤 ──
+    articles: list[dict] = []
+    for elem in root.findall(".//PubmedArticle"):
+        article = _extract_article(elem)
+        if article is None:
+            continue
+        stats["articles_total"] += 1
+        if not _is_oncology(article):
+            stats["articles_filtered"] += 1
+            continue
+        articles.append(article)
+
+    if not articles and not delete_pmids:
+        return stats
+
+    # ── 3. 事务处理 ──
+    async with async_session() as db:
+        # 3a. DeleteCitation → retracted
+        if delete_pmids:
+            result = await db.execute(
+                select(GlobalLiterature).where(GlobalLiterature.pmid.in_(delete_pmids))
+            )
+            to_retract = list(result.scalars().all())
+            for lit in to_retract:
+                lit.retracted = True
+            stats["articles_deleted"] = len(to_retract)
+            if to_retract:
+                logger.info("Marked %d retracted via DeleteCitation", len(to_retract))
+
+        # 3b. Upsert
+        from app.constants.study_design import classify_study_design
+        from app.services.pubmed_api import detect_rct
+
+        for article in articles:
+            result = await db.execute(
+                select(GlobalLiterature).where(GlobalLiterature.pmid == article["pmid"])
+            )
+            lit = result.scalar_one_or_none()
+
+            if lit:
+                _update_lit_from_article(lit, article)
+                await db.flush()
+                await tag_article(db, lit.id, article.get("mesh_headings", []))
+                stats["articles_updated"] += 1
+            else:
+                lit = GlobalLiterature(
+                    id=uuid.uuid4(),
+                    pmid=article["pmid"],
+                    title=article["title"],
+                    abstract=article.get("abstract"),
+                    authors=article.get("authors", []),
+                    doi=article.get("doi"),
+                    journal=article.get("journal"),
+                    journal_issn=article.get("journal_issn"),
+                    journal_iso=article.get("journal_iso"),
+                    volume=article.get("volume"),
+                    issue=article.get("issue"),
+                    pages=article.get("pages"),
+                    pub_date=article.get("pub_date"),
+                    pub_year=article.get("pub_year"),
+                    pub_types=article.get("pub_types", []),
+                    mesh_headings=article.get("mesh_headings", []),
+                    language=article.get("language", "en"),
+                    pmc_id=article.get("pmc_id"),
+                    is_oa=article.get("pmc_id") is not None,
+                    keywords=article.get("keywords", []),
+                    pubmed_revised=article.get("pubmed_revised"),
+                    citation_status=article.get("citation_status"),
+                    date_completed=article.get("date_completed"),
+                    meshed_date=article.get("date_completed"),
+                    grants=article.get("grants", []),
+                    trial_reg=article.get("trial_reg"),
+                    study_design=classify_study_design(article.get("pub_types", [])),
+                    rct_detection=detect_rct(
+                        article.get("pub_types"), article.get("title"),
+                        article.get("abstract"), article.get("pub_year"),
+                    ),
+                    chemical_list=article.get("chemical_list", []),
+                    gene_symbols=article.get("gene_symbols", []),
+                    num_refs=article.get("num_refs"),
+                    publication_status=article.get("publication_status"),
+                    article_date=article.get("article_date"),
+                    print_date=article.get("print_date"),
+                    create_date=article.get("create_date"),
+                    entrez_date=article.get("entrez_date"),
+                    databank_list=article.get("databank_list", []),
+                    suppl_mesh_list=article.get("suppl_mesh_list", []),
+                    pharmacological_actions=article.get("pharmacological_actions", []),
+                    cois_statement=article.get("cois_statement"),
+                    vernacular_title=article.get("vernacular_title"),
+                    is_preprint=article.get("is_preprint", False),
+                    investigators=article.get("investigators", []),
+                    personal_name_subjects=article.get("personal_name_subjects", []),
+                    publication_notes=article.get("publication_notes", []),
+                    source="ftp_daily_update",
+                    raw_xml_hash=hashlib.sha256(article["title"].encode()).hexdigest()[:16],
+                )
+                db.add(lit)
+                await db.flush()
+
+                await tag_article(db, lit.id, article.get("mesh_headings", []))
+                stats["articles_new"] += 1
+
+                # Feed
+                feeds = await generate_feeds_for_literature(db, lit.id)
+                stats["feeds_generated"] += feeds
+                if feeds > 0:
+                    try:
+                        from app.core.websocket import ws_manager
+                        await ws_manager.send_new_literature_alert(
+                            article["title"], article["pmid"], article.get("journal", ""))
+                    except Exception:
+                        logger.warning("ws_push_failed pmid=%s", article["pmid"], exc_info=True)
+
+                # 期刊去重
+                await ensure_journal_async(
+                    db, article.get("journal_issn"),
+                    article.get("journal"),
+                )
+
+        await db.commit()
+
+    logger.info("File %s: total=%d new=%d updated=%d deleted=%d filtered=%d feeds=%d",
+                filename, stats["articles_total"], stats["articles_new"],
+                stats["articles_updated"], stats["articles_deleted"],
+                stats["articles_filtered"], stats["feeds_generated"])
+    return stats
+
+
+# ─── E-utilities 降级 ───
+
+
+async def run_eutils_fallback(last_checkpoint: Optional[date]) -> dict:
+    """FTP 不可用时的降级路径：用 E-utilities reldate 拉取
+
+    使用 pubmed_api.py 的 search_pmids() + fetch_articles() + _process_article()
+    """
+    from app.services.pubmed_api import _run_pipeline
+
+    async with async_session() as db:
+        stats = await _run_pipeline(
+            db, max_per_query=30,
+            use_majr=True, use_broad=True,
+            precision_mode="majr",
+            update_citations=False,
+        )
+        logger.info("Fallback E-utilities pipeline done: %s", stats)
+        return stats
+
+
+# ─── 入口 ───
+
+
+async def run_daily_ftp_update(ctx: Optional[dict] = None) -> dict:
+    """每日 FTP 更新入口（ARQ cron task / 手动调用共用）
+
+    流程：
+    1. 读 processed_date 检查点
+    2. 确定 FTP 文件年份（优先基线年份，无基线则用当前年）
+    3. 列出 FTP 上该年份的所有 update 文件
+    4. 筛选未处理的（sequence > 上次处理的最大序号）
+    5. 按 sequence 顺序逐个下载 & 处理
+    6. 记录新的 PipelineRun（含年份信息）
+    """
+    logger.info("=== daily_ftp_update started ===")
+
+    last_checkpoint = await get_last_checkpoint()
+    logger.info("Last checkpoint: %s", last_checkpoint)
+
+    # 确定年份：已有检查点 → 从已处理文件名推断；首次运行 → 从基线记录获取
+    ftp_year = date.today().year
+    last_seq = 0
+    if last_checkpoint:
+        # 已跑过 → 年份沿用上次的
+        async with async_session() as db:
+            last_run = await db.execute(
+                select(PipelineRun.run_metadata)
+                .where(PipelineRun.run_type == "daily_ftp_update", PipelineRun.status == "success")
+                .order_by(PipelineRun.created_at.desc())
+                .limit(1)
+            )
+            meta = last_run.scalar()
+            if meta and "ftp_year" in meta:
+                ftp_year = int(meta["ftp_year"])
+            if meta and "last_sequence" in meta:
+                last_seq = int(meta["last_sequence"])
+    else:
+        # 首次运行 → 查基线年份
+        ftp_year = await get_baseline_year()
+        logger.info("First run: detected baseline year = %d", ftp_year)
+
+    logger.info("Querying FTP files for year %d (prefix: %s)", ftp_year, _ftp_file_prefix(ftp_year))
+
+    # 列出 FTP 文件
+    all_files = await list_update_files(ftp_year)
+    if not all_files:
+        # 当前年份还没文件？尝试上一年（年初过渡期）
+        prev_year = ftp_year - 1
+        logger.info("No files for year %d, trying %d", ftp_year, prev_year)
+        all_files = await list_update_files(prev_year)
+        if all_files:
+            ftp_year = prev_year
+
+    if not all_files:
+        logger.warning("No FTP update files found, falling back to E-utilities")
+        if FALLBACK_ENABLED:
+            return await run_eutils_fallback(last_checkpoint)
+        return {"status": "no_ftp_files"}
+
+    # 筛选未处理的（按 sequence 号比修改日期更准确）
+    all_files.sort(key=lambda f: f.sequence)
+    pending = [f for f in all_files if f.sequence > last_seq]
+    if not pending and last_checkpoint:
+        # sequence 没变，再用日期兜底（年更跨年时有用）
+        pending = [f for f in all_files
+                   if last_checkpoint is None or f.modified.date() > last_checkpoint]
+    if not pending:
+        logger.info("No new files since checkpoint %s (last_seq=%d)", last_checkpoint, last_seq)
+        return {"status": "up_to_date", "last_checkpoint": str(last_checkpoint), "last_sequence": last_seq}
+
+    logger.info("Processing %d file(s) (seq %d..%d, year %d)", len(pending),
+                pending[0].sequence, pending[-1].sequence, ftp_year)
+
+    # 逐个下载并处理
+    merged = {"articles_new": 0, "articles_updated": 0, "articles_deleted": 0,
+              "articles_filtered": 0, "articles_total": 0, "feeds_generated": 0,
+              "ftp_year": ftp_year}
+
+    last_modified = last_checkpoint or date(ftp_year, 1, 1)
+    for fi in pending:
+        raw = await download_file(fi.name)
+        if raw is None:
+            logger.warning("Skipping %s after download failure", fi.name)
+            continue
+        data = gzip.decompress(raw)
+        stats = await process_update_file(data, fi.name)
+        for k in ("articles_new", "articles_updated", "articles_deleted",
+                   "articles_filtered", "articles_total", "feeds_generated"):
+            merged[k] += stats.get(k, 0)
+        if fi.modified.date() > last_modified:
+            last_modified = fi.modified.date()
+        if fi.sequence > last_seq:
+            last_seq = fi.sequence
+
+    # 记录检查点（含年份和最后序号）
+    run_meta = {"ftp_year": ftp_year, "last_sequence": last_seq}
+    async with async_session() as db:
+        run = PipelineRun(
+            id=uuid.uuid4(),
+            run_type="daily_ftp_update",
+            status="success",
+            processed_date=last_modified,
+            files_processed=len(pending),
+            articles_total=merged["articles_total"],
+            articles_new=merged["articles_new"],
+            articles_updated=merged["articles_updated"],
+            articles_deleted=merged["articles_deleted"],
+            articles_filtered=merged["articles_filtered"],
+            feeds_generated=merged["feeds_generated"],
+            run_metadata=run_meta,
+            started_at=datetime.now(UTC),
+            completed_at=datetime.now(UTC),
+        )
+        db.add(run)
+        await db.commit()
+
+    logger.info("=== daily_ftp_update done: year=%d seq=%d new=%d updated=%d deleted=%d ===",
+                ftp_year, last_seq, merged["articles_new"], merged["articles_updated"],
+                merged["articles_deleted"])
+    return merged
@@ -22,22 +22,58 @@ import re
 from dataclasses import dataclass, field
 from enum import Enum, auto

+# ─── 查询复杂度限制 ───
+MAX_TERMS = 50       # 与 API 层 check_query_complexity 的 50 词限制对齐
+MAX_PAREN_DEPTH = 10  # 括号嵌套最深层数
+

 # ─── 字段标签映射 ───

 _FIELD_TAG_MAP: dict[str, str] = {
    "TI": "title",
    "AB": "abstract",
-    "TIAB": "all",       # 映射到 field=all（title+abstract covered）
+    "TIAB": "all",
    "AU": "author",
    "TA": "journal",
+    "JT": "journal",         # P1-2: Journal Title (同 TA)
+    "AD": "affiliation",
+    "CN": "author",
+    "FAU": "author",
+    "LAU": "author",
+    "TW": "all",
+    "OT": "all",
+    # 新增标量字段
+    "LA": "language",
+    "VI": "volume",
+    "IP": "issue",
+    "PG": "pages",
+    "LID": "lid",
 }

-# MeSH/PT/DP/PMID/DOI 需要特殊处理，不直接映射到 field 参数
-_SPECIAL_FIELDS = {"MH", "MAJR", "PT", "DP", "PMID", "DOI"}
+# 需要特殊处理的字段（不直接映射到 field 参数）
+_SPECIAL_FIELDS = {
+    "MH", "MAJR", "PT", "DP", "PMID", "DOI",
+    # 日期字段
+    "EDAT", "CRDT", "MHDA", "LR", "DCOM",
+    # JSONB 字段
+    "GR", "SH", "RN", "NM", "SI", "PA",
+    # 待补充抽取的字段（标记为特殊以便未来实现）
+    "AUID", "COIS", "ED", "IR", "PS", "PUBN", "TT",
+}

-# 所有合法字段标签
-_ALL_FIELD_TAGS = set(_FIELD_TAG_MAP.keys()) | _SPECIAL_FIELDS
+# 支持日期范围语法的字段
+_DATE_RANGE_FIELDS = {"DP", "EDAT", "CRDT", "MHDA", "LR", "DCOM"}
+
+# 所有合法字段标签（PubMed 全量字段）
+# P1-1: 移除了 BOOK/FILTER/ISBN（未实现，降级为 plain text 不如报错透明）
+_ALL_FIELD_TAGS = {
+    "AB", "AD", "AU", "CN", "FAU", "AUID", "LAU", "COIS",
+    "DCOM", "CRDT", "EDAT", "MHDA", "LR", "DP", "DOI",
+    "RN", "ED", "GR", "IR", "IP",
+    "TA", "JT", "LA", "LID", "MAJR", "SH", "MH", "MH:NOEXP", "OT", "PG",
+    "PA", "PT", "PMID", "PUBN", "SI", "PS", "NM", "TW",
+    "TI", "TIAB", "TT", "VI",
+}


 # ─── Token Types ───
@@ -53,6 +89,8 @@ class TokenType(Enum):
    COLON = auto()        # :
    WORD = auto()         # unquoted text
    NUMBER = auto()       # digits (for PMID, year)
+    DATE = auto()         # YYYY-MM-DD (normalized from YYYY/MM/DD)
+    UNKNOWN_FIELD = auto() # [...] 但内容不是已知字段标签
    EOF = auto()          # end of input


@@ -67,12 +105,14 @@ class Token:
 _TOKEN_PATTERNS: list[tuple[TokenType, str]] = [
    (TokenType.QUOTED, r'"(?:[^"\\]|\\.)*"'),
    (TokenType.FIELD, r'\[(?:' + '|'.join(_ALL_FIELD_TAGS) + r')\]'),
+    (TokenType.UNKNOWN_FIELD, r'\[[^\]]*\]'),  # 未识别的字段标签 → 触发降级
    (TokenType.AND, r'\bAND\b'),
    (TokenType.OR, r'\bOR\b'),
    (TokenType.NOT, r'\bNOT\b'),
    (TokenType.LPAREN, r'\('),
    (TokenType.RPAREN, r'\)'),
    (TokenType.COLON, r':'),
+    (TokenType.DATE, r'\d{4}-\d{2}-\d{2}'),
    (TokenType.NUMBER, r'\d+'),
    (TokenType.WORD, r'[^\s"\[\]():]+'),
 ]
@@ -90,7 +130,12 @@ def tokenise(query: str) -> list[Token]:
        for name, value in m.groupdict().items():
            if value is not None:
                ttype = TokenType[name]
+                if ttype == TokenType.UNKNOWN_FIELD:
+                    fname = value.strip('[]').upper()
+                    raise ParseError(f"不认识的字段标签 [{fname}]，降级为简单文本搜索")
                tokens.append(Token(ttype, value))
+                if len(tokens) > MAX_TERMS:
+                    raise ParseError(f"查询词过多（超过 {MAX_TERMS} 个），降级为简单文本搜索")
    tokens.append(Token(TokenType.EOF))
    return tokens

@@ -104,6 +149,8 @@ class Term:
    exact: bool = False          # True if quoted phrase
    field: str | None = None     # None = plain (no field tag); "title"/"abstract"/etc for mapped; "MH"/"PT"/etc for special
    is_not: bool = False         # True if preceded by NOT
+    group_id: int = -1           # -1 = top-level, >= 0 = parenthesized group index
+    _noexp: bool = False         # P1-4: [MH:noexp] 抑制树展开


@dataclass
@@ -119,14 +166,54 @@ class ParsedPubmedQuery:
    pub_types: list[str] = field(default_factory=list)             # [PT]
    doi_terms: list[str] = field(default_factory=list)             # [DOI]
    pmid_terms: list[int] = field(default_factory=list)            # [PMID]
+    affiliation_terms: list[Term] = field(default_factory=list)    # [AD]
+    language_terms: list[Term] = field(default_factory=list)       # [LA]
+    volume_terms: list[Term] = field(default_factory=list)         # [VI]
+    issue_terms: list[Term] = field(default_factory=list)          # [IP]
+    pages_terms: list[Term] = field(default_factory=list)          # [PG]
+    lid_terms: list[Term] = field(default_factory=list)            # [LID]
    date_from: str | None = None                                   # [DP] lower bound (YYYY-MM-DD)
    date_to: str | None = None                                     # [DP] upper bound (YYYY-MM-DD)
    year_from: int | None = None                                   # [DP] year lower
    year_to: int | None = None                                     # [DP] year upper
+    # 更多日期字段
+    edat_from: str | None = None                                   # [EDAT]
+    edat_to: str | None = None
+    crdt_from: str | None = None                                   # [CRDT]
+    crdt_to: str | None = None
+    mhda_from: str | None = None                                   # [MHDA]
+    mhda_to: str | None = None
+    lr_from: str | None = None                                     # [LR]
+    lr_to: str | None = None
+    dcom_from: str | None = None                                   # [DCOM]
+    dcom_to: str | None = None
+    # 特殊字段（存储 Term 以保留 is_not 标志）
+    mesh_terms: list[Term] = field(default_factory=list)           # [MH] (was list[str])
+    majr_terms: list[Term] = field(default_factory=list)           # [MAJR] (was list[str])
+    pub_types: list[Term] = field(default_factory=list)            # [PT] (was list[str])
+    doi_terms: list[Term] = field(default_factory=list)            # [DOI] (was list[str])
+    pmid_terms: list[Term] = field(default_factory=list)           # [PMID] (was list[int])
+    # JSONB 字段
+    grant_terms: list[Term] = field(default_factory=list)           # [GR]
+    subheading_terms: list[Term] = field(default_factory=list)      # [SH]
+    registry_terms: list[Term] = field(default_factory=list)        # [RN]
+    substance_terms: list[Term] = field(default_factory=list)       # [NM]
+    databank_terms: list[Term] = field(default_factory=list)        # [SI]
+    pharmaco_terms: list[Term] = field(default_factory=list)         # [PA]
+    ed_terms: list[Term] = field(default_factory=list)               # [ED]
+    investigator_terms: list[Term] = field(default_factory=list)     # [IR]
+    personal_name_terms: list[Term] = field(default_factory=list)    # [PS]
+    pubnote_terms: list[Term] = field(default_factory=list)          # [PUBN]
+    auid_terms: list[Term] = field(default_factory=list)             # [AUID]
+    cois_terms: list[Term] = field(default_factory=list)             # [COIS]
+    tt_terms: list[Term] = field(default_factory=list)               # [TT]
    plain_terms: list[Term] = field(default_factory=list)          # no field tag
    boolean_operator: str = "and"                                  # "and" | "or" | "mixed"
    has_not: bool = False                                          # contains NOT
    not_terms: list[Term] = field(default_factory=list)            # terms under NOT
+    groups: list[list[Term]] = field(default_factory=list)         # parenthesized sub-groups
+    group_operators: list[str] = field(default_factory=list)       # "and"/"or" per group (P2-2)
+    negated_date_ranges: set[str] = field(default_factory=set)     # date fields negated by NOT


 # ─── Parser ───
@@ -174,6 +261,7 @@ class PubmedQueryParser:
    def parse(self) -> ParsedPubmedQuery:
        """入口：解析完整的查询字符串。"""
        result = ParsedPubmedQuery()
+        self._depth = 0  # 括号嵌套深度计数器
        try:
            terms = self._parse_or_expr(result)
        except ParseError:
@@ -187,12 +275,12 @@ class PubmedQueryParser:
        elif has_or and not has_and:
            result.boolean_operator = "or"

-        result.has_not = any(t.is_not for t in terms if not getattr(t, '_is_range_end', False))
-        result.not_terms = [t for t in terms if t.is_not and not getattr(t, '_is_range_end', False)]
-
-        for t in terms:
-            if getattr(t, '_is_range_end', False):
-                continue
+        # 分组词不从 flat lists 走，避免括号内外的词被一起 AND/OR
+        # 同时 has_not/not_terms 也只考虑非分组词
+        _ungrouped = [t for t in terms if not getattr(t, '_is_range_end', False) and t.group_id < 0]
+        result.has_not = any(t.is_not for t in _ungrouped)
+        result.not_terms = [t for t in _ungrouped if t.is_not]
+        for t in _ungrouped:
            self._dispatch_term(result, t)

        return result
@@ -206,26 +294,75 @@ class PubmedQueryParser:
            result.tiab_terms.append(term)
        elif term.field == "author":
            result.author_terms.append(term)
+        elif term.field == "affiliation":
+            result.affiliation_terms.append(term)
        elif term.field == "journal":
            result.journal_terms.append(term)
+        elif term.field == "language":
+            result.language_terms.append(term)
+        elif term.field == "volume":
+            result.volume_terms.append(term)
+        elif term.field == "issue":
+            result.issue_terms.append(term)
+        elif term.field == "pages":
+            result.pages_terms.append(term)
+        elif term.field == "lid":
+            result.lid_terms.append(term)
        elif term.field == "MH":
-            result.mesh_terms.append(term.text)
+            result.mesh_terms.append(term)
        elif term.field == "MAJR":
-            result.majr_terms.append(term.text)
+            result.majr_terms.append(term)
        elif term.field == "PT":
-            result.pub_types.append(term.text)
+            result.pub_types.append(term)
        elif term.field == "PMID":
            try:
-                result.pmid_terms.append(int(term.text))
+                int(term.text)  # validate
            except ValueError:
                result.plain_terms.append(term)
+                return
+            result.pmid_terms.append(term)
        elif term.field == "DOI":
-            result.doi_terms.append(term.text)
+            result.doi_terms.append(term)
+        elif term.field == "GR":
+            result.grant_terms.append(term)
+        elif term.field == "SH":
+            result.subheading_terms.append(term)
+        elif term.field == "RN":
+            result.registry_terms.append(term)
+        elif term.field == "NM":
+            result.substance_terms.append(term)
+        elif term.field == "SI":
+            result.databank_terms.append(term)
+        elif term.field == "PA":
+            result.pharmaco_terms.append(term)
        elif term.field is None:
            result.plain_terms.append(term)
        elif term.field == "__RANGE_DP__":
-            # Already handled inline during parse; skip.
            pass
+        elif term.field == "__RANGE_EDAT__":
+            pass
+        elif term.field == "__RANGE_CRDT__":
+            pass
+        elif term.field == "__RANGE_MHDA__":
+            pass
+        elif term.field == "__RANGE_LR__":
+            pass
+        elif term.field == "__RANGE_DCOM__":
+            pass
+        elif term.field == "ED":
+            result.ed_terms.append(term)
+        elif term.field == "IR":
+            result.investigator_terms.append(term)
+        elif term.field == "PS":
+            result.personal_name_terms.append(term)
+        elif term.field == "PUBN":
+            result.pubnote_terms.append(term)
+        elif term.field == "AUID":
+            result.auid_terms.append(term)
+        elif term.field == "COIS":
+            result.cois_terms.append(term)
+        elif term.field == "TT":
+            result.tt_terms.append(term)
        else:
            result.plain_terms.append(term)

@@ -240,7 +377,7 @@ class PubmedQueryParser:

    def _is_primary_start(self, token: Token) -> bool:
        """Check if token could start a primary expression."""
-        return token.type in (TokenType.WORD, TokenType.QUOTED, TokenType.NUMBER,
+        return token.type in (TokenType.WORD, TokenType.QUOTED, TokenType.NUMBER, TokenType.DATE,
                              TokenType.LPAREN, TokenType.NOT)

    def _parse_and_expr(self, result: ParsedPubmedQuery) -> list[Term]:
@@ -272,9 +409,23 @@ class PubmedQueryParser:
    def _parse_primary(self, result: ParsedPubmedQuery, negated: bool = False) -> list[Term]:
        """primary → atom FIELD? | LPAREN query RPAREN"""
        if self.peek().type == TokenType.LPAREN:
+            self._depth += 1
+            if self._depth > MAX_PAREN_DEPTH:
+                raise ParseError(f"括号嵌套过深（超过 {MAX_PAREN_DEPTH} 层），降级为简单文本搜索")
            self.advance()
+            start_pos = self.pos  # P2-2: 记录组起始标记位置
            terms = self._parse_or_expr(result)
+            end_pos = self.pos    # P2-2: 记录组结束标记位置
            self.expect(TokenType.RPAREN)
+            self._depth -= 1
+            # 标记为子组，不放入 flat lists，保留括号分组结构
+            group_id = len(result.groups)
+            for t in terms:
+                t.group_id = group_id
+            result.groups.append(terms)
+            # P2-2: 检测组内是否有显式 OR
+            _has_or = any(t.type == TokenType.OR for t in self.tokens[start_pos:end_pos])
+            result.group_operators.append("or" if _has_or else "and")
            if negated:
                for t in terms:
                    t.is_not = True
@@ -289,17 +440,19 @@ class PubmedQueryParser:
          - "quoted phrase"[FIELD]
          - word[FIELD]
          - NUMBER:NUMBER[DP]  (year/date range)
+          - DATE:DATE[DP]     (full date range like 2024-01-01:2024-12-31)
          - NUMBER  (bare number)
          - word    (bare word)
        """
-        # Look ahead for NUMBER:NUMBER pattern
+        # Look ahead for range pattern (NUMBER:NUMBER or DATE:DATE)
        t0 = self.peek()
        t1 = self.peek_n(1)
        t2 = self.peek_n(2)

-        if (t0.type == TokenType.NUMBER
-                and t1 is not None and t1.type == TokenType.COLON
-                and t2 is not None and t2.type == TokenType.NUMBER):
+        if (t1 is not None and t1.type == TokenType.COLON
+                and t2 is not None
+                and t0.type in (TokenType.NUMBER, TokenType.DATE, TokenType.WORD)
+                and t2.type in (TokenType.NUMBER, TokenType.DATE, TokenType.WORD)):
            return self._parse_range(result, negated)

        # Normal atom
@@ -307,16 +460,23 @@ class PubmedQueryParser:
        text = token.value.strip('"') if token.type == TokenType.QUOTED else token.value
        is_exact = (token.type == TokenType.QUOTED)
        field = None
+        _noexp = False  # P1-4
        if self.peek().type == TokenType.FIELD:
            ft = self.advance()
-            field = ft.value[1:-1].upper()
-            if field in _FIELD_TAG_MAP:
-                field = _FIELD_TAG_MAP[field]
+            raw = ft.value[1:-1].upper()
+            # P1-4: [MH:noexp] → 抑制树展开
+            if raw == "MH:NOEXP":
+                field = "MH"
+                _noexp = True
+            else:
+                field = raw
+                if field in _FIELD_TAG_MAP:
+                    field = _FIELD_TAG_MAP[field]

-        return [Term(text, exact=is_exact, field=field, is_not=negated)]
+        return [Term(text, exact=is_exact, field=field, is_not=negated, _noexp=_noexp)]

    def _parse_range(self, result: ParsedPubmedQuery, negated: bool = False) -> list[Term]:
-        """Parse NUMBER:NUMBER[FIELD] — handles DP (year) ranges specially."""
+        """Parse NUMBER:NUMBER[FIELD] — handles date ranges specially."""
        start_val = self.advance().value  # NUMBER
        self.advance()                     # COLON
        end_val = self.advance().value     # NUMBER
@@ -326,25 +486,39 @@ class PubmedQueryParser:
            ft = self.advance()
            field = ft.value[1:-1].upper()

-        if field == "DP":
-            try:
-                sy, ey = int(start_val), int(end_val)
-            except ValueError:
-                sy = ey = None
-            # Year-only range (e.g., 2024:2026[DP])
+        if field in _DATE_RANGE_FIELDS:
+            attr_map = {
+                "DP": ("date_from", "date_to", "year_from", "year_to", "__RANGE_DP__"),
+                "EDAT": ("edat_from", "edat_to", None, None, "__RANGE_EDAT__"),
+                "CRDT": ("crdt_from", "crdt_to", None, None, "__RANGE_CRDT__"),
+                "MHDA": ("mhda_from", "mhda_to", None, None, "__RANGE_MHDA__"),
+                "LR": ("lr_from", "lr_to", None, None, "__RANGE_LR__"),
+                "DCOM": ("dcom_from", "dcom_to", None, None, "__RANGE_DCOM__"),
+            }
+            date_attr, date_attr_to, yr_from_attr, yr_to_attr, marker_field = attr_map[field]
+            # 反向范围自动交换（如 2026:2024[DP] → 2024:2026[DP]）
+            if start_val.isdigit() and end_val.isdigit() and int(start_val) > int(end_val):
+                start_val, end_val = end_val, start_val
+            # Year-only range (e.g., 2024:2026[EDAT])
            if start_val.isdigit() and len(start_val) == 4:
-                result.year_from = sy
-                result.year_to = ey
+                if yr_from_attr:
+                    setattr(result, yr_from_attr, int(start_val))
+                    setattr(result, yr_to_attr, int(end_val))
+                else:
+                    # For non-DP date fields: convert year to full date for consistency
+                    setattr(result, date_attr, f"{start_val}-01-01")
+                    setattr(result, date_attr_to, f"{end_val}-12-31")
            else:
-                # Full date range (e.g., 2024/01/01:2024/12/31[DP])
-                result.date_from = start_val
-                result.date_to = end_val
-            # Return a marker term so not-terms tracking knows about it, but _dispatch skips.
-            marker = Term(f"{start_val}:{end_val}", field="__RANGE_DP__", is_not=negated)
+                # Full date range (e.g., 2024-01-01:2024-12-31[EDAT])
+                setattr(result, date_attr, start_val)
+                setattr(result, date_attr_to, end_val)
+            marker = Term(f"{start_val}:{end_val}", field=marker_field, is_not=negated)
            marker._is_range_end = True
+            if negated:
+                result.negated_date_ranges.add(field)
            return [marker]

-        # Non-DP range or no field → plain text
+        # Non-date range or no field → plain text
        txt = f"{start_val}:{end_val}"
        if field:
            txt = f"{txt}[{field}]"
@@ -364,7 +538,7 @@ def is_pubmed_syntax(query: str) -> bool:
        return False
    if re.search(r'\[(' + '|'.join(_ALL_FIELD_TAGS) + r')\]', query, re.IGNORECASE):
        return True
-    if re.search(r'\b(AND|OR|NOT)\b', query):
+    if re.search(r'\b(AND|OR|NOT)\b', query, re.IGNORECASE):
        return True
    return False

@@ -378,6 +552,9 @@ def parse_pubmed_query(query: str) -> ParsedPubmedQuery:
        return ParsedPubmedQuery()

    try:
+        # 将 YYYY/MM/DD 格式的日期分隔符统一为 YYYY-MM-DD，使 tokeniser 正确识别为 DATE
+        import re as _re
+        query = _re.sub(r'(\d{4})/(\d{2})/(\d{2})', r'\1-\2-\3', query)
        tokens = tokenise(query)
        parser = PubmedQueryParser(tokens)
        return parser.parse()
@@ -392,7 +569,9 @@ def extract_pubmed_query_for_prisma(query: str) -> tuple[str, list[str]]:
        (normalized_query, mesh_terms_used)
    """
    parsed = parse_pubmed_query(query)
-    mesh_used = list(set(parsed.mesh_terms + parsed.majr_terms))
+    mesh_used = list(set(
+        [t.text for t in parsed.mesh_terms] + [t.text for t in parsed.majr_terms]
+    ))
    mesh_used.sort()

    # 标准化：统一字段大写
@@ -0,0 +1,174 @@
+"""ATM 自动术语映射引擎 — 让普通搜索框实现 PubMed 级别的 MeSH 展开
+
+流程:
+  1. MeSH Translation: exact entry term match (@>) → name_en ILIKE → per-word fallback
+  2. Tree Expansion: tree_number LIKE 前缀展开到所有子节点
+  3. SQL Condition: GlobalLiterature.id IN (tag_ids) — OR 合并 tsvector
+
+设计原则:
+  - 与 PubMed ATM 一致：先精确入口词匹配，再逐词拆解
+  - entry_terms 已全部小写标准化，用 @> 做精确 JSONB 元素匹配
+    杜绝 "breast cancer" 误匹配 "Breast Cancer Anti-Estrogen Resistance 1 Protein" 等问题
+
+用法:
+  cond = await expand_atm(db, "lung cancer")
+  if cond is not None:
+      conditions.append(sa.or_(cond, tsvector_cond))
+"""
+
+import logging
+from uuid import UUID
+
+from sqlalchemy import or_ as _or_, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.literature import GlobalTag, GlobalTagTreeNumber, GlobalLiteratureTag, GlobalLiterature
+
+logger = logging.getLogger(__name__)
+
+MIN_QUERY_LENGTH = 2
+
+
+def _escape_ilike(s: str) -> str:
+    """转义 ILIKE 模式中的通配符 _ 和 %"""
+    if not s:
+        return s
+    return s.replace('\\', '\\\\').replace('%', '\\%').replace('_', '\\_')
+
+
+async def expand_atm(db: AsyncSession, query: str):
+    """ATM 展开：输入纯文本查询，返回 SQLAlchemy condition 或 None"""
+    q = query.strip().replace('"', '').replace("'", '')
+    if not q or len(q) < MIN_QUERY_LENGTH:
+        return None
+
+    tag_ids = await _find_mesh_tags(db, q)
+    if not tag_ids:
+        tag_ids = await _find_partial_mesh_tags(db, q)
+    if not tag_ids:
+        return None
+
+    expanded_ids = await _expand_tree_numbers(db, tag_ids)
+    if not expanded_ids:
+        return None
+
+    return GlobalLiterature.id.in_(
+        select(GlobalLiteratureTag.literature_id)
+        .where(GlobalLiteratureTag.tag_id.in_(expanded_ids))
+        .distinct()
+    )
+
+
+async def _find_mesh_tags(db: AsyncSession, query: str) -> list[UUID]:
+    """通过精确入口词匹配 + name_en ILIKE 找 MeSH 标签"""
+    q = query.strip().lower()
+    seen: set[UUID] = set()
+    tag_ids: list[UUID] = []
+
+    # 方法 A: JSONB @> 精确元素匹配
+    # entry_terms 已全小写存储，可做 case-insensitive 精确匹配
+    # 只有 query 是 entry_terms 数组中的完整元素时才匹配
+    # 如 "breast cancer" 只匹配元素 "breast cancer"，不匹配 "breast cancer anti-estrogen resistance 1 protein"
+    stmt = select(GlobalTag.id).where(
+        GlobalTag.source.in_(["mesh", "manual"]),
+        GlobalTag.mesh_ui.isnot(None),
+        GlobalTag.entry_terms.isnot(None),
+        GlobalTag.entry_terms.contains([q]),
+    )
+    rows = await db.execute(stmt)
+    for (tid,) in rows:
+        if tid not in seen:
+            tag_ids.append(tid)
+            seen.add(tid)
+
+    # 方法 B: name_en ILIKE 匹配
+    like_pattern = f"%{_escape_ilike(query)}%"
+    stmt = select(GlobalTag.id).where(
+        GlobalTag.source.in_(["mesh", "manual"]),
+        GlobalTag.name_en.ilike(like_pattern),
+    )
+    rows = await db.execute(stmt)
+    for (tid,) in rows:
+        if tid not in seen:
+            tag_ids.append(tid)
+            seen.add(tid)
+
+    return tag_ids
+
+
+async def _find_partial_mesh_tags(db: AsyncSession, query: str) -> list[UUID]:
+    """降级匹配：多词查询按单词分别匹配，支持跨标签组合（如 "egfr mutation" → EGFR + Mutation）
+
+    PubMed 行为：当短语无法匹配 MeSH 入口词时，逐词尝试 MeSH 翻译。
+    """
+    words = [w.strip().lower() for w in query.strip().split() if len(w.strip()) >= MIN_QUERY_LENGTH][:10]
+    if len(words) < 2:
+        return []
+
+    seen: set[UUID] = set()
+    tag_ids: list[UUID] = []
+
+
+    # 批量精确入口词匹配（一次查询替代 N 次）
+    entry_terms_conds = []
+    name_ilike_conds = []
+    for word in words:
+        entry_terms_conds.append(
+            GlobalTag.entry_terms.contains([word])
+        )
+        name_ilike_conds.append(GlobalTag.name_en.ilike(f"%{_escape_ilike(word)}%"))
+
+    # 一次查询所有词的精确入口词匹配
+    if entry_terms_conds:
+        stmt = select(GlobalTag.id).where(
+            GlobalTag.source.in_(["mesh", "manual"]),
+            GlobalTag.mesh_ui.isnot(None),
+            GlobalTag.entry_terms.isnot(None),
+            _or_(*entry_terms_conds),
+        )
+        rows = await db.execute(stmt)
+        for (tid,) in rows:
+            if tid not in seen:
+                tag_ids.append(tid)
+                seen.add(tid)
+
+    # 一次查询所有词的 name_en ILIKE
+    if name_ilike_conds:
+        stmt = select(GlobalTag.id).where(
+            GlobalTag.source.in_(["mesh", "manual"]),
+            _or_(*name_ilike_conds),
+        )
+        rows = await db.execute(stmt)
+        for (tid,) in rows:
+            if tid not in seen:
+                tag_ids.append(tid)
+                seen.add(tid)
+
+    return tag_ids
+
+
+async def _expand_tree_numbers(db: AsyncSession, tag_ids: list[UUID]) -> list[UUID]:
+    """通过树号展开到所有子节点"""
+    rows = await db.execute(
+        select(GlobalTagTreeNumber.tree_number, GlobalTagTreeNumber.tag_id)
+        .where(GlobalTagTreeNumber.tag_id.in_(tag_ids))
+    )
+    tree_map: dict[str, list[UUID]] = {}
+    for tn, tid in rows:
+        tree_map.setdefault(tn, []).append(tid)
+
+    if not tree_map:
+        return tag_ids
+
+    expanded = set(tag_ids)
+    # 一次查询所有 tree_number 的子节点（用 OR 合并）
+    tn_conds = [GlobalTagTreeNumber.tree_number.like(f"{tn}%") for tn in tree_map]
+    rows = await db.execute(
+        select(GlobalTagTreeNumber.tag_id)
+        .where(_or_(*tn_conds))
+        .distinct()
+    )
+    for (tid,) in rows:
+        expanded.add(tid)
+
+    return list(expanded)
@@ -1,15 +1,24 @@
-"""共享标签加载工具"""
+"""共享标签加载工具（带单次请求级内存缓存）"""

 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession

 from app.models.literature import GlobalLiteratureTag, GlobalTag

+# 请求级缓存：同一批 literature_ids 在短时间内重复查询不走 DB
+_cache: dict[str, dict[str, list[dict]]] = {}
+_CACHE_MAX_KEYS = 5
+

 async def load_tags_for_literature(db: AsyncSession, literature_ids: list[str]) -> dict[str, list[dict]]:
    """批量加载文献标签，返回 {literature_id: [{name_zh, path, category, is_major}]}"""
    if not literature_ids:
        return {}
+    # 缓存 key 基于排序后的 ID 列表，确保相同集合命中
+    cache_key = ",".join(sorted(literature_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]
    result = await db.execute(
@@ -24,4 +33,9 @@ async def load_tags_for_literature(db: AsyncSession, literature_ids: list[str])
        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})
+
+    # 存入请求级缓存，限制大小
+    if len(_cache) >= _CACHE_MAX_KEYS:
+        _cache.pop(next(iter(_cache)), None)
+    _cache[cache_key] = tags_map
    return tags_map
@@ -1,9 +1,6 @@
 """文献打标服务：mesh_headings → global_tags → global_literature_tags

-策略：
-1. 按 mesh_ui 精确匹配（存量种子标签 + 懒创建标签）
-2. 按 name_en 回退匹配（import_mesh_tags.py 导入的 C04 标签无 mesh_ui）
-3. 仍未匹配 → 懒创建标签（确保打标覆盖）
+只关联已存在的 manual 标签，不懒创建新标签。
 """

 import uuid
@@ -12,19 +9,10 @@ import logging
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession

-from app.models.literature import GlobalTag, GlobalLiteratureTag, GlobalTagTreeNumber
+from app.models.literature import GlobalTag, GlobalLiteratureTag

 logger = logging.getLogger(__name__)

-# 新标签默认分类（按 mesh_ui 前缀推断）
-_CATEGORY_MAP = {
-    "C": "cancer", "D": "gene", "E": "treatment",
-    "F": "mesh_other", "G": "mesh_other", "H": "mesh_other",
-    "I": "mesh_other", "J": "mesh_other", "K": "mesh_other",
-    "L": "mesh_other", "M": "mesh_other", "N": "mesh_other",
-    "V": "study_type", "Z": "mesh_other",
-}
-

 async def tag_article(db: AsyncSession, lit_id: uuid.UUID,
                      mesh_headings: list[dict]) -> int:
@@ -61,35 +49,10 @@ async def tag_article(db: AsyncSession, lit_id: uuid.UUID,
            if t.id not in already_matched_ids:
                matched.append(t)

-    # 3. 仍未匹配的 → 懒创建
-    matched_uis = {t.mesh_ui for t in matched if t.mesh_ui}
-    created_tags: list[GlobalTag] = []
-    for mh in mesh_headings:
-        ui = mh.get("ui")
-        if not ui or ui in matched_uis:
-            continue
-        tag = GlobalTag(
-            mesh_ui=ui,
-            name_en=mh.get("descriptor", ""),
-            name_zh=None,            # 懒创建无中文名，后台补充
-            path=mh.get("descriptor", ""),
-            tag_category=_guess_category(mh.get("ui", "")),
-            level=1,
-            is_selectable=True,
-            source="auto",           # 标注为待审核来源
-            is_active=False,         # 未审核，用户端不可见
-        )
-        db.add(tag)
-        await db.flush()
-        matched.append(tag)
-        matched_uis.add(ui)
-        created_tags.append(tag)
-        logger.debug("懒创建标签: %s (%s)", tag.name_en, tag.mesh_ui)
-
    if not matched:
        return 0

-    # 4. 批量查询已有关系
+    # 3. 批量查询已有关系
    result = await db.execute(
        select(GlobalLiteratureTag.tag_id).where(
            GlobalLiteratureTag.literature_id == lit_id,
@@ -98,7 +61,7 @@ async def tag_article(db: AsyncSession, lit_id: uuid.UUID,
    )
    existing_tag_ids = {r for (r,) in result.all()}

-    # 5. 建立关联
+    # 4. 建立关联
    count = 0
    for tag in matched:
        if tag.id in existing_tag_ids:
@@ -145,9 +108,3 @@ async def retag_all_unindexed(db: AsyncSession, batch_size: int = 500) -> int:
        total += len(rows)
        logger.info("retag 进度: %d 篇", total)
    return total
-
-
-def _guess_category(mesh_ui: str) -> str:
-    """从 mesh_ui（D008175）推断分类"""
-    prefix = mesh_ui[0] if mesh_ui else ""
-    return _CATEGORY_MAP.get(prefix, "mesh_other")
@@ -0,0 +1,38 @@
+"""年份柱状图预计算缓存服务
+
+每天流水线跑完后刷新全库 year_counts 到 Redis，避免首页/空搜索时做全表 GROUP BY。
+"""
+
+import logging
+
+from sqlalchemy import func, select
+
+from app.core.cache import cache
+from app.db import async_session
+from app.models.literature import GlobalLiterature
+
+logger = logging.getLogger(__name__)
+
+CACHE_KEY = "search:year_counts:all"
+CACHE_TTL = 86400  # 24h（每天 pipeline 后 cron 刷新）
+
+
+async def precompute_year_counts():
+    """计算全库年份分布并写入 Redis 缓存"""
+    try:
+        async with async_session() as db:
+            rows = await db.execute(
+                select(GlobalLiterature.pub_year, func.count().label("cnt"))
+                .group_by(GlobalLiterature.pub_year)
+                .order_by(GlobalLiterature.pub_year.desc())
+            )
+            data = [{"year": y, "count": c} for y, c in rows if y is not None]
+
+        if not data:
+            logger.warning("year_counts 预计算：全库无文献数据")
+            return
+
+        await cache.set(CACHE_KEY, data, ttl=CACHE_TTL)
+        logger.info("year_counts 缓存已更新：%d 个年份", len(data))
+    except Exception:
+        logger.exception("year_counts 预计算失败")
@@ -1,9 +1,14 @@
-"""ARQ Worker 启动配置 + 定时任务"""
+"""ARQ Worker 启动配置 + 定时任务
+
+注意：旧的 daily_pubmed_pipeline(精搜) + weekly_broad_pipeline(宽搜) + daily_mesh_retagger 已合并为 daily_ftp_update。
+旧任务保留在此（标记 @deprecated），待稳定运行后移除。
+"""

 from arq import cron
 from arq.connections import RedisSettings

 from app.config import settings
+from app.services.pubmed_daily_update import run_daily_ftp_update
 from app.services.pubmed_api import pull_daily_oncology, pull_broad_oncology


@@ -17,9 +22,9 @@ async def shutdown(ctx):
    pass


-async def daily_pubmed_pipeline(ctx):
-    """每日 PubMed 管道任务"""
-    stats = await pull_daily_oncology()
+async def daily_ftp_update(ctx):
+    """每日 FTP 增量更新（取代旧的精搜+宽搜+retagger）"""
+    stats = await run_daily_ftp_update(ctx)
    return stats if isinstance(stats, dict) else {"status": "ok"}


@@ -35,12 +40,6 @@ async def daily_citation_update(ctx):
    return await run_citation_update(cached_hours=24)


-async def weekly_broad_pipeline(ctx):
-    """每周宽搜补充：Title/Abstract 覆盖 in-process + publisher"""
-    stats = await pull_broad_oncology()
-    return stats if isinstance(stats, dict) else {"status": "ok"}
-
-
 async def refresh_hot_articles_cache(ctx):
    """每 30 分钟刷新热搜缓存"""
    from app.services.hot_articles_cache import precompute_hot_articles
@@ -53,11 +52,20 @@ async def refresh_homepage_feed(ctx):
    await precompute_homepage_feed()


-async def daily_mesh_retagger(ctx):
-    """每日回查未标引文献（in-process/publisher→medline）"""
-    from app.services.mesh_retagger import run_retagger
-    stats = await run_retagger(max_articles=2000)
-    return stats
+async def refresh_year_counts_cache(ctx):
+    """刷新 year_counts 缓存（每日 pipeline 后）"""
+    from app.services.year_counts_cache import precompute_year_counts
+    await precompute_year_counts()
+
+
+async def refresh_filter_options_cache(ctx):
+    """刷新搜索筛选选项缓存（每日 pipeline 后，1h TTL）"""
+    from app.core.cache import cache
+    from app.db import async_session
+    from app.api.v1.features import _load_filter_options
+    async with async_session() as db:
+        data = await _load_filter_options(db)
+        await cache.set("filter-options", data, ttl=3600)


 async def weekly_drug_approval_sync(ctx):
@@ -73,23 +81,61 @@ async def daily_approval_notification(ctx):
    return await run_approval_notification()


+# ── @deprecated 旧管道任务（待移除） ──
+
+
+async def daily_pubmed_pipeline(ctx):
+    """@deprecated 已由 daily_ftp_update 取代"""
+    stats = await pull_daily_oncology()
+    return stats if isinstance(stats, dict) else {"status": "ok"}
+
+
+async def weekly_broad_pipeline(ctx):
+    """@deprecated 已由 daily_ftp_update 取代"""
+    stats = await pull_broad_oncology()
+    return stats if isinstance(stats, dict) else {"status": "ok"}
+
+
+async def daily_mesh_retagger(ctx):
+    """@deprecated 已由 daily_ftp_update 取代"""
+    from app.services.mesh_retagger import run_retagger
+    stats = await run_retagger(max_articles=2000)
+    return stats
+
+
 class WorkerSettings:
    on_startup = startup
    on_shutdown = shutdown
    redis_settings = RedisSettings.from_dsn(settings.REDIS_URL)
-    functions = [daily_pubmed_pipeline, daily_digest_task, daily_citation_update, weekly_broad_pipeline, refresh_hot_articles_cache, refresh_homepage_feed, daily_mesh_retagger, weekly_drug_approval_sync, daily_approval_notification]
+    functions = [
+        daily_ftp_update,
+        daily_digest_task,
+        daily_citation_update,
+        refresh_hot_articles_cache,
+        refresh_homepage_feed,
+        refresh_year_counts_cache,
+        refresh_filter_options_cache,
+        weekly_drug_approval_sync,
+        daily_approval_notification,
+        # @deprecated（保留但不在 cron 中激活，确保旧代码可引用）
+        daily_pubmed_pipeline,
+        weekly_broad_pipeline,
+        daily_mesh_retagger,
+    ]

    # 定时任务（UTC 时间）
    cron_jobs = [
-        cron(daily_pubmed_pipeline, hour=3, minute=7),      # 03:07 UTC = 11:07 Beijing
-        cron(weekly_broad_pipeline, hour=3, minute=37, weekday='sun'),  # 周日 03:37 UTC = 11:37 Beijing
+        cron(daily_ftp_update, hour=3, minute=7),           # 03:07 UTC = 11:07 Beijing（取代精搜+宽搜+retagger）
        cron(daily_digest_task, hour=22, minute=30),        # 22:30 UTC = 06:30 Beijing
        cron(daily_citation_update, hour=5, minute=13),     # 05:13 UTC = 13:13 Beijing
-        cron(daily_mesh_retagger, hour=4, minute=47),       # 04:47 UTC = 12:47 Beijing
-        cron(weekly_drug_approval_sync, hour=6, minute=23, weekday='mon'),  # 周一 06:23 UTC = 14:23 Beijing 同步FDA审批
-        cron(daily_approval_notification, hour=7, minute=47),  # 07:47 UTC = 15:47 Beijing 审批通知推送
-        cron(refresh_hot_articles_cache, minute=0),    # 每整点刷新热搜缓存
-        cron(refresh_hot_articles_cache, minute=30),   # 每半点刷新热搜缓存
-        cron(refresh_homepage_feed, minute=0),         # 每整点刷新首页 Feed
-        cron(refresh_homepage_feed, minute=30),        # 每半点刷新首页 Feed
+        cron(weekly_drug_approval_sync, hour=6, minute=23, weekday='mon'),  # 周一 06:23 UTC
+        cron(daily_approval_notification, hour=7, minute=47),  # 07:47 UTC = 15:47 Beijing
+        cron(refresh_hot_articles_cache, minute=0),         # 每整点刷新热搜
+        cron(refresh_hot_articles_cache, minute=30),        # 每半点刷新热搜
+        cron(refresh_homepage_feed, minute=0),              # 每整点刷新 Feed
+        cron(refresh_homepage_feed, minute=30),             # 每半点刷新 Feed
+        cron(refresh_year_counts_cache, hour=3, minute=17), # 03:17 UTC = 11:17 Beijing（FTP 后 10 分钟）
+        cron(refresh_year_counts_cache, minute=22),         # 开机 ~22 分钟后刷新一次
+        cron(refresh_filter_options_cache, hour=3, minute=18),  # 03:18 UTC（FTP 更新后，紧随 year_counts）
+        cron(refresh_filter_options_cache, minute=23),          # 开机 ~23 分钟后
    ]