fix: 防止重复推送通知 — 添加已发送检查 + GXGP 标记已发送

- Pipeline 新增 _exclude_sent() 方法，通知前查询数据库跳过 is_sent=True 的公告
- GXGP Spider mark_sent 改为 True，配合去重逻辑防止每次爬取重复通知

app/crawler/gxgp_spider.py

@@ -25,7 +25,7 @@ class GXGPSpider(BaseSpider):
             keywords=list(settings.crawler_keywords),
             dedup_enabled=True,
             notify_mode="filtered",
-            mark_sent=False,
+            mark_sent=True,
         )
 
     async def crawl(self, sources: list[str] | None = None,

app/services/pipeline.py

@@ -33,7 +33,11 @@ class PostCrawlPipeline:
                          if self._match_keywords(a, config.keywords)]
             result.filtered = before - len(to_notify)
 
-        # 4. Notify
+        # 4. Skip already-notified
+        if to_notify:
+            to_notify = await self._exclude_sent(to_notify)
+
+        # 5. Notify
         if config.notify_mode == "all":
             result.notified = await self._send_notifications(to_notify)
         elif config.notify_mode == "filtered":
@@ -42,7 +46,7 @@ class PostCrawlPipeline:
             elif not config.filter_enabled:
                 result.notified = await self._send_notifications(to_notify)
 
-        # 5. Mark sent
+        # 6. Mark sent
         if config.mark_sent and result.notified > 0:
             await self._mark_sent(to_notify)
 
@@ -99,6 +103,24 @@ class PostCrawlPipeline:
         await self.db.commit()
         return result.rowcount
 
+    async def _exclude_sent(self, announcements: list[dict[str, Any]]) -> list[dict[str, Any]]:
+        from sqlalchemy import select
+
+        from app.models.announcement import Announcement
+
+        hashes = [a["content_hash"] for a in announcements if a.get("content_hash")]
+        if not hashes:
+            return announcements
+
+        stmt = select(Announcement.content_hash).where(
+            Announcement.content_hash.in_(hashes),
+            Announcement.is_sent == True,  # noqa: E712
+        )
+        result = await self.db.execute(stmt)
+        sent_hashes = {row[0] for row in result.fetchall()}
+
+        return [a for a in announcements if a.get("content_hash") not in sent_hashes]
+
     @staticmethod
     def _match_keywords(announcement: dict[str, Any], keywords: list[str]) -> bool:
         text = f"{announcement.get('title', '')} {announcement.get('purchase_name', '')}"

tests/test_crawler/test_gxgp_spider.py

@@ -21,7 +21,7 @@ async def test_gxgp_spider_pipeline_config():
     assert isinstance(config, PipelineConfig)
     assert config.filter_enabled is True
     assert config.notify_mode == "filtered"
-    assert config.mark_sent is False
+    assert config.mark_sent is True
 
 
 @pytest.mark.asyncio

tests/test_services/test_pipeline.py

@@ -45,6 +45,9 @@ async def test_pipeline_filtered_mode():
     pipeline = PostCrawlPipeline(db_session=mock_db, notification_service=mock_notify)
 
     with patch.object(pipeline, "_save_to_db", AsyncMock(return_value=2)):
+        with patch.object(pipeline, "_exclude_sent", AsyncMock(
+            side_effect=lambda anns: [a for a in anns if "大化" in a["title"]]
+        )):
             with patch.object(pipeline, "_send_notifications", AsyncMock(return_value=1)):
                 pipe_result = await pipeline.process(
                     result.announcements, config
@@ -84,6 +87,9 @@ async def test_pipeline_all_mode():
     pipeline = PostCrawlPipeline(db_session=mock_db, notification_service=mock_notify)
 
     with patch.object(pipeline, "_save_to_db", AsyncMock(return_value=2)):
+        with patch.object(pipeline, "_exclude_sent", AsyncMock(
+            side_effect=lambda anns: anns
+        )):
             with patch.object(pipeline, "_send_notifications", AsyncMock(return_value=2)):
                 with patch.object(pipeline, "_mark_sent", AsyncMock(return_value=2)):
                     pipe_result = await pipeline.process(