255 lines
11 KiB
Python
255 lines
11 KiB
Python
import time
|
|
|
|
from rich.markup import escape
|
|
from textual import on, work
|
|
from textual.app import ComposeResult
|
|
from textual.binding import Binding
|
|
from textual.containers import Horizontal, VerticalScroll
|
|
from textual.widgets import DataTable, Static
|
|
|
|
import book_cache
|
|
import books
|
|
import goodreads
|
|
import goodreads_series
|
|
from tui.cache_indicator import cache_state, cache_text
|
|
from tui.content_pane import ContentPane
|
|
from tui.dialogs import ConfirmDialog, FormDialog, FormField
|
|
from tui.series_picker import SeriesPicker
|
|
|
|
REQUEST_DELAY_SECONDS = 3
|
|
|
|
|
|
def book_details(book: dict) -> str:
|
|
rows = [
|
|
("Shelf", book.get("shelf")),
|
|
("Series", " ".join(filter(None, [(book.get("series") or {}).get("name"), (book.get("series") or {}).get("position")]))),
|
|
("Your rating", f"{book['rating']}/10" if book.get("rating") is not None else None),
|
|
("Goodreads", f"{book['averageRating']} from {book.get('ratingCount') or 0:,} ratings" if book.get("averageRating") else None),
|
|
("Pages", book.get("pages")),
|
|
("Published", book.get("published")),
|
|
("Genres", ", ".join(book.get("genres") or [])),
|
|
("Added", book.get("added")),
|
|
("Cache", book_cache_state(book)),
|
|
]
|
|
lines = [f"[b]{escape(book.get('title', ''))}[/b]", escape(", ".join(book.get("authors") or [])), ""]
|
|
lines += [f"[dim]{label:<12}[/dim]{escape(str(value))}" for label, value in rows if value]
|
|
if book.get("description"):
|
|
lines += ["", escape(book["description"])]
|
|
lines += ["", f"[dim]{escape(book.get('url', ''))}[/dim]"]
|
|
return "\n".join(lines)
|
|
|
|
|
|
def book_cache_state(book: dict) -> str:
|
|
has_details = bool(book.get("authors")) or bool(book.get("description"))
|
|
return cache_state(has_details, book_cache.is_local_cover(book.get("cover")))
|
|
|
|
|
|
class BooksPane(ContentPane):
|
|
BINDINGS = [
|
|
Binding("a", "add", "Add"),
|
|
Binding("S", "add_series", "Add series"),
|
|
Binding("e", "rate", "Rate"),
|
|
Binding("s", "shelf", "Shelf"),
|
|
Binding("d", "delete", "Delete"),
|
|
Binding("u", "refresh", "Refresh"),
|
|
Binding("U", "refresh_all", "Refresh all"),
|
|
Binding("C", "cache_covers", "Cache covers"),
|
|
Binding("r", "reload", "Reload"),
|
|
]
|
|
|
|
content_file = "books.json"
|
|
|
|
def compose(self) -> ComposeResult:
|
|
with Horizontal(classes="split-layout"):
|
|
yield DataTable(cursor_type="row", zebra_stripes=True)
|
|
with VerticalScroll(classes="detail-scroll"):
|
|
yield Static("Add a book with [b]a[/b] and a Goodreads link.", id="book-detail")
|
|
|
|
def on_mount(self):
|
|
table = self.table()
|
|
table.add_column("Shelf", key="shelf")
|
|
table.add_column("Rating", key="rating")
|
|
table.add_column("Title", key="title")
|
|
table.add_column("Authors", key="authors")
|
|
table.add_column("Cache", key="cache")
|
|
self.reload()
|
|
|
|
def book_by_id(self, identifier: str | None) -> dict | None:
|
|
return next((book for book in books.load_books() if book.get("goodreadsId") == identifier), None)
|
|
|
|
def reload(self, focus_key: str | None = None):
|
|
table = self.table()
|
|
table.clear()
|
|
shelf_order = {shelf: index for index, shelf in enumerate(books.SHELVES)}
|
|
sorted_books = sorted(books.load_books(), key=lambda book: (shelf_order.get(book.get("shelf"), 99), -(book.get("rating") or -1)))
|
|
for book in sorted_books:
|
|
rating = f"{book['rating']}/10" if book.get("rating") is not None else "-"
|
|
authors = ", ".join(book.get("authors") or [])
|
|
table.add_row(book.get("shelf", ""), rating, book.get("title", ""), authors, cache_text(book_cache_state(book)), key=book["goodreadsId"])
|
|
self.restore_cursor(focus_key)
|
|
self.show_details(self.selected_key())
|
|
|
|
def show_details(self, identifier: str | None):
|
|
book = self.book_by_id(identifier)
|
|
detail = self.query_one("#book-detail", Static)
|
|
detail.update(book_details(book) if book else "Add a book with [b]a[/b] and a Goodreads link.")
|
|
|
|
@on(DataTable.RowHighlighted)
|
|
def highlight_changed(self, event: DataTable.RowHighlighted):
|
|
self.show_details(event.row_key.value if event.row_key else None)
|
|
|
|
def action_add(self):
|
|
fields = [
|
|
FormField("link", "Goodreads link", placeholder="https://www.goodreads.com/book/show/170448.Animal_Farm", required=True),
|
|
FormField("shelf", "Shelf", kind="select", options=books.SHELVES, value="read"),
|
|
FormField("rating", "Your rating (0 to 10, optional)", kind="rating"),
|
|
]
|
|
self.app.push_screen(FormDialog("Add a book", fields, "Add"), self.add_book)
|
|
|
|
def add_book(self, values: dict | None):
|
|
if not values:
|
|
return
|
|
|
|
def added(book: dict):
|
|
self.reload(book["goodreadsId"])
|
|
self.notify(f"Added {book['title']} to {book['shelf']}")
|
|
|
|
self.run_in_background("Reading the book from Goodreads...", lambda: books.add_book(values["link"], values["shelf"], values["rating"]), added)
|
|
|
|
def action_rate(self):
|
|
book = self.book_by_id(self.selected_key())
|
|
if not book:
|
|
return
|
|
current = "" if book.get("rating") is None else str(book["rating"])
|
|
fields = [FormField("rating", f"Rating for {book['title']} (empty to clear)", kind="rating", value=current)]
|
|
|
|
def rated(values: dict | None):
|
|
if values is not None:
|
|
self.run_now(lambda: books.rate_book(book["goodreadsId"], values["rating"]), f"Updated {book['title']}", book["goodreadsId"])
|
|
|
|
self.app.push_screen(FormDialog("Change rating", fields), rated)
|
|
|
|
def action_shelf(self):
|
|
book = self.book_by_id(self.selected_key())
|
|
if not book:
|
|
return
|
|
fields = [FormField("shelf", f"Shelf for {book['title']}", kind="select", options=books.SHELVES, value=book.get("shelf", "read"))]
|
|
|
|
def moved(values: dict | None):
|
|
if values:
|
|
self.run_now(lambda: books.move_book(book["goodreadsId"], values["shelf"]), f"Moved {book['title']} to {values['shelf']}", book["goodreadsId"])
|
|
|
|
self.app.push_screen(FormDialog("Move to shelf", fields, "Move"), moved)
|
|
|
|
def action_delete(self):
|
|
book = self.book_by_id(self.selected_key())
|
|
if not book:
|
|
return
|
|
|
|
def confirmed(should_delete: bool | None):
|
|
if should_delete:
|
|
self.run_now(lambda: books.remove_book(book["goodreadsId"]), f"Removed {book['title']}")
|
|
|
|
self.app.push_screen(ConfirmDialog(f"Remove {book['title']}?"), confirmed)
|
|
|
|
def action_refresh(self):
|
|
book = self.book_by_id(self.selected_key())
|
|
if not book:
|
|
return
|
|
|
|
def refreshed(updated: dict):
|
|
self.reload(updated["goodreadsId"])
|
|
self.notify(f"Refreshed {updated['title']}")
|
|
|
|
self.run_in_background(f"Refreshing {book['title']}...", lambda: books.refresh_book(book["goodreadsId"]), refreshed)
|
|
|
|
def action_refresh_all(self):
|
|
self.notify("Refreshing every book from Goodreads...")
|
|
self.refresh_every_book()
|
|
|
|
def report_blocked(self, error: Exception):
|
|
self.app.call_from_thread(self.notify, str(error), title="Stopped", severity="error", timeout=15)
|
|
|
|
@work(thread=True, exclusive=True, group="refresh")
|
|
def refresh_every_book(self):
|
|
refreshed = 0
|
|
all_books = books.load_books()
|
|
for index, book in enumerate(all_books):
|
|
if index:
|
|
time.sleep(REQUEST_DELAY_SECONDS)
|
|
try:
|
|
books.refresh_book(book["goodreadsId"])
|
|
except goodreads.BlockedError as error:
|
|
self.report_blocked(error)
|
|
break
|
|
except (OSError, ValueError) as error:
|
|
self.app.call_from_thread(self.notify, f"{book['title']}: {error}", title="Skipped", severity="warning")
|
|
continue
|
|
refreshed += 1
|
|
self.app.call_from_thread(lambda: self.reload(self.selected_key()))
|
|
self.app.call_from_thread(self.notify, f"Refreshed {refreshed} of {len(all_books)} books")
|
|
|
|
def action_cache_covers(self):
|
|
missing = books.books_missing_local_covers()
|
|
if not missing:
|
|
self.notify("Every cover is already saved in static/covers/books")
|
|
return
|
|
self.notify(f"Saving {len(missing)} covers...")
|
|
self.cache_missing_covers(missing)
|
|
|
|
@work(thread=True, exclusive=True, group="covers")
|
|
def cache_missing_covers(self, missing: list[dict]):
|
|
saved = 0
|
|
for book in missing:
|
|
try:
|
|
books.cache_book_cover(book["goodreadsId"])
|
|
except (OSError, ValueError) as error:
|
|
self.app.call_from_thread(self.notify, f"{book['title']}: {error}", title="Skipped", severity="warning")
|
|
continue
|
|
saved += 1
|
|
self.app.call_from_thread(lambda: self.reload(self.selected_key()))
|
|
self.app.call_from_thread(self.notify, f"Saved {saved} of {len(missing)} covers")
|
|
|
|
def action_add_series(self):
|
|
fields = [FormField("link", "Goodreads series link", placeholder="https://www.goodreads.com/series/92037-wings-of-fire", required=True)]
|
|
self.app.push_screen(FormDialog("Add books from a series", fields, "Load series"), self.load_series)
|
|
|
|
def load_series(self, values: dict | None):
|
|
if values:
|
|
self.run_in_background("Reading the series from Goodreads...", lambda: goodreads_series.scrape_series(values["link"]), self.pick_series_books)
|
|
|
|
def pick_series_books(self, series: dict):
|
|
def picked(choices: dict[str, str] | None):
|
|
if not choices:
|
|
return
|
|
self.notify(f"Adding {len(choices)} books from {series['name']}...")
|
|
self.add_series_books(series, choices)
|
|
|
|
self.app.push_screen(SeriesPicker(series, books.library_shelves()), picked)
|
|
|
|
@work(thread=True, exclusive=True, group="series")
|
|
def add_series_books(self, series: dict, choices: dict[str, str]):
|
|
added = 0
|
|
already_listed = set(books.library_shelves())
|
|
fetched_any = False
|
|
for entry in series["books"]:
|
|
shelf = choices.get(entry["goodreadsId"])
|
|
if not shelf:
|
|
continue
|
|
needs_fetch = entry["goodreadsId"] not in already_listed
|
|
if needs_fetch and fetched_any:
|
|
time.sleep(REQUEST_DELAY_SECONDS)
|
|
fetched_any = fetched_any or needs_fetch
|
|
try:
|
|
book = books.add_series_book(series, entry, shelf)
|
|
except goodreads.BlockedError as error:
|
|
self.report_blocked(error)
|
|
break
|
|
except (OSError, ValueError) as error:
|
|
self.app.call_from_thread(self.notify, f"{entry['title']}: {error}", title="Skipped", severity="warning")
|
|
continue
|
|
added += 1
|
|
self.app.call_from_thread(self.notify, f"Saved {book['title']} ({shelf})")
|
|
self.app.call_from_thread(lambda: self.reload(self.selected_key()))
|
|
self.app.call_from_thread(self.notify, f"Saved {added} of {len(choices)} books from {series['name']}")
|