eddy/buzz-rss
Watch
1
0
Fork
You've already forked buzz-rss
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

v1.5.0: entry-dedup in één query + conditional get (ETag/Last-Modified)

Browse source 
- filter_entries: één SELECT per feed i.p.v. per-entry EXISTS (1082 queries → 20)
- feed.etag/last_modified (migratie v9); parse_feed stuurt If-None-Match/If-Modified-Since, 304 → None
- _poll_once telt unchanged (304-feeds); summary-regel uitgebreid met unchanged
- _poll_feeds re-raised CancelledError (reload-stapeling duplicaten) + handler-cleanup
This commit is contained in:
Eddy de Vink 2026-09-06 00:11:22 +02:00
commit a03f76ca83
4 changed files with 59 additions and 25 deletions
Download patch file Download diff file Expand all files Collapse all files

46
buzzrss/bot.py
View file

@ -15,7 +15,7 @@
# along with this program. If not, see <https://www.gnu.org/licenses/>.
from __future__ import annotations
from typing import Any, Iterable
from typing import Any
from datetime import datetime, timezone
from string import Template
from time import mktime, monotonic, time
@ -235,24 +235,32 @@ class RSSBot(Plugin):
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)
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: Iterable[Entry]
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: {bool(entries)}"
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
@ -291,6 +299,7 @@ class RSSBot(Plugin):
placed_total,
filtered_total,
failed_total,
unchanged_total,
duration,
)
@ -301,11 +310,13 @@ class RSSBot(Plugin):
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} duration={duration:.2f}s"
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:
@ -323,14 +334,14 @@ class RSSBot(Plugin):
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]]:
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]]:
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} "
@ -349,18 +360,35 @@ class RSSBot(Plugin):
async def parse_feed(
self, *, feed: Feed | None = None, url: str | None = None
) -> tuple[Feed, list[Entry]]:
) -> 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=self._feed_get_headers,
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)

30
buzzrss/db.py
View file

@ -64,6 +64,8 @@ class Feed:
next_retry: int = 0
error_count: int = 0
etag: str = ""
last_modified: str = ""
subscriptions: list[Subscription] = attr.ib(factory=lambda: [])
@ -119,7 +121,7 @@ class DBManager:
async def get_feeds(self) -> list[Feed]:
q = """
SELECT id, url, title, subtitle, link, next_retry, error_count,
SELECT id, url, title, subtitle, link, next_retry, error_count, etag, last_modified,
room_id, user_id, notification_template, send_notice
FROM feed INNER JOIN subscription ON feed.id = subscription.feed_id
"""
@ -135,25 +137,19 @@ class DBManager:
async def get_feeds_by_room(self, room_id: RoomID) -> list[tuple[Feed, UserID]]:
q = """
SELECT id, url, title, subtitle, link, next_retry, error_count, user_id FROM feed
SELECT id, url, title, subtitle, link, next_retry, error_count, etag, last_modified,
user_id 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"]) 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:
if not entries:
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]
existing = await self.db.fetch("SELECT id FROM entry WHERE feed_id = $1", feed_id)
for row in existing:
entries.pop(row["id"], None)
async def add_entries(self, entries: list[Entry], override_feed_id: int | None = None) -> None:
if not entries:
@ -174,14 +170,14 @@ class DBManager:
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"
q = "SELECT id, url, title, subtitle, link, next_retry, error_count, etag, last_modified 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,
SELECT id, url, title, subtitle, link, next_retry, error_count, etag, last_modified,
room_id, user_id, notification_template, send_notice
FROM feed LEFT JOIN subscription ON feed.id = subscription.feed_id AND room_id = $2
WHERE feed.id = $1
@ -206,6 +202,10 @@ class DBManager:
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 set_feed_headers(self, info: Feed, etag: str, last_modified: str) -> None:
q = "UPDATE feed SET etag = $2, last_modified = $3 WHERE id = $1"
await self.db.execute(q, info.id, etag, last_modified)
async def subscribe(
self,
feed_id: int,

6
buzzrss/migrations.py
View file

@ -99,3 +99,9 @@ async def upgrade_v7(conn: Connection) -> None:
@upgrade_table.register(description="Add paywalled flag to entries")
async def upgrade_v8(conn: Connection) -> None:
await conn.execute("ALTER TABLE entry ADD COLUMN paywalled BOOLEAN NOT NULL DEFAULT false")
@upgrade_table.register(description="Add conditional-fetch headers to feeds")
async def upgrade_v9(conn: Connection) -> None:
await conn.execute("ALTER TABLE feed ADD COLUMN etag TEXT DEFAULT ''")
await conn.execute("ALTER TABLE feed ADD COLUMN last_modified TEXT DEFAULT ''")

2
maubot.yaml
View file

@ -1,5 +1,5 @@
id: nl.eddydevink.rss
version: 1.4.1
version: 1.5.0
modules:
- buzzrss
main_class: RSSBot