aa4f7144be
Deze feeds dragen geen paywall-marker (<premium>/<dpp:paid>); premium-artikelen leven onder /ps/. Gedeelde helper _is_paywalled_link voor RSS- en JSON-parsing.
730 lines
30 KiB
Python
730 lines
30 KiB
Python
# 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
|
||
from datetime import datetime, 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
|
||
from .migrations import upgrade_table
|
||
|
||
rss_change_level = EventType.find("xyz.maubot.rss", t_class=EventType.Class.STATE)
|
||
|
||
|
||
DEFAULT_MATCH_FILTER = (
|
||
r"(?i)"
|
||
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"
|
||
r"|\b(?:bus|tram|metro)(?:halte|haltes|lijn|lijnen|station|stations|verkeer|net|staking|sen)?\b"
|
||
r"|\btrein(?:en|verkeer|stel(?:len)?|staking|ongeluk|reizigers?|station|stations)?\b"
|
||
r"|\b(?:spoorweg|spoorbaan|spoorlijn|spoorverkeer|spoorwegovergang|spoorstoring|perron|perrons|overweg|overwegen|seinstoring|wisselstoring|sein|ontsporing|ontspoord|spoorstaking)\b"
|
||
r"|\b(?:openbaar vervoer|dienstregeling|reizigers?|inchecken|uitchecken|pont|ponten|lightrail|pendelbus|shuttlebus)\b"
|
||
r"|\b(?:ov-chipkaart|chipkaart|ovpay|ov-pay|reisproduct|studentenreisproduct|vervoersbewijs|vervoerbewijs|opwaarderen)\b"
|
||
r"|\b(?:ov-staking|conducteursstaking)\b"
|
||
)
|
||
|
||
|
||
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("exclude_paywalled")
|
||
helper.copy("match_filter")
|
||
helper.copy("run_log_file")
|
||
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_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:
|
||
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:
|
||
if self._run_logger:
|
||
for handler in list(self._run_logger.handlers):
|
||
self._run_logger.removeHandler(handler)
|
||
handler.close()
|
||
self._run_logger = None
|
||
self._run_handler = None
|
||
run_log_file = self.config["run_log_file"] if "run_log_file" in self.config else ""
|
||
if not run_log_file:
|
||
return
|
||
self._run_handler = logging.FileHandler(run_log_file, encoding="utf-8")
|
||
self._run_handler.setFormatter(logging.Formatter("%(asctime)s %(message)s"))
|
||
self._run_logger = logging.getLogger(f"{self.log.name}.runlog")
|
||
self._run_logger.setLevel(logging.INFO)
|
||
self._run_logger.propagate = False
|
||
self._run_logger.addHandler(self._run_handler)
|
||
|
||
async def stop(self) -> None:
|
||
await super().stop()
|
||
self.poll_task.cancel()
|
||
if self._run_handler:
|
||
self._run_handler.close()
|
||
|
||
async def poll_feeds(self) -> None:
|
||
try:
|
||
await self._poll_feeds()
|
||
except asyncio.CancelledError:
|
||
self.log.debug("Polling stopped")
|
||
except Exception:
|
||
self.log.exception("Fatal error while polling feeds")
|
||
|
||
@staticmethod
|
||
def _filter_score(title: str, summary: str, match_filter: str) -> int:
|
||
"""Score = 2× hits in titel + 1× hits in summary. Aaneengesloten matches
|
||
(bv 'NS-trein', 'openbaar vervoer') tellen als 1 cluster."""
|
||
def count(text: str) -> int:
|
||
spans = sorted((m.start(), m.end()) for m in re.finditer(match_filter, text))
|
||
if not spans:
|
||
return 0
|
||
clusters, prev_end = 1, spans[0][1]
|
||
for s, e in spans[1:]:
|
||
if s - prev_end <= 1:
|
||
prev_end = max(prev_end, e)
|
||
else:
|
||
clusters += 1
|
||
prev_end = e
|
||
return clusters
|
||
return 2 * count(title) + count(summary)
|
||
|
||
async def _send(
|
||
self, feed: Feed, entry: Entry, sub: Subscription, *, check_filter: bool = True
|
||
) -> EventID | None:
|
||
if check_filter:
|
||
match_filter = self.config["match_filter"]
|
||
if match_filter:
|
||
summary = html.unescape(re.sub(r"<[^>]+>", "", entry.summary))
|
||
if self._filter_score(entry.title, summary, match_filter) < 2:
|
||
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 _broadcast(
|
||
self, feed: Feed, entry: Entry, subscriptions: list[Subscription]
|
||
) -> tuple[int, int, int]:
|
||
"""Send entry to all subs. Returns (placed, filtered, failed).
|
||
|
||
placed = 1 when at least one room received the message
|
||
filtered = 1 when the global match filter rejected the entry
|
||
failed = number of subscriptions where sending failed
|
||
"""
|
||
self.log.debug(f"Broadcasting {entry.id} of {feed.id}")
|
||
match_filter = self.config["match_filter"]
|
||
if match_filter:
|
||
summary = html.unescape(re.sub(r"<[^>]+>", "", entry.summary))
|
||
if self._filter_score(entry.title, summary, match_filter) < 2:
|
||
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)
|
||
return (1 if sent 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)
|
||
return
|
||
now = int(time())
|
||
start = monotonic()
|
||
tasks = [self.try_parse_feed(feed=feed) for feed in subs if feed.next_retry < now]
|
||
feed: Feed
|
||
entries: list[Entry] | None
|
||
self.log.info(f"Polling {len(tasks)} feeds")
|
||
evaluated_total = 0
|
||
placed_total = 0
|
||
filtered_total = 0
|
||
failed_total = 0
|
||
unchanged_total = 0
|
||
for res in asyncio.as_completed(tasks):
|
||
feed, entries = await res
|
||
self.log.trace(
|
||
f"Fetching {feed.id} (backoff: {feed.error_count} / {feed.next_retry}) "
|
||
f"success: {entries is not None}"
|
||
)
|
||
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:
|
||
placed, filtered, failed = await self._broadcast(
|
||
feed, entry, feed.subscriptions
|
||
)
|
||
placed_total += placed
|
||
filtered_total += filtered
|
||
failed_total += failed
|
||
duration = monotonic() - start
|
||
self._log_run_summary(
|
||
len(tasks),
|
||
evaluated_total,
|
||
placed_total,
|
||
filtered_total,
|
||
failed_total,
|
||
unchanged_total,
|
||
duration,
|
||
)
|
||
|
||
def _log_run_summary(
|
||
self,
|
||
feeds: int,
|
||
evaluated: int,
|
||
selected: int,
|
||
filtered: int,
|
||
failed: int,
|
||
unchanged: int,
|
||
duration: float,
|
||
) -> None:
|
||
line = (
|
||
f"run complete feeds={feeds} evaluated={evaluated} selected={selected} "
|
||
f"filtered={filtered} failed={failed} unchanged={unchanged} "
|
||
f"duration={duration:.2f}s"
|
||
)
|
||
self.log.info(f"Finished polling {feeds} feeds in {duration:.2f} seconds: {line}")
|
||
if self._run_logger:
|
||
self._run_logger.info(line)
|
||
|
||
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: <premium>true</premium>
|
||
str(getattr(entry, "premium", "")).lower() == "true"
|
||
# DPG Media (AD/Volkskrant/Parool/Trouw): <dpp:paid>true</dpp:paid>
|
||
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="Set the global match filter: !rss set-filter <regex> | !rss set-filter clear"
|
||
)
|
||
@command.argument("match_filter", "match filter", pass_raw=True, required=False)
|
||
async def set_filter(self, evt: MessageEvent, 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 the match filter")
|
||
return
|
||
elif not await self.can_manage(evt):
|
||
return
|
||
if not match_filter:
|
||
# Lege aanroep: nooit het filter wissen. Toon usage + huidige filter.
|
||
f = self.config["match_filter"]
|
||
if f:
|
||
await evt.reply(
|
||
f"Usage: `!rss set-filter <regex>` or `!rss set-filter clear`\n\n"
|
||
f"**Current filter:**\n```\n{f}\n```"
|
||
)
|
||
else:
|
||
await evt.reply(
|
||
"Usage: `!rss set-filter <regex>` or `!rss set-filter clear`\n\n"
|
||
"No filter is currently set (all posts are sent)"
|
||
)
|
||
return
|
||
if match_filter.strip().lower() in ("clear", "none", "off", "remove"):
|
||
self.config["match_filter"] = ""
|
||
self.config.save()
|
||
await evt.reply("Global filter removed (all posts will be sent)")
|
||
return
|
||
try:
|
||
re.compile(match_filter)
|
||
except re.error:
|
||
await evt.reply("Filter is not a valid regular expression")
|
||
return
|
||
self.config["match_filter"] = match_filter
|
||
self.config.save()
|
||
await evt.reply(
|
||
f"Global filter updated: only posts matching `{match_filter}` will be sent"
|
||
)
|
||
|
||
@rss.subcommand(
|
||
"filter", aliases=("show-filter", "get-filter"), help="Show the current global match filter."
|
||
)
|
||
async def command_show_filter(self, evt: MessageEvent) -> None:
|
||
f = self.config["match_filter"]
|
||
if f:
|
||
await evt.reply(f"**Current global filter:**\n```\n{f}\n```")
|
||
else:
|
||
await evt.reply("No global filter is set (all posts are sent)")
|
||
|
||
@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) -> 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)
|
||
match_filter = self.config["match_filter"]
|
||
if len(subscriptions) == 0:
|
||
await evt.reply("There are no RSS subscriptions in this room")
|
||
return
|
||
filter_line = (
|
||
f"**Global filter:** `{match_filter}`"
|
||
if match_filter
|
||
else "**Global filter:** none (all posts are sent)"
|
||
)
|
||
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)
|