fix: avoid duplicate dispatch within a single rss batch

Author: RhoninSeiei

main.py

@@ -15,7 +15,7 @@
     "astrbot_rss",
     "AstrBot RSS Forwarder",
     "面向 AstrBot 的 RSS/RSSHub 推送编排插件",
-    "0.3.2",
+    "0.3.3",
 )
 class RSSPlugin(Star, RSSCommands):
     def __init__(self, context: Context, config=None):

metadata.yaml

@@ -1,6 +1,6 @@
 name: astrbot_plugin_rss_forwarder
 display_name: AstrBot RSS Forwarder
 desc: 面向 AstrBot 的 RSS/RSSHub 推送编排插件，支持去重、路由与后续智能增强。
-version: v0.3.2
+version: v0.3.3
 author: RhoninSeiei
 repo: https://github.com/RhoninSeiei/astrbot_plugin_rss_forwarder

scheduler.py

@@ -238,10 +238,12 @@ async def _run_job_once_guarded(self, job: JobConfig) -> None:
             pushed_count = 0
             parsed_count = 0
             skipped_seen_count = 0
+            skipped_batch_duplicate_count = 0
             skipped_history_count = 0
             skipped_invalid_target_count = 0
             dispatch_fail_count = 0
             error_summary = ""
+            seen_in_run: set[str] = set()
 
             try:
                 raw_items = await self._call_fetch(job)
@@ -255,9 +257,22 @@ async def _run_job_once_guarded(self, job: JobConfig) -> None:
                 parsed_count = len(items)
                 for item in items:
                     item_id = self._storage.build_dedup_key(item)
-                    if not item_id or await self._storage.has_seen(item_id):
+                    if not item_id:
                         skipped_seen_count += 1
                         continue
+                    if item_id in seen_in_run:
+                        skipped_batch_duplicate_count += 1
+                        logger.warning(
+                            "skip job=%s duplicate item in current batch: id=%s title=%s",
+                            job.id,
+                            item_id,
+                            str(item.get("title", "")).strip(),
+                        )
+                        continue
+                    if await self._storage.has_seen(item_id):
+                        skipped_seen_count += 1
+                        continue
+                    seen_in_run.add(item_id)
                     if self._should_mark_history_only(item, feed_state_map, bootstrap_only=True):
                         await self._storage.mark_seen(
                             item_id,
@@ -315,12 +330,13 @@ async def _run_job_once_guarded(self, job: JobConfig) -> None:
                 error_summary=error_summary,
             )
             logger.info(
-                "job=%s finished: fetched=%s parsed=%s pushed=%s skipped_seen=%s skipped_history=%s skipped_invalid_target=%s dispatch_fail=%s duration_ms=%s error=%s",
+                "job=%s finished: fetched=%s parsed=%s pushed=%s skipped_seen=%s skipped_batch_duplicate=%s skipped_history=%s skipped_invalid_target=%s dispatch_fail=%s duration_ms=%s error=%s",
                 job.id,
                 fetched_count,
                 parsed_count,
                 pushed_count,
                 skipped_seen_count,
+                skipped_batch_duplicate_count,
                 skipped_history_count,
                 skipped_invalid_target_count,
                 dispatch_fail_count,

tests/test_scheduler.py

@@ -7,7 +7,11 @@
 
 astrbot_module = types.ModuleType("astrbot")
 astrbot_api_module = types.ModuleType("astrbot.api")
-astrbot_api_module.logger = types.SimpleNamespace(info=lambda *a, **k: None)
+astrbot_api_module.logger = types.SimpleNamespace(
+    info=lambda *a, **k: None,
+    warning=lambda *a, **k: None,
+    error=lambda *a, **k: None,
+)
 sys.modules.setdefault("astrbot", astrbot_module)
 sys.modules["astrbot.api"] = astrbot_api_module
 
@@ -146,6 +150,69 @@ async def dispatch(self, item):
         self.assertEqual(storage.marked, [("item-1", 123)])
 
 
+class SchedulerBatchDedupTests(unittest.IsolatedAsyncioTestCase):
+    async def test_duplicate_items_in_same_batch_are_dispatched_once(self):
+        class FakeStorage:
+            def __init__(self):
+                self.marked = []
+
+            def build_dedup_key(self, item):
+                return item["guid"]
+
+            async def has_seen(self, item_id):
+                return False
+
+            async def mark_seen(self, item_id, ttl_seconds=0):
+                self.marked.append((item_id, ttl_seconds))
+
+            async def get_feed_state(self, feed_id):
+                return {"last_success_time": 0}
+
+            async def update_feed_state(self, *args, **kwargs):
+                return {}
+
+        class FakeFetcher:
+            async def fetch(self, job):
+                return [{"feed_id": "feed-1"}]
+
+        class FakeParser:
+            def parse(self, raw_items, job):
+                return [
+                    {"feed_id": "feed-1", "guid": "dup-1", "title": "Same Item", "published_at": ""},
+                    {"feed_id": "feed-1", "guid": "dup-1", "title": "Same Item", "published_at": ""},
+                ]
+
+        class FakeDispatcher:
+            def __init__(self):
+                self.calls = 0
+
+            async def dispatch(self, item):
+                self.calls += 1
+                return DispatchResult(success_count=1)
+
+        config = types.SimpleNamespace(
+            jobs=[],
+            dedup_ttl_seconds=123,
+            poll_interval_seconds=300,
+        )
+        job = types.SimpleNamespace(id="job-1", feed_ids=["feed-1"], enabled=True, interval_seconds=300)
+        storage = FakeStorage()
+        dispatcher = FakeDispatcher()
+        scheduler = RSSScheduler(
+            config=config,
+            fetcher=FakeFetcher(),
+            parser=FakeParser(),
+            dispatcher=dispatcher,
+            storage=storage,
+            pipeline=None,
+        )
+
+        await scheduler._run_job_once_guarded(job)
+
+        self.assertEqual(dispatcher.calls, 1)
+        self.assertEqual(storage.marked, [("dup-1", 123)])
+
+
 class SchedulerTranslationTest(unittest.IsolatedAsyncioTestCase):
     async def test_test_translation_returns_pipeline_error_when_missing(self):
         config = types.SimpleNamespace(jobs=[])