# 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 2×(titelscore) + 1×(samenvattingscore) >= MATCH_FILTER_THRESHOLD
# Gewicht 3 (sterk, onmiskenbaar OV): één woord in titel óf samenvatting volstaat.
# Gewicht 2 (medium): één woord in de titel volstaat; in de samenvatting moet er
# een tweede OV-woord bijkomen.
# Gewicht 1 (zwak/dubbelzinnig: los "bus", "kaartje", "rails"): moet met een ander
# OV-woord combineren. Weert zo "VW-bus"/"schoolbus", "weer op de rails" en
# 112-berichten met "bussen rijden om" in de lead, zonder eenduidige
# bushalte/overweg/NS/trein-berichten te raken.
MATCH_FILTER_FAMILIES: list[tuple[str, int]] = [
# sterk (3)
(r"\b(?:ns|arriva|qbuzz|connexxion|cxx|transdev|keolis|ret|gvb|ebs|htm|syntus|breng|u-ov|allgo|flixbus|blauwnet|r-net|valleilijn|waterbus|eurostar|thalys|randstadrail|gvu|ov)\b", 3), # vervoerders
(r"\b(?:spoorweg|spoorbaan|spoorlijn|spoorverkeer|spoorwegovergang|spoorstoring|perron|perrons|overweg|overwegen|seinstoring|wisselstoring|sein|ontsporing|ontspoord|spoorstaking)\b", 3), # spoorinfra
(r"\bstation\w*\b", 3), # station(s)
(r"\b(?:bus|tram|metro)(?:halte|haltes|lijn|lijnen|station|stations|verkeer|net|staking)\b", 3), # voertuig-samenstelling
(r"\btrein\w*\b", 3), # trein(verkeer/staking/ongeluk/...)
# medium (2)
(r"\btram\b", 2),
(r"\bmetro\b", 2),
(r"\b(?:openbaar vervoer|dienstregeling|inchecken|uitchecken|lightrail|pendelbus|shuttlebus)\b", 2),
(r"\b(?:ov-chipkaart|chipkaart|ovpay|ov-pay|reisproduct|studentenreisproduct|vervoersbewijs|vervoerbewijs|opwaarderen)\b", 2),
(r"\b(?:ov-staking|conducteursstaking)\b", 2),
(r"\b(?:pont|ponten)\b", 2), # waterbus/veerdienst
# zwak (1) — dubbelzinnig, moet combineren
(r"\bbus(?:sen)?\b", 1),
(r"\bkaart(?:je|jes)\b", 1),
(r"\brails\b", 1),
]
# Drempel voor doorlaten (score = 2×titel + 1×samenvatting).
MATCH_FILTER_THRESHOLD = 3
_compiled_match_families: list[tuple[re.Pattern, int]] = [
(re.compile(pat, re.IGNORECASE), w) for pat, w in MATCH_FILTER_FAMILIES
]
class Config(BaseProxyConfig):
def do_update(self, helper: ConfigUpdateHelper) -> 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
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())
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 _fam_score(text: str) -> int:
"""Som van gewichten over de OV-families die in text voorkomen (1× per familie)."""
return sum(weight for family, weight in _compiled_match_families if family.search(text))
@staticmethod
def _filter_passes(title: str, summary: str) -> bool:
"""Gewogen familiescore: doorlaten bij 2×(titelscore) + 1×(samenvattingscore) >= drempel.
Sterke families (gewicht 3) halen de drempel al met één woord in titel óf
samenvatting. Medium (2) in de titel ook; in de samenvatting moet een tweede
OV-woord bijkomen. Zwakke woorden (losse 'bus', 'kaartje', 'rails') moeten
combineren met een ander OV-woord. Weert oa 'VW-bus', 'schoolbus',
'bussen rijden om' in een woningbrand-lead en figuratief 'weer op de rails'.
Bewust doorgelaten: krant 'De Metro'/café 'De Tram' in de titel (medium)."""
return 2 * RSSBot._fam_score(title) + RSSBot._fam_score(summary) >= MATCH_FILTER_THRESHOLD
@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, Qbuzz, ...), spoorinfra (perron, overweg, sein), "
"station(s), bushalte/buslijn/treinstel, trein\n"
"- Medium (2): tram, metro, openbaar vervoer/dienstregeling/inchecken, "
"ov-chipkaart/ovpay/reisproduct, ov-staking, pont\n"
"- Zwak (1): losse 'bus', 'kaartje', 'rails' — moet combineren met een ander OV-woord\n\n"
"Doorlaten: `2×(titel-score) + 1×(samenvatting-score) ≥ 3`"
)
async def _send(
self, feed: Feed, entry: Entry, sub: Subscription, *, check_filter: bool = True
) -> EventID | None:
if check_filter:
summary = html.unescape(re.sub(r"<[^>]+>", "", 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 data.get("success"):
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 = html.unescape(re.sub(r"<[^>]+>", "", 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", "")
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)
else:
return await self._parse_rss(feed, resp)
@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", "")
return feed, [cls._parse_json_entry(feed.id, entry) for entry in content["items"]]
@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)