v1.5.3: filter vereist titel-match of totaalscore >= 3

Regel C: doorlaten bij (titel-match >= 1) of (totaalscore >= 3).
Een treffer in de titel is een sterke OV-intentie (score 2 volstaat);
treffers alleen in de samenvatting moeten dubbel bevestigd worden.
Weert bv. woningbrand-berichten die via 'bussen rijden om'/'busverkeer'
in de lead binnenkwamen, zonder de 112/trein/staking-berichten te raken.
Gevalideerd op 30d kamerhistorie: 96 doorlaat -> 7 geblokkeerd (allemaal ruis).

base-config.yaml

@@ -12,9 +12,10 @@ spam_sleep: 2
 command_prefix: "rss"
 # Default post notification template for new subscriptions
 notification_template: "New post in $feed_title: [$title]($link)"
-# Should non-admin users be allowed to set regex filters to drop posts from feeds?
-# Regexes are dangerous and may lead to DoS attacks.
-allow_filter: false
+# Per-run summary logfile: one line per poll run with feeds/evaluated/selected/filtered/failed counts.
+# Absoluut pad in de container aanraden: /data is de persistente maubot-mount (host: /matrix/maubot/data).
+# Leeg = uitgeschakeld.
+run_log_file: /data/buzzrss-runs.log
 # Exclude paywalled articles (Telegraaf/NHD markeren dit met <premium>true</premium>)
 exclude_paywalled: true
 # Global match filter (whitelist): applies to ALL feeds in this instance.

buzzrss/bot.py

@@ -15,7 +15,7 @@
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 from __future__ import annotations
 
-from typing import Any, Iterable
+from typing import Any
 from datetime import datetime, timezone
 from string import Template
 from time import mktime, monotonic, time
@@ -23,6 +23,7 @@ import asyncio
 import hashlib
 import html
 import io
+import logging
 import re
 
 import aiohttp
@@ -71,6 +72,7 @@ class Config(BaseProxyConfig):
         helper.copy("allow_filter")
         helper.copy("exclude_paywalled")
         helper.copy("match_filter")
+        helper.copy("run_log_file")
         helper.copy("admins")
 
 
@@ -94,6 +96,8 @@ class RSSBot(Plugin):
     exclude_paywalled: bool = False
     poll_task: asyncio.Future
     poll_sema: asyncio.Semaphore | None
+    _run_logger: logging.Logger | None = None
+    _run_handler: logging.FileHandler | None = None
     http: aiohttp.ClientSession
     power_level_cache: dict[RoomID, tuple[int, PowerLevelStateEventContent, StateEvent]]
 
@@ -121,10 +125,30 @@ class RSSBot(Plugin):
         )
         # single-instance fork: class-level since _parse_rss is a classmethod
         type(self).exclude_paywalled = bool(self.config["exclude_paywalled"])
+        self._setup_run_log()
+
+    def _setup_run_log(self) -> None:
+        if self._run_logger:
+            for handler in list(self._run_logger.handlers):
+                self._run_logger.removeHandler(handler)
+                handler.close()
+        self._run_logger = None
+        self._run_handler = None
+        run_log_file = self.config["run_log_file"] if "run_log_file" in self.config else ""
+        if not run_log_file:
+            return
+        self._run_handler = logging.FileHandler(run_log_file, encoding="utf-8")
+        self._run_handler.setFormatter(logging.Formatter("%(asctime)s %(message)s"))
+        self._run_logger = logging.getLogger(f"{self.log.name}.runlog")
+        self._run_logger.setLevel(logging.INFO)
+        self._run_logger.propagate = False
+        self._run_logger.addHandler(self._run_handler)
 
     async def stop(self) -> None:
         await super().stop()
         self.poll_task.cancel()
+        if self._run_handler:
+            self._run_handler.close()
 
     async def poll_feeds(self) -> None:
         try:
@@ -135,29 +159,47 @@ class RSSBot(Plugin):
             self.log.exception("Fatal error while polling feeds")
 
     @staticmethod
-    def _filter_score(title: str, summary: str, match_filter: str) -> int:
-        """Score = 2× hits in titel + 1× hits in summary. Aaneengesloten matches
-        (bv 'NS-trein', 'openbaar vervoer') tellen als 1 cluster."""
-        def count(text: str) -> int:
-            spans = sorted((m.start(), m.end()) for m in re.finditer(match_filter, text))
-            if not spans:
-                return 0
-            clusters, prev_end = 1, spans[0][1]
-            for s, e in spans[1:]:
-                if s - prev_end <= 1:
-                    prev_end = max(prev_end, e)
-                else:
-                    clusters += 1
-                    prev_end = e
-            return clusters
-        return 2 * count(title) + count(summary)
+    def _matches_in(text: str, match_filter: str) -> int:
+        """Aantal aaneengesloten match-clusters in text (0 als geen match).
+        Aaneengesloten matches (bv 'NS-trein', 'openbaar vervoer') tellen als 1."""
+        spans = sorted((m.start(), m.end()) for m in re.finditer(match_filter, text))
+        if not spans:
+            return 0
+        clusters, prev_end = 1, spans[0][1]
+        for s, e in spans[1:]:
+            if s - prev_end <= 1:
+                prev_end = max(prev_end, e)
+            else:
+                clusters += 1
+                prev_end = e
+        return clusters
 
-    async def _send(self, feed: Feed, entry: Entry, sub: Subscription) -> EventID | None:
-        match_filter = self.config["match_filter"]
-        if match_filter:
-            summary = html.unescape(re.sub(r"<[^>]+>", "", entry.summary))
-            if self._filter_score(entry.title, summary, match_filter) < 2:
-                return None
+    @staticmethod
+    def _filter_score(title: str, summary: str, match_filter: str) -> int:
+        """Score = 2× hits in titel + 1× hits in summary."""
+        return 2 * RSSBot._matches_in(title, match_filter) + RSSBot._matches_in(summary, match_filter)
+
+    @staticmethod
+    def _filter_passes(title: str, summary: str, match_filter: str) -> bool:
+        """Regel C: doorlaten bij (titel-match >= 1) of (totaalscore >= 3).
+
+        Een OV-trefwoord in de titel is een sterke intentieverklaring (score 2
+        volstaat). Treffers alleen in de samenvatting moeten dubbel bevestigd
+        worden (>= 3): zo houdt 'bussen rijden om' in de lead van een
+        woningbrand-bericht het artikel niet meer binnen."""
+        if RSSBot._matches_in(title, match_filter) >= 1:
+            return True
+        return RSSBot._filter_score(title, summary, match_filter) >= 3
+
+    async def _send(
+        self, feed: Feed, entry: Entry, sub: Subscription, *, check_filter: bool = True
+    ) -> EventID | None:
+        if check_filter:
+            match_filter = self.config["match_filter"]
+            if match_filter:
+                summary = html.unescape(re.sub(r"<[^>]+>", "", entry.summary))
+                if not self._filter_passes(entry.title, summary, match_filter):
+                    return None
 
         message = sub.notification_template.safe_substitute(
             {
@@ -175,36 +217,65 @@ class RSSBot(Plugin):
             )
         except Exception as e:
             self.log.warning(f"Failed to send {entry.id} of {feed.id} to {sub.room_id}: {e}")
+            return None
 
     async def _broadcast(
         self, feed: Feed, entry: Entry, subscriptions: list[Subscription]
-    ) -> None:
+    ) -> tuple[int, int, int]:
+        """Send entry to all subs. Returns (placed, filtered, failed).
+
+        placed   = 1 when at least one room received the message
+        filtered = 1 when the global match filter rejected the entry
+        failed   = number of subscriptions where sending failed
+        """
         self.log.debug(f"Broadcasting {entry.id} of {feed.id}")
+        match_filter = self.config["match_filter"]
+        if match_filter:
+            summary = html.unescape(re.sub(r"<[^>]+>", "", entry.summary))
+            if not self._filter_passes(entry.title, summary, match_filter):
+                return 0, 1, 0
         spam_sleep = self.config["spam_sleep"]
-        tasks = [self._send(feed, entry, sub) for sub in subscriptions]
         if spam_sleep >= 0:
-            for task in tasks:
-                await task
+            results: list[EventID | None] = []
+            for sub in subscriptions:
+                results.append(await self._send(feed, entry, sub, check_filter=False))
                 await asyncio.sleep(spam_sleep)
         else:
-            await asyncio.gather(*tasks)
+            results = await asyncio.gather(
+                *(self._send(feed, entry, sub, check_filter=False) for sub in subscriptions)
+            )
+        sent = sum(1 for result in results if result is not None)
+        return (1 if sent else 0, 0, len(results) - sent)
 
     async def _poll_once(self) -> None:
         subs = await self.dbm.get_feeds()
         if not subs:
+            self._log_run_summary(0, 0, 0, 0, 0, 0, 0.0)
             return
         now = int(time())
         start = monotonic()
         tasks = [self.try_parse_feed(feed=feed) for feed in subs if feed.next_retry < now]
         feed: Feed
-        entries: Iterable[Entry]
+        entries: list[Entry] | None
         self.log.info(f"Polling {len(tasks)} feeds")
+        evaluated_total = 0
+        placed_total = 0
+        filtered_total = 0
+        failed_total = 0
+        unchanged_total = 0
         for res in asyncio.as_completed(tasks):
             feed, entries = await res
             self.log.trace(
                 f"Fetching {feed.id} (backoff: {feed.error_count} / {feed.next_retry}) "
-                f"success: {bool(entries)}"
+                f"success: {entries is not None}"
             )
+            if entries is None:
+                # Feed returned 304: unchanged, no new entries, no error.
+                unchanged_total += 1
+                if feed.error_count > 0:
+                    self.log.debug(f"Resetting backoff of {feed.id}")
+                    await self.dbm.set_backoff(feed, error_count=0, next_retry=0)
+                continue
             if not entries:
                 error_count = feed.error_count + 1
                 next_retry_delay = self.config["update_interval"] * 60 * error_count
@@ -225,13 +296,46 @@ class RSSBot(Plugin):
             self.log.trace(
                 f"Feed {feed.id} had {len(new_entries)} new entries (of {len(entries)} total)"
             )
+            evaluated_total += len(entries)
             new_entry_list: list[Entry] = list(new_entries.values())
             new_entry_list.sort(key=lambda entry: (entry.date, entry.id))
             await self.dbm.add_entries(new_entry_list)
             for entry in new_entry_list:
-                await self._broadcast(feed, entry, feed.subscriptions)
+                placed, filtered, failed = await self._broadcast(
+                    feed, entry, feed.subscriptions
+                )
+                placed_total += placed
+                filtered_total += filtered
+                failed_total += failed
         duration = monotonic() - start
-        self.log.info(f"Finished polling {len(tasks)} feeds in {duration:.2f} seconds")
+        self._log_run_summary(
+            len(tasks),
+            evaluated_total,
+            placed_total,
+            filtered_total,
+            failed_total,
+            unchanged_total,
+            duration,
+        )
+
+    def _log_run_summary(
+        self,
+        feeds: int,
+        evaluated: int,
+        selected: int,
+        filtered: int,
+        failed: int,
+        unchanged: int,
+        duration: float,
+    ) -> None:
+        line = (
+            f"run complete feeds={feeds} evaluated={evaluated} selected={selected} "
+            f"filtered={filtered} failed={failed} unchanged={unchanged} "
+            f"duration={duration:.2f}s"
+        )
+        self.log.info(f"Finished polling {feeds} feeds in {duration:.2f} seconds: {line}")
+        if self._run_logger:
+            self._run_logger.info(line)
 
     async def _poll_feeds(self) -> None:
         self.log.debug("Polling started")
@@ -240,18 +344,19 @@ class RSSBot(Plugin):
                 await self._poll_once()
             except asyncio.CancelledError:
                 self.log.debug("Polling stopped")
+                raise
             except Exception:
                 self.log.exception("Error while polling feeds")
             await asyncio.sleep(self.config["update_interval"] * 60)
 
-    async def try_parse_feed(self, feed: Feed | None = None) -> tuple[Feed, list[Entry]]:
+    async def try_parse_feed(self, feed: Feed | None = None) -> tuple[Feed, list[Entry] | None]:
         if self.poll_sema is not None:
             async with self.poll_sema:
                 return await self._try_parse_feed(feed)
         else:
             return await self._try_parse_feed(feed)
 
-    async def _try_parse_feed(self, feed: Feed | None = None) -> tuple[Feed, list[Entry]]:
+    async def _try_parse_feed(self, feed: Feed | None = None) -> tuple[Feed, list[Entry] | None]:
         try:
             self.log.trace(
                 f"Trying to fetch {feed.id} / {feed.url} "
@@ -270,18 +375,35 @@ class RSSBot(Plugin):
 
     async def parse_feed(
         self, *, feed: Feed | None = None, url: str | None = None
-    ) -> tuple[Feed, list[Entry]]:
+    ) -> tuple[Feed, list[Entry] | None]:
         if feed is None:
             if url is None:
                 raise ValueError("Either feed or url must be set")
             feed = Feed(id=-1, url=url, title="", subtitle="", link="")
         elif url is not None:
             raise ValueError("Only one of feed or url must be set")
+        headers = self._feed_get_headers
+        if feed.id > 0:
+            if feed.etag:
+                headers["If-None-Match"] = feed.etag
+            if feed.last_modified:
+                headers["If-Modified-Since"] = feed.last_modified
         resp = await self.http.get(
             feed.url,
-            headers=self._feed_get_headers,
+            headers=headers,
             timeout=aiohttp.ClientTimeout(total=20),
         )
+        if resp.status == 304:
+            # Feed unchanged since last fetch: no new entries, no error.
+            return feed, None
+        etag = resp.headers.get("ETag", "")
+        last_modified = resp.headers.get("Last-Modified", "")
+        if feed.id > 0 and (etag or last_modified):
+            feed.etag, feed.last_modified = etag, last_modified
+            try:
+                await self.dbm.set_feed_headers(feed, etag, last_modified)
+            except Exception:
+                self.log.debug(f"Failed to store fetch headers of {feed.id}")
         ct = resp.headers["Content-Type"].split(";")[0].strip()
         if ct == "application/json" or ct == "application/feed+json":
             return await self._parse_json(feed, resp)
@@ -317,7 +439,10 @@ class RSSBot(Plugin):
         ).strip()
         id = str(entry["id"])
         link = entry.get("url") or id
-        return Entry(feed_id=feed_id, id=id, date=date, title=title, summary=summary, link=link)
+        return Entry(
+            feed_id=feed_id, id=id, date=date, title=title, summary=summary, link=link,
+            paywalled=cls._is_paywalled_link(link),
+        )
 
     @classmethod
     async def _parse_rss(
@@ -338,6 +463,12 @@ class RSSBot(Plugin):
             entries = [e for e in entries if not e.paywalled]
         return feed, entries
 
+    @staticmethod
+    def _is_paywalled_link(link: str) -> bool:
+        # DPG Media premium articles live under /ps/ (Parool/VK/Trouw feeds
+        # carry no paywall marker; the /ps/ path is the only reliable signal).
+        return "/ps/" in (link or "")
+
     @classmethod
     def _parse_rss_entry(cls, feed_id: int, entry: Any) -> Entry:
         return Entry(
@@ -358,7 +489,14 @@ class RSSBot(Plugin):
             title=getattr(entry, "title", ""),
             summary=getattr(entry, "description", "").strip(),
             link=getattr(entry, "link", ""),
-            paywalled=str(getattr(entry, "premium", "")).lower() == "true",
+            paywalled=(
+                # Telegraaf/NHD convention: <premium>true</premium>
+                str(getattr(entry, "premium", "")).lower() == "true"
+                # DPG Media (AD/Volkskrant/Parool/Trouw): <dpp:paid>true</dpp:paid>
+                or str(getattr(entry, "dpp_paid", "")).lower() == "true"
+                # DPG Media /ps/ premium path (Parool/VK/Trouw carry no marker)
+                or cls._is_paywalled_link(getattr(entry, "link", ""))
+            ),
         )
 
     @staticmethod

buzzrss/db.py

@@ -64,6 +64,8 @@ class Feed:
 
     next_retry: int = 0
     error_count: int = 0
+    etag: str = ""
+    last_modified: str = ""
 
     subscriptions: list[Subscription] = attr.ib(factory=lambda: [])
 
@@ -119,7 +121,7 @@ class DBManager:
 
     async def get_feeds(self) -> list[Feed]:
         q = """
-        SELECT id, url, title, subtitle, link, next_retry, error_count,
+        SELECT id, url, title, subtitle, link, next_retry, error_count, etag, last_modified,
                room_id, user_id, notification_template, send_notice
         FROM feed INNER JOIN subscription ON feed.id = subscription.feed_id
         """
@@ -135,25 +137,19 @@ class DBManager:
 
     async def get_feeds_by_room(self, room_id: RoomID) -> list[tuple[Feed, UserID]]:
         q = """
-        SELECT id, url, title, subtitle, link, next_retry, error_count, user_id FROM feed
+        SELECT id, url, title, subtitle, link, next_retry, error_count, etag, last_modified,
+               user_id FROM feed
         INNER JOIN subscription ON feed.id = subscription.feed_id AND subscription.room_id = $1
         """
         rows = await self.db.fetch(q, room_id)
         return [(Feed.from_row(row), row["user_id"]) for row in rows]
 
     async def filter_entries(self, feed_id: int, entries: dict[str, Entry]) -> None:
-        entry_keys = list(entries.keys())
-        if not entry_keys:
+        if not entries:
             return
-        async with self.db.acquire() as conn:
-            for key in entry_keys:
-                exists = await conn.fetchval(
-                    "SELECT EXISTS(SELECT 1 FROM entry WHERE feed_id = $1 AND id = $2)",
-                    feed_id,
-                    key,
-                )
-                if exists:
-                    del entries[key]
+        existing = await self.db.fetch("SELECT id FROM entry WHERE feed_id = $1", feed_id)
+        for row in existing:
+            entries.pop(row["id"], None)
 
     async def add_entries(self, entries: list[Entry], override_feed_id: int | None = None) -> None:
         if not entries:
@@ -174,14 +170,14 @@ class DBManager:
                 await conn.executemany(q, records)
 
     async def get_feed_by_url(self, url: str) -> Feed | None:
-        q = "SELECT id, url, title, subtitle, link, next_retry, error_count FROM feed WHERE url=$1"
+        q = "SELECT id, url, title, subtitle, link, next_retry, error_count, etag, last_modified FROM feed WHERE url=$1"
         return Feed.from_row(await self.db.fetchrow(q, url))
 
     async def get_subscription(
         self, feed_id: int, room_id: RoomID
     ) -> tuple[Subscription | None, Feed | None]:
         q = """
-        SELECT id, url, title, subtitle, link, next_retry, error_count,
+        SELECT id, url, title, subtitle, link, next_retry, error_count, etag, last_modified,
                room_id, user_id, notification_template, send_notice
         FROM feed LEFT JOIN subscription ON feed.id = subscription.feed_id AND room_id = $2
         WHERE feed.id = $1
@@ -206,6 +202,10 @@ class DBManager:
         q = "UPDATE feed SET error_count = $2, next_retry = $3 WHERE id = $1"
         await self.db.execute(q, info.id, error_count, next_retry)
 
+    async def set_feed_headers(self, info: Feed, etag: str, last_modified: str) -> None:
+        q = "UPDATE feed SET etag = $2, last_modified = $3 WHERE id = $1"
+        await self.db.execute(q, info.id, etag, last_modified)
+
     async def subscribe(
         self,
         feed_id: int,

buzzrss/migrations.py

@@ -99,3 +99,9 @@ async def upgrade_v7(conn: Connection) -> None:
 @upgrade_table.register(description="Add paywalled flag to entries")
 async def upgrade_v8(conn: Connection) -> None:
     await conn.execute("ALTER TABLE entry ADD COLUMN paywalled BOOLEAN NOT NULL DEFAULT false")
+
+
+@upgrade_table.register(description="Add conditional-fetch headers to feeds")
+async def upgrade_v9(conn: Connection) -> None:
+    await conn.execute("ALTER TABLE feed ADD COLUMN etag TEXT DEFAULT ''")
+    await conn.execute("ALTER TABLE feed ADD COLUMN last_modified TEXT DEFAULT ''")

maubot.yaml

@@ -1,5 +1,5 @@
 id: nl.eddydevink.rss
-version: 1.3.1
+version: 1.5.3
 modules:
 - buzzrss
 main_class: RSSBot