From 0628776469c09b4ba3928a8b3e7bc95f52f724fc Mon Sep 17 00:00:00 2001 From: eddy Date: Sat, 5 Sep 2026 17:13:39 +0200 Subject: [PATCH] Fork xyz.maubot.rss v0.4.1: whitelist match-filter op titel+summary - plugin id nl.eddydevink.rss, module buzzrss (uniek, voorkomt botsing met xyz.maubot.rss die dezelfde 'rss'-module importeert) - match_filter (v6-migratie, hernoemd van title_exclude_filter): post wordt alleen verstuurd als de regex matcht op titel OF (HTML-ontdaan) summary - config-default whitelist voor ov-only: gvb/ebs/connexxion/transdev/keolis/ ret/gvu/bus/metro/tram/trein(verwanten)/ov/ov-staking/staking - set-filter: lege filter valt terug op config-default --- base-config.yaml | 24 ++ buzzrss/__init__.py | 1 + buzzrss/bot.py | 548 ++++++++++++++++++++++++++++++++++++++++++ buzzrss/db.py | 243 +++++++++++++++++++ buzzrss/migrations.py | 91 +++++++ maubot.yaml | 16 ++ 6 files changed, 923 insertions(+) create mode 100644 base-config.yaml create mode 100644 buzzrss/__init__.py create mode 100644 buzzrss/bot.py create mode 100644 buzzrss/db.py create mode 100644 buzzrss/migrations.py create mode 100644 maubot.yaml diff --git a/base-config.yaml b/base-config.yaml new file mode 100644 index 0000000..e7ada45 --- /dev/null +++ b/base-config.yaml @@ -0,0 +1,24 @@ +# Feed update interval in minutes +update_interval: 60 +# Maximum backoff in minutes when failing to fetch feeds (defaults to 5 days) +max_backoff: 7200 +# How many feeds to poll in parallel? Set to 0 to disable limit. +poll_parallelism_limit: 10 +# The time to sleep between send requests when broadcasting a new feed entry. +# Set to 0 to disable sleep or -1 to run all requests asynchronously at once. +spam_sleep: 2 +# The prefix for all commands +# It has to be prefixed with ! in matrix to be recognised +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-subscription match filter (whitelist): set with `!rss set-filter `. +# A post is ONLY sent when the regex matches the title OR the (HTML-stripped) summary. +# Empty = fall back to the default below. +match_filter: "(?i)\\b(?:gvb|ebs|connexxion|cxx|transdev|keolis|ret|gvu|bus|metro|tram|ov)\\b|trein(?:en|verkeer|stel(?:len)?|staking|ongeluk)?|\\bov-staking|staking" +# Users who can bypass room permission checks +admins: +- "@user:example.com" diff --git a/buzzrss/__init__.py b/buzzrss/__init__.py new file mode 100644 index 0000000..95623a1 --- /dev/null +++ b/buzzrss/__init__.py @@ -0,0 +1 @@ +from .bot import RSSBot diff --git a/buzzrss/bot.py b/buzzrss/bot.py new file mode 100644 index 0000000..cefba65 --- /dev/null +++ b/buzzrss/bot.py @@ -0,0 +1,548 @@ +# 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, Iterable +from datetime import datetime, timezone +from string import Template +from time import mktime, monotonic, time +import asyncio +import hashlib +import html +import io +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 +from .migrations import upgrade_table + +rss_change_level = EventType.find("xyz.maubot.rss", t_class=EventType.Class.STATE) + + +DEFAULT_MATCH_FILTER = ( + r"(?i)\b(?:gvb|ebs|connexxion|cxx|transdev|keolis|ret|gvu|bus|metro|tram|ov)\b" + r"|trein(?:en|verkeer|stel(?:len)?|staking|ongeluk)?" + r"|\bov-staking|staking" +) + + +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("allow_filter") + helper.copy("match_filter") + 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 + poll_task: asyncio.Future + poll_sema: asyncio.Semaphore | 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: + 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 + ) + + async def stop(self) -> None: + await super().stop() + self.poll_task.cancel() + + 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") + + async def _send(self, feed: Feed, entry: Entry, sub: Subscription) -> EventID | None: + match_filter = sub.match_filter or self.config["match_filter"] + if match_filter: + haystack = f"{entry.title}\n{html.unescape(re.sub(r'<[^>]+>', '', entry.summary))}" + if not re.search(match_filter, haystack): + 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}") + + async def _broadcast( + self, feed: Feed, entry: Entry, subscriptions: list[Subscription] + ) -> None: + self.log.debug(f"Broadcasting {entry.id} of {feed.id}") + 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 + await asyncio.sleep(spam_sleep) + else: + await asyncio.gather(*tasks) + + async def _poll_once(self) -> None: + subs = await self.dbm.get_feeds() + if not subs: + 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] + self.log.info(f"Polling {len(tasks)} feeds") + 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)}" + ) + 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)" + ) + 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) + duration = monotonic() - start + self.log.info(f"Finished polling {len(tasks)} feeds in {duration:.2f} seconds") + + 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") + 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]]: + 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]]: + 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]]: + 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") + resp = await self.http.get(feed.url, headers=self._feed_get_headers) + 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) + + @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", "") + return feed, [cls._parse_rss_entry(feed.id, entry) for entry in parsed_data.entries] + + @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", ""), + ) + + @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", "filter"), help="Set match filter (whitelist regex) for a specific feed." + ) + @command.argument("feed_id", "feed ID", parser=int) + @command.argument("match_filter", "match filter", pass_raw=True) + async def set_filter(self, evt: MessageEvent, feed_id: int, match_filter: str) -> None: + if evt.sender in self.config["admins"]: + pass + elif not self.config["allow_filter"]: + await evt.reply("This bot does not allow non-admin users to set title filters") + return + elif 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 + try: + re.compile(match_filter) + except re.error: + await evt.reply(f"Filter is not a valid regular expression") + return + await self.dbm.update_match_filter(feed.id, evt.room_id, match_filter) + if match_filter: + await evt.reply( + f"Feed {feed_id} will now only send posts matching: `{match_filter}`" + ) + else: + await evt.reply( + f"Removed match filter from feed {feed_id} " + f"(falling back to default: `{self.config['match_filter']}`)" + ) + + @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, match_filter: str) -> str: + msg = ( + f"* {feed.id} - [{feed.title}]({feed.url}) " + f"(subscribed by [{subscriber}](https://matrix.to/#/{subscriber}))" + ) + if match_filter: + msg += f" (only posts matching: {match_filter})" + 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 + await evt.reply( + "**Subscriptions in this room:**\n\n" + + "\n".join( + self._format_subscription(feed, subscriber, match_filter) + for feed, subscriber, match_filter in subscriptions + ) + ) + + @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) diff --git a/buzzrss/db.py b/buzzrss/db.py new file mode 100644 index 0000000..ba71a0b --- /dev/null +++ b/buzzrss/db.py @@ -0,0 +1,243 @@ +# 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 datetime import datetime, timezone +from string import Template + +from asyncpg import Record +from attr import dataclass +import attr + +from mautrix.types import RoomID, UserID +from mautrix.util.async_db import Database, Scheme + + +@dataclass +class Subscription: + feed_id: int + room_id: RoomID + user_id: UserID + notification_template: Template + send_notice: bool + match_filter: str + + @classmethod + def from_row(cls, row: Record | None) -> Subscription | None: + if not row: + return None + feed_id = row["id"] + room_id = row["room_id"] + user_id = row["user_id"] + if not room_id or not user_id: + return None + send_notice = bool(row["send_notice"]) + tpl = Template(row["notification_template"]) + match_filter = row["match_filter"] + return cls( + feed_id=feed_id, + room_id=room_id, + user_id=user_id, + notification_template=tpl, + send_notice=send_notice, + match_filter=match_filter, + ) + + +@dataclass +class Feed: + id: int + url: str + title: str + subtitle: str + link: str + + next_retry: int = 0 + error_count: int = 0 + + subscriptions: list[Subscription] = attr.ib(factory=lambda: []) + + @classmethod + def from_row(cls, row: Record | None) -> Feed | None: + if not row: + return None + data = {**row} + data.pop("room_id", None) + data.pop("user_id", None) + data.pop("send_notice", None) + data.pop("notification_template", None) + data.pop("match_filter", None) + return cls(**data, subscriptions=[]) + + +date_fmt = "%Y-%m-%d %H:%M:%S" +date_fmt_microseconds = "%Y-%m-%d %H:%M:%S.%f" + + +@dataclass +class Entry: + feed_id: int + id: str + date: datetime + title: str + summary: str + link: str + + def __attrs_post_init__(self): + if self.date.tzinfo is None: + self.date = self.date.replace(tzinfo=timezone.utc) + + @classmethod + def from_row(cls, row: Record | None) -> Entry | None: + if not row: + return None + data = {**row} + date = data.pop("date") + if not isinstance(date, datetime): + try: + date = datetime.strptime(date, date_fmt_microseconds if "." in date else date_fmt) + except ValueError: + date = datetime.now(timezone.utc) + return cls(**data, date=date) + + +class DBManager: + db: Database + + def __init__(self, db: Database) -> None: + self.db = db + + async def get_feeds(self) -> list[Feed]: + q = """ + SELECT id, url, title, subtitle, link, next_retry, error_count, + room_id, user_id, notification_template, send_notice, match_filter + FROM feed INNER JOIN subscription ON feed.id = subscription.feed_id + """ + rows = await self.db.fetch(q) + feeds: dict[int, Feed] = {} + for row in rows: + try: + feed = feeds[row["id"]] + except KeyError: + feed = feeds[row["id"]] = Feed.from_row(row) + feed.subscriptions.append(Subscription.from_row(row)) + return list(feeds.values()) + + async def get_feeds_by_room(self, room_id: RoomID) -> list[tuple[Feed, UserID, str]]: + q = """ + SELECT id, url, title, subtitle, link, next_retry, error_count, user_id, match_filter 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"], row["match_filter"]) 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: + 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] + + async def add_entries(self, entries: list[Entry], override_feed_id: int | None = None) -> None: + if not entries: + return + if override_feed_id: + for entry in entries: + entry.feed_id = override_feed_id + records = [attr.astuple(entry) for entry in entries] + columns = ("feed_id", "id", "date", "title", "summary", "link") + async with self.db.acquire() as conn: + if self.db.scheme == Scheme.POSTGRES: + await conn.copy_records_to_table("entry", records=records, columns=columns) + else: + q = ( + "INSERT INTO entry (feed_id, id, date, title, summary, link) " + "VALUES ($1, $2, $3, $4, $5, $6)" + ) + 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" + 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, + room_id, user_id, notification_template, send_notice, match_filter + FROM feed LEFT JOIN subscription ON feed.id = subscription.feed_id AND room_id = $2 + WHERE feed.id = $1 + """ + row = await self.db.fetchrow(q, feed_id, room_id) + return Subscription.from_row(row), Feed.from_row(row) + + async def update_room_id(self, old: RoomID, new: RoomID) -> None: + await self.db.execute("UPDATE subscription SET room_id = $1 WHERE room_id = $2", new, old) + + async def create_feed(self, info: Feed) -> Feed: + q = ( + "INSERT INTO feed (url, title, subtitle, link, next_retry) " + "VALUES ($1, $2, $3, $4, $5) RETURNING (id)" + ) + info.id = await self.db.fetchval( + q, info.url, info.title, info.subtitle, info.link, info.next_retry + ) + return info + + async def set_backoff(self, info: Feed, error_count: int, next_retry: int) -> None: + 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 subscribe( + self, + feed_id: int, + room_id: RoomID, + user_id: UserID, + template: str | None = None, + send_notice: bool = True, + ) -> None: + q = """ + INSERT INTO subscription (feed_id, room_id, user_id, notification_template, send_notice) + VALUES ($1, $2, $3, $4, $5) + """ + template = template or "New post in $feed_title: [$title]($link)" + await self.db.execute(q, feed_id, room_id, user_id, template, send_notice) + + async def unsubscribe(self, feed_id: int, room_id: RoomID) -> None: + q = "DELETE FROM subscription WHERE feed_id = $1 AND room_id = $2" + await self.db.execute(q, feed_id, room_id) + + async def update_template(self, feed_id: int, room_id: RoomID, template: str) -> None: + q = "UPDATE subscription SET notification_template=$3 WHERE feed_id=$1 AND room_id=$2" + await self.db.execute(q, feed_id, room_id, template) + + async def set_send_notice(self, feed_id: int, room_id: RoomID, send_notice: bool) -> None: + q = "UPDATE subscription SET send_notice=$3 WHERE feed_id=$1 AND room_id=$2" + await self.db.execute(q, feed_id, room_id, send_notice) + + async def update_match_filter( + self, feed_id: int, room_id: RoomID, match_filter: str + ) -> None: + q = "UPDATE subscription SET match_filter=$3 WHERE feed_id=$1 AND room_id=$2" + await self.db.execute(q, feed_id, room_id, match_filter) diff --git a/buzzrss/migrations.py b/buzzrss/migrations.py new file mode 100644 index 0000000..7691ea7 --- /dev/null +++ b/buzzrss/migrations.py @@ -0,0 +1,91 @@ +# 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 mautrix.util.async_db import Connection, Scheme, UpgradeTable + +upgrade_table = UpgradeTable() + + +@upgrade_table.register(description="Latest revision", upgrades_to=5) +async def upgrade_latest(conn: Connection, scheme: Scheme) -> None: + gen = "GENERATED ALWAYS AS IDENTITY" if scheme != Scheme.SQLITE else "" + await conn.execute(f""" + CREATE TABLE IF NOT EXISTS feed ( + id INTEGER {gen}, + url TEXT NOT NULL, + title TEXT NOT NULL, + subtitle TEXT NOT NULL, + link TEXT NOT NULL, + + next_retry BIGINT DEFAULT 0, + error_count BIGINT DEFAULT 0, + + PRIMARY KEY (id), + UNIQUE (url) + ) + """) + await conn.execute(""" + CREATE TABLE IF NOT EXISTS subscription ( + feed_id INTEGER, + room_id TEXT, + user_id TEXT NOT NULL, + + notification_template TEXT, + send_notice BOOLEAN DEFAULT true, + title_exclude_filter TEXT DEFAULT '', + + PRIMARY KEY (feed_id, room_id), + FOREIGN KEY (feed_id) REFERENCES feed (id) + ) + """) + await conn.execute(""" + CREATE TABLE IF NOT EXISTS entry ( + feed_id INTEGER, + id TEXT, + date timestamp with time zone NOT NULL, + title TEXT NOT NULL, + summary TEXT NOT NULL, + link TEXT NOT NULL, + PRIMARY KEY (feed_id, id), + FOREIGN KEY (feed_id) REFERENCES feed (id) + ) + """) + + +@upgrade_table.register(description="Add send_notice field to subscriptions") +async def upgrade_v2(conn: Connection) -> None: + await conn.execute("ALTER TABLE subscription ADD COLUMN send_notice BOOLEAN DEFAULT true") + + +@upgrade_table.register(description="Add error counts to feeds") +async def upgrade_v3(conn: Connection) -> None: + await conn.execute("ALTER TABLE feed ADD COLUMN next_retry BIGINT DEFAULT 0") + await conn.execute("ALTER TABLE feed ADD COLUMN error_count BIGINT DEFAULT 0") + + +@upgrade_table.register(description="Change entry date to timezone-aware") +async def upgrade_v4(conn: Connection, scheme: Scheme) -> None: + if scheme == Scheme.POSTGRES: + await conn.execute("ALTER TABLE entry ALTER COLUMN date TYPE timestamp with time zone") + + +@upgrade_table.register(description="Add title exclude filter to subscriptions") +async def upgrade_v5(conn: Connection) -> None: + await conn.execute("ALTER TABLE subscription ADD COLUMN title_exclude_filter TEXT DEFAULT ''") + + +@upgrade_table.register(description="Rename title exclude filter to match filter (whitelist mode)") +async def upgrade_v6(conn: Connection) -> None: + await conn.execute("ALTER TABLE subscription RENAME COLUMN title_exclude_filter TO match_filter") diff --git a/maubot.yaml b/maubot.yaml new file mode 100644 index 0000000..d6e9fa6 --- /dev/null +++ b/maubot.yaml @@ -0,0 +1,16 @@ +id: nl.eddydevink.rss +version: 1.0.1 +modules: +- buzzrss +main_class: RSSBot +maubot: 0.3.0 +database: true +database_type: asyncpg +config: false +webapp: false +license: AGPL-3.0-or-later +extra_files: +- base-config.yaml +dependencies: +- feedparser>=5.1 +soft_dependencies: []