diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..92825f0 --- /dev/null +++ b/.env.example @@ -0,0 +1,16 @@ +# BookFusion API key. Get yours from BookFusion → Settings → Integrations → Calibre. +BOOKFUSION_API_KEY=your-bookfusion-api-key-here + +# BookFusion API base URL (rarely changed). +BOOKFUSION_API_BASE=https://www.bookfusion.com/calibre-api/v1 + +# Path to your Calibre library directory (the folder containing metadata.db). +# For Calibre-Web-Automated docker, this is typically the host path mapped to /calibre-library +# inside the container, e.g. /opt/cwa/library. +CALIBRE_LIBRARY_PATH=/path/to/your/calibre-library + +# Optional: how many parallel upload workers to run when calling sync_all. +BOOKFUSION_UPLOAD_THREADS=2 + +# Optional: request timeout in seconds for BookFusion API calls. +BOOKFUSION_TIMEOUT=120 diff --git a/.gitignore b/.gitignore index 9a47826..8aff515 100644 --- a/.gitignore +++ b/.gitignore @@ -52,3 +52,28 @@ m4/lt~obsolete.m4 # can automatically generate from config.status script # (which is called by configure script)) Makefile + +# Python +__pycache__/ +*.py[cod] +*$py.class +*.egg-info/ +.eggs/ +build/ +dist/ +.venv/ +venv/ +.pytest_cache/ +.ruff_cache/ +.mypy_cache/ + +# Local configuration / secrets +.env +.env.local +*.local + +# Editor +.vscode/ +.idea/ +*.swp +.DS_Store diff --git a/README.md b/README.md new file mode 100644 index 0000000..329a872 --- /dev/null +++ b/README.md @@ -0,0 +1,143 @@ +# BookFusion MCP + +An [MCP](https://modelcontextprotocol.io) server that lets Claude (or any other +MCP client) sync books from a [Calibre](https://calibre-ebook.com) library — +including the one managed by +[Calibre-Web-Automated](https://github.com/crocodilestick/Calibre-Web-Automated) +— to your [BookFusion](https://www.bookfusion.com) account. + +It reuses the exact same HTTP API that the official +[BookFusion Calibre plugin](https://github.com/BookFusion/calibre-plugin) talks +to, so anything you already have configured on the BookFusion side (account, +quota, integrations, API key) just works. + +## What it does + +The server reads metadata directly from the Calibre `metadata.db` SQLite +database and the on-disk book / cover files. It never writes to the database. +It exposes the following MCP tools: + +| Tool | Purpose | +| --- | --- | +| `library_status` | Show the configured library path, book count and BookFusion endpoint. | +| `list_books` | Paginated list of local books, optionally filtered by title/author. | +| `get_book` | Full details for one book (formats, identifiers, digests). | +| `get_bookfusion_limits` | Fetch upload limits (max filesize, etc.). | +| `check_book_in_bookfusion` | Look up a local book on BookFusion (by id, ISBN or digest). | +| `sync_book_to_bookfusion` | Upload or update a single book. | +| `sync_books_to_bookfusion` | Same, in parallel for a list of book ids. | +| `sync_all_books` | Sync every book matching a `list_books` filter. | + +## Configuration + +Copy `.env.example` to `.env` and fill it in: + +```bash +cp .env.example .env +``` + +| Env var | Required | Default | Notes | +| --- | --- | --- | --- | +| `BOOKFUSION_API_KEY` | yes | — | Get it from BookFusion → Settings → Integrations → Calibre. | +| `CALIBRE_LIBRARY_PATH` | yes | — | Host path to the directory containing `metadata.db`. For Calibre-Web-Automated this is the host folder you mapped to `/calibre-library`. | +| `BOOKFUSION_API_BASE` | no | `https://www.bookfusion.com/calibre-api/v1` | Override only if you self-host or test against a staging instance. | +| `BOOKFUSION_UPLOAD_THREADS` | no | `2` | Parallel uploads when running `sync_books_to_bookfusion` / `sync_all_books`. | +| `BOOKFUSION_TIMEOUT` | no | `120` | Request timeout in seconds. | + +> **Never commit your API key.** `.env` is already in `.gitignore`. + +## Install + +The server requires Python 3.10+. + +```bash +python -m venv .venv +source .venv/bin/activate +pip install -e ".[dev]" +``` + +Run the test suite to make sure everything is wired up: + +```bash +pytest -q +``` + +## Running the server + +```bash +bookfusion-mcp +``` + +This starts a stdio MCP server. Connect any MCP client to it. + +### Claude Desktop / Claude Code + +Add the server to your client config (Claude Desktop's +`claude_desktop_config.json` or Claude Code's `~/.claude.json`): + +```json +{ + "mcpServers": { + "bookfusion": { + "command": "bookfusion-mcp", + "env": { + "BOOKFUSION_API_KEY": "your-bookfusion-api-key", + "CALIBRE_LIBRARY_PATH": "/path/to/your/calibre-library" + } + } + } +} +``` + +If you installed inside a virtualenv, point `command` at the venv's +`bookfusion-mcp` binary, e.g. +`/home/you/FGU-bookfusion/.venv/bin/bookfusion-mcp`. + +### Calibre-Web-Automated specifics + +The server only needs read access to: + +* the `metadata.db` file produced by Calibre / CWA, and +* the book / cover files inside the same library directory. + +If you run CWA in Docker and want the MCP server on the same host, point +`CALIBRE_LIBRARY_PATH` at whatever you mounted into the container at +`/calibre-library`. Example: + +```bash +docker run -d \ + -v /opt/cwa/library:/calibre-library \ + ... \ + crocodilestick/calibre-web-automated +``` + +→ on the host, set `CALIBRE_LIBRARY_PATH=/opt/cwa/library`. + +## How the sync works + +For each book the server: + +1. Picks the first format the BookFusion plugin supports (EPUB, PDF, CBZ, CBR, + CBT, MOBI). +2. Computes the file digest the way the plugin does + (`sha256(filesize_bytes || 0x00 || file_contents)`) and a metadata digest + over the same fields the plugin hashes — so digests match what BookFusion + already has. +3. Looks up the book on BookFusion using the most specific identifier + available, in this order: existing `bookfusion` identifier → `isbn` → + file digest. +4. If the upstream digest matches and `reupload=false`, skips the book. +5. Otherwise either `PUT /uploads/{id}` to update an existing book, or runs the + three-step `init → upload to presigned URL → finalize` flow for a new one. + +## Notes & limits + +* Uploads are one-way (Calibre → BookFusion), exactly like the official plugin. +* `metadata.db` is opened read-only via `file:...?mode=ro` so Calibre / CWA can + keep the file open at the same time. +* The `bookshelves` field is wired through the API but not yet populated from + Calibre custom columns; PRs welcome. + +## License + +GPL-3.0-or-later, matching the upstream BookFusion Calibre plugin. diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..0bf5312 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,41 @@ +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "bookfusion-mcp" +version = "0.1.0" +description = "MCP server to sync a Calibre / Calibre-Web-Automated library to BookFusion" +readme = "README.md" +requires-python = ">=3.10" +license = { text = "GPL-3.0-or-later" } +authors = [ + { name = "Fguarien" } +] +keywords = ["mcp", "bookfusion", "calibre", "calibre-web-automated", "ebook"] +dependencies = [ + "mcp[cli]>=1.2.0", + "httpx>=0.27.0", + "python-dotenv>=1.0.0", +] + +[project.optional-dependencies] +dev = [ + "pytest>=8.0.0", + "pytest-asyncio>=0.23.0", + "ruff>=0.5.0", +] + +[project.scripts] +bookfusion-mcp = "bookfusion_mcp.__main__:main" + +[tool.hatch.build.targets.wheel] +packages = ["src/bookfusion_mcp"] + +[tool.ruff] +line-length = 100 +target-version = "py310" + +[tool.pytest.ini_options] +asyncio_mode = "auto" +testpaths = ["tests"] diff --git a/src/bookfusion_mcp/__init__.py b/src/bookfusion_mcp/__init__.py new file mode 100644 index 0000000..b2b1654 --- /dev/null +++ b/src/bookfusion_mcp/__init__.py @@ -0,0 +1,3 @@ +"""BookFusion MCP server for Calibre / Calibre-Web-Automated libraries.""" + +__version__ = "0.1.0" diff --git a/src/bookfusion_mcp/__main__.py b/src/bookfusion_mcp/__main__.py new file mode 100644 index 0000000..23b1d93 --- /dev/null +++ b/src/bookfusion_mcp/__main__.py @@ -0,0 +1,13 @@ +"""Entry point: run the BookFusion MCP server over stdio.""" + +from __future__ import annotations + +from .server import mcp + + +def main() -> None: + mcp.run() + + +if __name__ == "__main__": + main() diff --git a/src/bookfusion_mcp/bookfusion_api.py b/src/bookfusion_mcp/bookfusion_api.py new file mode 100644 index 0000000..894a086 --- /dev/null +++ b/src/bookfusion_mcp/bookfusion_api.py @@ -0,0 +1,268 @@ +"""Async BookFusion API client. + +This mirrors the request flow of the official BookFusion Calibre plugin +(https://github.com/BookFusion/calibre-plugin) and reuses the same endpoints: + +* ``GET /limits`` +* ``GET /uploads/{id_or_digest}`` +* ``GET /uploads?isbn=...`` +* ``POST /uploads/init`` (multipart: filename, digest) +* ``POST `` (multipart upload to S3-compatible storage) +* ``POST /uploads/finalize`` (multipart: key, digest, metadata[*]) +* ``PUT /uploads/{id}`` (multipart: optional file + metadata[*]) + +Authentication uses HTTP Basic with the API key as the username and an empty +password, exactly like the upstream plugin. +""" + +from __future__ import annotations + +import base64 +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +import httpx + +USER_AGENT = "BookFusion MCP Server/0.1.0" + + +def _basic_auth_header(api_key: str) -> str: + token = base64.b64encode(f"{api_key}:".encode("utf-8")).decode("ascii") + return f"Basic {token}" + + +@dataclass +class BookMetadata: + """Subset of book metadata accepted by the BookFusion finalize/update endpoints.""" + + calibre_metadata_digest: str + title: str + summary: str | None = None + language: str | None = None + isbn: str | None = None + issued_on: str | None = None # ISO date YYYY-MM-DD + authors: list[str] = field(default_factory=list) + tags: list[str] = field(default_factory=list) + bookshelves: list[str] | None = None + series: list[dict[str, Any]] = field(default_factory=list) # [{title, index}] + cover_path: Path | None = None + + def as_multipart(self) -> tuple[list[tuple[str, Any]], list[tuple[str, Any]]]: + """Return (data_fields, file_fields) ready for httpx multipart upload.""" + data: list[tuple[str, Any]] = [ + ("metadata[calibre_metadata_digest]", self.calibre_metadata_digest), + ("metadata[title]", self.title), + ] + if self.summary: + data.append(("metadata[summary]", self.summary)) + if self.language: + data.append(("metadata[language]", self.language)) + if self.isbn: + data.append(("metadata[isbn]", self.isbn)) + if self.issued_on: + data.append(("metadata[issued_on]", self.issued_on)) + for series_item in self.series: + data.append(("metadata[series][][title]", series_item["title"])) + if series_item.get("index") is not None: + data.append(("metadata[series][][index]", str(series_item["index"]))) + for author in self.authors: + data.append(("metadata[author_list][]", author)) + for tag in self.tags: + data.append(("metadata[tag_list][]", tag)) + if self.bookshelves is not None: + # The plugin always emits an empty marker first to clear the list server side. + data.append(("metadata[bookshelves][]", "")) + for shelf in self.bookshelves: + data.append(("metadata[bookshelves][]", shelf)) + + files: list[tuple[str, Any]] = [] + if self.cover_path and self.cover_path.exists(): + files.append( + ( + "metadata[cover]", + (self.cover_path.name, self.cover_path.read_bytes()), + ) + ) + return data, files + + +class BookFusionError(Exception): + """Raised when the BookFusion API returns an error response.""" + + def __init__(self, message: str, status_code: int | None = None, payload: Any = None): + super().__init__(message) + self.status_code = status_code + self.payload = payload + + +class BookFusionClient: + """Thin async wrapper around the BookFusion Calibre API.""" + + def __init__(self, api_key: str, api_base: str, timeout: float = 120.0): + if not api_key: + raise ValueError("api_key is required") + self._api_base = api_base.rstrip("/") + self._client = httpx.AsyncClient( + timeout=timeout, + headers={ + "User-Agent": USER_AGENT, + "Authorization": _basic_auth_header(api_key), + "Accept": "application/json", + }, + ) + + async def __aenter__(self) -> "BookFusionClient": + return self + + async def __aexit__(self, exc_type, exc, tb) -> None: + await self.aclose() + + async def aclose(self) -> None: + await self._client.aclose() + + # ------------------------------------------------------------------ helpers + def _url(self, path: str) -> str: + if not path.startswith("/"): + path = "/" + path + return f"{self._api_base}{path}" + + async def _request( + self, + method: str, + path: str, + *, + params: dict[str, Any] | None = None, + data: list[tuple[str, Any]] | None = None, + files: list[tuple[str, Any]] | None = None, + url_override: str | None = None, + expect_json: bool = True, + ) -> Any: + url = url_override or self._url(path) + try: + resp = await self._client.request( + method, + url, + params=params, + data=data, + files=files, + ) + except httpx.HTTPError as exc: + raise BookFusionError(f"Network error calling {method} {url}: {exc}") from exc + + if resp.status_code == 401: + raise BookFusionError("Invalid BookFusion API key", status_code=401) + + if resp.status_code == 404: + raise BookFusionError("Not found", status_code=404) + + if resp.status_code == 422: + try: + payload = resp.json() + msg = payload.get("error", "Unprocessable entity") + except ValueError: + payload = resp.text + msg = "Unprocessable entity" + raise BookFusionError(msg, status_code=422, payload=payload) + + if resp.status_code >= 400: + raise BookFusionError( + f"BookFusion error ({resp.status_code}): {resp.text[:500]}", + status_code=resp.status_code, + payload=resp.text, + ) + + if not expect_json or not resp.content: + return resp.text + try: + return resp.json() + except ValueError: + return resp.text + + # -------------------------------------------------------------------- API + async def get_limits(self) -> dict[str, Any]: + """``GET /limits`` — returns at least ``{"filesize": }``.""" + return await self._request("GET", "/limits") + + async def get_upload_by_id(self, bookfusion_id: str) -> dict[str, Any] | None: + try: + return await self._request("GET", f"/uploads/{bookfusion_id}") + except BookFusionError as exc: + if exc.status_code == 404: + return None + raise + + async def get_upload_by_digest(self, digest: str) -> dict[str, Any] | None: + try: + return await self._request("GET", f"/uploads/{digest}") + except BookFusionError as exc: + if exc.status_code == 404: + return None + raise + + async def search_uploads_by_isbn(self, isbn: str) -> list[dict[str, Any]]: + result = await self._request("GET", "/uploads", params={"isbn": isbn}) + if isinstance(result, list): + return result + return [] + + async def init_upload(self, *, filename: str, digest: str) -> dict[str, Any]: + """Initialise an upload — returns ``{"url": ..., "params": {...}}``.""" + return await self._request( + "POST", + "/uploads/init", + data=[("filename", filename), ("digest", digest)], + ) + + async def upload_to_storage( + self, + upload_url: str, + upload_params: dict[str, Any], + file_path: Path, + ) -> None: + """POST the file to the presigned URL returned by ``init_upload``.""" + data = [(key, str(value)) for key, value in upload_params.items()] + with file_path.open("rb") as fh: + files = [("file", (file_path.name, fh.read()))] + await self._request( + "POST", + "", + url_override=upload_url, + data=data, + files=files, + expect_json=False, + ) + + async def finalize_upload( + self, + *, + key: str, + digest: str, + metadata: BookMetadata, + ) -> dict[str, Any]: + data, files = metadata.as_multipart() + data = [("key", key), ("digest", digest), *data] + return await self._request( + "POST", + "/uploads/finalize", + data=data, + files=files or None, + ) + + async def update_upload( + self, + bookfusion_id: str, + *, + metadata: BookMetadata, + file_path: Path | None = None, + ) -> dict[str, Any] | str: + data, files = metadata.as_multipart() + if file_path is not None: + files = list(files) + files.append(("file", (file_path.name, file_path.read_bytes()))) + return await self._request( + "PUT", + f"/uploads/{bookfusion_id}", + data=data, + files=files or None, + ) diff --git a/src/bookfusion_mcp/calibre_library.py b/src/bookfusion_mcp/calibre_library.py new file mode 100644 index 0000000..c639384 --- /dev/null +++ b/src/bookfusion_mcp/calibre_library.py @@ -0,0 +1,366 @@ +"""Read-only access to a Calibre library (the same on-disk format used by +Calibre-Web-Automated). + +A Calibre library is a directory that contains: + +* ``metadata.db`` — SQLite database with the canonical schema +* ``/ ()/.`` — book files +* ``/ ()/cover.jpg`` — optional cover + +Calibre-Web-Automated mounts this directory at ``/calibre-library`` inside the +container by default; on the host it lives wherever the user mapped that +volume. This module only reads the SQLite metadata; it never writes to it. +""" + +from __future__ import annotations + +import sqlite3 +from contextlib import contextmanager +from dataclasses import dataclass, field +from datetime import datetime +from hashlib import sha256 +from pathlib import Path +from typing import Iterator + + +# Formats accepted by BookFusion (kept in sync with the calibre-plugin). +SUPPORTED_FORMATS = ("EPUB", "PDF", "CBZ", "CBR", "CBT", "MOBI") + + +@dataclass +class BookFormatFile: + format: str + path: Path + size: int + + +@dataclass +class CalibreBook: + id: int + title: str + sort: str | None + pubdate: datetime | None + timestamp: datetime | None + last_modified: datetime | None + isbn: str | None + bookfusion_id: str | None + series: str | None + series_index: float | None + summary: str | None + language: str | None + authors: list[str] = field(default_factory=list) + tags: list[str] = field(default_factory=list) + identifiers: dict[str, str] = field(default_factory=dict) + formats: list[BookFormatFile] = field(default_factory=list) + cover_path: Path | None = None + book_dir: Path | None = None + + def best_format(self) -> BookFormatFile | None: + """Pick the first supported format in the library, in BookFusion order.""" + by_format = {fmt.format.upper(): fmt for fmt in self.formats} + for candidate in SUPPORTED_FORMATS: + if candidate in by_format: + return by_format[candidate] + return self.formats[0] if self.formats else None + + def to_summary_dict(self) -> dict: + return { + "id": self.id, + "title": self.title, + "authors": self.authors, + "isbn": self.isbn, + "language": self.language, + "series": self.series, + "series_index": self.series_index, + "tags": self.tags, + "bookfusion_id": self.bookfusion_id, + "formats": [ + {"format": f.format, "size": f.size, "path": str(f.path)} for f in self.formats + ], + "cover_path": str(self.cover_path) if self.cover_path else None, + } + + +class CalibreLibrary: + """Read-only Calibre library accessor.""" + + def __init__(self, library_path: Path): + self.library_path = Path(library_path) + self.metadata_db = self.library_path / "metadata.db" + if not self.metadata_db.exists(): + raise FileNotFoundError( + f"metadata.db not found in {self.library_path!s}. " + "Set CALIBRE_LIBRARY_PATH to the directory that contains it." + ) + + # ------------------------------------------------------------------ db + @contextmanager + def _connect(self) -> Iterator[sqlite3.Connection]: + # ``uri=True`` + ``mode=ro`` ensures we never accidentally write to the DB, + # which is important because Calibre / Calibre-Web may have it open. + uri = f"file:{self.metadata_db}?mode=ro&immutable=0" + conn = sqlite3.connect(uri, uri=True) + conn.row_factory = sqlite3.Row + try: + yield conn + finally: + conn.close() + + # --------------------------------------------------------------- queries + def list_books( + self, + *, + limit: int = 50, + offset: int = 0, + search: str | None = None, + ) -> list[CalibreBook]: + with self._connect() as conn: + rows = self._select_books(conn, limit=limit, offset=offset, search=search) + return [self._hydrate_book(conn, row) for row in rows] + + def get_book(self, book_id: int) -> CalibreBook | None: + with self._connect() as conn: + row = conn.execute( + """ + SELECT id, title, sort, pubdate, timestamp, last_modified, series_index, path + FROM books + WHERE id = ? + """, + (book_id,), + ).fetchone() + if row is None: + return None + return self._hydrate_book(conn, row) + + def count_books(self) -> int: + with self._connect() as conn: + return conn.execute("SELECT COUNT(*) FROM books").fetchone()[0] + + # ------------------------------------------------------------- internals + def _select_books( + self, + conn: sqlite3.Connection, + *, + limit: int, + offset: int, + search: str | None, + ) -> list[sqlite3.Row]: + sql = ( + "SELECT id, title, sort, pubdate, timestamp, last_modified, series_index, path " + "FROM books" + ) + params: list = [] + if search: + sql += ( + " WHERE id IN (" + " SELECT book FROM books_authors_link bal" + " JOIN authors a ON a.id = bal.author" + " WHERE a.name LIKE ?" + " )" + " OR title LIKE ?" + ) + like = f"%{search}%" + params.extend([like, like]) + sql += " ORDER BY last_modified DESC LIMIT ? OFFSET ?" + params.extend([limit, offset]) + return conn.execute(sql, params).fetchall() + + def _hydrate_book(self, conn: sqlite3.Connection, row: sqlite3.Row) -> CalibreBook: + book_id: int = row["id"] + rel_path: str = row["path"] + book_dir = self.library_path / rel_path + + authors = [ + r["name"] + for r in conn.execute( + """ + SELECT a.name + FROM authors a + JOIN books_authors_link bal ON bal.author = a.id + WHERE bal.book = ? + ORDER BY bal.id + """, + (book_id,), + ) + ] + + tags = [ + r["name"] + for r in conn.execute( + """ + SELECT t.name + FROM tags t + JOIN books_tags_link btl ON btl.tag = t.id + WHERE btl.book = ? + ORDER BY t.name + """, + (book_id,), + ) + ] + + identifiers = { + r["type"]: r["val"] + for r in conn.execute( + "SELECT type, val FROM identifiers WHERE book = ?", + (book_id,), + ) + } + + languages_row = conn.execute( + """ + SELECT l.lang_code + FROM languages l + JOIN books_languages_link bll ON bll.lang_code = l.id + WHERE bll.book = ? + ORDER BY bll.item_order + LIMIT 1 + """, + (book_id,), + ).fetchone() + language = languages_row["lang_code"] if languages_row else None + + series_row = conn.execute( + """ + SELECT s.name + FROM series s + JOIN books_series_link bsl ON bsl.series = s.id + WHERE bsl.book = ? + LIMIT 1 + """, + (book_id,), + ).fetchone() + series = series_row["name"] if series_row else None + + comments_row = conn.execute( + "SELECT text FROM comments WHERE book = ?", (book_id,) + ).fetchone() + summary = comments_row["text"] if comments_row else None + + format_rows = conn.execute( + "SELECT format, name FROM data WHERE book = ?", (book_id,) + ).fetchall() + formats: list[BookFormatFile] = [] + for fr in format_rows: + fmt: str = fr["format"] + name: str = fr["name"] + file_path = book_dir / f"{name}.{fmt.lower()}" + if file_path.exists(): + formats.append( + BookFormatFile(format=fmt.upper(), path=file_path, size=file_path.stat().st_size) + ) + + cover_path = book_dir / "cover.jpg" + if not cover_path.exists(): + cover_path = None # type: ignore[assignment] + + return CalibreBook( + id=book_id, + title=row["title"], + sort=row["sort"], + pubdate=_parse_dt(row["pubdate"]), + timestamp=_parse_dt(row["timestamp"]), + last_modified=_parse_dt(row["last_modified"]), + isbn=identifiers.get("isbn"), + bookfusion_id=identifiers.get("bookfusion"), + series=series, + series_index=row["series_index"], + summary=summary, + language=language, + authors=authors, + tags=tags, + identifiers=identifiers, + formats=formats, + cover_path=cover_path, + book_dir=book_dir, + ) + + +# ----------------------------------------------------------------- digests + +def file_digest(path: Path) -> str: + """SHA-256 digest computed exactly the way the BookFusion plugin does it. + + The plugin first hashes the file size as raw bytes, then a single null byte, + then the file contents in 64 KiB blocks. We replicate that so digests match. + """ + h = sha256() + h.update(bytes(path.stat().st_size)) + h.update(b"\0") + with path.open("rb") as fh: + while True: + chunk = fh.read(65536) + if not chunk: + break + h.update(chunk) + return h.hexdigest() + + +def metadata_digest(book: CalibreBook) -> str: + """SHA-256 over the metadata fields BookFusion compares for change detection.""" + h = sha256() + h.update(book.title.encode("utf-8")) + if book.summary: + h.update(book.summary.encode("utf-8")) + if book.language: + h.update(book.language.encode("utf-8")) + if book.isbn: + h.update(book.isbn.encode("utf-8")) + issued_on = _issued_on(book.pubdate) + if issued_on: + h.update(issued_on.encode("utf-8")) + if book.series: + h.update(book.series.encode("utf-8")) + if book.series_index is not None: + h.update(str(book.series_index).encode("utf-8")) + for author in book.authors: + h.update(author.encode("utf-8")) + for tag in book.tags: + h.update(tag.encode("utf-8")) + if book.cover_path and book.cover_path.exists(): + h.update(bytes(book.cover_path.stat().st_size)) + h.update(b"\0") + with book.cover_path.open("rb") as fh: + while True: + chunk = fh.read(65536) + if not chunk: + break + h.update(chunk) + return h.hexdigest() + + +def _issued_on(pubdate: datetime | None) -> str | None: + if pubdate is None: + return None + iso = pubdate.date().isoformat() + if iso == "0101-01-01": + return None + return iso + + +def _parse_dt(value) -> datetime | None: + if value is None: + return None + if isinstance(value, datetime): + return value + if isinstance(value, (bytes, bytearray)): + value = value.decode("utf-8", errors="ignore") + text = str(value).strip() + if not text: + return None + # Calibre stores ISO-8601 with optional fractional seconds and timezone. + for fmt in ( + "%Y-%m-%d %H:%M:%S.%f%z", + "%Y-%m-%d %H:%M:%S%z", + "%Y-%m-%d %H:%M:%S.%f", + "%Y-%m-%d %H:%M:%S", + "%Y-%m-%dT%H:%M:%S.%f%z", + "%Y-%m-%dT%H:%M:%S%z", + "%Y-%m-%dT%H:%M:%S.%f", + "%Y-%m-%dT%H:%M:%S", + "%Y-%m-%d", + ): + try: + return datetime.strptime(text, fmt) + except ValueError: + continue + return None diff --git a/src/bookfusion_mcp/config.py b/src/bookfusion_mcp/config.py new file mode 100644 index 0000000..f29b346 --- /dev/null +++ b/src/bookfusion_mcp/config.py @@ -0,0 +1,55 @@ +"""Runtime configuration loaded from environment variables.""" + +from __future__ import annotations + +import os +from dataclasses import dataclass +from pathlib import Path + +from dotenv import load_dotenv + +DEFAULT_API_BASE = "https://www.bookfusion.com/calibre-api/v1" +DEFAULT_TIMEOUT = 120 +DEFAULT_THREADS = 2 + + +@dataclass(frozen=True) +class Settings: + api_key: str + api_base: str + library_path: Path + timeout: float + upload_threads: int + + @property + def metadata_db(self) -> Path: + return self.library_path / "metadata.db" + + +def load_settings() -> Settings: + """Load settings from .env (if present) and the process environment.""" + load_dotenv(override=False) + + api_key = os.environ.get("BOOKFUSION_API_KEY", "").strip() + if not api_key: + raise RuntimeError( + "BOOKFUSION_API_KEY is not set. Add it to your environment or .env file." + ) + + library_path_str = os.environ.get("CALIBRE_LIBRARY_PATH", "").strip() + if not library_path_str: + raise RuntimeError( + "CALIBRE_LIBRARY_PATH is not set. Point it at the directory that contains " + "metadata.db (e.g. the host path mapped to /calibre-library inside the " + "calibre-web-automated container)." + ) + + library_path = Path(library_path_str).expanduser().resolve() + + return Settings( + api_key=api_key, + api_base=os.environ.get("BOOKFUSION_API_BASE", DEFAULT_API_BASE).rstrip("/"), + library_path=library_path, + timeout=float(os.environ.get("BOOKFUSION_TIMEOUT", DEFAULT_TIMEOUT)), + upload_threads=int(os.environ.get("BOOKFUSION_UPLOAD_THREADS", DEFAULT_THREADS)), + ) diff --git a/src/bookfusion_mcp/server.py b/src/bookfusion_mcp/server.py new file mode 100644 index 0000000..d924f37 --- /dev/null +++ b/src/bookfusion_mcp/server.py @@ -0,0 +1,194 @@ +"""MCP server exposing BookFusion sync tools for a Calibre / CWA library.""" + +from __future__ import annotations + +import asyncio +from typing import Any + +from mcp.server.fastmcp import FastMCP + +from .bookfusion_api import BookFusionClient, BookFusionError +from .calibre_library import CalibreLibrary, file_digest, metadata_digest +from .config import Settings, load_settings +from .sync import sync_book + +mcp = FastMCP("bookfusion") + +_settings: Settings | None = None +_library: CalibreLibrary | None = None + + +def _get_settings() -> Settings: + global _settings + if _settings is None: + _settings = load_settings() + return _settings + + +def _get_library() -> CalibreLibrary: + global _library + if _library is None: + _library = CalibreLibrary(_get_settings().library_path) + return _library + + +def _client() -> BookFusionClient: + s = _get_settings() + return BookFusionClient(api_key=s.api_key, api_base=s.api_base, timeout=s.timeout) + + +# ---------------------------------------------------------------- read-only + +@mcp.tool() +def list_books(limit: int = 50, offset: int = 0, search: str | None = None) -> dict[str, Any]: + """List books from the local Calibre library. + + Args: + limit: Maximum number of books to return (default 50, max 500). + offset: Pagination offset. + search: Substring to match against title or author name. + """ + limit = max(1, min(int(limit), 500)) + offset = max(0, int(offset)) + library = _get_library() + books = library.list_books(limit=limit, offset=offset, search=search) + total = library.count_books() + return { + "total": total, + "returned": len(books), + "offset": offset, + "limit": limit, + "books": [b.to_summary_dict() for b in books], + } + + +@mcp.tool() +def get_book(book_id: int) -> dict[str, Any]: + """Return full details about a single book in the local library.""" + book = _get_library().get_book(int(book_id)) + if book is None: + return {"error": f"book {book_id} not found"} + data = book.to_summary_dict() + data["summary"] = book.summary + data["pubdate"] = book.pubdate.isoformat() if book.pubdate else None + data["last_modified"] = book.last_modified.isoformat() if book.last_modified else None + data["identifiers"] = book.identifiers + fmt = book.best_format() + if fmt is not None: + data["best_format"] = {"format": fmt.format, "size": fmt.size, "path": str(fmt.path)} + data["file_digest"] = file_digest(fmt.path) + data["metadata_digest"] = metadata_digest(book) + return data + + +@mcp.tool() +def library_status() -> dict[str, Any]: + """Return basic info about the configured Calibre library.""" + settings = _get_settings() + library = _get_library() + return { + "library_path": str(settings.library_path), + "metadata_db": str(settings.metadata_db), + "book_count": library.count_books(), + "api_base": settings.api_base, + } + + +# --------------------------------------------------------------- BookFusion + +@mcp.tool() +async def get_bookfusion_limits() -> dict[str, Any]: + """Fetch upload limits (max filesize, etc.) from BookFusion.""" + async with _client() as client: + try: + return await client.get_limits() + except BookFusionError as exc: + return {"error": str(exc), "status_code": exc.status_code} + + +@mcp.tool() +async def check_book_in_bookfusion(book_id: int) -> dict[str, Any]: + """Look up a local book on BookFusion by bookfusion id, ISBN, or file digest.""" + book = _get_library().get_book(int(book_id)) + if book is None: + return {"error": f"book {book_id} not found"} + fmt = book.best_format() + digest = file_digest(fmt.path) if fmt else None + + async with _client() as client: + try: + if book.bookfusion_id: + upstream = await client.get_upload_by_id(book.bookfusion_id) + strategy = "bookfusion_id" + elif book.isbn: + results = await client.search_uploads_by_isbn(book.isbn) + upstream = results[0] if results else None + strategy = "isbn" + elif digest: + upstream = await client.get_upload_by_digest(digest) + strategy = "digest" + else: + return {"error": "no identifier and no file to digest"} + except BookFusionError as exc: + return {"error": str(exc), "status_code": exc.status_code} + + return { + "book_id": book.id, + "title": book.title, + "lookup_strategy": strategy, + "found": upstream is not None, + "upstream": upstream, + "local_metadata_digest": metadata_digest(book), + } + + +@mcp.tool() +async def sync_book_to_bookfusion(book_id: int, reupload: bool = False) -> dict[str, Any]: + """Upload (or update) a single book on BookFusion. + + Args: + book_id: Calibre book id. + reupload: If true, also re-send the file even when only metadata changed. + """ + library = _get_library() + async with _client() as client: + result = await sync_book(library, client, int(book_id), reupload=bool(reupload)) + return result.to_dict() + + +@mcp.tool() +async def sync_books_to_bookfusion( + book_ids: list[int], + reupload: bool = False, +) -> dict[str, Any]: + """Sync several books, running up to ``BOOKFUSION_UPLOAD_THREADS`` in parallel.""" + settings = _get_settings() + library = _get_library() + sem = asyncio.Semaphore(max(1, settings.upload_threads)) + + async with _client() as client: + async def _one(bid: int) -> dict[str, Any]: + async with sem: + result = await sync_book(library, client, int(bid), reupload=bool(reupload)) + return result.to_dict() + + results = await asyncio.gather(*[_one(b) for b in book_ids]) + + summary = {"uploaded": 0, "updated": 0, "skipped": 0, "failed": 0} + for r in results: + summary[r["action"]] = summary.get(r["action"], 0) + 1 + return {"summary": summary, "results": results} + + +@mcp.tool() +async def sync_all_books( + limit: int = 50, + offset: int = 0, + search: str | None = None, + reupload: bool = False, +) -> dict[str, Any]: + """Sync every book that matches the same filters as ``list_books``.""" + library = _get_library() + books = library.list_books(limit=limit, offset=offset, search=search) + book_ids = [b.id for b in books] + return await sync_books_to_bookfusion(book_ids, reupload=reupload) diff --git a/src/bookfusion_mcp/sync.py b/src/bookfusion_mcp/sync.py new file mode 100644 index 0000000..aa842f3 --- /dev/null +++ b/src/bookfusion_mcp/sync.py @@ -0,0 +1,193 @@ +"""High-level sync flow used by MCP tools. + +Reproduces the behaviour of ``upload_worker.py`` from the BookFusion Calibre +plugin: check whether the book already exists upstream (by BookFusion id, ISBN, +or content digest), then either upload + finalize a new copy or update the +existing one's metadata (and optionally re-upload the file). +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Literal + +from .bookfusion_api import BookFusionClient, BookFusionError, BookMetadata +from .calibre_library import ( + CalibreBook, + CalibreLibrary, + file_digest, + metadata_digest, +) + + +SyncAction = Literal["uploaded", "updated", "skipped", "failed"] + + +@dataclass +class SyncResult: + book_id: int + title: str + action: SyncAction + bookfusion_id: str | None = None + message: str | None = None + + def to_dict(self) -> dict: + return { + "book_id": self.book_id, + "title": self.title, + "action": self.action, + "bookfusion_id": self.bookfusion_id, + "message": self.message, + } + + +def _build_metadata(book: CalibreBook, digest: str) -> BookMetadata: + pub = book.pubdate + issued_on = pub.date().isoformat() if pub else None + if issued_on == "0101-01-01": + issued_on = None + + series_items: list[dict] = [] + if book.series: + series_items.append({"title": book.series, "index": book.series_index}) + + return BookMetadata( + calibre_metadata_digest=digest, + title=book.title, + summary=book.summary, + language=book.language, + isbn=book.isbn, + issued_on=issued_on, + authors=list(book.authors), + tags=list(book.tags), + series=series_items, + cover_path=book.cover_path, + ) + + +async def _check_existing( + client: BookFusionClient, book: CalibreBook, digest: str +) -> dict | None: + """Mirror the priority order used by the upstream plugin.""" + if book.bookfusion_id: + try: + return await client.get_upload_by_id(book.bookfusion_id) + except BookFusionError: + return None + if book.isbn: + results = await client.search_uploads_by_isbn(book.isbn) + if results: + return results[0] + return None + return await client.get_upload_by_digest(digest) + + +async def sync_book( + library: CalibreLibrary, + client: BookFusionClient, + book_id: int, + *, + reupload: bool = False, +) -> SyncResult: + book = library.get_book(book_id) + if book is None: + return SyncResult(book_id=book_id, title="", action="failed", message="Book not found") + + fmt = book.best_format() + if fmt is None: + return SyncResult( + book_id=book.id, + title=book.title, + action="failed", + message="No supported file format on disk", + ) + + # Pre-flight against BookFusion's filesize limit so we fail early. + try: + limits = await client.get_limits() + except BookFusionError as exc: + return SyncResult( + book_id=book.id, + title=book.title, + action="failed", + message=f"Could not fetch BookFusion limits: {exc}", + ) + max_size = int(limits.get("filesize", 0)) + if max_size and fmt.size > max_size: + return SyncResult( + book_id=book.id, + title=book.title, + action="failed", + message=( + f"File {fmt.path.name} is {fmt.size} bytes, exceeds BookFusion limit " + f"of {max_size} bytes" + ), + ) + + digest = file_digest(fmt.path) + md_digest = metadata_digest(book) + metadata = _build_metadata(book, md_digest) + + try: + existing = await _check_existing(client, book, digest) + except BookFusionError as exc: + return SyncResult( + book_id=book.id, + title=book.title, + action="failed", + message=str(exc), + ) + + if existing and existing.get("calibre_metadata_digest") == md_digest and not reupload: + return SyncResult( + book_id=book.id, + title=book.title, + action="skipped", + bookfusion_id=str(existing.get("id")) if existing.get("id") is not None else None, + message="Already in sync", + ) + + if existing: + bookfusion_id = str(existing["id"]) + try: + await client.update_upload( + bookfusion_id, + metadata=metadata, + file_path=fmt.path if reupload else None, + ) + except BookFusionError as exc: + return SyncResult( + book_id=book.id, + title=book.title, + action="failed", + bookfusion_id=bookfusion_id, + message=f"Update failed: {exc}", + ) + return SyncResult( + book_id=book.id, + title=book.title, + action="updated", + bookfusion_id=bookfusion_id, + ) + + # Brand-new upload: init → upload to presigned URL → finalize. + try: + init = await client.init_upload(filename=fmt.path.name, digest=digest) + await client.upload_to_storage(init["url"], init["params"], fmt.path) + finalized = await client.finalize_upload( + key=init["params"]["key"], digest=digest, metadata=metadata + ) + except BookFusionError as exc: + return SyncResult( + book_id=book.id, + title=book.title, + action="failed", + message=f"Upload failed: {exc}", + ) + + return SyncResult( + book_id=book.id, + title=book.title, + action="uploaded", + bookfusion_id=str(finalized.get("id")) if finalized.get("id") is not None else None, + ) diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_bookfusion_api.py b/tests/test_bookfusion_api.py new file mode 100644 index 0000000..f030464 --- /dev/null +++ b/tests/test_bookfusion_api.py @@ -0,0 +1,227 @@ +"""Sanity tests for the BookFusion API client and Calibre helpers. + +These tests do not hit the network — they exercise the request shape and the +local helpers using an httpx ``MockTransport``. +""" + +from __future__ import annotations + +import base64 +from pathlib import Path + +import httpx +import pytest + +from bookfusion_mcp.bookfusion_api import ( + BookFusionClient, + BookFusionError, + BookMetadata, + _basic_auth_header, +) +from bookfusion_mcp.calibre_library import file_digest + + +def test_basic_auth_header_matches_calibre_plugin(): + header = _basic_auth_header("MY_KEY") + expected = "Basic " + base64.b64encode(b"MY_KEY:").decode() + assert header == expected + + +def test_metadata_multipart_includes_expected_fields(): + md = BookMetadata( + calibre_metadata_digest="d1g3st", + title="The Hobbit", + summary="There and back again", + language="en", + isbn="978-0-261-10221-7", + issued_on="1937-09-21", + authors=["J. R. R. Tolkien"], + tags=["Fantasy"], + bookshelves=["Classics"], + series=[{"title": "Middle-earth", "index": 1}], + ) + data, files = md.as_multipart() + keys = [k for k, _ in data] + assert ("metadata[title]", "The Hobbit") in data + assert ("metadata[author_list][]", "J. R. R. Tolkien") in data + assert ("metadata[tag_list][]", "Fantasy") in data + assert ("metadata[series][][title]", "Middle-earth") in data + assert ("metadata[series][][index]", "1") in data + # The bookshelves marker must be present so that BookFusion clears the list first. + assert keys.count("metadata[bookshelves][]") == 2 + assert files == [] + + +@pytest.mark.asyncio +async def test_get_limits_uses_basic_auth_and_returns_json(): + captured: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + captured["url"] = str(request.url) + captured["auth"] = request.headers.get("authorization") + captured["ua"] = request.headers.get("user-agent") + return httpx.Response(200, json={"filesize": 1024}) + + transport = httpx.MockTransport(handler) + client = BookFusionClient("KEY", "https://example.test/api/v1") + client._client = httpx.AsyncClient( + transport=transport, + headers=client._client.headers, + ) + try: + result = await client.get_limits() + finally: + await client.aclose() + + assert result == {"filesize": 1024} + assert captured["url"].endswith("/api/v1/limits") + assert captured["auth"] == _basic_auth_header("KEY") + assert "BookFusion MCP Server" in captured["ua"] + + +@pytest.mark.asyncio +async def test_404_returns_none_for_lookup_helpers(): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(404, text="not found") + + client = BookFusionClient("KEY", "https://example.test/api/v1") + client._client = httpx.AsyncClient( + transport=httpx.MockTransport(handler), headers=client._client.headers + ) + try: + assert await client.get_upload_by_id("123") is None + assert await client.get_upload_by_digest("abc") is None + finally: + await client.aclose() + + +@pytest.mark.asyncio +async def test_401_raises_invalid_api_key(): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(401, text="nope") + + client = BookFusionClient("KEY", "https://example.test/api/v1") + client._client = httpx.AsyncClient( + transport=httpx.MockTransport(handler), headers=client._client.headers + ) + try: + with pytest.raises(BookFusionError) as exc: + await client.get_limits() + finally: + await client.aclose() + assert exc.value.status_code == 401 + + +@pytest.mark.asyncio +async def test_init_upload_sends_filename_and_digest(monkeypatch): + """``init_upload`` should POST to ``/uploads/init`` with the filename + digest.""" + from unittest.mock import AsyncMock + + captured: dict = {} + + async def fake_request(method, url, **kwargs): + captured["method"] = method + captured["url"] = url + captured["data"] = kwargs.get("data") + captured["files"] = kwargs.get("files") + return httpx.Response( + 200, + json={"url": "https://s3.test/upload", "params": {"key": "abc/def"}}, + request=httpx.Request(method, url), + ) + + client = BookFusionClient("KEY", "https://example.test/api/v1") + client._client.request = AsyncMock(side_effect=fake_request) # type: ignore[method-assign] + try: + resp = await client.init_upload(filename="hobbit.epub", digest="DEAD") + finally: + await client.aclose() + + assert resp["url"] == "https://s3.test/upload" + assert captured["method"] == "POST" + assert captured["url"].endswith("/uploads/init") + assert ("filename", "hobbit.epub") in captured["data"] + assert ("digest", "DEAD") in captured["data"] + assert not captured["files"] + + +@pytest.mark.asyncio +async def test_finalize_upload_sends_metadata_fields(): + """``finalize_upload`` must include the key, digest and metadata[*] fields.""" + from unittest.mock import AsyncMock + + captured: dict = {} + + async def fake_request(method, url, **kwargs): + captured["method"] = method + captured["url"] = url + captured["data"] = kwargs.get("data") + return httpx.Response(200, json={"id": 42}, request=httpx.Request(method, url)) + + client = BookFusionClient("KEY", "https://example.test/api/v1") + client._client.request = AsyncMock(side_effect=fake_request) # type: ignore[method-assign] + try: + result = await client.finalize_upload( + key="abc/def", + digest="DEAD", + metadata=BookMetadata( + calibre_metadata_digest="MD", + title="Hobbit", + authors=["Tolkien"], + tags=["Fantasy"], + ), + ) + finally: + await client.aclose() + + assert result == {"id": 42} + assert captured["method"] == "POST" + assert captured["url"].endswith("/uploads/finalize") + assert ("key", "abc/def") in captured["data"] + assert ("digest", "DEAD") in captured["data"] + assert ("metadata[title]", "Hobbit") in captured["data"] + assert ("metadata[author_list][]", "Tolkien") in captured["data"] + assert ("metadata[tag_list][]", "Fantasy") in captured["data"] + + +def test_file_digest_matches_plugin_algorithm(tmp_path: Path): + target = tmp_path / "book.epub" + target.write_bytes(b"hello world") + expected = file_digest(target) + # Recompute manually using the same algorithm to guard against drift. + from hashlib import sha256 + + h = sha256() + h.update(bytes(target.stat().st_size)) + h.update(b"\0") + h.update(b"hello world") + assert expected == h.hexdigest() + + +def test_settings_loader_requires_api_key(monkeypatch, tmp_path: Path): + from bookfusion_mcp.config import load_settings + + monkeypatch.delenv("BOOKFUSION_API_KEY", raising=False) + monkeypatch.setenv("CALIBRE_LIBRARY_PATH", str(tmp_path)) + with pytest.raises(RuntimeError, match="BOOKFUSION_API_KEY"): + load_settings() + + +def test_settings_loader_requires_library_path(monkeypatch, tmp_path: Path): + from bookfusion_mcp.config import load_settings + + monkeypatch.setenv("BOOKFUSION_API_KEY", "x") + monkeypatch.delenv("CALIBRE_LIBRARY_PATH", raising=False) + with pytest.raises(RuntimeError, match="CALIBRE_LIBRARY_PATH"): + load_settings() + + +def test_settings_loader_happy_path(monkeypatch, tmp_path: Path): + from bookfusion_mcp.config import load_settings + + monkeypatch.setenv("BOOKFUSION_API_KEY", "X") + monkeypatch.setenv("CALIBRE_LIBRARY_PATH", str(tmp_path)) + s = load_settings() + assert s.api_key == "X" + assert s.library_path == tmp_path.resolve() + assert s.api_base.endswith("/calibre-api/v1") diff --git a/tests/test_calibre_library.py b/tests/test_calibre_library.py new file mode 100644 index 0000000..b2b1d08 --- /dev/null +++ b/tests/test_calibre_library.py @@ -0,0 +1,158 @@ +"""End-to-end test against a hand-rolled metadata.db that mirrors the +schema Calibre / Calibre-Web-Automated produce on disk.""" + +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +import pytest + +from bookfusion_mcp.calibre_library import CalibreLibrary, file_digest, metadata_digest + + +CALIBRE_SCHEMA = """ +CREATE TABLE books ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + title TEXT, + sort TEXT, + pubdate TEXT, + timestamp TEXT, + last_modified TEXT, + series_index REAL, + path TEXT +); +CREATE TABLE authors (id INTEGER PRIMARY KEY, name TEXT); +CREATE TABLE books_authors_link ( + id INTEGER PRIMARY KEY, + book INTEGER, + author INTEGER +); +CREATE TABLE tags (id INTEGER PRIMARY KEY, name TEXT); +CREATE TABLE books_tags_link ( + id INTEGER PRIMARY KEY, + book INTEGER, + tag INTEGER +); +CREATE TABLE identifiers ( + id INTEGER PRIMARY KEY, + book INTEGER, + type TEXT, + val TEXT +); +CREATE TABLE languages (id INTEGER PRIMARY KEY, lang_code TEXT); +CREATE TABLE books_languages_link ( + id INTEGER PRIMARY KEY, + book INTEGER, + lang_code INTEGER, + item_order INTEGER +); +CREATE TABLE series (id INTEGER PRIMARY KEY, name TEXT); +CREATE TABLE books_series_link ( + id INTEGER PRIMARY KEY, + book INTEGER, + series INTEGER +); +CREATE TABLE comments (id INTEGER PRIMARY KEY, book INTEGER, text TEXT); +CREATE TABLE data ( + id INTEGER PRIMARY KEY, + book INTEGER, + format TEXT, + name TEXT, + uncompressed_size INTEGER +); +""" + + +def _make_library(tmp_path: Path) -> Path: + library = tmp_path / "library" + library.mkdir() + db_path = library / "metadata.db" + + conn = sqlite3.connect(db_path) + conn.executescript(CALIBRE_SCHEMA) + + # Book on disk: library/Tolkien/The Hobbit (1)/hobbit.epub + book_dir = library / "Tolkien" / "The Hobbit (1)" + book_dir.mkdir(parents=True) + (book_dir / "hobbit.epub").write_bytes(b"epub-bytes") + (book_dir / "cover.jpg").write_bytes(b"jpeg-bytes") + + conn.execute( + """INSERT INTO books (id, title, sort, pubdate, timestamp, last_modified, + series_index, path) + VALUES (1, 'The Hobbit', 'Hobbit, The', '1937-09-21 00:00:00', + '2024-01-01 00:00:00', '2024-06-01 12:00:00', 1.0, + 'Tolkien/The Hobbit (1)')""", + ) + conn.execute("INSERT INTO authors (id, name) VALUES (1, 'J. R. R. Tolkien')") + conn.execute("INSERT INTO books_authors_link (book, author) VALUES (1, 1)") + conn.execute("INSERT INTO tags (id, name) VALUES (1, 'Fantasy')") + conn.execute("INSERT INTO books_tags_link (book, tag) VALUES (1, 1)") + conn.execute( + "INSERT INTO identifiers (book, type, val) VALUES (1, 'isbn', '9780261102217')" + ) + conn.execute( + "INSERT INTO identifiers (book, type, val) VALUES (1, 'bookfusion', '4242')" + ) + conn.execute("INSERT INTO languages (id, lang_code) VALUES (1, 'eng')") + conn.execute( + "INSERT INTO books_languages_link (book, lang_code, item_order) VALUES (1, 1, 0)" + ) + conn.execute("INSERT INTO series (id, name) VALUES (1, 'Middle-earth')") + conn.execute("INSERT INTO books_series_link (book, series) VALUES (1, 1)") + conn.execute( + "INSERT INTO comments (book, text) VALUES (1, 'There and back again')" + ) + conn.execute( + "INSERT INTO data (book, format, name, uncompressed_size) " + "VALUES (1, 'EPUB', 'hobbit', 10)" + ) + conn.commit() + conn.close() + return library + + +def test_list_and_get_book(tmp_path: Path): + library = _make_library(tmp_path) + cl = CalibreLibrary(library) + + assert cl.count_books() == 1 + + books = cl.list_books() + assert len(books) == 1 + book = books[0] + assert book.id == 1 + assert book.title == "The Hobbit" + assert book.authors == ["J. R. R. Tolkien"] + assert book.tags == ["Fantasy"] + assert book.isbn == "9780261102217" + assert book.bookfusion_id == "4242" + assert book.series == "Middle-earth" + assert book.series_index == 1.0 + assert book.language == "eng" + assert book.summary == "There and back again" + assert book.cover_path is not None and book.cover_path.exists() + + fmt = book.best_format() + assert fmt is not None + assert fmt.format == "EPUB" + assert fmt.size == len(b"epub-bytes") + + # Digests should match the algorithm exposed by the helper. + assert file_digest(fmt.path) + assert metadata_digest(book) + + +def test_search_filters_by_title_or_author(tmp_path: Path): + library = _make_library(tmp_path) + cl = CalibreLibrary(library) + + assert len(cl.list_books(search="hobbit")) == 1 + assert len(cl.list_books(search="tolkien")) == 1 + assert cl.list_books(search="nope") == [] + + +def test_missing_metadata_db(tmp_path: Path): + with pytest.raises(FileNotFoundError): + CalibreLibrary(tmp_path)