Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
cb6d00e707 | ||
|
|
aa4f7144be | ||
|
|
cbb9c2e55d | ||
|
|
a03f76ca83 | ||
|
|
1f92b38ba6 |
5 changed files with 202 additions and 57 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.
|
||||||
|
|
|
||||||
184
buzzrss/bot.py
184
buzzrss/bot.py
|
|
@ -15,7 +15,7 @@
|
||||||
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
from typing import Any, Iterable
|
from typing import Any
|
||||||
from datetime import datetime, timezone
|
from datetime import datetime, timezone
|
||||||
from string import Template
|
from string import Template
|
||||||
from time import mktime, monotonic, time
|
from time import mktime, monotonic, time
|
||||||
|
|
@ -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:
|
||||||
|
|
@ -135,10 +159,9 @@ class RSSBot(Plugin):
|
||||||
self.log.exception("Fatal error while polling feeds")
|
self.log.exception("Fatal error while polling feeds")
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def _filter_score(title: str, summary: str, match_filter: str) -> int:
|
def _matches_in(text: str, match_filter: str) -> int:
|
||||||
"""Score = 2× hits in titel + 1× hits in summary. Aaneengesloten matches
|
"""Aantal aaneengesloten match-clusters in text (0 als geen match).
|
||||||
(bv 'NS-trein', 'openbaar vervoer') tellen als 1 cluster."""
|
Aaneengesloten matches (bv 'NS-trein', 'openbaar vervoer') tellen als 1."""
|
||||||
def count(text: str) -> int:
|
|
||||||
spans = sorted((m.start(), m.end()) for m in re.finditer(match_filter, text))
|
spans = sorted((m.start(), m.end()) for m in re.finditer(match_filter, text))
|
||||||
if not spans:
|
if not spans:
|
||||||
return 0
|
return 0
|
||||||
|
|
@ -150,13 +173,32 @@ class RSSBot(Plugin):
|
||||||
clusters += 1
|
clusters += 1
|
||||||
prev_end = e
|
prev_end = e
|
||||||
return clusters
|
return clusters
|
||||||
return 2 * count(title) + count(summary)
|
|
||||||
|
|
||||||
async def _send(self, feed: Feed, entry: Entry, sub: Subscription) -> EventID | None:
|
@staticmethod
|
||||||
|
def _filter_score(title: str, summary: str, match_filter: str) -> int:
|
||||||
|
"""Score = 2× hits in titel + 1× hits in summary."""
|
||||||
|
return 2 * RSSBot._matches_in(title, match_filter) + RSSBot._matches_in(summary, match_filter)
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _filter_passes(title: str, summary: str, match_filter: str) -> bool:
|
||||||
|
"""Regel C: doorlaten bij (titel-match >= 1) of (totaalscore >= 3).
|
||||||
|
|
||||||
|
Een OV-trefwoord in de titel is een sterke intentieverklaring (score 2
|
||||||
|
volstaat). Treffers alleen in de samenvatting moeten dubbel bevestigd
|
||||||
|
worden (>= 3): zo houdt 'bussen rijden om' in de lead van een
|
||||||
|
woningbrand-bericht het artikel niet meer binnen."""
|
||||||
|
if RSSBot._matches_in(title, match_filter) >= 1:
|
||||||
|
return True
|
||||||
|
return RSSBot._filter_score(title, summary, match_filter) >= 3
|
||||||
|
|
||||||
|
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"]
|
match_filter = self.config["match_filter"]
|
||||||
if match_filter:
|
if match_filter:
|
||||||
summary = html.unescape(re.sub(r"<[^>]+>", "", entry.summary))
|
summary = html.unescape(re.sub(r"<[^>]+>", "", entry.summary))
|
||||||
if self._filter_score(entry.title, summary, match_filter) < 2:
|
if not self._filter_passes(entry.title, summary, match_filter):
|
||||||
return None
|
return None
|
||||||
|
|
||||||
message = sub.notification_template.safe_substitute(
|
message = sub.notification_template.safe_substitute(
|
||||||
|
|
@ -175,36 +217,65 @@ 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 not self._filter_passes(entry.title, summary, match_filter):
|
||||||
|
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.0)
|
||||||
return
|
return
|
||||||
now = int(time())
|
now = int(time())
|
||||||
start = monotonic()
|
start = monotonic()
|
||||||
tasks = [self.try_parse_feed(feed=feed) for feed in subs if feed.next_retry < now]
|
tasks = [self.try_parse_feed(feed=feed) for feed in subs if feed.next_retry < now]
|
||||||
feed: Feed
|
feed: Feed
|
||||||
entries: Iterable[Entry]
|
entries: list[Entry] | None
|
||||||
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
|
||||||
|
unchanged_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(
|
||||||
f"Fetching {feed.id} (backoff: {feed.error_count} / {feed.next_retry}) "
|
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:
|
if not entries:
|
||||||
error_count = feed.error_count + 1
|
error_count = feed.error_count + 1
|
||||||
next_retry_delay = self.config["update_interval"] * 60 * error_count
|
next_retry_delay = self.config["update_interval"] * 60 * error_count
|
||||||
|
|
@ -225,13 +296,46 @@ 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,
|
||||||
|
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:
|
async def _poll_feeds(self) -> None:
|
||||||
self.log.debug("Polling started")
|
self.log.debug("Polling started")
|
||||||
|
|
@ -240,18 +344,19 @@ 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)
|
||||||
|
|
||||||
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:
|
if self.poll_sema is not None:
|
||||||
async with self.poll_sema:
|
async with self.poll_sema:
|
||||||
return await self._try_parse_feed(feed)
|
return await self._try_parse_feed(feed)
|
||||||
else:
|
else:
|
||||||
return await self._try_parse_feed(feed)
|
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:
|
try:
|
||||||
self.log.trace(
|
self.log.trace(
|
||||||
f"Trying to fetch {feed.id} / {feed.url} "
|
f"Trying to fetch {feed.id} / {feed.url} "
|
||||||
|
|
@ -270,18 +375,35 @@ class RSSBot(Plugin):
|
||||||
|
|
||||||
async def parse_feed(
|
async def parse_feed(
|
||||||
self, *, feed: Feed | None = None, url: str | None = None
|
self, *, feed: Feed | None = None, url: str | None = None
|
||||||
) -> tuple[Feed, list[Entry]]:
|
) -> tuple[Feed, list[Entry] | None]:
|
||||||
if feed is None:
|
if feed is None:
|
||||||
if url is None:
|
if url is None:
|
||||||
raise ValueError("Either feed or url must be set")
|
raise ValueError("Either feed or url must be set")
|
||||||
feed = Feed(id=-1, url=url, title="", subtitle="", link="")
|
feed = Feed(id=-1, url=url, title="", subtitle="", link="")
|
||||||
elif url is not None:
|
elif url is not None:
|
||||||
raise ValueError("Only one of feed or url must be set")
|
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(
|
resp = await self.http.get(
|
||||||
feed.url,
|
feed.url,
|
||||||
headers=self._feed_get_headers,
|
headers=headers,
|
||||||
timeout=aiohttp.ClientTimeout(total=20),
|
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()
|
ct = resp.headers["Content-Type"].split(";")[0].strip()
|
||||||
if ct == "application/json" or ct == "application/feed+json":
|
if ct == "application/json" or ct == "application/feed+json":
|
||||||
return await self._parse_json(feed, resp)
|
return await self._parse_json(feed, resp)
|
||||||
|
|
@ -317,7 +439,10 @@ class RSSBot(Plugin):
|
||||||
).strip()
|
).strip()
|
||||||
id = str(entry["id"])
|
id = str(entry["id"])
|
||||||
link = entry.get("url") or id
|
link = entry.get("url") or id
|
||||||
return Entry(feed_id=feed_id, id=id, date=date, title=title, summary=summary, link=link)
|
return Entry(
|
||||||
|
feed_id=feed_id, id=id, date=date, title=title, summary=summary, link=link,
|
||||||
|
paywalled=cls._is_paywalled_link(link),
|
||||||
|
)
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
async def _parse_rss(
|
async def _parse_rss(
|
||||||
|
|
@ -338,6 +463,12 @@ class RSSBot(Plugin):
|
||||||
entries = [e for e in entries if not e.paywalled]
|
entries = [e for e in entries if not e.paywalled]
|
||||||
return feed, entries
|
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
|
@classmethod
|
||||||
def _parse_rss_entry(cls, feed_id: int, entry: Any) -> Entry:
|
def _parse_rss_entry(cls, feed_id: int, entry: Any) -> Entry:
|
||||||
return Entry(
|
return Entry(
|
||||||
|
|
@ -358,7 +489,14 @@ class RSSBot(Plugin):
|
||||||
title=getattr(entry, "title", ""),
|
title=getattr(entry, "title", ""),
|
||||||
summary=getattr(entry, "description", "").strip(),
|
summary=getattr(entry, "description", "").strip(),
|
||||||
link=getattr(entry, "link", ""),
|
link=getattr(entry, "link", ""),
|
||||||
paywalled=str(getattr(entry, "premium", "")).lower() == "true",
|
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
|
@staticmethod
|
||||||
|
|
|
||||||
|
|
@ -64,6 +64,8 @@ class Feed:
|
||||||
|
|
||||||
next_retry: int = 0
|
next_retry: int = 0
|
||||||
error_count: int = 0
|
error_count: int = 0
|
||||||
|
etag: str = ""
|
||||||
|
last_modified: str = ""
|
||||||
|
|
||||||
subscriptions: list[Subscription] = attr.ib(factory=lambda: [])
|
subscriptions: list[Subscription] = attr.ib(factory=lambda: [])
|
||||||
|
|
||||||
|
|
@ -119,7 +121,7 @@ class DBManager:
|
||||||
|
|
||||||
async def get_feeds(self) -> list[Feed]:
|
async def get_feeds(self) -> list[Feed]:
|
||||||
q = """
|
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
|
room_id, user_id, notification_template, send_notice
|
||||||
FROM feed INNER JOIN subscription ON feed.id = subscription.feed_id
|
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]]:
|
async def get_feeds_by_room(self, room_id: RoomID) -> list[tuple[Feed, UserID]]:
|
||||||
q = """
|
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
|
INNER JOIN subscription ON feed.id = subscription.feed_id AND subscription.room_id = $1
|
||||||
"""
|
"""
|
||||||
rows = await self.db.fetch(q, room_id)
|
rows = await self.db.fetch(q, room_id)
|
||||||
return [(Feed.from_row(row), row["user_id"]) for row in rows]
|
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:
|
async def filter_entries(self, feed_id: int, entries: dict[str, Entry]) -> None:
|
||||||
entry_keys = list(entries.keys())
|
if not entries:
|
||||||
if not entry_keys:
|
|
||||||
return
|
return
|
||||||
async with self.db.acquire() as conn:
|
existing = await self.db.fetch("SELECT id FROM entry WHERE feed_id = $1", feed_id)
|
||||||
for key in entry_keys:
|
for row in existing:
|
||||||
exists = await conn.fetchval(
|
entries.pop(row["id"], None)
|
||||||
"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:
|
async def add_entries(self, entries: list[Entry], override_feed_id: int | None = None) -> None:
|
||||||
if not entries:
|
if not entries:
|
||||||
|
|
@ -174,14 +170,14 @@ class DBManager:
|
||||||
await conn.executemany(q, records)
|
await conn.executemany(q, records)
|
||||||
|
|
||||||
async def get_feed_by_url(self, url: str) -> Feed | None:
|
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))
|
return Feed.from_row(await self.db.fetchrow(q, url))
|
||||||
|
|
||||||
async def get_subscription(
|
async def get_subscription(
|
||||||
self, feed_id: int, room_id: RoomID
|
self, feed_id: int, room_id: RoomID
|
||||||
) -> tuple[Subscription | None, Feed | None]:
|
) -> tuple[Subscription | None, Feed | None]:
|
||||||
q = """
|
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
|
room_id, user_id, notification_template, send_notice
|
||||||
FROM feed LEFT JOIN subscription ON feed.id = subscription.feed_id AND room_id = $2
|
FROM feed LEFT JOIN subscription ON feed.id = subscription.feed_id AND room_id = $2
|
||||||
WHERE feed.id = $1
|
WHERE feed.id = $1
|
||||||
|
|
@ -206,6 +202,10 @@ class DBManager:
|
||||||
q = "UPDATE feed SET error_count = $2, next_retry = $3 WHERE id = $1"
|
q = "UPDATE feed SET error_count = $2, next_retry = $3 WHERE id = $1"
|
||||||
await self.db.execute(q, info.id, error_count, next_retry)
|
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(
|
async def subscribe(
|
||||||
self,
|
self,
|
||||||
feed_id: int,
|
feed_id: int,
|
||||||
|
|
|
||||||
|
|
@ -99,3 +99,9 @@ async def upgrade_v7(conn: Connection) -> None:
|
||||||
@upgrade_table.register(description="Add paywalled flag to entries")
|
@upgrade_table.register(description="Add paywalled flag to entries")
|
||||||
async def upgrade_v8(conn: Connection) -> None:
|
async def upgrade_v8(conn: Connection) -> None:
|
||||||
await conn.execute("ALTER TABLE entry ADD COLUMN paywalled BOOLEAN NOT NULL DEFAULT false")
|
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 ''")
|
||||||
|
|
|
||||||
|
|
@ -1,5 +1,5 @@
|
||||||
id: nl.eddydevink.rss
|
id: nl.eddydevink.rss
|
||||||
version: 1.3.1
|
version: 1.5.3
|
||||||
modules:
|
modules:
|
||||||
- buzzrss
|
- buzzrss
|
||||||
main_class: RSSBot
|
main_class: RSSBot
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue