diff --git a/base-config.yaml b/base-config.yaml
index a0f595f..2111378 100644
--- a/base-config.yaml
+++ b/base-config.yaml
@@ -12,10 +12,9 @@ spam_sleep: 2
command_prefix: "rss"
# Default post notification template for new subscriptions
notification_template: "New post in $feed_title: [$title]($link)"
-# 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
+# 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
# Exclude paywalled articles (Telegraaf/NHD markeren dit met true)
exclude_paywalled: true
# Global match filter (whitelist): applies to ALL feeds in this instance.
diff --git a/buzzrss/bot.py b/buzzrss/bot.py
index 1a4d48e..5c6f894 100644
--- a/buzzrss/bot.py
+++ b/buzzrss/bot.py
@@ -15,7 +15,7 @@
# along with this program. If not, see .
from __future__ import annotations
-from typing import Any
+from typing import Any, Iterable
from datetime import datetime, timezone
from string import Template
from time import mktime, monotonic, time
@@ -23,7 +23,6 @@ import asyncio
import hashlib
import html
import io
-import logging
import re
import aiohttp
@@ -72,7 +71,6 @@ class Config(BaseProxyConfig):
helper.copy("allow_filter")
helper.copy("exclude_paywalled")
helper.copy("match_filter")
- helper.copy("run_log_file")
helper.copy("admins")
@@ -96,8 +94,6 @@ 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]]
@@ -125,30 +121,10 @@ 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:
@@ -158,48 +134,30 @@ class RSSBot(Plugin):
except Exception:
self.log.exception("Fatal error while polling feeds")
- @staticmethod
- 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
-
@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)
+ """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)
- @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
+ 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
message = sub.notification_template.safe_substitute(
{
@@ -217,65 +175,36 @@ 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]
- ) -> 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
- """
+ ) -> None:
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:
- results: list[EventID | None] = []
- for sub in subscriptions:
- results.append(await self._send(feed, entry, sub, check_filter=False))
+ for task in tasks:
+ await task
await asyncio.sleep(spam_sleep)
else:
- 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)
+ await asyncio.gather(*tasks)
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: list[Entry] | None
+ entries: Iterable[Entry]
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: {entries is not None}"
+ f"success: {bool(entries)}"
)
- 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
@@ -296,46 +225,13 @@ 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:
- placed, filtered, failed = await self._broadcast(
- feed, entry, feed.subscriptions
- )
- placed_total += placed
- filtered_total += filtered
- failed_total += failed
+ await self._broadcast(feed, entry, feed.subscriptions)
duration = monotonic() - start
- 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)
+ self.log.info(f"Finished polling {len(tasks)} feeds in {duration:.2f} seconds")
async def _poll_feeds(self) -> None:
self.log.debug("Polling started")
@@ -344,19 +240,18 @@ 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] | None]:
+ async def try_parse_feed(self, feed: Feed | None = None) -> tuple[Feed, list[Entry]]:
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] | None]:
+ async def _try_parse_feed(self, feed: Feed | None = None) -> tuple[Feed, list[Entry]]:
try:
self.log.trace(
f"Trying to fetch {feed.id} / {feed.url} "
@@ -375,35 +270,18 @@ class RSSBot(Plugin):
async def parse_feed(
self, *, feed: Feed | None = None, url: str | None = None
- ) -> tuple[Feed, list[Entry] | None]:
+ ) -> tuple[Feed, list[Entry]]:
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=headers,
+ headers=self._feed_get_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)
@@ -439,10 +317,7 @@ 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,
- paywalled=cls._is_paywalled_link(link),
- )
+ return Entry(feed_id=feed_id, id=id, date=date, title=title, summary=summary, link=link)
@classmethod
async def _parse_rss(
@@ -463,12 +338,6 @@ 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(
@@ -489,14 +358,7 @@ class RSSBot(Plugin):
title=getattr(entry, "title", ""),
summary=getattr(entry, "description", "").strip(),
link=getattr(entry, "link", ""),
- paywalled=(
- # Telegraaf/NHD convention: true
- str(getattr(entry, "premium", "")).lower() == "true"
- # DPG Media (AD/Volkskrant/Parool/Trouw): true
- 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", ""))
- ),
+ paywalled=str(getattr(entry, "premium", "")).lower() == "true",
)
@staticmethod
diff --git a/buzzrss/db.py b/buzzrss/db.py
index b043edb..65e33b3 100644
--- a/buzzrss/db.py
+++ b/buzzrss/db.py
@@ -64,8 +64,6 @@ class Feed:
next_retry: int = 0
error_count: int = 0
- etag: str = ""
- last_modified: str = ""
subscriptions: list[Subscription] = attr.ib(factory=lambda: [])
@@ -121,7 +119,7 @@ class DBManager:
async def get_feeds(self) -> list[Feed]:
q = """
- SELECT id, url, title, subtitle, link, next_retry, error_count, etag, last_modified,
+ SELECT id, url, title, subtitle, link, next_retry, error_count,
room_id, user_id, notification_template, send_notice
FROM feed INNER JOIN subscription ON feed.id = subscription.feed_id
"""
@@ -137,19 +135,25 @@ 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, etag, last_modified,
- user_id FROM feed
+ SELECT id, url, title, subtitle, link, next_retry, error_count, 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:
- if not entries:
+ entry_keys = list(entries.keys())
+ if not entry_keys:
return
- 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 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]
async def add_entries(self, entries: list[Entry], override_feed_id: int | None = None) -> None:
if not entries:
@@ -170,14 +174,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, etag, last_modified FROM feed WHERE url=$1"
+ q = "SELECT id, url, title, subtitle, link, next_retry, error_count 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, etag, last_modified,
+ SELECT id, url, title, subtitle, link, next_retry, error_count,
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
@@ -202,10 +206,6 @@ 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,
diff --git a/buzzrss/migrations.py b/buzzrss/migrations.py
index 9432cf9..6d15ec8 100644
--- a/buzzrss/migrations.py
+++ b/buzzrss/migrations.py
@@ -99,9 +99,3 @@ 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 ''")
diff --git a/maubot.yaml b/maubot.yaml
index a30c5b7..adb1f76 100644
--- a/maubot.yaml
+++ b/maubot.yaml
@@ -1,5 +1,5 @@
id: nl.eddydevink.rss
-version: 1.5.3
+version: 1.3.1
modules:
- buzzrss
main_class: RSSBot