From c791c060967ced67a7bb76111ef8b76ad4f7ddcc Mon Sep 17 00:00:00 2001
From: "34047007@qq.com" <34047007@qq.com>
Date: Mon, 27 Jul 2026 16:16:17 +0800
Subject: [PATCH] =?UTF-8?q?fix:=20FTP=20pipeline=20MeSH=20=E6=89=93?=
 =?UTF-8?q?=E6=A0=87=E9=93=BE=20=E2=80=94=20=E4=BF=AE=E5=A4=8D=20=5Fupsert?=
 =?UTF-8?q?=5Farticle=5Fimpl=20=E7=BC=BA=E5=B0=91=20tag=5Farticle()=20?=
 =?UTF-8?q?=E8=B0=83=E7=94=A8?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

FTP 增量更新写入 mesh_headings JSONB 后未调用 tag_article()，导致
GlobalLiteratureTag 无关联记录，[MH]/[MAJR] 搜索返回空。新增 tag_ids
数组维护 + 标签缓存失效。
---
 backend/app/api/v1/admin.py          |  4 ++--
 backend/app/services/tag_service.py  | 17 ++++++++++++++++-
 backend/app/tasks/pubmed_pipeline.py | 15 +++++++++++++++
 3 files changed, 33 insertions(+), 3 deletions(-)

diff --git a/backend/app/api/v1/admin.py b/backend/app/api/v1/admin.py
index 2b1a9f5..4515526 100644
--- a/backend/app/api/v1/admin.py
+++ b/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
diff --git a/backend/app/services/tag_service.py b/backend/app/services/tag_service.py
index 18041a8..a83a453 100644
--- a/backend/app/services/tag_service.py
+++ b/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
 
 
diff --git a/backend/app/tasks/pubmed_pipeline.py b/backend/app/tasks/pubmed_pipeline.py
index fc7b79f..9afad95 100644
--- a/backend/app/tasks/pubmed_pipeline.py
+++ b/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