tim
312dc3efc5
Bindery mounts no CWA database. It searches the OPDS feed by title and reads the story URL inside each candidate epub to find the library book. Updates go to the ingest folder with a .cwa.json sidecar, so CWA replaces the epub of the existing book.
310 lines
12 KiB
Python
310 lines
12 KiB
Python
"""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
|