v1.4.1: per-run summary log (feeds/evaluated/selected/filtered/failed)
- _broadcast checks match filter and returns (placed, filtered, failed) - _poll_once accumulates per-run counters, logs one summary line per run - run_log_file config writes to /data/buzzrss-runs.log (persistent mount) - _poll_feeds re-raises CancelledError so reloads stop old poll loops - _setup_run_log fully resets handlers; empty config disables logging
This commit is contained in:
parent
dff304ba56
commit
1f92b38ba6
3 changed files with 97 additions and 17 deletions
|
|
@ -12,9 +12,10 @@ spam_sleep: 2
|
||||||
command_prefix: "rss"
|
command_prefix: "rss"
|
||||||
# Default post notification template for new subscriptions
|
# Default post notification template for new subscriptions
|
||||||
notification_template: "New post in $feed_title: [$title]($link)"
|
notification_template: "New post in $feed_title: [$title]($link)"
|
||||||
# Should non-admin users be allowed to set regex filters to drop posts from feeds?
|
# Per-run summary logfile: one line per poll run with feeds/evaluated/selected/filtered/failed counts.
|
||||||
# Regexes are dangerous and may lead to DoS attacks.
|
# Absoluut pad in de container aanraden: /data is de persistente maubot-mount (host: /matrix/maubot/data).
|
||||||
allow_filter: false
|
# Leeg = uitgeschakeld.
|
||||||
|
run_log_file: /data/buzzrss-runs.log
|
||||||
# Exclude paywalled articles (Telegraaf/NHD markeren dit met <premium>true</premium>)
|
# Exclude paywalled articles (Telegraaf/NHD markeren dit met <premium>true</premium>)
|
||||||
exclude_paywalled: true
|
exclude_paywalled: true
|
||||||
# Global match filter (whitelist): applies to ALL feeds in this instance.
|
# Global match filter (whitelist): applies to ALL feeds in this instance.
|
||||||
|
|
|
||||||
105
buzzrss/bot.py
105
buzzrss/bot.py
|
|
@ -23,6 +23,7 @@ import asyncio
|
||||||
import hashlib
|
import hashlib
|
||||||
import html
|
import html
|
||||||
import io
|
import io
|
||||||
|
import logging
|
||||||
import re
|
import re
|
||||||
|
|
||||||
import aiohttp
|
import aiohttp
|
||||||
|
|
@ -71,6 +72,7 @@ class Config(BaseProxyConfig):
|
||||||
helper.copy("allow_filter")
|
helper.copy("allow_filter")
|
||||||
helper.copy("exclude_paywalled")
|
helper.copy("exclude_paywalled")
|
||||||
helper.copy("match_filter")
|
helper.copy("match_filter")
|
||||||
|
helper.copy("run_log_file")
|
||||||
helper.copy("admins")
|
helper.copy("admins")
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -94,6 +96,8 @@ class RSSBot(Plugin):
|
||||||
exclude_paywalled: bool = False
|
exclude_paywalled: bool = False
|
||||||
poll_task: asyncio.Future
|
poll_task: asyncio.Future
|
||||||
poll_sema: asyncio.Semaphore | None
|
poll_sema: asyncio.Semaphore | None
|
||||||
|
_run_logger: logging.Logger | None = None
|
||||||
|
_run_handler: logging.FileHandler | None = None
|
||||||
http: aiohttp.ClientSession
|
http: aiohttp.ClientSession
|
||||||
power_level_cache: dict[RoomID, tuple[int, PowerLevelStateEventContent, StateEvent]]
|
power_level_cache: dict[RoomID, tuple[int, PowerLevelStateEventContent, StateEvent]]
|
||||||
|
|
||||||
|
|
@ -121,10 +125,30 @@ class RSSBot(Plugin):
|
||||||
)
|
)
|
||||||
# single-instance fork: class-level since _parse_rss is a classmethod
|
# single-instance fork: class-level since _parse_rss is a classmethod
|
||||||
type(self).exclude_paywalled = bool(self.config["exclude_paywalled"])
|
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:
|
async def stop(self) -> None:
|
||||||
await super().stop()
|
await super().stop()
|
||||||
self.poll_task.cancel()
|
self.poll_task.cancel()
|
||||||
|
if self._run_handler:
|
||||||
|
self._run_handler.close()
|
||||||
|
|
||||||
async def poll_feeds(self) -> None:
|
async def poll_feeds(self) -> None:
|
||||||
try:
|
try:
|
||||||
|
|
@ -152,12 +176,15 @@ class RSSBot(Plugin):
|
||||||
return clusters
|
return clusters
|
||||||
return 2 * count(title) + count(summary)
|
return 2 * count(title) + count(summary)
|
||||||
|
|
||||||
async def _send(self, feed: Feed, entry: Entry, sub: Subscription) -> EventID | None:
|
async def _send(
|
||||||
match_filter = self.config["match_filter"]
|
self, feed: Feed, entry: Entry, sub: Subscription, *, check_filter: bool = True
|
||||||
if match_filter:
|
) -> EventID | None:
|
||||||
summary = html.unescape(re.sub(r"<[^>]+>", "", entry.summary))
|
if check_filter:
|
||||||
if self._filter_score(entry.title, summary, match_filter) < 2:
|
match_filter = self.config["match_filter"]
|
||||||
return None
|
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(
|
message = sub.notification_template.safe_substitute(
|
||||||
{
|
{
|
||||||
|
|
@ -175,23 +202,40 @@ class RSSBot(Plugin):
|
||||||
)
|
)
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
self.log.warning(f"Failed to send {entry.id} of {feed.id} to {sub.room_id}: {e}")
|
self.log.warning(f"Failed to send {entry.id} of {feed.id} to {sub.room_id}: {e}")
|
||||||
|
return None
|
||||||
|
|
||||||
async def _broadcast(
|
async def _broadcast(
|
||||||
self, feed: Feed, entry: Entry, subscriptions: list[Subscription]
|
self, feed: Feed, entry: Entry, subscriptions: list[Subscription]
|
||||||
) -> None:
|
) -> 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}")
|
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"]
|
spam_sleep = self.config["spam_sleep"]
|
||||||
tasks = [self._send(feed, entry, sub) for sub in subscriptions]
|
|
||||||
if spam_sleep >= 0:
|
if spam_sleep >= 0:
|
||||||
for task in tasks:
|
results: list[EventID | None] = []
|
||||||
await task
|
for sub in subscriptions:
|
||||||
|
results.append(await self._send(feed, entry, sub, check_filter=False))
|
||||||
await asyncio.sleep(spam_sleep)
|
await asyncio.sleep(spam_sleep)
|
||||||
else:
|
else:
|
||||||
await asyncio.gather(*tasks)
|
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:
|
async def _poll_once(self) -> None:
|
||||||
subs = await self.dbm.get_feeds()
|
subs = await self.dbm.get_feeds()
|
||||||
if not subs:
|
if not subs:
|
||||||
|
self._log_run_summary(0, 0, 0, 0, 0, 0.0)
|
||||||
return
|
return
|
||||||
now = int(time())
|
now = int(time())
|
||||||
start = monotonic()
|
start = monotonic()
|
||||||
|
|
@ -199,6 +243,10 @@ class RSSBot(Plugin):
|
||||||
feed: Feed
|
feed: Feed
|
||||||
entries: Iterable[Entry]
|
entries: Iterable[Entry]
|
||||||
self.log.info(f"Polling {len(tasks)} feeds")
|
self.log.info(f"Polling {len(tasks)} feeds")
|
||||||
|
evaluated_total = 0
|
||||||
|
placed_total = 0
|
||||||
|
filtered_total = 0
|
||||||
|
failed_total = 0
|
||||||
for res in asyncio.as_completed(tasks):
|
for res in asyncio.as_completed(tasks):
|
||||||
feed, entries = await res
|
feed, entries = await res
|
||||||
self.log.trace(
|
self.log.trace(
|
||||||
|
|
@ -225,13 +273,43 @@ class RSSBot(Plugin):
|
||||||
self.log.trace(
|
self.log.trace(
|
||||||
f"Feed {feed.id} had {len(new_entries)} new entries (of {len(entries)} total)"
|
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: list[Entry] = list(new_entries.values())
|
||||||
new_entry_list.sort(key=lambda entry: (entry.date, entry.id))
|
new_entry_list.sort(key=lambda entry: (entry.date, entry.id))
|
||||||
await self.dbm.add_entries(new_entry_list)
|
await self.dbm.add_entries(new_entry_list)
|
||||||
for entry in new_entry_list:
|
for entry in new_entry_list:
|
||||||
await self._broadcast(feed, entry, feed.subscriptions)
|
placed, filtered, failed = await self._broadcast(
|
||||||
|
feed, entry, feed.subscriptions
|
||||||
|
)
|
||||||
|
placed_total += placed
|
||||||
|
filtered_total += filtered
|
||||||
|
failed_total += failed
|
||||||
duration = monotonic() - start
|
duration = monotonic() - start
|
||||||
self.log.info(f"Finished polling {len(tasks)} feeds in {duration:.2f} seconds")
|
self._log_run_summary(
|
||||||
|
len(tasks),
|
||||||
|
evaluated_total,
|
||||||
|
placed_total,
|
||||||
|
filtered_total,
|
||||||
|
failed_total,
|
||||||
|
duration,
|
||||||
|
)
|
||||||
|
|
||||||
|
def _log_run_summary(
|
||||||
|
self,
|
||||||
|
feeds: int,
|
||||||
|
evaluated: int,
|
||||||
|
selected: int,
|
||||||
|
filtered: int,
|
||||||
|
failed: int,
|
||||||
|
duration: float,
|
||||||
|
) -> None:
|
||||||
|
line = (
|
||||||
|
f"run complete feeds={feeds} evaluated={evaluated} selected={selected} "
|
||||||
|
f"filtered={filtered} failed={failed} 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:
|
async def _poll_feeds(self) -> None:
|
||||||
self.log.debug("Polling started")
|
self.log.debug("Polling started")
|
||||||
|
|
@ -240,6 +318,7 @@ class RSSBot(Plugin):
|
||||||
await self._poll_once()
|
await self._poll_once()
|
||||||
except asyncio.CancelledError:
|
except asyncio.CancelledError:
|
||||||
self.log.debug("Polling stopped")
|
self.log.debug("Polling stopped")
|
||||||
|
raise
|
||||||
except Exception:
|
except Exception:
|
||||||
self.log.exception("Error while polling feeds")
|
self.log.exception("Error while polling feeds")
|
||||||
await asyncio.sleep(self.config["update_interval"] * 60)
|
await asyncio.sleep(self.config["update_interval"] * 60)
|
||||||
|
|
|
||||||
|
|
@ -1,5 +1,5 @@
|
||||||
id: nl.eddydevink.rss
|
id: nl.eddydevink.rss
|
||||||
version: 1.3.1
|
version: 1.4.1
|
||||||
modules:
|
modules:
|
||||||
- buzzrss
|
- buzzrss
|
||||||
main_class: RSSBot
|
main_class: RSSBot
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue