diff options
Diffstat (limited to 'src/epy_reader/state.py')
| -rw-r--r-- | src/epy_reader/state.py | 195 |
1 files changed, 195 insertions, 0 deletions
diff --git a/src/epy_reader/state.py b/src/epy_reader/state.py new file mode 100644 index 0000000..5129394 --- /dev/null +++ b/src/epy_reader/state.py @@ -0,0 +1,195 @@ +import dataclasses +import hashlib +import os +import sqlite3 +from datetime import datetime +from typing import List, Tuple + +from epy_reader.ebooks import Ebook +from epy_reader.models import AppData, LibraryItem, Optional, ReadingState + + +class State(AppData): + """ + Use sqlite3 instead of JSON (in older version) + to shift the weight from memory to process + """ + + def __init__(self): + if not os.path.isfile(self.filepath): + self.init_db() + + @property + def filepath(self) -> str: + return os.path.join(self.prefix, "states.db") if self.prefix else os.devnull + + def get_from_history(self) -> List[LibraryItem]: + try: + conn = sqlite3.connect(self.filepath) + cur = conn.cursor() + cur.execute( + """ + SELECT last_read, filepath, title, author, reading_progress + FROM library ORDER BY last_read DESC + """ + ) + results = cur.fetchall() + library_items: List[LibraryItem] = [] + for result in results: + library_items.append( + LibraryItem( + last_read=datetime.fromisoformat(result[0]), + filepath=result[1], + title=result[2], + author=result[3], + reading_progress=result[4], + ) + ) + return library_items + finally: + conn.close() + + def delete_from_library(self, filepath: str) -> None: + try: + conn = sqlite3.connect(self.filepath) + conn.execute("PRAGMA foreign_keys = ON") + conn.execute("DELETE FROM reading_states WHERE filepath=?", (filepath,)) + conn.commit() + finally: + conn.close() + + def get_last_read(self) -> Optional[str]: + library = self.get_from_history() + return library[0].filepath if library else None + + def update_library(self, ebook: Ebook, reading_progress: Optional[float]) -> None: + try: + metadata = ebook.get_meta() + conn = sqlite3.connect(self.filepath) + conn.execute( + """ + INSERT OR REPLACE INTO library (filepath, title, author, reading_progress) + VALUES (?, ?, ?, ?) + """, + (ebook.path, metadata.title, metadata.creator, reading_progress), + ) + conn.commit() + finally: + conn.close() + + def get_last_reading_state(self, ebook: Ebook) -> ReadingState: + try: + conn = sqlite3.connect(self.filepath) + conn.row_factory = sqlite3.Row + cur = conn.cursor() + cur.execute("SELECT * FROM reading_states WHERE filepath=?", (ebook.path,)) + result = cur.fetchone() + if result: + result = dict(result) + del result["filepath"] + return ReadingState(**result, section=None) + return ReadingState(content_index=0, textwidth=80, row=0, rel_pctg=None, section=None) + finally: + conn.close() + + def set_last_reading_state(self, ebook: Ebook, reading_state: ReadingState) -> None: + try: + conn = sqlite3.connect(self.filepath) + conn.execute( + """ + INSERT OR REPLACE INTO reading_states + VALUES (:filepath, :content_index, :textwidth, :row, :rel_pctg) + """, + {"filepath": ebook.path, **dataclasses.asdict(reading_state)}, + ) + conn.commit() + finally: + conn.close() + + def insert_bookmark(self, ebook: Ebook, name: str, reading_state: ReadingState) -> None: + try: + conn = sqlite3.connect(self.filepath) + conn.execute( + """ + INSERT INTO bookmarks + VALUES (:id, :filepath, :name, :content_index, :textwidth, :row, :rel_pctg) + """, + { + "id": hashlib.sha1(f"{ebook.path}{name}".encode()).hexdigest()[:10], + "filepath": ebook.path, + "name": name, + **dataclasses.asdict(reading_state), + }, + ) + conn.commit() + finally: + conn.close() + + def delete_bookmark(self, ebook: Ebook, name: str) -> None: + try: + conn = sqlite3.connect(self.filepath) + conn.execute("DELETE FROM bookmarks WHERE filepath=? AND name=?", (ebook.path, name)) + conn.commit() + finally: + conn.close() + + def get_bookmarks(self, ebook: Ebook) -> List[Tuple[str, ReadingState]]: + try: + conn = sqlite3.connect(self.filepath) + conn.row_factory = sqlite3.Row + cur = conn.cursor() + cur.execute("SELECT * FROM bookmarks WHERE filepath=?", (ebook.path,)) + results = cur.fetchall() + bookmarks: List[Tuple[str, ReadingState]] = [] + for result in results: + tmp_dict = dict(result) + name = tmp_dict["name"] + tmp_dict = { + k: v + for k, v in tmp_dict.items() + if k in ("content_index", "textwidth", "row", "rel_pctg") + } + bookmarks.append((name, ReadingState(**tmp_dict))) + return bookmarks + finally: + conn.close() + + def init_db(self) -> None: + try: + conn = sqlite3.connect(self.filepath) + conn.executescript( + """ + CREATE TABLE reading_states ( + filepath TEXT PRIMARY KEY, + content_index INTEGER, + textwidth INTEGER, + row INTEGER, + rel_pctg REAL + ); + + CREATE TABLE library ( + last_read DATETIME DEFAULT (datetime('now','localtime')), + filepath TEXT PRIMARY KEY, + title TEXT, + author TEXT, + reading_progress REAL, + FOREIGN KEY (filepath) REFERENCES reading_states(filepath) + ON DELETE CASCADE + ); + + CREATE TABLE bookmarks ( + id TEXT PRIMARY KEY, + filepath TEXT, + name TEXT, + content_index INTEGER, + textwidth INTEGER, + row INTEGER, + rel_pctg REAL, + FOREIGN KEY (filepath) REFERENCES reading_states(filepath) + ON DELETE CASCADE + ); + """ + ) + conn.commit() + finally: + conn.close() |