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
This commit is contained in:
eddy
commit
0628776469
6 changed files with 923 additions and 0 deletions
1
buzzrss/__init__.py
Normal file
1
buzzrss/__init__.py
Normal file
|
|
@ -0,0 +1 @@
|
|||
from .bot import RSSBot
|
||||
548
buzzrss/bot.py
Normal file
548
buzzrss/bot.py
Normal file
|
|
@ -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 <https://www.gnu.org/licenses/>.
|
||||
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(
|
||||
'<p>Current template in this room:</p><pre><code language="markdown">'
|
||||
f"{html.escape(sub.notification_template.template)}"
|
||||
"</code></pre>",
|
||||
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)
|
||||
243
buzzrss/db.py
Normal file
243
buzzrss/db.py
Normal file
|
|
@ -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 <https://www.gnu.org/licenses/>.
|
||||
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)
|
||||
91
buzzrss/migrations.py
Normal file
91
buzzrss/migrations.py
Normal file
|
|
@ -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 <https://www.gnu.org/licenses/>.
|
||||
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")
|
||||
Loading…
Reference in a new issue