175 lines
5.9 KiB
Python
175 lines
5.9 KiB
Python
import sqlite3
|
|
from hashlib import sha256
|
|
|
|
|
|
class Database:
|
|
db = None
|
|
c = None
|
|
|
|
def __init__(self, dbfile):
|
|
self.db = sqlite3.connect(dbfile, timeout=30)
|
|
self.db.row_factory = sqlite3.Row
|
|
self.c = self.db.cursor()
|
|
|
|
def init_db(self):
|
|
with self.db:
|
|
self.c.execute(
|
|
"CREATE TABLE IF NOT EXISTS feeds "
|
|
"(url TEXT PRIMARY KEY, name TEXT COLLATE NOCASE, "
|
|
"custom_name INTEGER, site_url TEXT, "
|
|
"category TEXT COLLATE NOCASE, table_name TEXT)"
|
|
)
|
|
|
|
def insert_feed(self, data, from_opml=False):
|
|
with self.db:
|
|
table_name = "table" + sha256(data[0].encode()).hexdigest()
|
|
data.append(table_name)
|
|
self.c.execute(
|
|
f"CREATE TABLE IF NOT EXISTS {table_name} "
|
|
"(id TEXT PRIMARY KEY, date DATE, title TEXT, "
|
|
"url TEXT, content TEXT, unread INTEGER)"
|
|
)
|
|
if from_opml:
|
|
self.c.execute(
|
|
"INSERT OR IGNORE INTO feeds VALUES (?, ?, ?, ?, ?, ?)",
|
|
tuple(data),
|
|
)
|
|
else:
|
|
self.c.execute(
|
|
"INSERT OR IGNORE INTO feeds VALUES (?, NULL, NULL, NULL, NULL, ?)",
|
|
tuple(data),
|
|
)
|
|
|
|
def update_feed(self, name, category, url):
|
|
with self.db:
|
|
if name:
|
|
self.c.execute(
|
|
"UPDATE feeds SET name = ?, custom_name = 1 WHERE url = ?",
|
|
(name, url),
|
|
)
|
|
if category:
|
|
self.c.execute(
|
|
"UPDATE feeds SET category = ? WHERE url = ?",
|
|
(category, url),
|
|
)
|
|
|
|
def get_all_feeds(self):
|
|
return self.c.execute(
|
|
"SELECT * FROM feeds ORDER BY category, name, url"
|
|
).fetchall()
|
|
|
|
def get_feed_by_url(self, url):
|
|
return self.c.execute("SELECT * FROM feeds WHERE url = ?", (url,)).fetchone()
|
|
|
|
def get_feed_by_index(self, index):
|
|
return self.c.execute(
|
|
f"SELECT * FROM feeds ORDER BY category, name, url LIMIT 1 OFFSET {index}"
|
|
).fetchone()
|
|
|
|
def get_all_entries(self, table):
|
|
return self.c.execute(f"SELECT * FROM {table} ORDER BY date").fetchall()
|
|
|
|
def get_entry_by_index(self, table, index):
|
|
return self.c.execute(
|
|
f"SELECT * FROM {table} ORDER BY date LIMIT 1 OFFSET {index}"
|
|
).fetchone()
|
|
|
|
def get_entry_by_id(self, table, id):
|
|
return self.c.execute(f"SELECT * FROM {table} WHERE id = ?", (id,)).fetchone()
|
|
|
|
def get_unread_count(self, table):
|
|
return self.c.execute(
|
|
f"SELECT COUNT(unread) FROM {table} WHERE unread = 1"
|
|
).fetchone()[0]
|
|
|
|
def get_unread_entries(self, table):
|
|
return self.c.execute(f"SELECT * FROM {table} WHERE unread = 1").fetchall()
|
|
|
|
def get_num_feeds(self):
|
|
return self.c.execute("SELECT COUNT(1) FROM feeds").fetchone()[0]
|
|
|
|
def get_num_entries(self, table):
|
|
return self.c.execute(f"SELECT COUNT(1) FROM {table}").fetchone()[0]
|
|
|
|
def search_feeds(self, query):
|
|
return self.c.execute(
|
|
"SELECT * FROM feeds WHERE url LIKE ? OR name LIKE ? ORDER BY category, name, url",
|
|
(query, query),
|
|
).fetchall()
|
|
|
|
def search_entries(self, table, query):
|
|
return self.c.execute(
|
|
f"SELECT * FROM {table} WHERE "
|
|
"date LIKE ? OR title LIKE ? OR url LIKE ? ORDER BY date ASC",
|
|
(query, query, query),
|
|
).fetchall()
|
|
|
|
def feed_missing_name(self, url):
|
|
if self.c.execute(
|
|
"SELECT url FROM feeds WHERE url = ? AND name IS NULL", (url,)
|
|
).fetchone():
|
|
return True
|
|
return False
|
|
|
|
def feed_missing_site_url(self, url):
|
|
if self.c.execute(
|
|
"SELECT url FROM feeds WHERE url = ? AND site_url IS NULL", (url,)
|
|
).fetchone():
|
|
return True
|
|
return False
|
|
|
|
def populate_feed_name(self, url, name):
|
|
with self.db:
|
|
self.c.execute(
|
|
"UPDATE feeds SET name = ?, custom_name = 0 WHERE url = ?", (name, url)
|
|
)
|
|
|
|
def populate_feed_site_url(self, url, site_url):
|
|
with self.db:
|
|
self.c.execute(
|
|
"UPDATE feeds SET site_url = ? WHERE url = ?", (site_url, url)
|
|
)
|
|
|
|
def remove_feeds(self, urls, tables):
|
|
urls = [(url,) for url in urls]
|
|
with self.db:
|
|
self.c.executemany("DELETE FROM feeds WHERE url = ?", urls)
|
|
for table in tables:
|
|
self.c.execute(f"DROP TABLE {table}")
|
|
|
|
def change_unread_status_of_feeds(self, feeds, unread_status):
|
|
with self.db:
|
|
for feed in feeds:
|
|
self.c.execute(
|
|
f'UPDATE {feed["table_name"]} SET unread = ?', (unread_status,)
|
|
)
|
|
|
|
def change_unread_status_of_entries(self, entries, unread_status):
|
|
with self.db:
|
|
for entry in entries:
|
|
self.c.execute(
|
|
f'UPDATE {entry["table"]} SET unread = ? WHERE id = ?',
|
|
(unread_status, entry["id"]),
|
|
)
|
|
|
|
def insert_entries(self, table, entries):
|
|
with self.db:
|
|
try:
|
|
self.c.executemany(
|
|
f"INSERT INTO {table} VALUES (?, ?, ?, ?, ?, 1)", entries
|
|
)
|
|
except sqlite3.IntegrityError:
|
|
pass # This is likely just a duplicate entry i.e. unique
|
|
# constraint failed on table<n>.id which doesn't matter
|
|
|
|
def update_entries(self, table, entries):
|
|
with self.db:
|
|
self.c.executemany(
|
|
f"UPDATE {table} SET title = ?, url = ?, content = ? WHERE id = ?",
|
|
entries,
|
|
)
|
|
|
|
def vacuum(self):
|
|
with self.db:
|
|
self.c.execute("VACUUM")
|