tim
312dc3efc5
Bindery mounts no CWA database. It searches the OPDS feed by title and reads the story URL inside each candidate epub to find the library book. Updates go to the ingest folder with a .cwa.json sidecar, so CWA replaces the epub of the existing book.
84 lines
3.3 KiB
Python
84 lines
3.3 KiB
Python
"""Optional, read-only access to Calibre-Web accounts and the calibre library."""
|
|
|
|
import sqlite3
|
|
from collections.abc import Iterator
|
|
from contextlib import closing, contextmanager
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from werkzeug.security import check_password_hash
|
|
|
|
# Calibre-Web marks anonymous accounts with this role bit.
|
|
ROLE_ANONYMOUS = 1 << 5
|
|
|
|
|
|
class DatabaseUnavailable(Exception):
|
|
"""The database cannot answer. The caller can try CWA over HTTP."""
|
|
|
|
|
|
@contextmanager
|
|
def _connect(path: Path) -> Iterator[sqlite3.Connection]:
|
|
try:
|
|
# Do not use immutable=1. CWA can change the database and its WAL at any time.
|
|
with closing(sqlite3.connect(path.resolve().as_uri() + "?mode=ro", uri=True, timeout=1)) as db:
|
|
db.row_factory = sqlite3.Row
|
|
db.execute("PRAGMA query_only = ON")
|
|
db.execute("BEGIN")
|
|
yield db
|
|
except (OSError, sqlite3.Error) as e:
|
|
raise DatabaseUnavailable(f"Cannot read {path}: {e}") from e
|
|
|
|
|
|
def login_ok(path: Path, username: str, password: str) -> bool:
|
|
"""False means rejected credentials, not a reason to try another login source."""
|
|
with _connect(path) as db:
|
|
users = db.execute(
|
|
"SELECT name, password, role FROM user WHERE name = ? COLLATE NOCASE LIMIT 2", (username,)
|
|
).fetchall()
|
|
if len(users) != 1 or not password:
|
|
return False
|
|
user = users[0]
|
|
if not isinstance(user["role"], int):
|
|
raise DatabaseUnavailable("The Calibre-Web account has no valid role")
|
|
if user["name"].casefold() == "guest" or user["role"] & ROLE_ANONYMOUS or not user["password"]:
|
|
return False
|
|
if not isinstance(user["password"], str):
|
|
raise DatabaseUnavailable("The Calibre-Web password hash is not text")
|
|
try:
|
|
return check_password_hash(user["password"], password)
|
|
except ValueError as e:
|
|
raise DatabaseUnavailable("The Calibre-Web password hash is not supported") from e
|
|
|
|
|
|
def _book(db: sqlite3.Connection, rows: list[sqlite3.Row]) -> dict[str, Any] | None:
|
|
# A duplicate UUID or story identifier is not enough to choose an update target.
|
|
if len(rows) != 1 or not rows[0]["uuid"]:
|
|
return None
|
|
book = dict(rows[0])
|
|
book["authors"] = [
|
|
row[0]
|
|
for row in db.execute(
|
|
"SELECT a.name FROM authors a JOIN books_authors_link l ON l.author = a.id WHERE l.book = ? ORDER BY l.id",
|
|
(book["id"],),
|
|
)
|
|
]
|
|
book["formats"] = [row[0] for row in db.execute("SELECT format FROM data WHERE book = ?", (book["id"],))]
|
|
return book
|
|
|
|
|
|
def book_by_uuid(path: Path, book_uuid: str) -> dict[str, Any] | None:
|
|
with _connect(path) as db:
|
|
rows = db.execute("SELECT id, uuid, title FROM books WHERE uuid = ? LIMIT 2", (book_uuid,)).fetchall()
|
|
return _book(db, rows)
|
|
|
|
|
|
def book_by_url(path: Path, story_url: str) -> dict[str, Any] | None:
|
|
"""Match an exact FanFicFare URL identifier. Other URLs use the OPDS search."""
|
|
with _connect(path) as db:
|
|
rows = db.execute(
|
|
"SELECT DISTINCT b.id, b.uuid, b.title FROM books b JOIN identifiers i ON i.book = b.id "
|
|
"WHERE i.type = 'url' COLLATE NOCASE AND i.val = ? COLLATE BINARY LIMIT 2",
|
|
(story_url,),
|
|
).fetchall()
|
|
return _book(db, rows)
|