diff --git a/bindery/calibre_db.py b/bindery/calibre_db.py new file mode 100644 index 0000000..953b8c1 --- /dev/null +++ b/bindery/calibre_db.py @@ -0,0 +1,83 @@ +"""Optional, read-only access to Calibre-Web accounts and the calibre library.""" + +import sqlite3 +from collections.abc import Iterator +from contextlib import closing, contextmanager +from pathlib import Path +from typing import Any + +from werkzeug.security import check_password_hash + +# Calibre-Web marks anonymous accounts with this role bit. +ROLE_ANONYMOUS = 1 << 5 + + +class DatabaseUnavailable(Exception): + """The database cannot answer. The caller can try CWA over HTTP.""" + + +@contextmanager +def _connect(path: Path) -> Iterator[sqlite3.Connection]: + try: + # Do not use immutable=1. CWA can change the database and its WAL at any time. + with closing(sqlite3.connect(path.resolve().as_uri() + "?mode=ro", uri=True, timeout=1)) as db: + db.row_factory = sqlite3.Row + db.execute("PRAGMA query_only = ON") + db.execute("BEGIN") + yield db + except (OSError, sqlite3.Error) as e: + raise DatabaseUnavailable(f"Cannot read {path}: {e}") from e + + +def login_ok(path: Path, username: str, password: str) -> bool: + """False means rejected credentials, not a reason to try another login source.""" + with _connect(path) as db: + users = db.execute( + "SELECT name, password, role FROM user WHERE name = ? COLLATE NOCASE LIMIT 2", (username,) + ).fetchall() + if len(users) != 1 or not password: + return False + user = users[0] + if not isinstance(user["role"], int): + raise DatabaseUnavailable("The Calibre-Web account has no valid role") + if user["name"].casefold() == "guest" or user["role"] & ROLE_ANONYMOUS or not user["password"]: + return False + if not isinstance(user["password"], str): + raise DatabaseUnavailable("The Calibre-Web password hash is not text") + try: + return check_password_hash(user["password"], password) + except ValueError as e: + raise DatabaseUnavailable("The Calibre-Web password hash is not supported") from e + + +def _book(db: sqlite3.Connection, rows: list[sqlite3.Row]) -> dict[str, Any] | None: + # A duplicate UUID or story identifier is not enough to choose an update target. + if len(rows) != 1 or not rows[0]["uuid"]: + return None + book = dict(rows[0]) + book["authors"] = [ + row[0] + for row in db.execute( + "SELECT a.name FROM authors a JOIN books_authors_link l ON l.author = a.id WHERE l.book = ? ORDER BY l.id", + (book["id"],), + ) + ] + book["formats"] = [row[0] for row in db.execute("SELECT format FROM data WHERE book = ?", (book["id"],))] + return book + + +def book_by_uuid(path: Path, book_uuid: str) -> dict[str, Any] | None: + with _connect(path) as db: + rows = db.execute("SELECT id, uuid, title FROM books WHERE uuid = ? LIMIT 2", (book_uuid,)).fetchall() + return _book(db, rows) + + +def book_by_url(path: Path, story_url: str) -> dict[str, Any] | None: + """Match an exact FanFicFare URL identifier. Other URLs use the OPDS search.""" + with _connect(path) as db: + rows = db.execute( + "SELECT DISTINCT b.id, b.uuid, b.title FROM books b JOIN identifiers i ON i.book = b.id " + "WHERE i.type = 'url' COLLATE NOCASE AND i.val = ? COLLATE BINARY LIMIT 2", + (story_url,), + ).fetchall() + return _book(db, rows) diff --git a/bindery/cwa.py b/bindery/cwa.py new file mode 100644 index 0000000..e488fb7 --- /dev/null +++ b/bindery/cwa.py @@ -0,0 +1,309 @@ +"""CWA's OPDS feed, over HTTP. + +An optional read-only metadata.db supplies book IDs. Without a database match, +Bindery uses CWA over HTTP. OPDS does not show identifiers, so Bindery searches +by title, then reads the story URL inside each candidate epub. +""" + +import logging +import re +import threading +import uuid +from collections.abc import Callable +from dataclasses import dataclass, field +from difflib import SequenceMatcher +from pathlib import Path +from typing import Any +from urllib.parse import urljoin + +import requests +from defusedxml import ElementTree + +from bindery import calibre_db, config +from bindery.epub import EpubError, epub_urls + +logger = logging.getLogger(__name__) + +ATOM = "{http://www.w3.org/2005/Atom}" +ACQUISITION = "http://opds-spec.org/acquisition" + +# The most epubs that one search downloads to find a story. +MAX_DOWNLOADS = 5 +MAX_PAGES = 100 + + +class CwaError(Exception): + pass + + +class CwaLoginError(CwaError): + """CWA refused the user name or password.""" + + +@dataclass +class Book: + """A book as the OPDS feed shows it.""" + + id: int + uuid: str + title: str + authors: list[str] = field(default_factory=list) + publisher: str = "" + formats: dict[str, str] = field(default_factory=dict) # "epub" -> download href + + def to_dict(self) -> dict[str, Any]: + return { + "id": self.id, + "uuid": self.uuid, + "title": self.title, + "authors": " & ".join(self.authors), + "publisher": self.publisher, + "formats": sorted(self.formats), + } + + +Normalizer = Callable[[str], tuple[str, str] | None] + + +class CwaClient: + def __init__(self, base_url: str, username: str = "", password: str = "", timeout: int = 60) -> None: + self.base_url = base_url.strip().rstrip("/") + "/" if base_url and base_url.strip() else "" + self.auth = (username, password or "") if username else None + self.timeout = timeout + self.session = requests.Session() + + def available(self) -> bool: + return bool(self.base_url) + + def _get(self, href: str, auth: Any = None, **kwargs: Any) -> requests.Response: + if not self.available(): + raise CwaError("CWA address is not set. Set CWA_URL.") + # hrefs from the feed are absolute paths, and include any path prefix of CWA. + url = urljoin(self.base_url, href) + try: + response = self.session.get(url, auth=auth or self.auth, timeout=self.timeout, **kwargs) + except requests.RequestException as e: + raise CwaError(f"Cannot reach CWA at {self.base_url}: {e}") from e + if response.status_code == 401: + raise CwaLoginError("CWA refused the login") + if response.status_code >= 400: + raise CwaError(f"CWA answered {response.status_code} for {url}") + return response + + def check(self) -> str: + """Log in to the OPDS feed. Returns the catalog title.""" + root = ElementTree.fromstring(self._get("opds").content) + return (root.findtext(f"{ATOM}title") or "CWA").strip() + + def login_ok(self, username: str, password: str) -> bool: + """True if CWA accepts this user name and password.""" + try: + self._get("opds", auth=(username, password)) + except CwaLoginError: + return False + return True + + def _feed(self, href: str, params: dict[str, str] | None = None) -> tuple[list[Book], list[tuple[str, str]]]: + """Books and navigation entries (title, href) of an OPDS feed, with all its pages.""" + books: list[Book] = [] + navigation: list[tuple[str, str]] = [] + for _ in range(MAX_PAGES): + try: + root = ElementTree.fromstring(self._get(href, params=params).content) + except ElementTree.ParseError as e: + raise CwaError(f"CWA sent invalid OPDS XML: {e}") from e + for entry in root.findall(f"{ATOM}entry"): + title = (entry.findtext(f"{ATOM}title") or "").strip() + formats: dict[str, str] = {} + book_id, subsection = None, None + for link in entry.findall(f"{ATOM}link"): + rel, link_href = link.get("rel", ""), link.get("href", "") + if rel == ACQUISITION: + match = re.search(r"/opds/download/(\d+)/(\w+)/?$", link_href) + if match: + book_id = int(match.group(1)) + formats[match.group(2).lower()] = link_href + elif rel == "subsection": + subsection = link_href + entry_id = entry.findtext(f"{ATOM}id") or "" + if book_id is not None and entry_id.startswith("urn:uuid:"): + books.append( + Book( + id=book_id, + uuid=entry_id[len("urn:uuid:") :], + title=title, + authors=[(a.findtext(f"{ATOM}name") or "").strip() for a in entry.findall(f"{ATOM}author")], + publisher=(entry.findtext(f"{ATOM}publisher/{ATOM}name") or "").strip(), + formats=formats, + ) + ) + elif subsection: + navigation.append((title, subsection)) + next_href = next( + (link.get("href") for link in root.findall(f"{ATOM}link") if link.get("rel") == "next"), None + ) + if not next_href: + break + href, params = next_href, None + return books, navigation + + def search(self, query: str) -> list[Book]: + """CWA's search, which matches title, authors, tags, series and publisher.""" + if not query.strip(): + return [] + return self._feed("opds/search", params={"query": query.strip()})[0] + + def get_book(self, book_uuid: str) -> dict[str, Any] | None: + """A book by its calibre uuid, or None if CWA has no such book.""" + # The uuid can come from a request. Other characters could change the path. + if not re.fullmatch(r"[0-9A-Za-z-]+", book_uuid): + return None + if config.CALIBRE_DB_PATH: + try: + book = calibre_db.book_by_uuid(config.CALIBRE_DB_PATH, book_uuid) + if book: + return book + except calibre_db.DatabaseUnavailable as e: + logger.warning(f"Cannot read the calibre library database. Trying CWA: {e}") + response = self._get(f"ajax/book/{book_uuid}") + if not response.text.strip(): + return None + try: + data = response.json() + except ValueError as e: + raise CwaError(f"CWA sent invalid book record: {e}") from e + if data.get("uuid") != book_uuid: + return None + return { + "id": int(data["application_id"]), + "uuid": book_uuid, + "title": data.get("title") or "", + "authors": data.get("authors") or [], + "formats": data.get("formats") or [], + } + + def book_by_url(self, story_url: str) -> dict[str, Any] | None: + """Try the library database. A miss leaves the OPDS search to the caller.""" + if config.CALIBRE_DB_PATH: + try: + return calibre_db.book_by_url(config.CALIBRE_DB_PATH, story_url) + except calibre_db.DatabaseUnavailable as e: + logger.warning(f"Cannot read the calibre library database. Trying OPDS: {e}") + return None + + def download_epub(self, book: Book, dest: Path) -> Path: + href = book.formats.get("epub") + if not href: + raise CwaError(f"Library book {book.id} has no epub") + dest.parent.mkdir(parents=True, exist_ok=True) + with self._get(href, stream=True) as response, open(dest, "wb") as f: + for chunk in response.iter_content(chunk_size=1 << 16): + f.write(chunk) + return dest + + def site_books(self, is_site: Callable[[str], bool]) -> list[Book]: + """Books whose publisher is a story site. FanFicFare sets the publisher to the site name.""" + try: + _, publishers = self._feed("opds/publisher") + except CwaError as e: + raise CwaError(f'{e}. Turn on "Show Publisher Section" for Bindery in CWA.') from e + books: list[Book] = [] + for name, href in publishers: + if is_site(name): + books.extend(b for b in self._feed(href)[0] if "epub" in b.formats) + return books + + +# A part of a title in parentheses or brackets, such as "[Tag A, Tag B]" or "(Book 1 Complete)". +BRACKETED = re.compile(r"\([^()]*\)|\[[^\[\]]*\]") +# Characters that a title can keep at an end after its bracketed parts go. +TITLE_EDGES = " -\u2013\u2014:|,." + + +def plain_title(title: str) -> str: + """A title without its bracketed parts, in lower case, for comparisons.""" + previous = None + while previous != title: + previous, title = title, BRACKETED.sub(" ", title) + return " ".join(title.split()).strip(TITLE_EDGES).casefold() + + +def search_text(title: str) -> str: + """The text that CWA searches for. CWA finds titles that contain the text, so the text before the first + bracket finds the title with and without its bracketed parts.""" + head = re.split(r"[(\[]", title, maxsplit=1)[0].strip(TITLE_EDGES) + return head or plain_title(title) + + +def title_rank(book_title: str, titles: list[str]) -> tuple[int, float] | None: + """How close a library title is to one of the story titles. A lower rank is closer, and None is no match. + + Rank 0 is the same title. Rank 1 is the same title without the bracketed parts. Rank 2 is a title that + starts with the other title without the bracketed parts, such as "Story A Part I" for "Story A". In a + rank, the more similar text comes first. + """ + full, plain = book_title.casefold(), plain_title(book_title) + ranks = [] + for title in titles: + other = plain_title(title) + if full == title.casefold(): + rank = 0 + elif plain and plain == other: + rank = 1 + elif plain and other and (plain.startswith(other + " ") or other.startswith(plain + " ")): + rank = 2 + else: + continue + ranks.append((rank, -SequenceMatcher(None, full, title.casefold()).ratio())) + return min(ranks, default=None) + + +class BookFinder: + """Finds the library book of a story, and remembers books that are not it.""" + + def __init__(self) -> None: + self._not_story: dict[str, set[int]] = {} + self._lock = threading.Lock() + + def find( + self, + client: CwaClient, + story_url: str, + titles: list[str], + normalize: Normalizer, + work_dir: Path, + exclude_ids: set[int] = frozenset(), + ) -> tuple[Book, Path] | None: + """Return the book and its downloaded epub, or None. The caller deletes the epub. + + Candidates are books with an epub and a close title (see title_rank). Bindery downloads + them, closest title first and then newest first, and compares the story URL in each epub. + """ + wanted = [t for t in titles if t] + candidates: dict[int, tuple[tuple[int, float], Book]] = {} + for text in dict.fromkeys(search_text(t) for t in wanted): + for book in client.search(text): + rank = title_rank(book.title, wanted) + if rank is not None and "epub" in book.formats: + candidates[book.id] = (rank, book) + with self._lock: + rejected = set(self._not_story.get(story_url, set())) + checked = 0 + for _, book in sorted(candidates.values(), key=lambda c: (c[0], -c[1].id)): + if book.id in exclude_ids or book.id in rejected: + continue + if checked >= MAX_DOWNLOADS: + break + checked += 1 + path = client.download_epub(book, work_dir / f"cwa-{book.id}-{uuid.uuid4().hex[:8]}.epub") + try: + urls = epub_urls(path) + except EpubError as e: + logger.warning(f"Cannot read the epub of library book {book.id}: {e}") + urls = [] + if any((normalize(u) or (None,))[0] == story_url for u in urls): + return book, path + with self._lock: + self._not_story.setdefault(story_url, set()).add(book.id) + path.unlink(missing_ok=True) + return None diff --git a/bindery/epub.py b/bindery/epub.py new file mode 100644 index 0000000..34f0847 --- /dev/null +++ b/bindery/epub.py @@ -0,0 +1,73 @@ +"""Read the OPF metadata of an epub that came from CWA.""" + +import re +import zipfile +from pathlib import Path +from typing import Any +from xml.etree.ElementTree import Element + +from defusedxml import ElementTree + +OPF = "{http://www.idpf.org/2007/opf}" +DC = "{http://purl.org/dc/elements/1.1/}" +CONTAINER = "{urn:oasis:names:tc:opendocument:xmlns:container}" + + +class EpubError(Exception): + pass + + +def _metadata(path: Path) -> Element: + try: + with zipfile.ZipFile(path) as z: + container = ElementTree.fromstring(z.read("META-INF/container.xml")) + rootfile = container.find(f".//{CONTAINER}rootfile") + if rootfile is None or not rootfile.get("full-path"): + raise EpubError(f"{path.name} has no OPF file") + opf = ElementTree.fromstring(z.read(rootfile.get("full-path"))) + except (zipfile.BadZipFile, KeyError, ElementTree.ParseError) as e: + raise EpubError(f"{path.name} is not a readable epub: {e}") from e + metadata = opf.find(f"{OPF}metadata") + if metadata is None: + raise EpubError(f"{path.name} has no OPF metadata") + return metadata + + +def epub_urls(path: Path) -> list[str]: + """Every URL in the identifiers and the source of an epub's OPF. + + FanFicFare writes the story URL as dc:source and as a URL identifier. CWA + keeps the identifier when it writes its own metadata into a download. + """ + urls: list[str] = [] + for element in _metadata(path): + if element.tag not in (f"{DC}identifier", f"{DC}source"): + continue + match = re.search(r"https?://\S+", element.text or "") + if match and match.group(0) not in urls: + urls.append(match.group(0)) + return urls + + +def epub_metadata(path: Path) -> dict[str, Any]: + """The metadata fields of an epub's OPF, in Bindery's field format.""" + metadata = _metadata(path) + + def texts(tag: str) -> list[str]: + return [(e.text or "").strip() for e in metadata.findall(f"{DC}{tag}") if (e.text or "").strip()] + + meta = {m.get("name"): m.get("content") for m in metadata.findall(f"{OPF}meta") if m.get("name")} + try: + series_index = float(meta["calibre:series_index"]) if meta.get("calibre:series_index") else None + except ValueError: + series_index = None + return { + "title": (texts("title") or [""])[0], + "authors": texts("creator"), + "comments": (texts("description") or [""])[0], + "tags": texts("subject"), + "series": meta.get("calibre:series") or "", + "series_index": series_index, + "pubdate": (texts("date") or [""])[0], + "languages": (texts("language") or [""])[0], + } diff --git a/bindery/ingest.py b/bindery/ingest.py new file mode 100644 index 0000000..8c91eca --- /dev/null +++ b/bindery/ingest.py @@ -0,0 +1,68 @@ +"""Hand-off to CWA through its ingest folder. CWA's own calibredb does every write. + +- A new story goes in as a plain epub, and CWA imports it as a new book. +- An update goes in with a sidecar file, .epub.cwa.json, that holds + {"action": "add_format", "book_id": N}. CWA then replaces the epub of book N. + CWA's own web UI uses the same sidecar, in cps/editbooks.py. +""" + +import json +import logging +import os +import re +import shutil +import uuid +from pathlib import Path + +logger = logging.getLogger(__name__) + +# CWA shortens ingest file names over 150 characters. That would break the +# match between an epub and its sidecar file, so names stay well under it. +MAX_NAME_LENGTH = 100 + + +def sanitize_filename(name: str) -> str: + name = re.sub(r'[\\/:*?"<>|\x00-\x1f]', "_", name).strip(" .") + return name[:MAX_NAME_LENGTH].strip(" .") or "story" + + +class IngestFolder: + def __init__(self, path: Path) -> None: + self.path = Path(path) + + def _place(self, epub: Path, final_path: Path) -> Path: + """Copy an epub in under a temporary .part name, then rename it. + + CWA ignores .part files, and reacts to the rename. A file without an + extension is not safe: CWA deletes files it cannot import. + """ + partial = final_path.with_name(final_path.name + ".part") + shutil.copyfile(epub, partial) + os.replace(partial, final_path) + return final_path + + def ingest(self, epub: Path, name: str) -> Path: + """Send a new book to CWA.""" + self.path.mkdir(parents=True, exist_ok=True) + # A short unique suffix, so a file that CWA has not imported yet is never replaced. + stem = sanitize_filename(f"{name[:80]} {uuid.uuid4().hex[:6]}") + final_path = self._place(epub, self.path / f"{stem}.epub") + logger.info(f"Sent {final_path.name} to the ingest folder as a new book") + return final_path + + def send_update(self, epub: Path, book_id: int, name: str) -> Path: + """Send a new epub for an existing book. CWA replaces the book's epub.""" + self.path.mkdir(parents=True, exist_ok=True) + # A unique name, so two updates of one book never share a sidecar file. + stem = sanitize_filename(f"{name[:60]} {book_id}-{uuid.uuid4().hex[:8]}") + final_path = self.path / f"{stem}.epub" + manifest = final_path.with_name(final_path.name + ".cwa.json") + # The sidecar must exist before the epub arrives. CWA skips .cwa.json files. + manifest.write_text(json.dumps({"action": "add_format", "book_id": int(book_id)}), encoding="utf-8") + try: + self._place(epub, final_path) + except Exception: + manifest.unlink(missing_ok=True) + raise + logger.info(f"Sent {final_path.name} to the ingest folder as an update of book {book_id}") + return final_path diff --git a/tests/fake_cwa.py b/tests/fake_cwa.py new file mode 100644 index 0000000..5f0dca2 --- /dev/null +++ b/tests/fake_cwa.py @@ -0,0 +1,139 @@ +"""A small stand-in for CWA's OPDS feed, book JSON and downloads.""" + +import base64 +import json +import threading +from dataclasses import dataclass, field +from html import escape +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from urllib.parse import parse_qs, urlparse + + +@dataclass +class FakeBook: + id: int + uuid: str + title: str + authors: list[str] = field(default_factory=lambda: ["Writer"]) + publisher: str = "" + epub: bytes | None = None + + +class FakeCwa: + def __init__(self, username: str = "bindery", password: str = "secret", page_size: int = 50) -> None: + self.username, self.password, self.page_size = username, password, page_size + self.books: list[FakeBook] = [] + self.requests: list[str] = [] + # CWA's "Show Publisher Section" setting. Without it, the publisher feed is 404. + self.publisher_view = True + fake = self + + class Handler(BaseHTTPRequestHandler): + def log_message(self, *args): # quiet + pass + + def do_GET(self): + fake.requests.append(self.path) + fake.handle(self) + + self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + self.url = f"http://127.0.0.1:{self.server.server_address[1]}" + threading.Thread(target=self.server.serve_forever, daemon=True).start() + + def close(self) -> None: + if self.server.socket.fileno() == -1: + return + self.server.shutdown() + self.server.server_close() + + def _authorized(self, h) -> bool: + header = h.headers.get("Authorization", "") + if not header.startswith("Basic "): + return False + user, _, password = base64.b64decode(header[6:]).decode().partition(":") + return user == self.username and password == self.password + + def _send(self, h, status: int, body: bytes, ctype: str) -> None: + h.send_response(status) + h.send_header("Content-Type", ctype) + h.send_header("Content-Length", str(len(body))) + h.end_headers() + h.wfile.write(body) + + def _feed(self, entries: list[str], next_href: str | None = None) -> bytes: + link = f'' if next_href else "" + return ( + f'' + f"Fake CWA{link}{''.join(entries)}" + ).encode() + + def _book_entry(self, b: FakeBook) -> str: + authors = "".join(f"{escape(a)}" for a in b.authors) + publisher = f"{escape(b.publisher)}" if b.publisher else "" + link = ( + ( + f'' + ) + if b.epub is not None + else "" + ) + return f"{escape(b.title)}urn:uuid:{b.uuid}{authors}{publisher}{link}" + + def _paged(self, h, books: list[FakeBook], path: str, query: dict) -> None: + offset = int(query.get("offset", ["0"])[0]) + page = books[offset : offset + self.page_size] + next_href = f"{path}?offset={offset + self.page_size}" if offset + self.page_size < len(books) else None + self._send(h, 200, self._feed([self._book_entry(b) for b in page], next_href), "application/atom+xml") + + def handle(self, h) -> None: + if not self._authorized(h): + self._send(h, 401, b"Unauthorized", "text/plain") + return + url = urlparse(h.path) + query = parse_qs(url.query) + path = url.path + if path == "/opds": + self._send(h, 200, self._feed([]), "application/atom+xml") + elif path == "/opds/search": + term = query.get("query", [""])[0].lower() + hits = [b for b in self.books if term in b.title.lower() or any(term in a.lower() for a in b.authors)] + # CWA returns every search result on one page. + self._send(h, 200, self._feed([self._book_entry(b) for b in hits]), "application/atom+xml") + elif path.startswith("/opds/publisher") and not self.publisher_view: + self._send(h, 404, b"Not Found", "text/plain") + elif path == "/opds/publisher": + names = sorted({b.publisher for b in self.books if b.publisher}) + entries = [ + f"{escape(n)}/opds/publisher/{i}" + f'' + for i, n in enumerate(names) + ] + self._send(h, 200, self._feed(entries), "application/atom+xml") + elif path.startswith("/opds/publisher/"): + names = sorted({b.publisher for b in self.books if b.publisher}) + name = names[int(path.rsplit("/", 1)[1])] + self._paged(h, [b for b in self.books if b.publisher == name], path, query) + elif path.startswith("/ajax/book/"): + uuid = path.rsplit("/", 1)[1] + book = next((b for b in self.books if b.uuid == uuid), None) + if not book: + self._send(h, 200, b"", "text/html") + return + data = { + "uuid": book.uuid, + "application_id": book.id, + "title": book.title, + "authors": book.authors, + "formats": ["EPUB"] if book.epub is not None else [], + } + self._send(h, 200, json.dumps(data).encode(), "application/json") + elif path.startswith("/opds/download/"): + book_id = int(path.split("/")[3]) + book = next((b for b in self.books if b.id == book_id and b.epub is not None), None) + if not book: + self._send(h, 404, b"Not found", "text/plain") + return + self._send(h, 200, book.epub, "application/epub+zip") + else: + self._send(h, 404, b"Not found", "text/plain") diff --git a/tests/test_cwa.py b/tests/test_cwa.py new file mode 100644 index 0000000..3794fa2 --- /dev/null +++ b/tests/test_cwa.py @@ -0,0 +1,197 @@ +"""The OPDS client, the book finder, the login check, the ingest folder and the epub reader.""" + +import json + +import pytest +from fake_cwa import FakeBook, FakeCwa +from test_build import URL, chapters, metahtml + +from bindery import fff +from bindery.cwa import BookFinder, CwaClient, CwaError, plain_title, search_text, title_rank +from bindery.epub import epub_metadata, epub_urls +from bindery.ingest import MAX_NAME_LENGTH, IngestFolder + +STORY_URL = "https://www.royalroad.com/fiction/1234" + + +@pytest.fixture(scope="module") +def story_epub(tmp_path_factory): + path = tmp_path_factory.mktemp("epub") / "story.epub" + fff.build_epub(URL, metahtml(), chapters(), [], None, [], {"series": ["Saga", 2.0]}, path) + return path.read_bytes() + + +@pytest.fixture(scope="module") +def other_epub(tmp_path_factory): + path = tmp_path_factory.mktemp("epub") / "other.epub" + other_url = "https://www.royalroad.com/fiction/9999/other-story" + other_chapters = [dict(c, url=c["url"].replace("1234/test-story", "9999/other-story")) for c in chapters()] + fff.build_epub(other_url, metahtml(other_url), other_chapters, [], None, [], {}, path) + return path.read_bytes() + + +@pytest.fixture +def fake(story_epub, other_epub): + server = FakeCwa(page_size=1) + other = other_epub + server.books = [ + FakeBook(1, "uuid-1", "Test Story", publisher="www.royalroad.com", epub=other), + FakeBook(2, "uuid-2", "Test Story", publisher="www.royalroad.com", epub=story_epub), + FakeBook(3, "uuid-3", "Another Book", ["Someone"], publisher="Tor", epub=b"not a zip"), + FakeBook(4, "uuid-4", "Test Story Two", publisher="www.royalroad.com"), + ] + yield server + server.close() + + +def client(fake, password="secret"): + return CwaClient(fake.url, "bindery", password) + + +def test_search_reads_ids_and_uuids(fake): + books = client(fake).search("test story") + # Book 4 has no files, so OPDS shows no download link and no book id for it. + assert [(b.id, b.uuid) for b in books] == [(1, "uuid-1"), (2, "uuid-2")] + assert books[0].formats == {"epub": "/opds/download/1/epub/"} + assert books[0].publisher == "www.royalroad.com" + + +def test_wrong_password_and_missing_address(fake): + with pytest.raises(CwaError, match="refused"): + client(fake, "wrong").search("x") + with pytest.raises(CwaError, match="not set"): + CwaClient("").search("x") + + +def test_get_book(fake): + c = client(fake) + assert c.get_book("uuid-2") == { + "id": 2, + "uuid": "uuid-2", + "title": "Test Story", + "authors": ["Writer"], + "formats": ["EPUB"], + } + assert c.get_book("nope") is None + + +@pytest.mark.parametrize("book_uuid", ["../opds", "uuid-2?x=1", "uuid-2#x", "uuid%2F2", ""]) +def test_get_book_refuses_a_uuid_that_changes_the_path(book_uuid): + # Nothing listens on this port, so a request would raise CwaError. + assert CwaClient("http://127.0.0.1:9").get_book(book_uuid) is None + + +def test_site_books(fake): + books = client(fake).site_books(fff.is_supported_site) + assert [b.id for b in books] == [1, 2] # book 4 has no epub, book 3 is not from a story site + assert any("offset=2" in r for r in fake.requests) # the publisher feed has three pages + + +def test_site_books_names_the_cwa_setting_that_shows_publishers(fake, monkeypatch): + monkeypatch.setattr(fake, "publisher_view", False) + with pytest.raises(CwaError, match='Turn on "Show Publisher Section" for Bindery in CWA'): + client(fake).site_books(fff.is_supported_site) + + +def test_finder_checks_the_url_in_each_epub(fake, tmp_path): + finder = BookFinder() + found = finder.find(client(fake), STORY_URL, ["test story"], fff.normalize_url, tmp_path) + assert found is not None + book, path = found + assert book.id == 2 and path.exists() + # With book 2 excluded, book 1 is checked once and remembered as another story. + assert finder.find(client(fake), STORY_URL, ["Test Story"], fff.normalize_url, tmp_path, {2}) is None + fake.requests.clear() + assert finder.find(client(fake), STORY_URL, ["Test Story"], fff.normalize_url, tmp_path, {2}) is None + assert not [r for r in fake.requests if r.startswith("/opds/download/")] + assert sorted(p.name for p in tmp_path.iterdir()) == [path.name] # rejected downloads are deleted + + +TAGGED = "Story A [Tag A, Tag B] (Book 1 Complete)" + + +@pytest.mark.parametrize( + "title, plain, search", + [ + (TAGGED, "story a", "Story A"), + ("Story A - (Book 1)", "story a", "Story A"), + ("[AU] Story A [Tag] Part I", "story a part i", "story a part i"), + ("Story (A [B]) C", "story c", "Story"), + ("Plain Title", "plain title", "Plain Title"), + ], +) +def test_titles_without_their_bracketed_parts(title, plain, search): + assert (plain_title(title), search_text(title)) == (plain, search) + + +def test_title_rank_prefers_the_closest_title(): + titles = [ + "Story A [Tag A, Tag B] Part I", + "Story A", + "Story A [Tag A, Tag B]", + TAGGED, + "Story Ab", + "Another Story A", + ] + ranked = sorted((t for t in titles if title_rank(t, [TAGGED])), key=lambda t: title_rank(t, [TAGGED])) + assert ranked == [TAGGED, "Story A [Tag A, Tag B]", "Story A", "Story A [Tag A, Tag B] Part I"] + assert title_rank("Story A [Tag A, Tag B]", ["Story A"])[0] == 1 + assert title_rank("Story A Part I", ["Story A Part II"]) is None + + +def test_finder_downloads_the_closest_titles_first(fake, other_epub, tmp_path): + fake.books = [ + FakeBook(9, "uuid-9", TAGGED, epub=other_epub), + FakeBook(10, "uuid-10", "Story A", epub=other_epub), + FakeBook(11, "uuid-11", "Story A [Tag A, Tag B] Part I", epub=other_epub), + FakeBook(12, "uuid-12", "Story A [Tag A, Tag B]", epub=other_epub), + FakeBook(13, "uuid-13", "Story Ab", epub=other_epub), + FakeBook(14, "uuid-14", "Another Story A", epub=other_epub), + ] + assert BookFinder().find(client(fake), STORY_URL, [TAGGED], fff.normalize_url, tmp_path) is None + assert [r for r in fake.requests if r.startswith("/opds/search")] == ["/opds/search?query=Story+A"] + downloads = [int(r.split("/")[3]) for r in fake.requests if r.startswith("/opds/download/")] + assert downloads == [9, 12, 10, 11] + + +def test_login_ok(fake): + c = CwaClient(fake.url) + assert c.login_ok("bindery", "secret") + assert not c.login_ok("bindery", "wrong") + fake.close() + with pytest.raises(CwaError, match="Cannot reach"): + c.login_ok("bindery", "secret") + + +def test_epub_reader(story_epub, tmp_path): + path = tmp_path / "s.epub" + path.write_bytes(story_epub) + assert STORY_URL in [fff.normalize_url(u)[0] for u in epub_urls(path)] + meta = epub_metadata(path) + assert meta["title"] == "Test Story" + assert meta["authors"] == ["Writer"] + assert meta["series"] == "Saga" and meta["series_index"] == 2.0 + assert meta["pubdate"].startswith("2024-01-02") + + +def test_ingest_uses_part_names_and_unique_names(tmp_path): + folder = IngestFolder(tmp_path / "ingest") + epub = tmp_path / "x.epub" + epub.write_bytes(b"PK") + first = folder.ingest(epub, 'A: "Story" / Name') + second = folder.ingest(epub, 'A: "Story" / Name') + assert first != second + assert first.name.startswith("A_ _Story_ _ Name ") and first.suffix == ".epub" + assert sorted(p.name for p in folder.path.iterdir()) == sorted([first.name, second.name]) + + +def test_send_update_writes_sidecar(tmp_path): + folder = IngestFolder(tmp_path / "ingest") + epub = tmp_path / "x.epub" + epub.write_bytes(b"PK") + path = folder.send_update(epub, 7, "T" * 300) + assert sorted(p.name for p in folder.path.iterdir()) == [path.name, path.name + ".cwa.json"] + assert len(path.name) <= MAX_NAME_LENGTH + len(".epub") + manifest = json.loads((folder.path / (path.name + ".cwa.json")).read_text()) + assert manifest == {"action": "add_format", "book_id": 7} + assert path.read_bytes() == b"PK"