vrr/FGU-bookfusion
2
0
Fork 0
mirror of https://github.com/Fguarien/FGU-bookfusion.git synced 2026-10-19 02:06:16 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Add BookFusion MCP server for Calibre / Calibre-Web-Automated

Browse source 
Introduces a stdio MCP server that lets Claude sync books from a Calibre
library (including the one managed by Calibre-Web-Automated) to a BookFusion
account. Reuses the same HTTP API, request shape and digest algorithms as the
official BookFusion Calibre plugin so existing accounts and metadata stay in
sync.

- bookfusion_api: async httpx client mirroring /uploads/init, /uploads/finalize,
  /uploads/{id}, /uploads?isbn=..., /limits with HTTP Basic auth.
- calibre_library: read-only reader for metadata.db plus the file/metadata
  digest helpers from the upstream plugin.
- sync: high-level flow (lookup by id/ISBN/digest -> upload or update).
- server: 8 MCP tools (library_status, list_books, get_book,
  get_bookfusion_limits, check_book_in_bookfusion, sync_book_to_bookfusion,
  sync_books_to_bookfusion, sync_all_books).
- config: env-var based settings (BOOKFUSION_API_KEY, CALIBRE_LIBRARY_PATH...).
- 14 pytest tests covering API client, settings loader and library reader.
This commit is contained in:
Claude 2026-04-14 12:19:55 +00:00
parent 2c8d787771
commit 53292df67d
No known key found for this signature in database
14 changed files with 1702 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

16
.env.example Normal file
View file

@ -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

25
.gitignore vendored
View file

@ -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

143
README.md Normal file
View file

@ -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.

41
pyproject.toml Normal file
View file

@ -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"]

3
src/bookfusion_mcp/__init__.py Normal file
View file

@ -0,0 +1,3 @@
"""BookFusion MCP server for Calibre / Calibre-Web-Automated libraries."""
__version__ = "0.1.0"

13
src/bookfusion_mcp/__main__.py Normal file
View file

@ -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()

268
src/bookfusion_mcp/bookfusion_api.py Normal file
View file

@ -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 <presigned upload URL>`` (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": <max bytes>}``."""
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,
)

366
src/bookfusion_mcp/calibre_library.py Normal file
View file

@ -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
* ``<Author>/<Book Title> (<id>)/<filename>.<ext>`` — book files
* ``<Author>/<Book Title> (<id>)/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

55
src/bookfusion_mcp/config.py Normal file
View file

@ -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)),
)

194
src/bookfusion_mcp/server.py Normal file
View file

@ -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)

193
src/bookfusion_mcp/sync.py Normal file
View file

@ -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,
)

0
tests/__init__.py Normal file
View file

227
tests/test_bookfusion_api.py Normal file
View file

@ -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")

158
tests/test_calibre_library.py Normal file
View file

@ -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)