diff --git a/app/crawler/gxgp_spider.py b/app/crawler/gxgp_spider.py index eeb528c..2bdff09 100644 --- a/app/crawler/gxgp_spider.py +++ b/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, diff --git a/app/services/pipeline.py b/app/services/pipeline.py index 663e074..b5049fe 100644 --- a/app/services/pipeline.py +++ b/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', '')}" diff --git a/tests/test_crawler/test_gxgp_spider.py b/tests/test_crawler/test_gxgp_spider.py index 221e763..5c5253b 100644 --- a/tests/test_crawler/test_gxgp_spider.py +++ b/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 diff --git a/tests/test_services/test_pipeline.py b/tests/test_services/test_pipeline.py index 7326560..c6e86be 100644 --- a/tests/test_services/test_pipeline.py +++ b/tests/test_services/test_pipeline.py @@ -45,13 +45,16 @@ 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, "_send_notifications", AsyncMock(return_value=1)): - pipe_result = await pipeline.process( - result.announcements, config - ) - assert pipe_result.stored == 2 - assert pipe_result.filtered == 1 - assert pipe_result.notified == 1 + 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 + ) + assert pipe_result.stored == 2 + assert pipe_result.filtered == 1 + assert pipe_result.notified == 1 @pytest.mark.asyncio @@ -84,12 +87,15 @@ 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, "_send_notifications", AsyncMock(return_value=2)): - with patch.object(pipeline, "_mark_sent", AsyncMock(return_value=2)): - pipe_result = await pipeline.process( - result.announcements, config - ) - assert pipe_result.stored == 2 - assert pipe_result.filtered == 0 - assert pipe_result.notified == 2 - pipeline._mark_sent.assert_awaited_once() + 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( + result.announcements, config + ) + assert pipe_result.stored == 2 + assert pipe_result.filtered == 0 + assert pipe_result.notified == 2 + pipeline._mark_sent.assert_awaited_once()