|
import sqlite3 |
|
from pathlib import Path |
|
from typing import List, Tuple |
|
import json |
|
|
|
|
|
class Database: |
|
def __init__(self, db_path=None): |
|
if db_path is None: |
|
raise ValueError("db_path must be provided") |
|
self.db_path = db_path |
|
self.db_file = self.db_path / "cache.db" |
|
if not self.db_file.exists(): |
|
print("Creating database") |
|
print("DB_FILE", self.db_file) |
|
db = sqlite3.connect(self.db_file) |
|
with open(Path("schema.sql"), "r") as f: |
|
db.executescript(f.read()) |
|
db.commit() |
|
db.close() |
|
|
|
def get_db(self): |
|
db = sqlite3.connect(self.db_file, check_same_thread=False) |
|
db.row_factory = sqlite3.Row |
|
return db |
|
|
|
def __enter__(self): |
|
self.db = self.get_db() |
|
return self.db |
|
|
|
def __exit__(self, exc_type, exc_value, traceback): |
|
self.db.close() |
|
|
|
def __call__(self): |
|
return self |
|
|
|
def insert(self, data: List[Tuple[str, str, str]]): |
|
with self() as db: |
|
cursor = db.cursor() |
|
try: |
|
for entry in data: |
|
url, title, entries = entry |
|
cursor.execute( |
|
"INSERT OR REPLACE INTO cache (url, title, entries) VALUES (?, ?, ?)", |
|
(url, title, entries), |
|
) |
|
except Exception as e: |
|
print(e) |
|
db.commit() |
|
|
|
def filter(self, category: str): |
|
with self() as db: |
|
entries = db.execute("SELECT url, title, entries FROM cache").fetchall() |
|
out = [] |
|
for row in entries: |
|
|
|
data = json.loads(row["entries"]) |
|
try: |
|
data = [ |
|
{ |
|
"title": entry["title"], |
|
"link": entry["link"], |
|
"published": entry["published"] |
|
if "published" in entry |
|
else entry["pubDate"] |
|
if "pubDate" in entry |
|
else "", |
|
"summary": entry["summary"] if "summary" in entry else "", |
|
} |
|
for entry in data["entries"] |
|
] |
|
if len(data) > 0: |
|
out.append( |
|
{ |
|
"entries": data, |
|
"url": row["url"], |
|
"title": row["title"], |
|
} |
|
) |
|
except Exception as e: |
|
print(f"Errro on {row['url']}: {e}") |
|
return out |
|
|
