mirror of
https://github.com/denizsafak/abogen.git
synced 2026-07-18 13:40:27 +02:00
feat: Add Calibre OPDS feed and import endpoints for enhanced integration
This commit is contained in:
JB
+140
-43
@@ -1,11 +1,14 @@
|
||||
from typing import Any, Dict, Mapping, List, Optional
|
||||
import base64
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
|
||||
from flask import Blueprint, request, jsonify, send_file, url_for
|
||||
from flask import Blueprint, request, jsonify, send_file, url_for, current_app
|
||||
from flask.typing import ResponseReturnValue
|
||||
|
||||
from abogen.web.routes.utils.settings import (
|
||||
load_settings,
|
||||
load_integration_settings,
|
||||
coerce_float,
|
||||
coerce_bool,
|
||||
)
|
||||
@@ -13,6 +16,7 @@ from abogen.voice_profiles import (
|
||||
load_profiles,
|
||||
save_profiles,
|
||||
delete_profile,
|
||||
serialize_profiles,
|
||||
)
|
||||
from abogen.web.routes.utils.preview import synthesize_preview, generate_preview_audio
|
||||
from abogen.normalization_settings import (
|
||||
@@ -23,8 +27,14 @@ from abogen.normalization_settings import (
|
||||
from abogen.llm_client import list_models, LLMClientError
|
||||
from abogen.kokoro_text_normalization import normalize_for_pipeline
|
||||
from abogen.integrations.audiobookshelf import AudiobookshelfClient, AudiobookshelfConfig
|
||||
from abogen.integrations.calibre_opds import CalibreOPDSClient
|
||||
from abogen.integrations.calibre_opds import (
|
||||
CalibreOPDSClient,
|
||||
CalibreOPDSError,
|
||||
feed_to_dict,
|
||||
)
|
||||
from abogen.web.routes.utils.service import get_service
|
||||
from abogen.web.routes.utils.form import build_pending_job_from_extraction
|
||||
from abogen.text_extractor import extract_from_path
|
||||
from werkzeug.utils import secure_filename
|
||||
|
||||
api_bp = Blueprint("api", __name__)
|
||||
@@ -79,6 +89,53 @@ def api_speaker_preview() -> ResponseReturnValue:
|
||||
|
||||
# --- Integration Routes ---
|
||||
|
||||
@api_bp.get("/integrations/calibre-opds/feed")
|
||||
def api_calibre_opds_feed() -> ResponseReturnValue:
|
||||
integrations = load_integration_settings()
|
||||
calibre_settings = integrations.get("calibre_opds", {})
|
||||
|
||||
payload = {
|
||||
"base_url": calibre_settings.get("base_url"),
|
||||
"username": calibre_settings.get("username"),
|
||||
"password": calibre_settings.get("password"),
|
||||
"verify_ssl": calibre_settings.get("verify_ssl", True),
|
||||
}
|
||||
|
||||
if not payload.get("base_url"):
|
||||
return jsonify({"error": "Calibre OPDS base URL is not configured."}), 400
|
||||
|
||||
try:
|
||||
client = CalibreOPDSClient(
|
||||
base_url=payload.get("base_url") or "",
|
||||
username=payload.get("username"),
|
||||
password=payload.get("password"),
|
||||
verify=bool(payload.get("verify_ssl", True)),
|
||||
)
|
||||
except ValueError as exc:
|
||||
return jsonify({"error": str(exc)}), 400
|
||||
|
||||
href = request.args.get("href", type=str)
|
||||
query = request.args.get("q", type=str)
|
||||
letter = request.args.get("letter", type=str)
|
||||
|
||||
try:
|
||||
if letter:
|
||||
feed = client.browse_letter(letter, start_href=href)
|
||||
elif query:
|
||||
feed = client.search(query)
|
||||
else:
|
||||
feed = client.fetch_feed(href)
|
||||
except CalibreOPDSError as exc:
|
||||
return jsonify({"error": str(exc)}), 502
|
||||
except Exception as exc:
|
||||
return jsonify({"error": f"Unexpected error: {str(exc)}"}), 500
|
||||
|
||||
return jsonify({
|
||||
"feed": feed_to_dict(feed),
|
||||
"href": href or "",
|
||||
"query": query or "",
|
||||
})
|
||||
|
||||
@api_bp.post("/integrations/audiobookshelf/folders")
|
||||
def api_abs_folders() -> ResponseReturnValue:
|
||||
payload = request.get_json(force=True, silent=True) or {}
|
||||
@@ -147,68 +204,108 @@ def api_calibre_opds_import() -> ResponseReturnValue:
|
||||
if not request.is_json:
|
||||
return jsonify({"error": "Expected JSON payload."}), 400
|
||||
|
||||
payload = request.get_json(force=True, silent=True) or {}
|
||||
href = payload.get("href")
|
||||
data = request.get_json(force=True, silent=True) or {}
|
||||
href = str(data.get("href") or "").strip()
|
||||
|
||||
if not href:
|
||||
return jsonify({"error": "Download URL (href) is required."}), 400
|
||||
|
||||
metadata_payload = data.get("metadata") if isinstance(data, Mapping) else None
|
||||
metadata_overrides: Dict[str, Any] = {}
|
||||
|
||||
if isinstance(metadata_payload, Mapping):
|
||||
def _stringify_metadata_value(value: Any) -> str:
|
||||
if value is None:
|
||||
return ""
|
||||
if isinstance(value, (list, tuple, set)):
|
||||
parts = [str(item).strip() for item in value if item is not None]
|
||||
parts = [part for part in parts if part]
|
||||
return ", ".join(parts)
|
||||
text = str(value).strip()
|
||||
return text
|
||||
|
||||
raw_series = metadata_payload.get("series") or metadata_payload.get("series_name")
|
||||
series_name = str(raw_series or "").strip()
|
||||
if series_name:
|
||||
metadata_overrides["series"] = series_name
|
||||
metadata_overrides.setdefault("series_name", series_name)
|
||||
|
||||
series_index_value = (
|
||||
metadata_payload.get("series_index")
|
||||
or metadata_payload.get("series_position")
|
||||
or metadata_payload.get("series_sequence")
|
||||
or metadata_payload.get("book_number")
|
||||
)
|
||||
if series_index_value is not None:
|
||||
series_index_text = str(series_index_value).strip()
|
||||
if series_index_text:
|
||||
metadata_overrides.setdefault("series_index", series_index_text)
|
||||
metadata_overrides.setdefault("series_position", series_index_text)
|
||||
metadata_overrides.setdefault("series_sequence", series_index_text)
|
||||
metadata_overrides.setdefault("book_number", series_index_text)
|
||||
|
||||
tags_value = metadata_payload.get("tags") or metadata_payload.get("keywords")
|
||||
if tags_value:
|
||||
tags_text = _stringify_metadata_value(tags_value)
|
||||
if tags_text:
|
||||
metadata_overrides.setdefault("tags", tags_text)
|
||||
metadata_overrides.setdefault("keywords", tags_text)
|
||||
metadata_overrides.setdefault("genre", tags_text)
|
||||
|
||||
description_value = metadata_payload.get("description") or metadata_payload.get("summary")
|
||||
if description_value:
|
||||
description_text = _stringify_metadata_value(description_value)
|
||||
if description_text:
|
||||
metadata_overrides.setdefault("description", description_text)
|
||||
metadata_overrides.setdefault("summary", description_text)
|
||||
|
||||
settings = load_settings()
|
||||
integrations = settings.get("integrations", {})
|
||||
integrations = load_integration_settings()
|
||||
calibre_settings = integrations.get("calibre_opds", {})
|
||||
|
||||
# We need to download the file
|
||||
# For now, let's just return a success message as a placeholder
|
||||
# In a real implementation, this would trigger a download job
|
||||
|
||||
# Re-using the logic from books.py if possible, or implementing it here
|
||||
# It seems books.py had a placeholder for this too.
|
||||
|
||||
# Let's try to actually download it using the service
|
||||
try:
|
||||
service = get_service()
|
||||
|
||||
client = CalibreOPDSClient(
|
||||
base_url=calibre_settings.get("base_url", ""),
|
||||
base_url=calibre_settings.get("base_url") or "",
|
||||
username=calibre_settings.get("username"),
|
||||
password=calibre_settings.get("password"),
|
||||
verify=calibre_settings.get("verify_ssl", True),
|
||||
verify=bool(calibre_settings.get("verify_ssl", True)),
|
||||
)
|
||||
|
||||
with client._open_client() as http_client:
|
||||
response = http_client.get(href, follow_redirects=True)
|
||||
response.raise_for_status()
|
||||
temp_dir = Path(current_app.config.get("UPLOAD_FOLDER", "uploads"))
|
||||
temp_dir.mkdir(exist_ok=True)
|
||||
|
||||
# Try to get filename from content-disposition
|
||||
filename = "downloaded_book.epub"
|
||||
if "content-disposition" in response.headers:
|
||||
import re
|
||||
fname = re.findall("filename=(.+)", response.headers["content-disposition"])
|
||||
if fname:
|
||||
filename = fname[0].strip('"')
|
||||
resource = client.download(href)
|
||||
filename = resource.filename
|
||||
content = resource.content
|
||||
|
||||
filename = secure_filename(filename)
|
||||
if not filename:
|
||||
filename = f"{uuid.uuid4().hex}.epub"
|
||||
|
||||
# Save to uploads folder
|
||||
uploads_dir = service._uploads_root
|
||||
target_path = uploads_dir / filename
|
||||
file_path = temp_dir / f"{uuid.uuid4().hex}_{filename}"
|
||||
file_path.write_bytes(content)
|
||||
|
||||
# Ensure unique filename
|
||||
stem = target_path.stem
|
||||
suffix = target_path.suffix
|
||||
counter = 1
|
||||
while target_path.exists():
|
||||
target_path = uploads_dir / f"{stem}_{counter}{suffix}"
|
||||
counter += 1
|
||||
extraction = extract_from_path(file_path)
|
||||
|
||||
with open(target_path, "wb") as f:
|
||||
for chunk in response.iter_bytes(chunk_size=8192):
|
||||
f.write(chunk)
|
||||
if metadata_overrides:
|
||||
extraction.metadata.update(metadata_overrides)
|
||||
|
||||
result = build_pending_job_from_extraction(
|
||||
stored_path=file_path,
|
||||
original_name=filename,
|
||||
extraction=extraction,
|
||||
form={},
|
||||
settings=settings,
|
||||
profiles=serialize_profiles(),
|
||||
metadata_overrides=metadata_overrides,
|
||||
)
|
||||
|
||||
get_service().store_pending_job(result.pending)
|
||||
|
||||
return jsonify({
|
||||
"success": True,
|
||||
"message": "Book downloaded successfully.",
|
||||
"redirect": url_for("main.prepare_page", filename=target_path.name)
|
||||
"status": "imported",
|
||||
"pending_id": result.pending.id,
|
||||
"redirect": url_for("main.wizard_step", step="chapters", pending_id=result.pending.id)
|
||||
})
|
||||
|
||||
except Exception as e:
|
||||
|
||||
+2
-183
@@ -1,8 +1,6 @@
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, Mapping, Optional
|
||||
from typing import Any, Dict
|
||||
|
||||
from flask import Blueprint, render_template, request, jsonify, current_app, url_for
|
||||
from flask import Blueprint, render_template
|
||||
from flask.typing import ResponseReturnValue
|
||||
|
||||
from abogen.web.routes.utils.settings import (
|
||||
@@ -10,15 +8,6 @@ from abogen.web.routes.utils.settings import (
|
||||
load_integration_settings,
|
||||
)
|
||||
from abogen.web.routes.utils.voice import template_options
|
||||
from abogen.web.routes.utils.form import build_pending_job_from_extraction
|
||||
from abogen.web.routes.utils.service import get_service
|
||||
from abogen.integrations.calibre_opds import (
|
||||
CalibreOPDSClient,
|
||||
CalibreOPDSError,
|
||||
feed_to_dict,
|
||||
)
|
||||
from abogen.text_extractor import extract_from_path
|
||||
from abogen.voice_profiles import serialize_profiles
|
||||
|
||||
books_bp = Blueprint("books", __name__)
|
||||
|
||||
@@ -26,14 +15,6 @@ def _calibre_integration_enabled(integrations: Dict[str, Any]) -> bool:
|
||||
calibre = integrations.get("calibre_opds", {})
|
||||
return bool(calibre.get("enabled") and calibre.get("base_url"))
|
||||
|
||||
def _build_calibre_client(payload: Dict[str, Any]) -> CalibreOPDSClient:
|
||||
return CalibreOPDSClient(
|
||||
base_url=payload.get("base_url") or "",
|
||||
username=payload.get("username"),
|
||||
password=payload.get("password"),
|
||||
verify=bool(payload.get("verify_ssl", True)),
|
||||
)
|
||||
|
||||
@books_bp.get("/")
|
||||
def find_books_page() -> ResponseReturnValue:
|
||||
settings = load_settings()
|
||||
@@ -48,168 +29,6 @@ def find_books_page() -> ResponseReturnValue:
|
||||
|
||||
@books_bp.get("/search")
|
||||
def search_books() -> ResponseReturnValue:
|
||||
# This seems to be handled by the feed endpoint in the original code
|
||||
# But let's see if there is a separate search page or if it's all JS driven
|
||||
return find_books_page()
|
||||
|
||||
@books_bp.get("/calibre/feed")
|
||||
def calibre_opds_feed() -> ResponseReturnValue:
|
||||
integrations = load_integration_settings()
|
||||
calibre_settings = integrations.get("calibre_opds", {})
|
||||
|
||||
payload = {
|
||||
"base_url": calibre_settings.get("base_url"),
|
||||
"username": calibre_settings.get("username"),
|
||||
"password": calibre_settings.get("password"),
|
||||
"verify_ssl": calibre_settings.get("verify_ssl", True),
|
||||
}
|
||||
|
||||
if not payload.get("base_url"):
|
||||
return jsonify({"error": "Calibre OPDS base URL is not configured."}), 400
|
||||
|
||||
try:
|
||||
client = _build_calibre_client(payload)
|
||||
except ValueError as exc:
|
||||
return jsonify({"error": str(exc)}), 400
|
||||
|
||||
href = request.args.get("href", type=str)
|
||||
query = request.args.get("q", type=str)
|
||||
letter = request.args.get("letter", type=str)
|
||||
|
||||
try:
|
||||
if letter:
|
||||
feed = client.browse_letter(letter, start_href=href)
|
||||
elif query:
|
||||
feed = client.search(query)
|
||||
else:
|
||||
feed = client.fetch_feed(href)
|
||||
except CalibreOPDSError as exc:
|
||||
return jsonify({"error": str(exc)}), 502
|
||||
|
||||
return jsonify({
|
||||
"feed": feed_to_dict(feed),
|
||||
"href": href or "",
|
||||
"query": query or "",
|
||||
})
|
||||
|
||||
@books_bp.post("/calibre/import")
|
||||
def calibre_opds_import() -> ResponseReturnValue:
|
||||
if not request.is_json:
|
||||
return jsonify({"error": "Expected JSON payload."}), 400
|
||||
|
||||
data = request.get_json(silent=True) or {}
|
||||
href = str(data.get("href") or "").strip()
|
||||
title = str(data.get("title") or "").strip()
|
||||
|
||||
if not href:
|
||||
return jsonify({"error": "Download link missing."}), 400
|
||||
|
||||
metadata_payload = data.get("metadata") if isinstance(data, Mapping) else None
|
||||
metadata_overrides: Dict[str, Any] = {}
|
||||
|
||||
if isinstance(metadata_payload, Mapping):
|
||||
def _stringify_metadata_value(value: Any) -> str:
|
||||
if value is None:
|
||||
return ""
|
||||
if isinstance(value, (list, tuple, set)):
|
||||
parts = [str(item).strip() for item in value if item is not None]
|
||||
parts = [part for part in parts if part]
|
||||
return ", ".join(parts)
|
||||
text = str(value).strip()
|
||||
return text
|
||||
|
||||
raw_series = metadata_payload.get("series") or metadata_payload.get("series_name")
|
||||
series_name = str(raw_series or "").strip()
|
||||
if series_name:
|
||||
metadata_overrides["series"] = series_name
|
||||
metadata_overrides.setdefault("series_name", series_name)
|
||||
|
||||
series_index_value = (
|
||||
metadata_payload.get("series_index")
|
||||
or metadata_payload.get("series_position")
|
||||
or metadata_payload.get("series_sequence")
|
||||
or metadata_payload.get("book_number")
|
||||
)
|
||||
if series_index_value is not None:
|
||||
series_index_text = str(series_index_value).strip()
|
||||
if series_index_text:
|
||||
metadata_overrides.setdefault("series_index", series_index_text)
|
||||
metadata_overrides.setdefault("series_position", series_index_text)
|
||||
metadata_overrides.setdefault("series_sequence", series_index_text)
|
||||
metadata_overrides.setdefault("book_number", series_index_text)
|
||||
|
||||
tags_value = metadata_payload.get("tags") or metadata_payload.get("keywords")
|
||||
if tags_value:
|
||||
tags_text = _stringify_metadata_value(tags_value)
|
||||
if tags_text:
|
||||
metadata_overrides.setdefault("tags", tags_text)
|
||||
metadata_overrides.setdefault("keywords", tags_text)
|
||||
metadata_overrides.setdefault("genre", tags_text)
|
||||
|
||||
description_value = metadata_payload.get("description") or metadata_payload.get("summary")
|
||||
if description_value:
|
||||
description_text = _stringify_metadata_value(description_value)
|
||||
if description_text:
|
||||
metadata_overrides.setdefault("description", description_text)
|
||||
metadata_overrides.setdefault("summary", description_text)
|
||||
|
||||
settings = load_settings()
|
||||
integrations = settings.get("integrations", {})
|
||||
calibre_settings = integrations.get("calibre", {})
|
||||
|
||||
payload = {
|
||||
"base_url": calibre_settings.get("url"),
|
||||
"username": calibre_settings.get("username"),
|
||||
"password": calibre_settings.get("password"),
|
||||
"verify_ssl": True,
|
||||
}
|
||||
|
||||
try:
|
||||
client = _build_calibre_client(payload)
|
||||
temp_dir = Path(current_app.config.get("UPLOAD_FOLDER", "uploads"))
|
||||
temp_dir.mkdir(exist_ok=True)
|
||||
|
||||
# We don't know the filename yet, so we'll use a temp name and rename later if possible
|
||||
# Or rely on content-disposition if the client supports it, but here we just download content
|
||||
# The client.download_book returns bytes or path?
|
||||
# Let's check CalibreClient.download_book
|
||||
|
||||
# Assuming it returns bytes for now based on typical usage
|
||||
# But wait, I need to check abogen/integrations/calibre_opds.py
|
||||
|
||||
resource = client.download(href)
|
||||
filename = resource.filename
|
||||
content = resource.content
|
||||
|
||||
if not filename:
|
||||
filename = f"{uuid.uuid4().hex}.epub" # Default to epub if unknown
|
||||
|
||||
file_path = temp_dir / f"{uuid.uuid4().hex}_{filename}"
|
||||
file_path.write_bytes(content)
|
||||
|
||||
extraction = extract_from_path(file_path)
|
||||
|
||||
# Apply metadata overrides to extraction if possible, or pass them to build_pending_job
|
||||
if metadata_overrides:
|
||||
extraction.metadata.update(metadata_overrides)
|
||||
|
||||
result = build_pending_job_from_extraction(
|
||||
stored_path=file_path,
|
||||
original_name=filename,
|
||||
extraction=extraction,
|
||||
form={}, # No form data for defaults
|
||||
settings=settings,
|
||||
profiles=serialize_profiles(),
|
||||
metadata_overrides=metadata_overrides,
|
||||
)
|
||||
|
||||
get_service().store_pending_job(result.pending)
|
||||
|
||||
return jsonify({
|
||||
"status": "imported",
|
||||
"pending_id": result.pending.id,
|
||||
"redirect": url_for("main.wizard_step", step="chapters", pending_id=result.pending.id)
|
||||
})
|
||||
|
||||
except Exception as exc:
|
||||
return jsonify({"error": str(exc)}), 500
|
||||
|
||||
Reference in New Issue
Block a user