| | 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 |
| |
|
