# rss - A maubot plugin to subscribe to RSS/Atom feeds.
# Copyright (C) 2022 Tulir Asokan
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see .
from __future__ import annotations
from typing import Any
from datetime import datetime, timedelta, timezone
from string import Template
from time import mktime, monotonic, time
import asyncio
import hashlib
import html
import io
import logging
import re
import aiohttp
import attr
import feedparser
from maubot import MessageEvent, Plugin, __version__ as maubot_version
from maubot.handlers import command, event
from mautrix.types import (
EventID,
EventType,
MessageType,
PowerLevelStateEventContent,
RoomID,
StateEvent,
)
from mautrix.util.async_db import UpgradeTable
from mautrix.util.config import BaseProxyConfig, ConfigUpdateHelper
from .db import DBManager, Entry, Feed, Subscription, normalize_title
from .migrations import upgrade_table
rss_change_level = EventType.find("xyz.maubot.rss", t_class=EventType.Class.STATE)
# Gewogen OV-trefwoord-families voor het match-filter. Elke tuple = (regex, gewicht).
# Per familie wordt 1× geteld per tekst (niet per cluster/herhaling). Drempel (3):
# doorlaten als titelscore + samenvattingscore >= MATCH_FILTER_THRESHOLD
# (titel weegt NIET extra — besluit 2026-09-11: 'restaurant naast busstation'
# als enige medium-hit in de titel is geen OV-bericht).
# Gewicht 3 (sterk, onmiskenbaar OV): één woord in titel óf samenvatting volstaat.
# Gewicht 2 (medium): moet combineren met een andere OV-familie; één medium-hit,
# in titel óf samenvatting, haalt de drempel niet.
# Gewicht 1 (zwak/dubbelzinnig: los "bus", "breng", "overweg", "overwegen", "sein",
# "kaartje", "opwaarderen", "rails", "pendelbus"): moet met een ander OV-woord
# combineren (een tweede, andere familie — dezelfde zwakke familie in titel én
# samenvatting telt niet als combinatie) EN minstens één zwak woord moet in de
# TITEL staan: een combinatie van alleen zwakke woorden in de samenvatting is
# geen bewijs (het golfbericht met "station" + "pendelbus" + "bus" in de
# samenvatting is geen OV-nieuws). Weert zo "VW-bus"/"schoolbus", "weer op de rails", het
# werkwoord "brengen" ("hoe breng je ... in beeld"), het idioom "goed met elkaar
# overweg kunnen", het werkwoord "overwegen" ("gratieverlening overwegen"), het
# werkwoord "opwaarderen" ("een wijk opwaarderen") en brandweerjargon
# ("sein brandmeester"), zonder eenduidige bushalte/overweg/NS/trein-berichten
# te raken.
# Inflectie: families matchen met \w* aan het eind, zodat meervoud en vervoeging
# meegenomen worden ("spoorlijnen", "seinstoringen", "ontspoorde", "chipkaarten").
# Een koppelteken vóór een match is géén OV-signaal: "VW-bus" en "radio-station"
# vallen af (lookbehind bij 'bus' resp. 'station').
# MATCH_FILTER_BLACKLIST: gematchte woorden die tóch geen OV-waarde hebben (o.a.
# straatnamen rond stations — "de Stationsstraat afsluiten voor een festival"
# is geen OV-nieuws). Zwartgelijste woorden leveren geen score op, maar
# blokkeren de familie niet: een écht "station"-woord ernaast scoort gewoon.
# Bewust geaccepteerde botsingen: krant "De Metro"/café "De Tram" (medium).
# Losse algemene woorden zijn zwak (moeten combineren met een andere familie):
# "trein", "sprinter" (atleet), "vervoerder" (logistiek), "reiziger",
# "inchecken"/"uitchecken" (Schiphol/hotel), "station(s)" (omroepen), "bus",
# "sein" (brandweer), "overweg", "breng", "kaartje", "opwaarderen", "rails".
# =====================================================================
# FILTERCLASSIFICATIE-BELEID (bij elke aanpassing aan het filter volgen)
# =====================================================================
# Doel: precisie boven recall. Liever een OV-bericht te veel gefilterd dan
# een niet-OV-bericht te veel doorgelaten (gebruikersbesluit, 2026-09-10).
#
# Drie lagen, elk met een eigen regel:
# 1. WHITELIST -> ALTIJD door. Alleen termen die in élke context OV
# betekenen (merken: NS/ProRail/Arriva; onmiskenbaar jargon: spoorstoring,
# perron, bushalte, conducteur, ovpay; multi-woord termen).
# REGEL: twijfel je of de term ook buiten het OV kan voorkomen -> NIET hier.
# NIET: station-samenstellingen (busstation/metrostation) — een bericht over
# een pand 'naast het busstation' is geen OV-nieuws (besluit 2026-09-11);
# die wegen mee via de gewogen families (medium).
# 2. GEWOGEN FAMILIES -> score = titel + samenvatting >= drempel (3).
# - sterk (3): onmiskenbaar OV-jargon/merken/samenvoegingen (spoorlijn,
# bushalte, busverkeer, intercity, lijn 12, ...).
# - medium (2): onmiskenbaar OV maar iets algemener (tram, metro, halte,
# pont, ovpay, openbaar vervoer, trein-/bus-/metrostation).
# Moet COMBINEREN met een andere OV-familie: één medium-hit,
# in titel óf samenvatting, is onvoldoende — de titel weegt
# niet extra ('restaurant naast busstation' in de titel is
# geen OV-bericht, besluit 2026-09-11).
# - zwak (1): DUBBELZINNIGHEID. Woorden die (ook) buiten het OV
# voorkomen. REGEL: elk dubbelzinnig woord krijgt een EIGEN
# familie (niet samengevoegd), zodat twee verschillende
# zwakke woorden wél als combinatie tellen, maar één enkel
# woord nooit de doorslag kan geven. Bekende dubbelzinnige:
# trein, bus, station(s), sprinter, reiziger, vervoerder,
# inchecken/uitchecken, sein, overweg, breng, kaartje,
# opwaarderen, rails, dienstregeling, chipkaart, ontsporen,
# pendelbus/shuttlebus (evenement-/hotel-/parkeerpendel;
# synoniemen -> bewust één familie, anders tellen twee
# synoniemen als twee onafhankelijke signalen).
# EXTRA REGEL voor zwak-alleen: staat er uitsluitend zwak
# bewijs in de SAMENVATTING (geen enkele zwakke hit in de
# titel), dan niet doorlaten. Twee/meer zwakke woorden in
# de samenvatting zijn anders samen al 3 punten
# ("station" + "pendelbus" + "bus" in een golfverslag).
# 3. BLACKLIST -> gematchte woorden zonder OV-waarde (stationsstraat,
# stationair, stationcar, ...); blokkeren de familie niet.
#
# BEOORDELEN VAN EEN NIEUW WOORD:
# a) Kan het ook buiten het OV voorkomen? ja -> zwak (1), eigen familie.
# b) Is het een samenvoeging/merk die alleen OV kan betekenen? -> sterk (3)
# of whitelist.
# c) Bij twijfel: zwak (1). Conservatief = precision-first.
# d) Ter controle: leg nieuwe woordkandidaten voor aan minimaal één extra
# taalmodel (de audit op 2026-09-10 gebruikte 3 modellen; enige stem
# 'TWIJFEL'/'NIET-OV' => behandelen als dubbelzinnig).
# =====================================================================
MATCH_FILTER_FAMILIES: list[tuple[str, int]] = [
# sterk (3)
(r"\b(?:ns|arriva|qbuzz|connexxion|cxx|transdev|keolis|ret|gvb|ebs|htm|syntus|u-ov|allgo|flixbus|blauwnet|r-net|valleilijn|waterbus|eurostar|thalys|randstadrail|gvu|prorail|hermes|rrreis|qliner|ov)\b", 3), # vervoerders ('Breng' staat in de zwakke groep: botst met het werkwoord "brengen")
(r"\b(?:spoorweg|spoorba(?:an|nen)|spoorlijn|spoorverkeer|spoorwegovergang|spoorbrug|spoorstoring|perron|seinstoring|wisselstoring|spoorstaking)\w*\b", 3), # spoorinfra; let op de klinkerwisseling: 'ontsporing' (open lettergreep, één o) vs 'ontspoord'/'ontspoorde' (dicht, dubbel o) — vandaar beide stammen
(r"\b(?:treinstation|treinstations|metrostation|metrostations|busstation|busstations)\b", 2), # station-samenstellingen: onmiskenbaar OV, maar MEDIUM — één station-woord náást een ander onderwerp ('pand naast busstation', besluit 2026-09-11) is geen OV-bericht; in de samenvatting moet een tweede OV-woord bijkomen
(r"(? None:
helper.copy("update_interval")
helper.copy("max_backoff")
helper.copy("poll_parallelism_limit")
helper.copy("spam_sleep")
helper.copy("command_prefix")
helper.copy("notification_template")
helper.copy("exclude_paywalled")
helper.copy("title_dedup_days")
helper.copy("run_log_file")
helper.copy("whatsapp_api")
helper.copy("whatsapp_session")
helper.copy("whatsapp_chat_id")
helper.copy("admins")
class BoolArgument(command.Argument):
def __init__(self, name: str, label: str = None, *, required: bool = False) -> None:
super().__init__(name, label, required=required, pass_raw=False)
def match(self, val: str, **kwargs) -> tuple[str, Any]:
part = val.split(" ")[0].lower()
if part in ("f", "false", "n", "no", "0"):
res = False
elif part in ("t", "true", "y", "yes", "1"):
res = True
else:
raise ValueError("invalid boolean")
return val[len(part) :], res
class RSSBot(Plugin):
dbm: DBManager
exclude_paywalled: bool = False
poll_task: asyncio.Future
_poll_started: bool = False
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]]
@classmethod
def get_config_class(cls) -> type[BaseProxyConfig]:
return Config
@classmethod
def get_db_upgrade_table(cls) -> UpgradeTable:
return upgrade_table
async def start(self) -> None:
# Idempotentie-guard: maubot kan internal_start() (her)aanroepen via
# boot/client/upload/update-paden; Instance.start() zet zijn eigen
# started-flag pas ná plugin.start(). Zonder guard ontstaat bij een
# tweede (concurrente) aanroep een tweede poll-loop. Check+set is
# synchronous (geen await ertussen) → atoom in de event-loop.
if self._poll_started:
self.log.warning("start() called twice, ignoring duplicate poll task")
return
self._poll_started = True
try:
await super().start()
self.on_external_config_update()
self.dbm = DBManager(self.database)
self.http = self.client.api.session
self.power_level_cache = {}
self.poll_task = asyncio.create_task(self.poll_feeds())
except Exception:
# Niets geïnitialiseerd (geen dbm/http/poll_task): guard terugdraaien,
# anders wordt een geldige herstart genegeerd en blijft de instance dood.
self._poll_started = False
raise
def on_external_config_update(self) -> None:
self.config.load_and_update()
poll_parallelism_limit = self.config["poll_parallelism_limit"]
self.poll_sema = (
asyncio.Semaphore(poll_parallelism_limit) if poll_parallelism_limit > 0 else None
)
# 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:
# De logger zelf is gedeeld over reloads (zelfde log-naam). Bij een
# maubot-reload is self._run_logger op de nieuwe instance weer None,
# terwijl de oude handler aan de geregistreerde logger blijft hangen.
# Daarom ALTIJD op naam opruimen, niet op het instance-attribuut:
# anders wordt elke poll-run eenmalig per reload gelogd (N× duplicaat).
self._run_logger = logging.getLogger(f"{self.log.name}.runlog")
for handler in list(self._run_logger.handlers):
self._run_logger.removeHandler(handler)
handler.close()
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.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_started = False
self.poll_task.cancel()
if self._run_handler:
self._run_handler.close()
async def poll_feeds(self) -> None:
try:
await self._poll_feeds()
except asyncio.CancelledError:
self.log.debug("Polling stopped")
except Exception:
self.log.exception("Fatal error while polling feeds")
@staticmethod
def _filter_text(raw: str) -> str:
"""Platte tekst voor het match-filter: eerst entities decoderen, dán tags
strippen. Andersom (strip → unescape) laat dubbel-geëscapete feeds hun HTML
als tekst houden — inclusief URL's en attributen, die dan vals scoren."""
return re.sub(r"<[^>]+>", "", html.unescape(raw))
@staticmethod
def _whitelist_hit(text: str) -> bool:
"""True als een onmiskenbare OV-term uit MATCH_FILTER_WHITELIST in text zit.
Enkele-woord termen matchen op woordgrens ('ns' telt NIET binnen 'kunstenaar'),
multi-woord termen ('nederlandse spoorwegen') als exacte frase, beide
case-insensitive."""
if _WHITELIST_RE.search(text):
return True
low = text.lower()
return any(ph in low for ph in _WHITELIST_PHRASES)
@staticmethod
def _fam_hits(text: str) -> list[tuple[int, int]]:
"""(familie-index, gewicht) voor elke OV-familie die in text voorkomt (1× per familie).
Gematchte woorden uit MATCH_FILTER_BLACKLIST (o.a. 'stationsstraat')
tellen niet mee en blokkeren de familie niet: een écht OV-woord in
dezelfde tekst scoort gewoon verder."""
hits: list[tuple[int, int]] = []
for idx, (family, weight) in enumerate(_compiled_match_families):
for m in family.finditer(text):
if m.group(0).lower() in MATCH_FILTER_BLACKLIST:
continue
hits.append((idx, weight))
break
return hits
@staticmethod
def _filter_passes(title: str, summary: str) -> bool:
"""Gewogen familiescore: doorlaten bij titelscore + samenvattingscore >= drempel
(titel weegt NIET extra — één medium-hit in de titel, 'restaurant naast
busstation', is geen OV-bewijs; besluit 2026-09-11).
Sterke families (gewicht 3) halen de drempel al met één woord in titel óf
samenvatting. Medium (2) moet combineren met een andere OV-familie.
Zwakke woorden (losse 'bus', 'breng', 'overweg', 'overwegen', 'sein',
'opwaarderen', 'kaartje', 'rails', 'pendelbus') moeten combineren met een
ander OV-woord én minstens één zwak woord moet in de titel staan; dezelfde
zwakke familie in titel én samenvatting is géén combinatie. Weert oa
'VW-bus' (bus matcht niet ná een koppelteken), 'schoolbus', 'bussen rijden
om' in een woningbrand-lead, het werkwoord 'brengen', het idioom 'goed met
elkaar overweg kunnen', 'een wijk opwaarderen', brandweerjargon
'sein brandmeester', figuratief 'weer op de rails' en een golfverslag
waarin de samenvatting 'station' + 'pendelbus' + 'bus' noemt.
Bewust doorgelaten: krant 'De Metro'/café 'De Tram' (medium) en atleet-
'sprinter' in combinatie met ander OV-woord."""
# Whitelist: één onmiskenbare OV-term (titel óf samenvatting) laat altijd door.
if RSSBot._whitelist_hit(title) or RSSBot._whitelist_hit(summary):
return True
t_hits = RSSBot._fam_hits(title)
s_hits = RSSBot._fam_hits(summary)
score = sum(w for _, w in t_hits) + sum(w for _, w in s_hits)
if score < MATCH_FILTER_THRESHOLD:
return False
if max(w for _, w in t_hits + s_hits) == 1:
# Alleen zwakke families: dan moeten ≥2 verschillende families meetellen
# én minstens één zwak woord moet in de TITEL staan. Zwakke woorden in
# de samenvatting alleen zijn geen bewijs ('station' + 'pendelbus' +
# 'bus' in een golfverslag telt anders op tot de drempel).
if not t_hits or len({i for i, _ in t_hits + s_hits}) < 2:
return False
return True
@staticmethod
def _filter_summary() -> str:
"""Mensvriendelijke weergave van het vaste gewogen match-filter."""
return (
"**OV-matchfilter (vast in code, gewogen families):**\n"
"- Sterk (3): vervoerders (NS, Arriva, ProRail, Qbuzz, ...), spoorinfra (spoorlijn, spoorwegovergang, perron, spoorstoring), "
"samenstellingen als bushalte/buslijn/bushokje/busbaan/busverkeer, treinstoring/treinverkeer/treinstaking, "
"intercity/streekvervoer/veerdienst/buurtbus/belbus, 'lijn 12'\n"
"- Medium (2): tram, metro, openbaar vervoer, lightrail, halte, pont, ov-chipkaart/ovpay, ov-staking, conducteur, "
"trein-/bus-/metrostation — één medium-hit volstaat niet, moet combineren met een andere OV-familie\n"
"- Zwak (1): losse 'bus' (niet ná een koppelteken), 'pendelbus'/'shuttlebus', 'trein', 'station(s)', 'breng', 'overweg', 'overwegen', "
"'sein', 'kaartje', 'opwaarderen', 'rails', 'inchecken'/'uitchecken', 'dienstregeling', 'chipkaart', 'reisproduct', 'vervoersbewijs', "
"'ontsporen', 'reiziger', 'vervoerder', 'sprinter' — moet combineren met een ander OV-woord, met minstens één zwak woord in de titel\n\n"
"**Whitelist (altijd door):** één onmiskenbare OV-term in titel óf samenvatting is genoeg "
"(NS, ProRail, Nederlandse Spoorwegen, spoorstoring, perron, bushalte, conducteur, "
"ov-chipkaart, ovpay, openbaar vervoer, ...)\n\n"
"Uitgesloten (blacklist): woorden die wél matchen maar geen OV-waarde hebben, "
"o.a. straatnamen rond stations ('stationsstraat', 'stationsweg', ...) en niet-OV-vormen als 'stationcar' en 'stationair'\n\n"
"Doorlaten: `titel-score + samenvatting-score ≥ 3` (titel weegt niet extra); "
"bij uitsluitend zwakke woorden tellen alleen "
"≥2 verschillende families én een zwak woord in de titel"
)
async def _send(
self, feed: Feed, entry: Entry, sub: Subscription, *, check_filter: bool = True
) -> EventID | None:
if check_filter:
summary = self._filter_text(entry.summary)
if not self._filter_passes(entry.title, summary):
return None
message = sub.notification_template.safe_substitute(
{
"feed_url": feed.url,
"feed_title": feed.title,
"feed_subtitle": feed.subtitle,
"feed_link": feed.link,
**attr.asdict(entry),
}
)
msgtype = MessageType.NOTICE if sub.send_notice else MessageType.TEXT
try:
return await self.client.send_markdown(
sub.room_id, message, msgtype=msgtype, allow_html=True
)
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 _send_whatsapp(self, feed: Feed, entry: Entry) -> bool:
"""Post entry.link naar de WhatsApp-groep via wwebjs-api (met dedup-check). True = nieuw geplaatst."""
api = self.config.get("whatsapp_api", "")
chat_id = self.config.get("whatsapp_chat_id", "")
if not api or not chat_id:
self.log.debug("WhatsApp forwarding disabled")
return False
link = entry.link
if not link:
self.log.warning(f"No link for {entry.id}, skipping WhatsApp")
return False
if await self.dbm.is_whatsapp_posted(link):
self.log.info(f"{entry.id} already posted to WhatsApp before, skipping")
return False
session = self.config.get("whatsapp_session", "EdGpt")
base = api.rstrip("/")
# Dedup: sla over als de URL al in de doelgroep staat. Eerst de lokale
# tabel (eigen posts; daarna de globale search (vreemde posts), gefilterd
# op chat-id van het gevonden bericht.
try:
resp = await self.http.post(
f"{base}/client/searchMessages/{session}",
json={"query": link, "options": {"limit": 1}},
timeout=aiohttp.ClientTimeout(total=20),
)
data = await resp.json()
if not data.get("success"):
# Checkfout (wwebjs antwoordt ook met HTTP 200 + success:false) =
# niet posten: anders omzeilt de fout de dedup → dubbele link.
self.log.warning(f"WhatsApp dedup check failed for {entry.id}: {data}")
return False
for msg in data.get("messages", []):
msg_data = msg.get("_data", {})
if (
msg_data.get("id", {}).get("remote") == chat_id
and link in msg_data.get("body", "")
):
self.log.info(f"{entry.id} already in WhatsApp group, skipping")
return False
except Exception as e:
self.log.warning(f"WhatsApp dedup check failed for {entry.id}: {e}")
return False
try:
resp = await self.http.post(
f"{base}/client/sendMessage/{session}",
json={"chatId": chat_id, "content": link, "contentType": "string"},
timeout=aiohttp.ClientTimeout(total=20),
)
data = await resp.json()
ok = bool(data.get("success"))
if ok:
try:
await self.dbm.mark_whatsapp_posted(link)
except Exception as e:
self.log.warning(f"Failed to record WhatsApp post {entry.id}: {e}")
self.log.info(f"WhatsApp post {entry.id} to {chat_id}: {'ok' if ok else 'failed: ' + str(data)}")
return ok
except Exception as e:
self.log.warning(f"Failed to send {entry.id} to WhatsApp {chat_id}: {e}")
return False
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 the entry was posted to at least one channel (Matrix room or newly posted WhatsApp link)
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}")
summary = self._filter_text(entry.summary)
if not self._filter_passes(entry.title, summary):
return 0, 1, 0
spam_sleep = self.config["spam_sleep"]
if spam_sleep >= 0:
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:
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)
whatsapp_ok = await self._send_whatsapp(feed, entry)
return (1 if (sent or whatsapp_ok) 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.0)
return
now = int(time())
start = monotonic()
title_dedup_days = int(self.config.get("title_dedup_days", 30))
posted_titles: set[str] = set()
if title_dedup_days > 0:
cutoff = datetime.now(timezone.utc) - timedelta(days=title_dedup_days)
posted_titles = await self.dbm.get_recent_posted_titles(cutoff)
tasks = [self.try_parse_feed(feed=feed) for feed in subs if feed.next_retry < now]
feed: Feed
entries: list[Entry] | None
self.log.info(f"Polling {len(tasks)} feeds")
evaluated_total = 0
placed_total = 0
filtered_total = 0
deduped_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}"
)
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
next_retry_delay = min(next_retry_delay, self.config["max_backoff"] * 60)
next_retry = int(time() + next_retry_delay)
self.log.debug(f"Setting backoff of {feed.id} to {error_count} / {next_retry}")
await self.dbm.set_backoff(feed, error_count, next_retry)
continue
elif 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)
try:
new_entries = {entry.id: entry for entry in entries}
except Exception:
self.log.exception(f"Weird error in items of {feed.url}")
continue
await self.dbm.filter_entries(feed.id, new_entries)
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:
norm = normalize_title(entry.title)
if title_dedup_days > 0 and norm and norm in posted_titles:
deduped_total += 1
self.log.info(
f"{entry.id} of feed {feed.id} title already posted ({norm[:60]}), skipping"
)
continue
placed, filtered, failed = await self._broadcast(
feed, entry, feed.subscriptions
)
placed_total += placed
filtered_total += filtered
failed_total += failed
if placed and norm:
# In-memory set: vangt ook duplicaten binnen dezelfde poll-run
# (twee sites van dezelfde uitgever in één run).
posted_titles.add(norm)
try:
await self.dbm.mark_entry_posted(feed.id, entry.id)
except Exception:
self.log.warning(
f"Failed to mark {entry.id} of feed {feed.id} as posted"
)
duration = monotonic() - start
self._log_run_summary(
len(tasks),
evaluated_total,
placed_total,
filtered_total,
deduped_total,
failed_total,
unchanged_total,
duration,
)
def _log_run_summary(
self,
feeds: int,
evaluated: int,
selected: int,
filtered: int,
deduped: int,
failed: int,
unchanged: int,
duration: float,
) -> None:
line = (
f"run complete feeds={feeds} evaluated={evaluated} selected={selected} "
f"filtered={filtered} deduped={deduped} 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")
while True:
try:
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]:
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]:
try:
self.log.trace(
f"Trying to fetch {feed.id} / {feed.url} "
f"(backoff: {feed.error_count} / {feed.next_retry})"
)
return await self.parse_feed(feed=feed)
except Exception as e:
self.log.warning(f"Failed to parse feed {feed.id} / {feed.url}: {e}")
return feed, []
@property
def _feed_get_headers(self) -> dict[str, str]:
return {
"User-Agent": f"maubot/{maubot_version} +https://github.com/maubot/rss",
}
async def parse_feed(
self, *, feed: Feed | None = None, url: str | None = None
) -> 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=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", "")
ct = resp.headers["Content-Type"].split(";")[0].strip()
if ct == "application/json" or ct == "application/feed+json":
feed, entries = await self._parse_json(feed, resp)
else:
feed, entries = await self._parse_rss(feed, resp)
# Validators pas bewaren als de body volledig gelezen én geparseerd is:
# anders stuurt de volgende poll If-None-Match op een respons die nooit
# verwerkt is (304 → 'unchanged', entries stil weg).
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}")
return feed, entries
@classmethod
async def _parse_json(
cls, feed: Feed, resp: aiohttp.ClientResponse
) -> tuple[Feed, list[Entry]]:
content = await resp.json()
if content["version"] not in (
"https://jsonfeed.org/version/1",
"https://jsonfeed.org/version/1.1",
):
raise ValueError("Unsupported JSON feed version")
if not isinstance(content["items"], list):
raise ValueError("Feed is not a valid JSON feed (items is not a list)")
feed.title = content["title"]
feed.subtitle = content.get("subtitle", "")
feed.link = content.get("home_page_url", "")
entries = [cls._parse_json_entry(feed.id, entry) for entry in content["items"]]
if cls.exclude_paywalled:
entries = [e for e in entries if not e.paywalled]
return feed, entries
@classmethod
def _parse_json_entry(cls, feed_id: int, entry: dict[str, Any]) -> Entry:
try:
date = datetime.fromisoformat(entry["date_published"])
except (ValueError, KeyError):
date = datetime.now(timezone.utc)
title = entry.get("title", "")
summary = (
entry.get("summary") or entry.get("content_html") or entry.get("content_text") or ""
).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),
)
@classmethod
async def _parse_rss(
cls, feed: Feed, resp: aiohttp.ClientResponse
) -> tuple[Feed, list[Entry]]:
content = await resp.read()
headers = {"Content-Location": feed.url, **resp.headers, "Content-Encoding": "identity"}
parsed_data = feedparser.parse(io.BytesIO(content), response_headers=headers)
if parsed_data.bozo:
if not isinstance(parsed_data.bozo_exception, feedparser.ThingsNobodyCaresAboutButMe):
raise parsed_data.bozo_exception
feed_data = parsed_data.get("feed", {})
feed.title = feed_data.get("title", feed.url)
feed.subtitle = feed_data.get("description", "")
feed.link = feed_data.get("link", "")
entries = [cls._parse_rss_entry(feed.id, entry) for entry in parsed_data.entries]
if cls.exclude_paywalled:
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(
feed_id=feed_id,
id=(
getattr(entry, "id", None)
or getattr(entry, "link", None)
or hashlib.sha1(
" ".join(
[
getattr(entry, "title", ""),
getattr(entry, "description", ""),
]
).encode("utf-8")
).hexdigest()
),
date=cls._parse_rss_date(entry),
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", ""))
),
)
@staticmethod
def _parse_rss_date(entry: Any) -> datetime:
try:
return datetime.fromtimestamp(mktime(entry["published_parsed"]))
except (KeyError, TypeError, ValueError):
pass
try:
return datetime.fromtimestamp(mktime(entry["date_parsed"]))
except (KeyError, TypeError, ValueError):
pass
return datetime.now(timezone.utc)
async def get_power_levels(
self, room_id: RoomID
) -> tuple[PowerLevelStateEventContent, StateEvent]:
try:
expiry, pls, create = self.power_level_cache[room_id]
if expiry < int(time()):
return pls, create
except KeyError:
pass
pls = await self.client.get_state_event(room_id, EventType.ROOM_POWER_LEVELS)
create = await self.client.get_state_event(room_id, EventType.ROOM_CREATE, format="event")
self.power_level_cache[room_id] = (int(time()) + 5 * 60, pls, create)
return pls, create
async def can_manage(self, evt: MessageEvent) -> bool:
if evt.sender in self.config["admins"]:
return True
levels, create = await self.get_power_levels(evt.room_id)
user_level = levels.get_user_level(evt.sender, create=create)
state_level = levels.get_event_level(rss_change_level)
if not isinstance(state_level, int):
state_level = 50
if user_level < state_level:
await evt.reply(
"You don't have the permission to manage the subscriptions of this room."
)
return False
return True
@command.new(
name=lambda self: self.config["command_prefix"],
help="Manage this RSS bot",
require_subcommand=True,
)
async def rss(self) -> None:
pass
@rss.subcommand("subscribe", aliases=("s", "sub"), help="Subscribe this room to a feed.")
@command.argument("url", "feed URL", pass_raw=True)
async def subscribe(self, evt: MessageEvent, url: str) -> None:
if not await self.can_manage(evt):
return
feed = await self.dbm.get_feed_by_url(url)
if not feed:
try:
info, entries = await self.parse_feed(url=url)
except Exception as e:
await evt.reply(f"Failed to load feed: {e}")
return
feed = await self.dbm.create_feed(info)
await self.dbm.add_entries(entries, override_feed_id=feed.id)
elif feed.error_count > 0:
await self.dbm.set_backoff(feed, error_count=feed.error_count, next_retry=0)
feed_info = f"feed ID {feed.id}: [{feed.title}]({feed.url})"
sub, _ = await self.dbm.get_subscription(feed.id, evt.room_id)
if sub is not None:
subscriber = (
"You"
if sub.user_id == evt.sender
else f"[{sub.user_id}](https://matrix.to/#/{sub.user_id})"
)
await evt.reply(f"{subscriber} had already subscribed this room to {feed_info}")
else:
await self.dbm.subscribe(
feed.id, evt.room_id, evt.sender, self.config["notification_template"]
)
await evt.reply(f"Subscribed to {feed_info}")
@rss.subcommand(
"set-filter", aliases=("f",), help="Show the fixed global OV match filter."
)
async def set_filter(self, evt: MessageEvent) -> None:
await evt.reply(self._filter_summary())
@rss.subcommand(
"filter", aliases=("show-filter", "get-filter"), help="Show the current global match filter."
)
async def command_show_filter(self, evt: MessageEvent) -> None:
await evt.reply(self._filter_summary())
@rss.subcommand(
"unsubscribe", aliases=("u", "unsub"), help="Unsubscribe this room from a feed."
)
@command.argument("feed_id", "feed ID", parser=int)
async def unsubscribe(self, evt: MessageEvent, feed_id: int) -> None:
if not await self.can_manage(evt):
return
sub, feed = await self.dbm.get_subscription(feed_id, evt.room_id)
if not sub:
await evt.reply("This room is not subscribed to that feed")
return
await self.dbm.unsubscribe(feed.id, evt.room_id)
await evt.reply(f"Unsubscribed from feed ID {feed.id}: [{feed.title}]({feed.url})")
@rss.subcommand(
"template",
aliases=("t", "tpl"),
help="Change the notification template for a subscription in this room",
)
@command.argument("feed_id", "feed ID", parser=int)
@command.argument("template", "new template", pass_raw=True, required=False)
async def command_template(self, evt: MessageEvent, feed_id: int, template: str) -> None:
if not await self.can_manage(evt):
return
sub, feed = await self.dbm.get_subscription(feed_id, evt.room_id)
if not sub:
await evt.reply("This room is not subscribed to that feed")
return
if not template:
await evt.reply(
'Current template in this room:
'
f"{html.escape(sub.notification_template.template)}"
"
",
allow_html=True,
markdown=False,
)
return
await self.dbm.update_template(feed.id, evt.room_id, template)
sub = Subscription(
feed_id=feed.id,
room_id=sub.room_id,
user_id=sub.user_id,
notification_template=Template(template),
send_notice=sub.send_notice,
)
sample_entry = Entry(
feed_id=feed.id,
id="SAMPLE",
date=datetime.now(timezone.utc),
title="Sample entry",
summary="This is a sample entry to demonstrate your new template",
link="http://example.com",
)
await evt.reply(f"Template for feed ID {feed.id} updated. Sample notification:")
await self._send(feed, sample_entry, sub)
@rss.subcommand(
"notice", aliases=("n",), help="Set whether or not the bot should send updates as m.notice"
)
@command.argument("feed_id", "feed ID", parser=int)
@BoolArgument("setting", "true/false")
async def command_notice(self, evt: MessageEvent, feed_id: int, setting: bool) -> None:
if not await self.can_manage(evt):
return
sub, feed = await self.dbm.get_subscription(feed_id, evt.room_id)
if not sub:
await evt.reply("This room is not subscribed to that feed")
return
await self.dbm.set_send_notice(feed.id, evt.room_id, setting)
send_type = "m.notice" if setting else "m.text"
await evt.reply(f"Updates for feed ID {feed.id} will now be sent as `{send_type}`")
@staticmethod
def _format_subscription(feed: Feed, subscriber: str) -> str:
msg = (
f"* {feed.id} - [{feed.title}]({feed.url}) "
f"(subscribed by [{subscriber}](https://matrix.to/#/{subscriber}))"
)
if feed.error_count > 1:
msg += f" \n ⚠️ The last {feed.error_count} attempts to fetch the feed have failed!"
return msg
@rss.subcommand(
"subscriptions",
aliases=("ls", "list", "subs"),
help="List the subscriptions in the current room.",
)
async def command_subscriptions(self, evt: MessageEvent) -> None:
subscriptions = await self.dbm.get_feeds_by_room(evt.room_id)
if len(subscriptions) == 0:
await evt.reply("There are no RSS subscriptions in this room")
return
filter_line = self._filter_summary().replace("\n", " \n")
await evt.reply(
"**Subscriptions in this room:**\n\n"
+ "\n".join(
self._format_subscription(feed, subscriber)
for feed, subscriber in subscriptions
)
+ f"\n\n{filter_line}"
)
@event.on(EventType.ROOM_TOMBSTONE)
async def tombstone(self, evt: StateEvent) -> None:
if not evt.content.replacement_room:
return
await self.dbm.update_room_id(evt.room_id, evt.content.replacement_room)