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 794ed86ffe
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 sqlite3 import Connection as SQL, Cursor
from contextlib import contextmanager
@ -6,27 +8,33 @@ from sys import argv, stderr
from dataclasses import dataclass
from typing import *
from subprocess import Popen, PIPE, run
from os import execvp, environ
from os import execvp, environ, makedirs
from json import dumps, loads
from pathlib import Path
@dataclass
class Feed:
id: int
url: str
title: Optional[str] = None
content: Optional[Soup] = None
def fetch(self):
def fetch_content(self):
res = get(self.url)
res.raise_for_status()
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
def entries(self):
if content := self.content:
return content.find_all('entry')
else:
return self.fetch().entries()
return self.fetch_content().entries()
@classmethod
def from_record(cls, record):
@ -42,7 +50,7 @@ class Feed:
return hash(self.id)
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:
return dumps(self.to_dict())
@ -50,7 +58,7 @@ class Feed:
@classmethod
def from_json(cls, text: str) -> 'Feed':
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']
@ -65,10 +73,16 @@ class FeedEntry:
@classmethod
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(
FeedEntry(id, Feed(feed_id, feed_url), upstream_id, title, link, read)
for id, feed_id, upstream_id, title, link, read, feed_url,
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, feed_title
in feed_entries
)
@ -83,7 +97,10 @@ class FeedEntry:
def to_json(self) -> str:
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
def from_json(cls, data) -> 'FeedEntry':
@ -118,7 +135,7 @@ class FeedEntry:
self.upstream_id,
self.title,
self.link,
self.mark_read
self.read
)
)
@ -133,9 +150,13 @@ class RSSNotifier:
def create_tables(self):
db = self.db_connection.cursor()
db.execute(
'create table if not exists feeds (id integer primary key autoincrement, url text)'
)
db.execute('''
create table if not exists feeds (
id integer primary key autoincrement,
url text unique not null,
title text
);
''')
db.execute('''
create table if not exists entries (
id integer primary key autoincrement,
@ -154,9 +175,9 @@ class RSSNotifier:
return map(Feed, feeds)
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.execute("insert into feeds (url) values (?)", (url,))
cursor.execute("insert into feeds (url, title) values (?, ?)", (url, feed.title))
feed.id = cursor.lastrowid
for entry in feed.entries():
FeedEntry.from_rss(entry, feed, mark_read).insert(cursor)
@ -166,7 +187,7 @@ class RSSNotifier:
def parse_args(self, args=None):
mark_read = True
if args is None:
args = argv
args = argv[1:]
feeds_to_add = []
try:
while True:
@ -241,7 +262,7 @@ class RSSNotifier:
'--expire-time', '0',
'--app-name', 'RSS Notifier',
*ACTIONS,
f'New RSS Story: {entry.title}'
f'New RSS Story from "{entry.feed.title}": {entry.title}'
],
stdout=PIPE,
stderr=PIPE
@ -261,4 +282,11 @@ class RSSNotifier:
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()