use nix-shell shebang and a cronjob to get this off my back for now

This commit is contained in:
D. Scott Boggs 2023-07-03 12:25:55 -04:00
parent 7d74067d88
commit 3c6fddf9ac
2 changed files with 49 additions and 18 deletions

3
README.md Normal file
View file

@ -0,0 +1,3 @@
## RSS Notifier
A python script which checks an RSS feed for new entries and sends
notifications to any libnotify-compatible Linux Desktop when it finds them.

64
rss_notifier.py Normal file → Executable file
View file

@ -1,3 +1,5 @@
#!/usr/bin/env nix-shell
#! nix-shell -i python -p python3 libnotify python3Packages.beautifulsoup4 python3Packages.lxml python3Packages.requests
from requests import get from requests import get
from sqlite3 import Connection as SQL, Cursor from sqlite3 import Connection as SQL, Cursor
from contextlib import contextmanager from contextlib import contextmanager
@ -6,27 +8,33 @@ from sys import argv, stderr
from dataclasses import dataclass from dataclasses import dataclass
from typing import * from typing import *
from subprocess import Popen, PIPE, run from subprocess import Popen, PIPE, run
from os import execvp, environ from os import execvp, environ, makedirs
from json import dumps, loads from json import dumps, loads
from pathlib import Path
@dataclass @dataclass
class Feed: class Feed:
id: int id: int
url: str url: str
title: Optional[str] = None
content: Optional[Soup] = None content: Optional[Soup] = None
def fetch(self): def fetch_content(self):
res = get(self.url) res = get(self.url)
res.raise_for_status() res.raise_for_status()
self.content = Soup(res.text, features="xml") self.content = Soup(res.text, features="xml")
if title_tag := self.content.find('title'):
self.title = title_tag.text
else:
print(f'warning: no title for feed {self.url!r}')
return self return self
def entries(self): def entries(self):
if content := self.content: if content := self.content:
return content.find_all('entry') return content.find_all('entry')
else: else:
return self.fetch().entries() return self.fetch_content().entries()
@classmethod @classmethod
def from_record(cls, record): def from_record(cls, record):
@ -42,7 +50,7 @@ class Feed:
return hash(self.id) return hash(self.id)
def to_dict(self): def to_dict(self):
return {'id': self.id, 'url': self.url} return {'id': self.id, 'url': self.url, 'title': self.title}
def to_json(self) -> str: def to_json(self) -> str:
return dumps(self.to_dict()) return dumps(self.to_dict())
@ -50,7 +58,7 @@ class Feed:
@classmethod @classmethod
def from_json(cls, text: str) -> 'Feed': def from_json(cls, text: str) -> 'Feed':
data = loads(text) data = loads(text)
return cls(id=data['id'], url=data['url']) return cls(id=data['id'], url=data['url'], title=data['title'])
ACTIONS = ['--action', 'open=Open link', '--action', 'read=Mark read'] ACTIONS = ['--action', 'open=Open link', '--action', 'read=Mark read']
@ -65,10 +73,16 @@ class FeedEntry:
@classmethod @classmethod
def select_all(cls, db_connection) -> List['FeedEntry']: def select_all(cls, db_connection) -> List['FeedEntry']:
feed_entries = db_connection.cursor().execute("select entry.*, feed.url from entries entry join feeds feed on feed.id = entry.feed_id;").fetchall() query = '''
select entry.*, feed.url, feed.title
from entries entry
join feeds feed
on feed.id = entry.feed_id;
'''
feed_entries = db_connection.cursor().execute(query).fetchall()
return set( return set(
FeedEntry(id, Feed(feed_id, feed_url), upstream_id, title, link, read) FeedEntry(id, Feed(feed_id, feed_url, feed_title), upstream_id, title, link, read)
for id, feed_id, upstream_id, title, link, read, feed_url, for id, feed_id, upstream_id, title, link, read, feed_url, feed_title
in feed_entries in feed_entries
) )
@ -83,7 +97,10 @@ class FeedEntry:
def to_json(self) -> str: def to_json(self) -> str:
assert list(self.__dict__.keys()) == "id feed upstream_id title link read".split() assert list(self.__dict__.keys()) == "id feed upstream_id title link read".split()
return dumps({ k: v.to_dict() if k == 'feed' else v for k, v in self.__dict__.items() }) return dumps({
k: v.to_dict() if k == 'feed' else v
for k, v in self.__dict__.items()
})
@classmethod @classmethod
def from_json(cls, data) -> 'FeedEntry': def from_json(cls, data) -> 'FeedEntry':
@ -118,7 +135,7 @@ class FeedEntry:
self.upstream_id, self.upstream_id,
self.title, self.title,
self.link, self.link,
self.mark_read self.read
) )
) )
@ -133,9 +150,13 @@ class RSSNotifier:
def create_tables(self): def create_tables(self):
db = self.db_connection.cursor() db = self.db_connection.cursor()
db.execute( db.execute('''
'create table if not exists feeds (id integer primary key autoincrement, url text)' create table if not exists feeds (
) id integer primary key autoincrement,
url text unique not null,
title text
);
''')
db.execute(''' db.execute('''
create table if not exists entries ( create table if not exists entries (
id integer primary key autoincrement, id integer primary key autoincrement,
@ -154,9 +175,9 @@ class RSSNotifier:
return map(Feed, feeds) return map(Feed, feeds)
def add_feed(self, url: str, mark_read: bool): def add_feed(self, url: str, mark_read: bool):
feed = Feed(-1, url).fetch() feed = Feed(-1, url).fetch_content()
cursor = self.db_connection.cursor() cursor = self.db_connection.cursor()
cursor.execute("insert into feeds (url) values (?)", (url,)) cursor.execute("insert into feeds (url, title) values (?, ?)", (url, feed.title))
feed.id = cursor.lastrowid feed.id = cursor.lastrowid
for entry in feed.entries(): for entry in feed.entries():
FeedEntry.from_rss(entry, feed, mark_read).insert(cursor) FeedEntry.from_rss(entry, feed, mark_read).insert(cursor)
@ -166,7 +187,7 @@ class RSSNotifier:
def parse_args(self, args=None): def parse_args(self, args=None):
mark_read = True mark_read = True
if args is None: if args is None:
args = argv args = argv[1:]
feeds_to_add = [] feeds_to_add = []
try: try:
while True: while True:
@ -241,7 +262,7 @@ class RSSNotifier:
'--expire-time', '0', '--expire-time', '0',
'--app-name', 'RSS Notifier', '--app-name', 'RSS Notifier',
*ACTIONS, *ACTIONS,
f'New RSS Story: {entry.title}' f'New RSS Story from {entry.feed.title}: {entry.title}'
], ],
stdout=PIPE, stdout=PIPE,
stderr=PIPE stderr=PIPE
@ -261,4 +282,11 @@ class RSSNotifier:
if __name__ == "__main__": if __name__ == "__main__":
RSSNotifier(environ.get("RSS_NOTIFIER_DATABASE_LOCATION") or Path(environ.get("HOME")) / ".local/state/rss-notifier/db.sqlite3").parse_args() db_path = environ.get("RSS_NOTIFIER_DATABASE_LOCATION")
if db_path is None:
db_path = Path(environ.get("HOME")) / ".local/state/rss-notifier/db.sqlite3"
else:
db_path = Path(db_path)
if not db_path.parent.exists():
makedirs(db_path.parent)
RSSNotifier(db_path).parse_args()