Add Goodreads series and group them on the about page

tools/content/books.py

@@ -73,6 +73,30 @@ def update_book(text: str, change) -> dict:
     return books[index]
 
 
+def series_tag(series: dict, entry: dict) -> dict:
+    tag = {"name": series["name"], "url": series["url"]}
+    if entry.get("position"):
+        tag["position"] = entry["position"]
+    return tag
+
+
+def add_series_book(series: dict, entry: dict, shelf: str) -> dict:
+    check_shelf(shelf)
+    tag = series_tag(series, entry)
+    books = load_books()
+    index = find_index(books, entry["goodreadsId"])
+    if index is not None:
+        books[index] = {**books[index], "shelf": shelf, "series": tag}
+        save_books(books)
+        return books[index]
+    book = add_book(entry["url"], shelf)
+    return update_book(book["goodreadsId"], lambda stored: {**stored, "series": tag})
+
+
+def library_shelves() -> dict[str, str]:
+    return {book["goodreadsId"]: book.get("shelf", "read") for book in load_books()}
+
+
 def add_book(url: str, shelf: str = "read", rating: int | None = None) -> dict:
     check_shelf(shelf)
     if rating is not None:

tools/content/goodreads.py

@@ -214,8 +214,10 @@ def print_summary(book: dict):
 
 
 def main() -> int:
-    parser = argparse.ArgumentParser(description="Scrape book info from a Goodreads book page")
-    parser.add_argument("urls", nargs="+", help="Goodreads book urls, for example https://www.goodreads.com/book/show/170448.Animal_Farm")
+    import goodreads_series
+
+    parser = argparse.ArgumentParser(description="Scrape book or series info from Goodreads")
+    parser.add_argument("urls", nargs="+", help="Goodreads book or series urls, for example https://www.goodreads.com/book/show/170448.Animal_Farm")
     parser.add_argument("--json", action="store_true", help="print the raw result as JSON")
     parser.add_argument("--debug", action="store_true", help="show what the downloaded page contains instead of parsing it")
     parser.add_argument("--save-html", type=Path, help="with --debug, save the downloaded page to this file")
@@ -223,19 +225,24 @@ def main() -> int:
 
     exit_code = 0
     for url in arguments.urls:
+        is_series = goodreads_series.series_id(url) is not None
         try:
             if arguments.debug:
                 print_diagnostics(url, arguments.save_html)
+                print(f"React data {len(goodreads_series.REACT_PROPS_PATTERN.findall(download_page(url).text))} blocks")
                 continue
-            book = scrape_book(url)
+            result = goodreads_series.scrape_series(url) if is_series else scrape_book(url)
         except (OSError, ValueError) as error:
             print(f"Error for {url}: {error}", file=sys.stderr)
             exit_code = 1
             continue
         if arguments.json:
-            print(json.dumps(book, indent="\t", ensure_ascii=False))
+            print(json.dumps(result, indent="\t", ensure_ascii=False))
+        elif is_series:
+            goodreads_series.print_series(result)
+            print()
         else:
-            print_summary(book)
+            print_summary(result)
             print()
     return exit_code
 

tools/content/goodreads_series.py

@@ -0,0 +1,169 @@
+import html
+import json
+import re
+
+import goodreads
+
+SERIES_URL_PATTERN = re.compile(r"goodreads\.com/series/(\d+)")
+REACT_PROPS_PATTERN = re.compile(r"data-react-props=\"([^\"]*)\"", re.S)
+BOOK_LINK_PATTERN = re.compile(r"<a[^>]+href=\"(?:https://www\.goodreads\.com)?/book/show/(\d+)[^\"]*\"[^>]*>\s*(?:<span[^>]*>)?\s*([^<]+?)\s*<", re.S)
+SERIES_TITLE_SUFFIX = re.compile(r"\s+(?:Series\s+)?by\s+.*$", re.I)
+
+
+def series_id(url: str) -> str | None:
+    match = SERIES_URL_PATTERN.search(url)
+    return match.group(1) if match else None
+
+
+def first_text(data: dict, keys: list[str]) -> str:
+    for key in keys:
+        value = data.get(key)
+        if isinstance(value, str) and value.strip():
+            return html.unescape(value.strip())
+    return ""
+
+
+def author_names(book: dict) -> list[str]:
+    author = book.get("author")
+    if isinstance(author, dict) and author.get("name"):
+        return [html.unescape(author["name"])]
+    authors = book.get("authors")
+    if isinstance(authors, list):
+        return [html.unescape(item["name"]) for item in authors if isinstance(item, dict) and item.get("name")]
+    name = first_text(book, ["authorName"])
+    return [name] if name else []
+
+
+def entry_from_book(book: dict, position: str) -> dict | None:
+    identifier = str(book.get("bookId") or book.get("legacyId") or book.get("id") or "")
+    link = first_text(book, ["bookUrl", "url", "webUrl"])
+    if not identifier.isdigit():
+        identifier = goodreads.book_id(link) or ""
+    if not identifier:
+        return None
+    return {
+        "goodreadsId": identifier,
+        "title": first_text(book, ["bookTitleBare", "title", "titleComplete"]),
+        "position": position,
+        "authors": author_names(book),
+        "cover": first_text(book, ["imageUrl", "image"]) or None,
+        "url": f"https://www.goodreads.com/book/show/{identifier}",
+    }
+
+
+def clean_header(value) -> str:
+    return html.unescape(value).strip() if isinstance(value, str) else ""
+
+
+def own_header(item: dict) -> str:
+    return next((clean_header(value) for key, value in item.items() if "header" in key.lower() and isinstance(value, str)), "")
+
+
+def is_book_list(value) -> bool:
+    return isinstance(value, list) and bool(value) and all(isinstance(item, dict) and isinstance(item.get("book"), dict) for item in value)
+
+
+def collect_entries(node, found: list[dict]):
+    if isinstance(node, list):
+        for item in node:
+            collect_entries(item, found)
+        return
+    if not isinstance(node, dict):
+        return
+
+    if isinstance(node.get("book"), dict):
+        entry = entry_from_book(node["book"], own_header(node))
+        if entry:
+            found.append(entry)
+        return
+
+    header_lists = [value for key, value in node.items() if "header" in key.lower() and isinstance(value, list)]
+    for value in node.values():
+        if not is_book_list(value):
+            collect_entries(value, found)
+            continue
+        headers = next((headers for headers in header_lists if len(headers) == len(value)), [])
+        for index, item in enumerate(value):
+            position = clean_header(headers[index]) if index < len(headers) else ""
+            entry = entry_from_book(item["book"], position or own_header(item))
+            if entry:
+                found.append(entry)
+
+
+def entries_from_react_props(page: str) -> list[dict]:
+    found: list[dict] = []
+    for raw in REACT_PROPS_PATTERN.findall(page):
+        try:
+            collect_entries(json.loads(html.unescape(raw)), found)
+        except ValueError:
+            continue
+    return found
+
+
+def entries_from_next_data(page: str) -> list[dict]:
+    state = goodreads.apollo_state(page)
+    books = [value for value in state.values() if isinstance(value, dict) and value.get("__typename") == "Book"]
+    entries = []
+    for book in books:
+        entry = entry_from_book(book, "")
+        if entry:
+            entry["authors"] = goodreads.book_authors(state, book) or entry["authors"]
+            entries.append(entry)
+    return entries
+
+
+def entries_from_links(page: str) -> list[dict]:
+    entries = []
+    for identifier, title in BOOK_LINK_PATTERN.findall(page):
+        title = html.unescape(title).strip()
+        if title:
+            entries.append({"goodreadsId": identifier, "title": title, "position": "", "authors": [], "cover": None, "url": f"https://www.goodreads.com/book/show/{identifier}"})
+    return entries
+
+
+def unique_entries(entries: list[dict]) -> list[dict]:
+    seen = set()
+    unique = []
+    for entry in entries:
+        if entry["goodreadsId"] in seen:
+            continue
+        seen.add(entry["goodreadsId"])
+        unique.append(entry)
+    return unique
+
+
+def series_title(page: str) -> str:
+    meta = goodreads.meta_tags(page)
+    title = meta.get("og:title") or goodreads.page_title(page)
+    return SERIES_TITLE_SUFFIX.sub("", title).strip()
+
+
+def scrape_series(url: str) -> dict:
+    identifier = series_id(url)
+    if not identifier:
+        raise ValueError(f"not a Goodreads series url: {url}")
+    page = goodreads.download_page(url)
+    reason = goodreads.blocked_reason(page.text)
+    if reason:
+        raise ValueError(reason)
+
+    entries = entries_from_react_props(page.text) or entries_from_next_data(page.text) or entries_from_links(page.text)
+    entries = unique_entries(entries)
+    if not entries:
+        raise ValueError("Could not find any books on the series page")
+
+    return {
+        "seriesId": identifier,
+        "name": series_title(page.text),
+        "url": f"https://www.goodreads.com/series/{identifier}",
+        "books": entries,
+    }
+
+
+def print_series(series: dict):
+    print(f"{series['name']}  ({len(series['books'])} books)")
+    print("-" * 60)
+    for entry in series["books"]:
+        position = entry["position"] or "-"
+        authors = ", ".join(entry["authors"]) or "unknown author"
+        print(f"{position:<12} {entry['title']}  by {authors}  [{entry['goodreadsId']}]")

tools/content/tui/app.tcss

@@ -250,3 +250,19 @@ RecordEditor {
     width: 100%;
     margin: 0 0 1 0;
 }
+
+SeriesPicker {
+    background: $background;
+}
+
+#series-table {
+    height: 1fr;
+    margin: 1 2 0 2;
+    background: $surface;
+    border: solid $primary;
+}
+
+.series-summary {
+    width: 100%;
+    padding: 0 2;
+}

tools/content/tui/books_pane.py

@@ -6,13 +6,16 @@ from textual.containers import Horizontal, VerticalScroll
 from textual.widgets import DataTable, Static
 
 import books
+import goodreads_series
 from tui.content_pane import ContentPane
 from tui.dialogs import ConfirmDialog, FormDialog, FormField
+from tui.series_picker import SeriesPicker
 
 
 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")),
@@ -31,6 +34,7 @@ def book_details(book: dict) -> str:
 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"),
@@ -160,3 +164,37 @@ class BooksPane(ContentPane):
         self.app.call_from_thread(lambda: self.reload(self.selected_key()))
         summary = f"Refreshed {len(all_books) - failures} of {len(all_books)} books"
         self.app.call_from_thread(self.notify, summary)
+
+    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
+        for entry in series["books"]:
+            shelf = choices.get(entry["goodreadsId"])
+            if not shelf:
+                continue
+            try:
+                book = books.add_series_book(series, entry, shelf)
+            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']}")

tools/content/tui/series_picker.py

@@ -0,0 +1,97 @@
+from rich.text import Text
+from textual.app import ComposeResult
+from textual.binding import Binding
+from textual.screen import Screen
+from textual.widgets import DataTable, Footer, Label
+
+SKIP = "skip"
+CHOICE_STYLES = {"read": "green", "reading": "blue", "want-to-read": "magenta", SKIP: "dim"}
+CHOICE_ORDER = ["read", "reading", "want-to-read", SKIP]
+
+
+class SeriesPicker(Screen[dict[str, str] | None]):
+    BINDINGS = [
+        Binding("r", "choose('read')", "Read"),
+        Binding("c", "choose('reading')", "Reading"),
+        Binding("w", "choose('want-to-read')", "Want to read"),
+        Binding("x", "choose('skip')", "Skip"),
+        Binding("space", "cycle", "Cycle"),
+        Binding("R", "choose_all('read')", "All read"),
+        Binding("W", "choose_all('want-to-read')", "All want"),
+        Binding("ctrl+s", "save", "Save"),
+        Binding("escape", "cancel", "Cancel"),
+    ]
+
+    def __init__(self, series: dict, current_shelves: dict[str, str]):
+        super().__init__()
+        self.series = series
+        self.choices = {entry["goodreadsId"]: current_shelves.get(entry["goodreadsId"], SKIP) for entry in series["books"]}
+        self.already_listed = set(current_shelves)
+
+    def compose(self) -> ComposeResult:
+        yield Label(f"{self.series['name']}  [dim]{len(self.series['books'])} books, pick a shelf for each one[/dim]", classes="editor-title")
+        yield DataTable(cursor_type="row", zebra_stripes=True, id="series-table")
+        yield Label("", id="series-summary", classes="series-summary")
+        yield Footer()
+
+    def on_mount(self):
+        table = self.query_one(DataTable)
+        table.add_column("Shelf", key="shelf")
+        table.add_column("#", key="position")
+        table.add_column("Title", key="title")
+        table.add_column("Author", key="author")
+        table.add_column("", key="listed")
+        for entry in self.series["books"]:
+            identifier = entry["goodreadsId"]
+            listed = "in library" if identifier in self.already_listed else ""
+            table.add_row(self.choice_text(identifier), entry["position"] or "-", entry["title"], ", ".join(entry["authors"]), listed, key=identifier)
+        table.focus()
+        self.update_summary()
+
+    def choice_text(self, identifier: str) -> Text:
+        choice = self.choices[identifier]
+        return Text(choice, style=CHOICE_STYLES[choice])
+
+    def update_summary(self):
+        counts = {choice: list(self.choices.values()).count(choice) for choice in CHOICE_ORDER}
+        parts = [f"[{CHOICE_STYLES[choice]}]{counts[choice]} {choice}[/]" for choice in CHOICE_ORDER]
+        self.query_one("#series-summary", Label).update("  ".join(parts) + "   [dim]ctrl+s to save[/dim]")
+
+    def selected_id(self) -> str | None:
+        table = self.query_one(DataTable)
+        if table.row_count == 0:
+            return None
+        row_key, _ = table.coordinate_to_cell_key(table.cursor_coordinate)
+        return row_key.value
+
+    def set_choice(self, identifier: str, choice: str):
+        self.choices[identifier] = choice
+        self.query_one(DataTable).update_cell(identifier, "shelf", self.choice_text(identifier))
+
+    def action_choose(self, choice: str):
+        identifier = self.selected_id()
+        if not identifier:
+            return
+        self.set_choice(identifier, choice)
+        self.update_summary()
+        self.query_one(DataTable).action_cursor_down()
+
+    def action_cycle(self):
+        identifier = self.selected_id()
+        if not identifier:
+            return
+        next_choice = CHOICE_ORDER[(CHOICE_ORDER.index(self.choices[identifier]) + 1) % len(CHOICE_ORDER)]
+        self.set_choice(identifier, next_choice)
+        self.update_summary()
+
+    def action_choose_all(self, choice: str):
+        for identifier in self.choices:
+            self.set_choice(identifier, choice)
+        self.update_summary()
+
+    def action_save(self):
+        picked = {identifier: choice for identifier, choice in self.choices.items() if choice != SKIP}
+        self.dismiss(picked)
+
+    def action_cancel(self):
+        self.dismiss(None)