fix: FTP pipeline MeSH 打标链 — 修复 _upsert_article_impl 缺少 tag_article() 调用

FTP 增量更新写入 mesh_headings JSONB 后未调用 tag_article()，导致
GlobalLiteratureTag 无关联记录，[MH]/[MAJR] 搜索返回空。新增 tag_ids
数组维护 + 标签缓存失效。

backend/app/api/v1/admin.py

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

backend/app/services/tag_service.py

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

backend/app/tasks/pubmed_pipeline.py

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