feed: add support to filter posts

The post filter is kept very simple, it works by checking if a fixed
string exists in the configured field.  This could easily be expanded
allowing for regex or even more complex filters like date-time
comparisons.
This commit is contained in:
ducklet 2020-11-07 21:25:36 +01:00
parent 78777a4da9
commit 15c3cb0221
2 changed files with 23 additions and 2 deletions

View file

@ -15,10 +15,15 @@ FeedId = str
PostId = str PostId = str
def no_filter(x):
return True
@dataclass @dataclass
class Feed: class Feed:
id: FeedId id: FeedId
url: str url: str
filter: Callable[["Post"], bool] = no_filter
title: Optional[str] = None title: Optional[str] = None
posts: List["Post"] = field(default_factory=list) posts: List["Post"] = field(default_factory=list)
etag: Optional[str] = None etag: Optional[str] = None
@ -55,13 +60,14 @@ class Feed:
if "title" in r.feed: if "title" in r.feed:
self.title = r.feed.title self.title = r.feed.title
posts = [Post.from_entry(e) for e in r.entries] posts = [p for e in r.entries if self.filter(p := Post.from_entry(e))]
for post in posts: for post in posts:
if post.date is None: if post.date is None:
post.date = pubdate(r.feed) post.date = pubdate(r.feed)
posts.sort(key=lambda e: e.date, reverse=True) posts.sort(key=lambda e: e.date, reverse=True)
self.posts = posts self.posts = posts
# Find link to next feed page.
for link in r.feed.get("links", []): for link in r.feed.get("links", []):
if link.get("rel") == "next": if link.get("rel") == "next":
self.next_url = link.get("href") self.next_url = link.get("href")

View file

@ -2,6 +2,7 @@ import asyncio
import logging import logging
from datetime import datetime, timezone from datetime import datetime, timezone
from html import escape from html import escape
from typing import *
import feeder import feeder
import postillon import postillon
@ -12,12 +13,26 @@ from ..models import Job, Message
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
def make_filter(config: Mapping[str, str]) -> Callable[[feeder.Post], bool]:
def filter(post):
return all(
text in getattr(post, attrname, "") for attrname, text in config.items()
)
return filter
def init(bot): def init(bot):
bot.on_command("feed", handle) bot.on_command("feed", handle)
if "feeder" not in bot.shared: if "feeder" not in bot.shared:
feeds = ( feeds = (
feeder.Feed(fid, f["url"], title=f["display"]) feeder.Feed(
fid,
f["url"],
title=f["display"],
filter=make_filter(f.get("filter", {})),
)
for fid, f in bot.config.get("feeder.feeds").items() for fid, f in bot.config.get("feeder.feeds").items()
) )
feedstore = feeder.Store(bot.config.get("feeder.storage")) feedstore = feeder.Store(bot.config.get("feeder.storage"))