Add voice management functionality and voice synthesis preview

- Implemented voice management routes in `voices.py` for listing, saving, and deleting speaker configurations.
- Added a test endpoint for voice synthesis preview, allowing users to test voice settings with provided text and speed.
- Introduced utility functions in `voice.py` for building voice catalogs, resolving voice settings, and synthesizing audio from normalized text.
- Enhanced speaker roster building and configuration application logic to support dynamic voice settings.
This commit is contained in:
JB
2025-11-28 14:57:23 -08:00
parent 0a2b3533f4
commit d08cbcfdc9
23 changed files with 4780 additions and 5463 deletions
+15 -2
View File
@@ -95,9 +95,22 @@ def create_app(config: Optional[dict[str, Any]] = None) -> Flask:
)
app.extensions["conversion_service"] = service
from .routes import web_bp, api_bp
from abogen.web.routes import (
main_bp,
jobs_bp,
settings_bp,
voices_bp,
entities_bp,
books_bp,
api_bp,
)
app.register_blueprint(web_bp)
app.register_blueprint(main_bp)
app.register_blueprint(jobs_bp, url_prefix="/jobs")
app.register_blueprint(settings_bp, url_prefix="/settings")
app.register_blueprint(voices_bp, url_prefix="/voices")
app.register_blueprint(entities_bp, url_prefix="/entities")
app.register_blueprint(books_bp, url_prefix="/find-books")
app.register_blueprint(api_bp, url_prefix="/api")
atexit.register(service.shutdown)
-5442
View File
File diff suppressed because it is too large Load Diff
+18
View File
@@ -0,0 +1,18 @@
from abogen.web.routes.main import main_bp
from abogen.web.routes.jobs import jobs_bp
from abogen.web.routes.settings import settings_bp
from abogen.web.routes.voices import voices_bp
from abogen.web.routes.entities import entities_bp
from abogen.web.routes.books import books_bp
from abogen.web.routes.api import api_bp
__all__ = [
"main_bp",
"jobs_bp",
"settings_bp",
"voices_bp",
"entities_bp",
"books_bp",
"api_bp",
]
+213
View File
@@ -0,0 +1,213 @@
from typing import Any, Dict, Mapping, List, Optional
from flask import Blueprint, request, jsonify, send_file
from flask.typing import ResponseReturnValue
from abogen.web.routes.utils.settings import (
load_settings,
coerce_float,
)
from abogen.voice_profiles import (
load_profiles,
save_profiles,
delete_profile,
)
from abogen.web.routes.utils.preview import synthesize_preview
from abogen.normalization_settings import (
build_llm_configuration,
build_apostrophe_config,
apply_overrides,
)
from abogen.llm_client import list_models, LLMClientError
from abogen.kokoro_text_normalization import normalize_for_pipeline
from abogen.integrations.audiobookshelf import AudiobookshelfClient, AudiobookshelfConfig
api_bp = Blueprint("api", __name__)
# --- Voice Profile Routes ---
@api_bp.get("/voice-profiles")
def api_get_voice_profiles() -> ResponseReturnValue:
profiles = load_profiles()
return jsonify(profiles)
@api_bp.post("/voice-profiles")
def api_save_voice_profile() -> ResponseReturnValue:
payload = request.get_json(force=True, silent=True) or {}
name = payload.get("name")
profile = payload.get("profile")
if not name or not profile:
return jsonify({"error": "Name and profile are required"}), 400
profiles = load_profiles()
profiles[name] = profile
save_profiles(profiles)
return jsonify({"success": True})
@api_bp.delete("/voice-profiles/<path:name>")
def api_delete_voice_profile(name: str) -> ResponseReturnValue:
delete_profile(name)
return jsonify({"success": True})
@api_bp.post("/speaker-preview")
def api_speaker_preview() -> ResponseReturnValue:
payload = request.get_json(force=True, silent=True) or {}
text = payload.get("text", "Hello world")
voice = payload.get("voice", "af_heart")
language = payload.get("language", "a")
speed = coerce_float(payload.get("speed"), 1.0)
settings = load_settings()
use_gpu = settings.get("use_gpu", False)
try:
return synthesize_preview(
text=text,
voice_spec=voice,
language=language,
speed=speed,
use_gpu=use_gpu
)
except Exception as e:
return jsonify({"error": str(e)}), 500
# --- Integration Routes ---
@api_bp.post("/integrations/audiobookshelf/folders")
def api_abs_folders() -> ResponseReturnValue:
payload = request.get_json(force=True, silent=True) or {}
host = payload.get("host")
token = payload.get("token")
if not host or not token:
return jsonify({"error": "Host and token are required"}), 400
try:
config = AudiobookshelfConfig(base_url=host, api_token=token)
client = AudiobookshelfClient(config)
folders = client.get_libraries()
return jsonify({"folders": folders})
except Exception as e:
return jsonify({"error": str(e)}), 400
@api_bp.post("/integrations/audiobookshelf/test")
def api_abs_test() -> ResponseReturnValue:
payload = request.get_json(force=True, silent=True) or {}
host = payload.get("host")
token = payload.get("token")
if not host or not token:
return jsonify({"error": "Host and token are required"}), 400
try:
config = AudiobookshelfConfig(base_url=host, api_token=token)
client = AudiobookshelfClient(config)
# Just getting libraries is a good enough test
client.get_libraries()
return jsonify({"success": True})
except Exception as e:
return jsonify({"error": str(e)}), 400
# --- LLM Routes ---
@api_bp.post("/llm/models")
def api_llm_models() -> ResponseReturnValue:
payload = request.get_json(force=True, silent=False) or {}
current_settings = load_settings()
base_url = str(payload.get("base_url") or payload.get("llm_base_url") or current_settings.get("llm_base_url") or "").strip()
if not base_url:
return jsonify({"error": "LLM base URL is required."}), 400
api_key = str(payload.get("api_key") or payload.get("llm_api_key") or current_settings.get("llm_api_key") or "")
timeout = coerce_float(payload.get("timeout"), current_settings.get("llm_timeout", 30.0))
overrides = {
"llm_base_url": base_url,
"llm_api_key": api_key,
"llm_timeout": timeout,
}
merged = apply_overrides(current_settings, overrides)
configuration = build_llm_configuration(merged)
try:
models = list_models(configuration)
except LLMClientError as exc:
return jsonify({"error": str(exc)}), 400
return jsonify({"models": models})
@api_bp.post("/llm/preview")
def api_llm_preview() -> ResponseReturnValue:
payload = request.get_json(force=True, silent=False) or {}
sample_text = str(payload.get("text") or "").strip()
if not sample_text:
return jsonify({"error": "Text is required."}), 400
base_settings = load_settings()
overrides: Dict[str, Any] = {
"llm_base_url": str(
payload.get("base_url")
or payload.get("llm_base_url")
or base_settings.get("llm_base_url")
or ""
).strip(),
"llm_api_key": str(
payload.get("api_key")
or payload.get("llm_api_key")
or base_settings.get("llm_api_key")
or ""
),
"llm_model": str(
payload.get("model")
or payload.get("llm_model")
or base_settings.get("llm_model")
or ""
),
"llm_prompt": payload.get("prompt") or payload.get("llm_prompt") or base_settings.get("llm_prompt"),
"llm_context_mode": payload.get("context_mode") or base_settings.get("llm_context_mode"),
"llm_timeout": coerce_float(payload.get("timeout"), base_settings.get("llm_timeout", 30.0)),
"normalization_apostrophe_mode": "llm",
}
merged = apply_overrides(base_settings, overrides)
if not merged.get("llm_base_url"):
return jsonify({"error": "LLM base URL is required."}), 400
if not merged.get("llm_model"):
return jsonify({"error": "Select an LLM model before previewing."}), 400
apostrophe_config = build_apostrophe_config(settings=merged)
try:
normalized_text = normalize_for_pipeline(sample_text, config=apostrophe_config, settings=merged)
except LLMClientError as exc:
return jsonify({"error": str(exc)}), 400
context = {
"text": sample_text,
"normalized_text": normalized_text,
}
return jsonify(context)
# --- Normalization Routes ---
@api_bp.post("/normalization/preview")
def api_normalization_preview() -> ResponseReturnValue:
payload = request.get_json(force=True, silent=False) or {}
sample_text = str(payload.get("text") or "").strip()
if not sample_text:
return jsonify({"error": "Sample text is required."}), 400
base_settings = load_settings()
# We might want to apply overrides from payload if any normalization settings are passed
# For now, just use base settings as in original code (presumably)
apostrophe_config = build_apostrophe_config(settings=base_settings)
try:
normalized_text = normalize_for_pipeline(sample_text, config=apostrophe_config, settings=base_settings)
except Exception as exc:
return jsonify({"error": str(exc)}), 400
return jsonify({
"text": sample_text,
"normalized_text": normalized_text,
})
+216
View File
@@ -0,0 +1,216 @@
import uuid
from pathlib import Path
from typing import Any, Dict, Mapping, Optional
from flask import Blueprint, render_template, request, jsonify, current_app, url_for
from flask.typing import ResponseReturnValue
from abogen.web.routes.utils.settings import (
load_settings,
stored_integration_config,
)
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__)
def _calibre_integration_enabled(integrations: Dict[str, Any]) -> bool:
calibre = integrations.get("calibre", {})
return bool(calibre.get("enabled") and calibre.get("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()
integrations = settings.get("integrations", {})
return render_template(
"find_books.html",
integrations=integrations,
opds_available=_calibre_integration_enabled(integrations),
options=template_options(),
settings=settings,
)
@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:
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, # Default
}
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
+96
View File
@@ -0,0 +1,96 @@
from typing import Mapping
from flask import Blueprint, request, jsonify, abort
from flask.typing import ResponseReturnValue
from abogen.web.routes.utils.service import require_pending_job, get_service
from abogen.web.routes.utils.entity import (
refresh_entity_summary,
pending_entities_payload,
upsert_manual_override,
delete_manual_override,
search_manual_override_candidates,
)
from abogen.web.routes.utils.settings import coerce_int
entities_bp = Blueprint("entities", __name__)
@entities_bp.post("/analyze")
def analyze_entities() -> ResponseReturnValue:
# This might be triggered via wizard update, but if there's a specific route:
# In original routes.py, it was likely part of wizard logic or API.
# I'll assume this is for the API endpoint /api/pending/<id>/entities/refresh
pending_id = request.form.get("pending_id") or request.args.get("pending_id")
if not pending_id:
abort(400, "Pending ID required")
pending = require_pending_job(pending_id)
refresh_entity_summary(pending, pending.chapters)
get_service().store_pending_job(pending)
return jsonify(pending_entities_payload(pending))
@entities_bp.get("/pending/<pending_id>")
def get_entities(pending_id: str) -> ResponseReturnValue:
pending = require_pending_job(pending_id)
refresh_flag = (request.args.get("refresh") or "").strip().lower()
expected_cache = (request.args.get("cache_key") or "").strip()
refresh_requested = refresh_flag in {"1", "true", "yes", "force"}
if expected_cache and expected_cache != (pending.entity_cache_key or ""):
refresh_requested = True
if refresh_requested or not pending.entity_summary:
refresh_entity_summary(pending, pending.chapters)
get_service().store_pending_job(pending)
return jsonify(pending_entities_payload(pending))
@entities_bp.post("/pending/<pending_id>/refresh")
def refresh_entities(pending_id: str) -> ResponseReturnValue:
pending = require_pending_job(pending_id)
refresh_entity_summary(pending, pending.chapters)
get_service().store_pending_job(pending)
return jsonify(pending_entities_payload(pending))
@entities_bp.get("/pending/<pending_id>/overrides")
def list_manual_overrides(pending_id: str) -> ResponseReturnValue:
pending = require_pending_job(pending_id)
return jsonify({
"overrides": pending.manual_overrides or [],
"pronunciation_overrides": pending.pronunciation_overrides or [],
"language": pending.language or "en",
})
@entities_bp.post("/pending/<pending_id>/overrides")
def upsert_override(pending_id: str) -> ResponseReturnValue:
pending = require_pending_job(pending_id)
payload = request.get_json(silent=True) or {}
if not isinstance(payload, Mapping):
abort(400, "Invalid override payload")
try:
override = upsert_manual_override(pending, payload)
except ValueError as exc:
abort(400, str(exc))
get_service().store_pending_job(pending)
return jsonify({"override": override, **pending_entities_payload(pending)})
@entities_bp.delete("/pending/<pending_id>/overrides/<override_id>")
def delete_override(pending_id: str, override_id: str) -> ResponseReturnValue:
pending = require_pending_job(pending_id)
deleted = delete_manual_override(pending, override_id)
if not deleted:
abort(404)
get_service().store_pending_job(pending)
return jsonify({"deleted": True, **pending_entities_payload(pending)})
@entities_bp.get("/pending/<pending_id>/overrides/search")
def search_candidates(pending_id: str) -> ResponseReturnValue:
pending = require_pending_job(pending_id)
query = (request.args.get("q") or request.args.get("query") or "").strip()
limit_param = request.args.get("limit")
limit_value = coerce_int(limit_param, 15, minimum=1, maximum=50) if limit_param is not None else 15
results = search_manual_override_candidates(pending, query, limit=limit_value)
return jsonify({"query": query, "limit": limit_value, "results": results})
+276
View File
@@ -0,0 +1,276 @@
import json
import logging
from pathlib import Path
from typing import Any, Dict, Optional
from flask import Blueprint, Response, abort, redirect, render_template, request, url_for, send_file
from flask.typing import ResponseReturnValue
from abogen.web.service import (
JobStatus,
load_audiobookshelf_chapters,
build_audiobookshelf_metadata,
)
from abogen.web.routes.utils.service import get_service
from abogen.web.routes.utils.form import render_jobs_panel
from abogen.web.routes.utils.voice import template_options
from abogen.web.routes.utils.epub import (
job_download_flags,
locate_job_epub,
locate_job_audio,
)
from abogen.web.routes.utils.settings import (
stored_integration_config,
build_audiobookshelf_config,
coerce_bool,
)
from abogen.web.routes.utils.common import existing_paths
from abogen.integrations.audiobookshelf import AudiobookshelfClient, AudiobookshelfUploadError
logger = logging.getLogger(__name__)
jobs_bp = Blueprint("jobs", __name__)
@jobs_bp.get("/<job_id>")
def job_detail(job_id: str) -> str:
job = get_service().get_job(job_id)
if not job:
abort(404)
return render_template(
"job_detail.html",
job=job,
options=template_options(),
JobStatus=JobStatus,
downloads=job_download_flags(job),
)
@jobs_bp.post("/<job_id>/pause")
def pause_job(job_id: str) -> ResponseReturnValue:
get_service().pause(job_id)
if request.headers.get("HX-Request"):
return render_jobs_panel()
return redirect(url_for("jobs.job_detail", job_id=job_id))
@jobs_bp.post("/<job_id>/resume")
def resume_job(job_id: str) -> ResponseReturnValue:
get_service().resume(job_id)
if request.headers.get("HX-Request"):
return render_jobs_panel()
return redirect(url_for("jobs.job_detail", job_id=job_id))
@jobs_bp.post("/<job_id>/cancel")
def cancel_job(job_id: str) -> ResponseReturnValue:
get_service().cancel(job_id)
if request.headers.get("HX-Request"):
return render_jobs_panel()
return redirect(url_for("jobs.job_detail", job_id=job_id))
@jobs_bp.post("/<job_id>/delete")
def delete_job(job_id: str) -> ResponseReturnValue:
get_service().delete(job_id)
if request.headers.get("HX-Request"):
return render_jobs_panel()
return redirect(url_for("main.index"))
@jobs_bp.post("/<job_id>/retry")
def retry_job(job_id: str) -> ResponseReturnValue:
new_job = get_service().retry(job_id)
if request.headers.get("HX-Request"):
return render_jobs_panel()
if new_job:
return redirect(url_for("jobs.job_detail", job_id=new_job.id))
return redirect(url_for("jobs.job_detail", job_id=job_id))
@jobs_bp.post("/<job_id>/audiobookshelf")
def send_job_to_audiobookshelf(job_id: str) -> ResponseReturnValue:
service = get_service()
job = service.get_job(job_id)
if job is None:
abort(404)
def _panel_response() -> ResponseReturnValue:
if request.headers.get("HX-Request"):
return render_jobs_panel()
return redirect(url_for("jobs.job_detail", job_id=job.id))
if job.status != JobStatus.COMPLETED:
return _panel_response()
settings = stored_integration_config("audiobookshelf")
if not settings or not coerce_bool(settings.get("enabled"), False):
job.add_log("Audiobookshelf upload skipped: integration is disabled.", level="warning")
service._persist_state()
return _panel_response()
config = build_audiobookshelf_config(settings)
if config is None:
job.add_log(
"Audiobookshelf upload skipped: configure base URL, API token, and library ID first.",
level="warning",
)
service._persist_state()
return _panel_response()
if not config.folder_id:
job.add_log(
"Audiobookshelf upload skipped: enter the folder name or ID in the Audiobookshelf settings.",
level="warning",
)
service._persist_state()
return _panel_response()
audio_path = locate_job_audio(job)
if not audio_path or not audio_path.exists():
job.add_log("Audiobookshelf upload skipped: audio output not found.", level="warning")
service._persist_state()
return _panel_response()
cover_path = None
if config.send_cover and job.cover_image_path:
cover_candidate = job.cover_image_path
if not isinstance(cover_candidate, Path):
cover_candidate = Path(str(cover_candidate))
if cover_candidate.exists():
cover_path = cover_candidate
subtitles = existing_paths(job.result.subtitle_paths) if config.send_subtitles else None
chapters = load_audiobookshelf_chapters(job) if config.send_chapters else None
metadata = build_audiobookshelf_metadata(job)
display_title = metadata.get("title") or audio_path.stem
overwrite_requested = request.form.get("overwrite") == "true" or request.args.get("overwrite") == "true"
try:
client = AudiobookshelfClient(config)
except ValueError as exc:
job.add_log(f"Audiobookshelf configuration error: {exc}", level="error")
service._persist_state()
return _panel_response()
try:
existing_items = client.find_existing_items(display_title, folder_id=config.folder_id)
except AudiobookshelfUploadError as exc:
job.add_log(f"Audiobookshelf lookup failed: {exc}", level="error")
service._persist_state()
return _panel_response()
if existing_items and not overwrite_requested:
job.add_log(
f"Audiobookshelf already contains '{display_title}'. Awaiting overwrite confirmation.",
level="warning",
)
service._persist_state()
if request.headers.get("HX-Request"):
detail = {
"jobId": job.id,
"title": display_title,
"url": url_for("jobs.send_job_to_audiobookshelf", job_id=job.id),
"target": request.headers.get("HX-Target") or "#jobs-panel",
"message": f'Audiobookshelf already contains "{display_title}". Overwrite?',
}
headers = {"HX-Trigger": json.dumps({"audiobookshelf-overwrite-prompt": detail})}
return Response("", status=204, headers=headers)
return _panel_response()
if existing_items and overwrite_requested:
try:
client.delete_items(existing_items)
except AudiobookshelfUploadError as exc:
job.add_log(f"Audiobookshelf overwrite aborted: {exc}", level="error")
service._persist_state()
return _panel_response()
else:
job.add_log(
f"Removed {len(existing_items)} existing Audiobookshelf item(s) prior to overwrite.",
level="info",
)
job.add_log("Audiobookshelf upload triggered manually.", level="info")
try:
client.upload_audiobook(
audio_path,
metadata=metadata,
cover_path=cover_path,
chapters=chapters,
subtitles=subtitles,
)
except AudiobookshelfUploadError as exc:
job.add_log(f"Audiobookshelf upload failed: {exc}", level="error")
except Exception as exc:
job.add_log(f"Audiobookshelf integration error: {exc}", level="error")
else:
job.add_log("Audiobookshelf upload queued.", level="success")
finally:
service._persist_state()
return _panel_response()
@jobs_bp.post("/clear-finished")
def clear_finished_jobs() -> ResponseReturnValue:
get_service().clear_finished()
if request.headers.get("HX-Request"):
return render_jobs_panel()
return redirect(url_for("main.index", _anchor="queue"))
@jobs_bp.get("/<job_id>/epub")
def job_epub(job_id: str) -> ResponseReturnValue:
job = get_service().get_job(job_id)
if job is None or job.status != JobStatus.COMPLETED:
abort(404)
epub_path = locate_job_epub(job)
if not epub_path:
abort(404)
return send_file(
epub_path,
as_attachment=True,
download_name=epub_path.name,
mimetype="application/epub+zip",
)
@jobs_bp.get("/<job_id>/download/<file_type>")
def download_file(job_id: str, file_type: str) -> ResponseReturnValue:
job = get_service().get_job(job_id)
if not job or job.status != JobStatus.COMPLETED:
abort(404)
if file_type == "audio":
path = locate_job_audio(job)
if not path or not path.exists():
abort(404)
return send_file(
path,
as_attachment=True,
download_name=path.name,
)
# Handle other file types if needed (subtitles, etc.)
# For now, just audio and epub are explicitly handled
abort(404)
@jobs_bp.get("/<job_id>/logs")
def job_logs(job_id: str) -> str:
job = get_service().get_job(job_id)
if not job:
abort(404)
return render_template("job_logs_static.html", job=job)
@jobs_bp.get("/<job_id>/logs/stream")
def stream_logs(job_id: str) -> ResponseReturnValue:
job = get_service().get_job(job_id)
if not job:
abort(404)
def generate():
last_index = 0
while True:
current_logs = job.logs
if len(current_logs) > last_index:
for log in current_logs[last_index:]:
yield f"data: {json.dumps({'timestamp': log.timestamp, 'level': log.level, 'message': log.message})}\n\n"
last_index = len(current_logs)
if job.status in {JobStatus.COMPLETED, JobStatus.FAILED, JobStatus.CANCELLED}:
break
import time
time.sleep(0.5)
return Response(generate(), mimetype="text/event-stream")
+329
View File
@@ -0,0 +1,329 @@
import logging
import time
import uuid
from pathlib import Path
from typing import Any, Dict, Optional, cast
from flask import Blueprint, redirect, render_template, request, url_for, jsonify, current_app
from werkzeug.utils import secure_filename
from abogen.web.service import PendingJob
from abogen.web.routes.utils.service import get_service, remove_pending_job, submit_job
from abogen.web.routes.utils.settings import load_settings
from abogen.web.routes.utils.voice import template_options
from abogen.web.routes.utils.form import (
normalize_wizard_step,
wants_wizard_json,
render_wizard_partial,
wizard_json_response,
build_pending_job_from_extraction,
apply_book_step_form,
apply_prepare_form,
render_jobs_panel,
)
from abogen.text_extractor import extract_from_path
from abogen.voice_profiles import serialize_profiles
logger = logging.getLogger(__name__)
main_bp = Blueprint("main", __name__)
@main_bp.app_template_filter("datetimeformat")
def datetimeformat(value: float, fmt: str = "%Y-%m-%d %H:%M:%S") -> str:
if not value:
return ""
from datetime import datetime
return datetime.fromtimestamp(value).strftime(fmt)
@main_bp.route("/")
def index():
pending_id = request.args.get("pending_id")
pending = get_service().get_pending_job(pending_id) if pending_id else None
# If we have a pending job, redirect to the wizard
if pending:
step_index = getattr(pending, "wizard_max_step_index", 0)
# Map index to step name roughly
steps = ["book", "chapters", "entities"]
step_name = steps[min(step_index, len(steps)-1)]
return redirect(url_for("main.wizard_step", step=step_name, pending_id=pending.id))
return render_template(
"index.html",
options=template_options(),
settings=load_settings(),
jobs_panel=render_jobs_panel(),
)
@main_bp.route("/wizard")
def wizard_start():
pending_id = request.args.get("pending_id")
if pending_id:
return redirect(url_for("main.wizard_step", step="book", pending_id=pending_id))
return redirect(url_for("main.wizard_step", step="book"))
@main_bp.route("/wizard/<step>")
def wizard_step(step: str):
pending_id = request.args.get("pending_id")
pending = get_service().get_pending_job(pending_id) if pending_id else None
normalized_step = normalize_wizard_step(step, pending)
if normalized_step != step:
return redirect(url_for("main.wizard_step", step=normalized_step, pending_id=pending_id))
if wants_wizard_json():
return wizard_json_response(pending, normalized_step)
return render_template(
"index.html",
options=template_options(),
settings=load_settings(),
jobs_panel=render_jobs_panel(),
wizard_mode=True,
wizard_step=normalized_step,
wizard_partial=render_wizard_partial(pending, normalized_step),
)
@main_bp.route("/wizard/upload", methods=["POST"])
def wizard_upload():
file = request.files.get("file")
if not file or not file.filename:
if wants_wizard_json():
return wizard_json_response(None, "book", error="No file selected", status=400)
return redirect(url_for("main.wizard_step", step="book"))
filename = secure_filename(file.filename)
temp_dir = Path(current_app.config.get("UPLOAD_FOLDER", "uploads"))
temp_dir.mkdir(exist_ok=True)
file_path = temp_dir / f"{uuid.uuid4().hex}_{filename}"
file.save(file_path)
settings = load_settings()
profiles = serialize_profiles()
try:
extraction = extract_from_path(file_path)
result = build_pending_job_from_extraction(
stored_path=file_path,
original_name=filename,
extraction=extraction,
form=request.form,
settings=settings,
profiles=profiles,
)
get_service().store_pending_job(result.pending)
if wants_wizard_json():
return wizard_json_response(result.pending, "chapters")
return redirect(url_for("main.wizard_step", step="chapters", pending_id=result.pending.id))
except Exception as e:
logger.exception("Error processing upload")
if file_path.exists():
try:
file_path.unlink()
except OSError:
pass
error_msg = f"Failed to process file: {str(e)}"
if wants_wizard_json():
return wizard_json_response(None, "book", error=error_msg, status=500)
return render_template(
"index.html",
options=template_options(),
settings=settings,
jobs_panel=render_jobs_panel(),
wizard_mode=True,
wizard_step="book",
wizard_partial=render_wizard_partial(None, "book", error=error_msg),
)
@main_bp.route("/wizard/text", methods=["POST"])
def wizard_text():
text = request.form.get("text", "").strip()
title = request.form.get("title", "").strip() or "Pasted Text"
if not text:
if wants_wizard_json():
return wizard_json_response(None, "book", error="No text provided", status=400)
return redirect(url_for("main.wizard_step", step="book"))
temp_dir = Path(current_app.config.get("UPLOAD_FOLDER", "uploads"))
temp_dir.mkdir(exist_ok=True)
file_path = temp_dir / f"{uuid.uuid4().hex}.txt"
file_path.write_text(text, encoding="utf-8")
settings = load_settings()
profiles = serialize_profiles()
try:
extraction = extract_from_path(file_path)
# Override title since text extraction might not find one
extraction.metadata["title"] = title
result = build_pending_job_from_extraction(
stored_path=file_path,
original_name=f"{title}.txt",
extraction=extraction,
form=request.form,
settings=settings,
profiles=profiles,
)
get_service().store_pending_job(result.pending)
if wants_wizard_json():
return wizard_json_response(result.pending, "chapters")
return redirect(url_for("main.wizard_step", step="chapters", pending_id=result.pending.id))
except Exception as e:
logger.exception("Error processing text")
if file_path.exists():
try:
file_path.unlink()
except OSError:
pass
error_msg = f"Failed to process text: {str(e)}"
if wants_wizard_json():
return wizard_json_response(None, "book", error=error_msg, status=500)
return render_template(
"index.html",
options=template_options(),
settings=settings,
jobs_panel=render_jobs_panel(),
wizard_mode=True,
wizard_step="book",
wizard_partial=render_wizard_partial(None, "book", error=error_msg),
)
@main_bp.route("/wizard/update", methods=["POST"])
def wizard_update():
pending_id = request.form.get("pending_id")
if not pending_id:
if wants_wizard_json():
return wizard_json_response(None, "book", error="Missing job ID", status=400)
return redirect(url_for("main.wizard_step", step="book"))
pending = get_service().get_pending_job(pending_id)
if not pending:
if wants_wizard_json():
return wizard_json_response(None, "book", error="Job expired or not found", status=404)
return redirect(url_for("main.wizard_step", step="book"))
current_step = request.form.get("step", "book")
next_step = request.form.get("next_step")
settings = load_settings()
profiles = serialize_profiles()
try:
if current_step == "book":
apply_book_step_form(pending, request.form, settings=settings, profiles=profiles)
target_step = next_step or "chapters"
elif current_step == "chapters":
# This step involves re-analyzing chunks if needed
(
chunk_level,
overrides,
enabled_overrides,
errors,
selected_total,
selected_config,
apply_config_requested,
persist_config_requested,
) = apply_prepare_form(pending, request.form)
if errors:
if wants_wizard_json():
return wizard_json_response(pending, current_step, error="\n".join(errors), status=400)
return render_template(
"index.html",
options=template_options(),
settings=settings,
jobs_panel=render_jobs_panel(),
wizard_mode=True,
wizard_step=current_step,
wizard_partial=render_wizard_partial(pending, current_step, error="\n".join(errors)),
)
target_step = next_step or "entities"
elif current_step == "entities":
# Just saving entity overrides
apply_prepare_form(pending, request.form)
target_step = next_step or "entities" # Stay or finish
else:
target_step = "book"
get_service().store_pending_job(pending)
if wants_wizard_json():
return wizard_json_response(pending, target_step)
return redirect(url_for("main.wizard_step", step=target_step, pending_id=pending.id))
except Exception as e:
logger.exception(f"Error updating wizard step {current_step}")
error_msg = f"Update failed: {str(e)}"
if wants_wizard_json():
return wizard_json_response(pending, current_step, error=error_msg, status=500)
return render_template(
"index.html",
options=template_options(),
settings=settings,
jobs_panel=render_jobs_panel(),
wizard_mode=True,
wizard_step=current_step,
wizard_partial=render_wizard_partial(pending, current_step, error=error_msg),
)
@main_bp.route("/wizard/cancel", methods=["POST"])
def wizard_cancel():
pending_id = request.form.get("pending_id")
if pending_id:
remove_pending_job(pending_id)
if wants_wizard_json():
return jsonify({"status": "cancelled", "redirect": url_for("main.index")})
return redirect(url_for("main.index"))
@main_bp.route("/wizard/finish", methods=["POST"])
def wizard_finish():
pending_id = request.form.get("pending_id")
if not pending_id:
if wants_wizard_json():
return jsonify({"error": "Missing job ID"}), 400
return redirect(url_for("main.index"))
pending = get_service().get_pending_job(pending_id)
if not pending:
if wants_wizard_json():
return jsonify({"error": "Job not found"}), 404
return redirect(url_for("main.index"))
# Final update from form
apply_prepare_form(pending, request.form)
# Submit job
job_id = submit_job(pending)
if wants_wizard_json():
return jsonify({
"status": "submitted",
"job_id": job_id,
"redirect": url_for("main.index"),
"jobs_panel": render_jobs_panel()
})
return redirect(url_for("main.index"))
+121
View File
@@ -0,0 +1,121 @@
from flask import Blueprint, render_template, request, redirect, url_for, flash
from flask.typing import ResponseReturnValue
from abogen.web.routes.utils.settings import (
load_settings,
save_settings,
coerce_bool,
coerce_int,
_NORMALIZATION_BOOLEAN_KEYS,
_NORMALIZATION_STRING_KEYS,
_DEFAULT_ANALYSIS_THRESHOLD,
)
from abogen.web.routes.utils.voice import template_options
settings_bp = Blueprint("settings", __name__)
@settings_bp.get("/")
def settings_page() -> str:
return render_template(
"settings.html",
settings=load_settings(),
options=template_options(),
)
@settings_bp.post("/update")
def update_settings() -> ResponseReturnValue:
current = load_settings()
form = request.form
# General settings
current["language"] = (form.get("language") or "en").strip()
current["default_voice"] = (form.get("default_voice") or "").strip()
current["output_format"] = (form.get("output_format") or "mp3").strip()
current["subtitle_mode"] = (form.get("subtitle_mode") or "Disabled").strip()
current["subtitle_format"] = (form.get("subtitle_format") or "srt").strip()
current["save_mode"] = (form.get("save_mode") or "save_next_to_input").strip()
current["replace_single_newlines"] = coerce_bool(form.get("replace_single_newlines"), False)
current["use_gpu"] = coerce_bool(form.get("use_gpu"), False)
current["save_chapters_separately"] = coerce_bool(form.get("save_chapters_separately"), False)
current["merge_chapters_at_end"] = coerce_bool(form.get("merge_chapters_at_end"), True)
current["save_as_project"] = coerce_bool(form.get("save_as_project"), False)
current["separate_chapters_format"] = (form.get("separate_chapters_format") or "wav").strip()
try:
current["silence_between_chapters"] = max(0.0, float(form.get("silence_between_chapters", 2.0)))
except ValueError:
pass
try:
current["chapter_intro_delay"] = max(0.0, float(form.get("chapter_intro_delay", 0.5)))
except ValueError:
pass
current["read_title_intro"] = coerce_bool(form.get("read_title_intro"), False)
current["read_closing_outro"] = coerce_bool(form.get("read_closing_outro"), True)
current["normalize_chapter_opening_caps"] = coerce_bool(form.get("normalize_chapter_opening_caps"), True)
current["auto_prefix_chapter_titles"] = coerce_bool(form.get("auto_prefix_chapter_titles"), True)
try:
current["max_subtitle_words"] = max(1, int(form.get("max_subtitle_words", 50)))
except ValueError:
pass
current["chunk_level"] = (form.get("chunk_level") or "paragraph").strip()
current["generate_epub3"] = coerce_bool(form.get("generate_epub3"), False)
current["speaker_analysis_threshold"] = coerce_int(
form.get("speaker_analysis_threshold"),
_DEFAULT_ANALYSIS_THRESHOLD,
minimum=1,
maximum=25,
)
# Normalization settings
for key in _NORMALIZATION_BOOLEAN_KEYS:
current[key] = coerce_bool(form.get(key), False)
for key in _NORMALIZATION_STRING_KEYS:
current[key] = (form.get(key) or "").strip()
# Integrations
# Audiobookshelf
abs_enabled = coerce_bool(form.get("audiobookshelf_enabled"), False)
abs_url = (form.get("audiobookshelf_url") or "").strip()
abs_token = (form.get("audiobookshelf_token") or "").strip()
abs_library = (form.get("audiobookshelf_library_id") or "").strip()
abs_folder = (form.get("audiobookshelf_folder_id") or "").strip()
abs_cover = coerce_bool(form.get("audiobookshelf_send_cover"), True)
abs_chapters = coerce_bool(form.get("audiobookshelf_send_chapters"), True)
abs_subtitles = coerce_bool(form.get("audiobookshelf_send_subtitles"), True)
current["integrations"] = current.get("integrations", {})
current["integrations"]["audiobookshelf"] = {
"enabled": abs_enabled,
"url": abs_url,
"token": abs_token,
"library_id": abs_library,
"folder_id": abs_folder,
"send_cover": abs_cover,
"send_chapters": abs_chapters,
"send_subtitles": abs_subtitles,
}
# Calibre
calibre_enabled = coerce_bool(form.get("calibre_enabled"), False)
calibre_url = (form.get("calibre_url") or "").strip()
calibre_user = (form.get("calibre_username") or "").strip()
calibre_pass = (form.get("calibre_password") or "").strip()
calibre_library = (form.get("calibre_library_id") or "").strip()
current["integrations"]["calibre"] = {
"enabled": calibre_enabled,
"url": calibre_url,
"username": calibre_user,
"password": calibre_pass,
"library_id": calibre_library,
}
save_settings(current)
flash("Settings updated successfully.", "success")
return redirect(url_for("settings.settings_page"))
+17
View File
@@ -0,0 +1,17 @@
from typing import Any, Optional, Tuple, Iterable, List
from pathlib import Path
def split_profile_spec(value: Any) -> Tuple[str, Optional[str]]:
text = str(value or "").strip()
if not text:
return "", None
if text.lower().startswith("profile:"):
_, _, remainder = text.partition(":")
name = remainder.strip()
return "", name or None
return text, None
def existing_paths(paths: Optional[Iterable[Path]]) -> List[Path]:
if not paths:
return []
return [p for p in paths if p.exists()]
+348
View File
@@ -0,0 +1,348 @@
import time
import uuid
from typing import Any, Dict, Iterable, List, Mapping, Optional
from abogen.web.service import PendingJob
from abogen.entity_analysis import (
extract_entities,
merge_override,
normalize_token as normalize_entity_token,
search_tokens as search_entity_tokens,
)
from abogen.pronunciation_store import (
delete_override as delete_pronunciation_override,
load_overrides as load_pronunciation_overrides,
save_override as save_pronunciation_override,
search_overrides as search_pronunciation_overrides,
)
from abogen.web.routes.utils.settings import load_settings
def collect_pronunciation_overrides(pending: PendingJob) -> List[Dict[str, Any]]:
language = pending.language or "en"
collected: Dict[str, Dict[str, Any]] = {}
summary = pending.entity_summary or {}
for group in ("people", "entities"):
entries = summary.get(group)
if not isinstance(entries, list):
continue
for entry in entries:
if not isinstance(entry, Mapping):
continue
override_payload = entry.get("override")
if not isinstance(override_payload, Mapping):
continue
token_value = str(entry.get("label") or override_payload.get("token") or "").strip()
pronunciation_value = str(override_payload.get("pronunciation") or "").strip()
if not token_value or not pronunciation_value:
continue
normalized = normalize_entity_token(entry.get("normalized") or token_value)
if not normalized:
continue
collected[normalized] = {
"token": token_value,
"normalized": normalized,
"pronunciation": pronunciation_value,
"voice": str(override_payload.get("voice") or "").strip() or None,
"notes": str(override_payload.get("notes") or "").strip() or None,
"context": str(override_payload.get("context") or "").strip() or None,
"source": f"{group}-override",
"language": language,
}
if isinstance(pending.speakers, Mapping):
for speaker_payload in pending.speakers.values():
if not isinstance(speaker_payload, Mapping):
continue
token_value = str(speaker_payload.get("label") or "").strip()
pronunciation_value = str(speaker_payload.get("pronunciation") or "").strip()
if not token_value or not pronunciation_value:
continue
normalized = normalize_entity_token(token_value)
if not normalized:
continue
collected[normalized] = {
"token": token_value,
"normalized": normalized,
"pronunciation": pronunciation_value,
"voice": str(
speaker_payload.get("resolved_voice")
or speaker_payload.get("voice")
or pending.voice
).strip()
or None,
"notes": None,
"context": None,
"source": "speaker",
"language": language,
}
for manual_entry in pending.manual_overrides or []:
if not isinstance(manual_entry, Mapping):
continue
token_value = str(manual_entry.get("token") or "").strip()
pronunciation_value = str(manual_entry.get("pronunciation") or "").strip()
if not token_value or not pronunciation_value:
continue
normalized = manual_entry.get("normalized") or normalize_entity_token(token_value)
if not normalized:
continue
collected[normalized] = {
"token": token_value,
"normalized": normalized,
"pronunciation": pronunciation_value,
"voice": str(manual_entry.get("voice") or "").strip() or None,
"notes": str(manual_entry.get("notes") or "").strip() or None,
"context": str(manual_entry.get("context") or "").strip() or None,
"source": str(manual_entry.get("source") or "manual"),
"language": language,
}
return list(collected.values())
def sync_pronunciation_overrides(pending: PendingJob) -> None:
pending.pronunciation_overrides = collect_pronunciation_overrides(pending)
if not pending.pronunciation_overrides:
return
summary = pending.entity_summary or {}
manual_map: Dict[str, Mapping[str, Any]] = {}
for override in pending.manual_overrides or []:
if not isinstance(override, Mapping):
continue
normalized = override.get("normalized") or normalize_entity_token(override.get("token") or "")
pronunciation_value = str(override.get("pronunciation") or "").strip()
if not normalized or not pronunciation_value:
continue
manual_map[normalized] = override
for group in ("people", "entities"):
entries = summary.get(group)
if not isinstance(entries, list):
continue
for entry in entries:
if not isinstance(entry, dict):
continue
normalized = normalize_entity_token(entry.get("normalized") or entry.get("label") or "")
manual_override = manual_map.get(normalized)
if manual_override:
entry["override"] = {
"token": manual_override.get("token"),
"pronunciation": manual_override.get("pronunciation"),
"voice": manual_override.get("voice"),
"notes": manual_override.get("notes"),
"context": manual_override.get("context"),
"source": manual_override.get("source"),
}
def refresh_entity_summary(pending: PendingJob, chapters: Iterable[Mapping[str, Any]]) -> None:
settings = load_settings()
if not bool(settings.get("enable_entity_recognition", True)):
pending.entity_summary = {}
pending.entity_cache_key = ""
pending.pronunciation_overrides = pending.pronunciation_overrides or []
return
language = pending.language or "en"
chapter_list: List[Mapping[str, Any]] = [chapter for chapter in chapters if isinstance(chapter, Mapping)]
if not chapter_list:
pending.entity_summary = {}
pending.entity_cache_key = ""
pending.pronunciation_overrides = pending.pronunciation_overrides or []
return
enabled_only = [chapter for chapter in chapter_list if chapter.get("enabled")]
target_chapters = enabled_only or chapter_list
result = extract_entities(target_chapters, language=language)
summary = dict(result.summary)
tokens: List[str] = []
for group in ("people", "entities"):
entries = summary.get(group)
if not isinstance(entries, list):
continue
for entry in entries:
if not isinstance(entry, Mapping):
continue
token_value = str(entry.get("normalized") or entry.get("label") or "").strip()
if token_value:
tokens.append(token_value)
overrides_from_store = load_pronunciation_overrides(language=language, tokens=tokens)
merged_summary = merge_override(summary, overrides_from_store)
if result.errors:
merged_summary["errors"] = list(result.errors)
merged_summary["cache_key"] = result.cache_key
pending.entity_summary = merged_summary
pending.entity_cache_key = result.cache_key
sync_pronunciation_overrides(pending)
def find_manual_override(pending: PendingJob, identifier: str) -> Optional[Dict[str, Any]]:
for entry in pending.manual_overrides or []:
if not isinstance(entry, dict):
continue
if entry.get("id") == identifier or entry.get("normalized") == identifier:
return entry
return None
def upsert_manual_override(pending: PendingJob, payload: Mapping[str, Any]) -> Dict[str, Any]:
token_value = str(payload.get("token") or "").strip()
if not token_value:
raise ValueError("Token is required")
pronunciation_value = str(payload.get("pronunciation") or "").strip()
voice_value = str(payload.get("voice") or "").strip()
notes_value = str(payload.get("notes") or "").strip()
context_value = str(payload.get("context") or "").strip()
normalized = payload.get("normalized") or normalize_entity_token(token_value)
if not normalized:
raise ValueError("Token is required")
existing = find_manual_override(pending, payload.get("id", "")) or find_manual_override(pending, normalized)
timestamp = time.time()
language = pending.language or "en"
if existing:
existing.update(
{
"token": token_value,
"normalized": normalized,
"pronunciation": pronunciation_value,
"voice": voice_value,
"notes": notes_value,
"context": context_value,
"updated_at": timestamp,
}
)
manual_entry = existing
else:
manual_entry = {
"id": payload.get("id") or uuid.uuid4().hex,
"token": token_value,
"normalized": normalized,
"pronunciation": pronunciation_value,
"voice": voice_value,
"notes": notes_value,
"context": context_value,
"language": language,
"source": payload.get("source") or "manual",
"created_at": timestamp,
"updated_at": timestamp,
}
if isinstance(pending.manual_overrides, list):
pending.manual_overrides.append(manual_entry)
else:
pending.manual_overrides = [manual_entry]
save_pronunciation_override(
language=language,
token=token_value,
pronunciation=pronunciation_value or None,
voice=voice_value or None,
notes=notes_value or None,
context=context_value or None,
)
sync_pronunciation_overrides(pending)
return dict(manual_entry)
def delete_manual_override(pending: PendingJob, override_id: str) -> bool:
if not override_id:
return False
entries = pending.manual_overrides or []
for index, entry in enumerate(entries):
if not isinstance(entry, dict):
continue
if entry.get("id") == override_id:
token_value = entry.get("token") or ""
language = pending.language or "en"
delete_pronunciation_override(language=language, token=token_value)
entries.pop(index)
pending.manual_overrides = entries
sync_pronunciation_overrides(pending)
return True
return False
def search_manual_override_candidates(pending: PendingJob, query: str, *, limit: int = 15) -> List[Dict[str, Any]]:
normalized_query = (query or "").strip()
summary_index = (pending.entity_summary or {}).get("index", {})
matches = search_entity_tokens(summary_index, normalized_query, limit=limit)
registry: Dict[str, Dict[str, Any]] = {}
for entry in matches:
normalized = normalize_entity_token(entry.get("normalized") or entry.get("token") or "")
if not normalized:
continue
registry.setdefault(
normalized,
{
"token": entry.get("token"),
"normalized": normalized,
"category": entry.get("category") or "entity",
"count": entry.get("count", 0),
"samples": entry.get("samples", []),
"source": "entity",
},
)
language = pending.language or "en"
store_matches = search_pronunciation_overrides(language=language, query=normalized_query, limit=limit)
for entry in store_matches:
normalized = entry.get("normalized")
if not normalized:
continue
registry.setdefault(
normalized,
{
"token": entry.get("token"),
"normalized": normalized,
"category": "history",
"count": entry.get("usage_count", 0),
"samples": [entry.get("context")] if entry.get("context") else [],
"source": "history",
"pronunciation": entry.get("pronunciation"),
"voice": entry.get("voice"),
},
)
for entry in pending.manual_overrides or []:
if not isinstance(entry, Mapping):
continue
normalized = entry.get("normalized")
if not normalized:
continue
registry.setdefault(
normalized,
{
"token": entry.get("token"),
"normalized": normalized,
"category": "manual",
"count": 0,
"samples": [entry.get("context")] if entry.get("context") else [],
"source": "manual",
"pronunciation": entry.get("pronunciation"),
"voice": entry.get("voice"),
},
)
ordered = sorted(registry.values(), key=lambda item: (-int(item.get("count") or 0), item.get("token") or ""))
if limit:
return ordered[:limit]
return ordered
def pending_entities_payload(pending: PendingJob) -> Dict[str, Any]:
settings = load_settings()
recognition_enabled = bool(settings.get("enable_entity_recognition", True))
return {
"summary": pending.entity_summary or {},
"manual_overrides": pending.manual_overrides or [],
"pronunciation_overrides": pending.pronunciation_overrides or [],
"cache_key": pending.entity_cache_key,
"language": pending.language or "en",
"recognition_enabled": recognition_enabled,
}
+434
View File
@@ -0,0 +1,434 @@
import json
import math
import posixpath
import zipfile
from html.parser import HTMLParser
from pathlib import Path
from typing import Any, Dict, Iterable, List, Mapping, Optional, Set, Tuple
from xml.etree import ElementTree as ET
from abogen.web.service import Job, JobStatus
def _coerce_path(value: Any) -> Optional[Path]:
if isinstance(value, Path):
return value
if isinstance(value, str):
candidate = Path(value)
return candidate
return None
def normalize_epub_path(base_dir: str, href: str) -> str:
if not href:
return ""
sanitized = href.split("#", 1)[0].split("?", 1)[0].strip()
sanitized = sanitized.replace("\\", "/")
if not sanitized:
return ""
if sanitized.startswith("/"):
sanitized = sanitized[1:]
base_dir = ""
normalized_base = base_dir.strip("/")
sanitized_lower = sanitized.lower()
if normalized_base:
base_lower = normalized_base.lower()
prefix = base_lower + "/"
if sanitized_lower.startswith(prefix):
remainder = sanitized[len(prefix):]
if remainder.lower().startswith(prefix):
sanitized = remainder
sanitized_lower = sanitized.lower()
base_dir = ""
elif sanitized_lower == base_lower:
base_dir = ""
base = base_dir.strip("/")
combined = posixpath.join(base, sanitized) if base else sanitized
normalized = posixpath.normpath(combined)
if normalized in {"", "."}:
return ""
normalized = normalized.replace("\\", "/")
segments = [segment for segment in normalized.split("/") if segment and segment != "."]
if not segments:
return ""
deduped: List[str] = []
last_lower: Optional[str] = None
for segment in segments:
segment_lower = segment.lower()
if last_lower == segment_lower:
continue
deduped.append(segment)
last_lower = segment_lower
normalized = "/".join(deduped)
if normalized.startswith("../") or normalized == "..":
return ""
return normalized
def decode_text(payload: bytes) -> str:
for encoding in ("utf-8", "utf-16", "windows-1252"):
try:
return payload.decode(encoding)
except UnicodeDecodeError:
continue
return payload.decode("utf-8", "ignore")
def coerce_positive_time(value: Any) -> Optional[float]:
try:
numeric = float(value)
except (TypeError, ValueError):
return None
if not math.isfinite(numeric) or numeric < 0:
return None
return numeric
def load_job_metadata(job: Job) -> Dict[str, Any]:
result = getattr(job, "result", None)
artifacts = getattr(result, "artifacts", None)
if not isinstance(artifacts, Mapping):
return {}
metadata_ref = artifacts.get("metadata")
if isinstance(metadata_ref, Path):
metadata_path = metadata_ref
elif isinstance(metadata_ref, str):
metadata_path = Path(metadata_ref)
else:
return {}
if not metadata_path.exists():
return {}
try:
return json.loads(metadata_path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError, UnicodeDecodeError):
return {}
def resolve_book_title(job: Job, *metadata_sources: Mapping[str, Any]) -> str:
for source in metadata_sources:
if not isinstance(source, Mapping):
continue
for key in ("title", "book_title", "name", "album", "album_title"):
value = source.get(key)
if isinstance(value, str):
candidate = value.strip()
if candidate:
return candidate
filename = job.original_filename or ""
stem = Path(filename).stem if filename else ""
return stem or filename
class _NavMapParser(HTMLParser):
def __init__(self, base_dir: str) -> None:
super().__init__()
self._base_dir = base_dir
self._in_nav = False
self._nav_depth = 0
self._current_href: Optional[str] = None
self._buffer: List[str] = []
self.links: Dict[str, str] = {}
def handle_starttag(self, tag: str, attrs: List[Tuple[str, Optional[str]]]) -> None:
tag_lower = tag.lower()
if tag_lower == "nav":
attributes = dict(attrs)
nav_type = (attributes.get("epub:type") or attributes.get("type") or "").strip().lower()
nav_role = (attributes.get("role") or "").strip().lower()
type_tokens = {token.strip() for token in nav_type.split() if token}
role_tokens = {token.strip() for token in nav_role.split() if token}
if "toc" in type_tokens or "doc-toc" in role_tokens:
self._in_nav = True
self._nav_depth = 1
return
if self._in_nav:
self._nav_depth += 1
return
if not self._in_nav:
return
if tag_lower == "a":
attributes = dict(attrs)
href = attributes.get("href") or ""
normalized = normalize_epub_path(self._base_dir, href)
if normalized:
self._current_href = normalized
self._buffer = []
def handle_endtag(self, tag: str) -> None:
tag_lower = tag.lower()
if tag_lower == "nav" and self._in_nav:
self._nav_depth -= 1
if self._nav_depth <= 0:
self._in_nav = False
return
if not self._in_nav:
return
if tag_lower == "a" and self._current_href:
text = "".join(self._buffer).strip()
if text:
self.links.setdefault(self._current_href, text)
self._current_href = None
self._buffer = []
def handle_data(self, data: str) -> None:
if self._in_nav and self._current_href and data:
self._buffer.append(data)
def parse_nav_document(payload: bytes, base_dir: str) -> Dict[str, str]:
parser = _NavMapParser(base_dir)
parser.feed(decode_text(payload))
parser.close()
return parser.links
def parse_ncx_document(payload: bytes, base_dir: str) -> Dict[str, str]:
try:
root = ET.fromstring(payload)
except ET.ParseError:
return {}
nav_map: Dict[str, str] = {}
for nav_point in root.findall(".//{*}navPoint"):
content = nav_point.find(".//{*}content")
if content is None:
continue
src = content.attrib.get("src", "")
normalized = normalize_epub_path(base_dir, src)
if not normalized:
continue
label_el = nav_point.find(".//{*}text")
label = (label_el.text or "").strip() if label_el is not None and label_el.text else ""
if not label:
label = posixpath.basename(normalized) or f"Section {len(nav_map) + 1}"
nav_map.setdefault(normalized, label)
return nav_map
def extract_epub_chapters(epub_path: Path) -> List[Dict[str, str]]:
chapters: List[Dict[str, str]] = []
if not epub_path or not epub_path.exists():
return chapters
try:
with zipfile.ZipFile(epub_path, "r") as archive:
container_bytes = archive.read("META-INF/container.xml")
container_root = ET.fromstring(container_bytes)
rootfile = container_root.find(".//{*}rootfile")
if rootfile is None:
return chapters
opf_path = (rootfile.attrib.get("full-path") or "").strip()
if not opf_path:
return chapters
opf_dir = posixpath.dirname(opf_path)
opf_bytes = archive.read(opf_path)
opf_root = ET.fromstring(opf_bytes)
manifest: Dict[str, Dict[str, str]] = {}
for item in opf_root.findall(".//{*}manifest/{*}item"):
item_id = item.attrib.get("id")
href = item.attrib.get("href")
if not item_id or not href:
continue
manifest[item_id] = {
"href": normalize_epub_path(opf_dir, href),
"properties": item.attrib.get("properties", ""),
"media_type": item.attrib.get("media-type", ""),
}
spine_hrefs: List[str] = []
nav_id: Optional[str] = None
spine = opf_root.find(".//{*}spine")
if spine is not None:
nav_id = spine.attrib.get("toc")
for itemref in spine.findall(".//{*}itemref"):
idref = itemref.attrib.get("idref")
if not idref:
continue
entry = manifest.get(idref)
if not entry:
continue
href = entry["href"]
if href and href not in spine_hrefs:
spine_hrefs.append(href)
nav_href: Optional[str] = None
for entry in manifest.values():
properties = entry.get("properties") or ""
if "nav" in {token.strip() for token in properties.split() if token}:
nav_href = entry["href"]
break
if not nav_href and nav_id:
toc_entry = manifest.get(nav_id)
if toc_entry:
nav_href = toc_entry["href"]
nav_titles: Dict[str, str] = {}
if nav_href:
nav_base = posixpath.dirname(nav_href)
try:
nav_bytes = archive.read(nav_href)
except KeyError:
nav_bytes = None
if nav_bytes is not None:
if nav_href.lower().endswith(".ncx"):
nav_titles = parse_ncx_document(nav_bytes, nav_base)
else:
nav_titles = parse_nav_document(nav_bytes, nav_base)
if not nav_titles and nav_id and nav_id in manifest:
toc_entry = manifest[nav_id]
nav_base = posixpath.dirname(toc_entry["href"])
try:
nav_bytes = archive.read(toc_entry["href"])
except KeyError:
nav_bytes = None
if nav_bytes is not None:
nav_titles = parse_ncx_document(nav_bytes, nav_base)
for index, href in enumerate(spine_hrefs, start=1):
normalized = href
if not normalized:
continue
title = (
nav_titles.get(normalized)
or nav_titles.get(normalized.split("#", 1)[0])
or posixpath.basename(normalized)
or f"Chapter {index}"
)
chapters.append({"href": normalized, "title": title})
if not chapters and nav_titles:
for index, (href, title) in enumerate(nav_titles.items(), start=1):
normalized = href
if not normalized:
continue
label = title or posixpath.basename(normalized) or f"Chapter {index}"
chapters.append({"href": normalized, "title": label})
return chapters
except (FileNotFoundError, zipfile.BadZipFile, KeyError, ET.ParseError, UnicodeDecodeError):
return []
return chapters
def read_epub_bytes(epub_path: Path, raw_href: str) -> bytes:
normalized = normalize_epub_path("", raw_href)
if not normalized:
raise ValueError("Invalid resource path")
with zipfile.ZipFile(epub_path, "r") as archive:
return archive.read(normalized)
def iter_job_result_paths(job: Job) -> List[Path]:
result = getattr(job, "result", None)
if result is None:
return []
resolved_seen: Set[Path] = set()
collected: List[Path] = []
def _remember(candidate: Optional[Path]) -> None:
if not candidate:
return
try:
resolved = candidate.resolve()
except OSError:
return
if resolved in resolved_seen:
return
resolved_seen.add(resolved)
collected.append(candidate)
artifacts = getattr(result, "artifacts", None)
if isinstance(artifacts, Mapping):
for value in artifacts.values():
candidate = _coerce_path(value)
if candidate and candidate.exists() and candidate.is_file():
_remember(candidate)
for attr in ("audio_path", "epub_path"):
candidate = _coerce_path(getattr(result, attr, None))
if candidate and candidate.exists() and candidate.is_file():
_remember(candidate)
return collected
def iter_job_artifact_dirs(job: Job) -> List[Path]:
result = getattr(job, "result", None)
if result is None:
return []
artifacts = getattr(result, "artifacts", None)
directories: List[Path] = []
if isinstance(artifacts, Mapping):
for value in artifacts.values():
candidate = _coerce_path(value)
if candidate and candidate.exists() and candidate.is_dir():
directories.append(candidate)
return directories
def normalize_suffixes(suffixes: Iterable[str]) -> List[str]:
normalized: List[str] = []
for suffix in suffixes:
if not suffix:
continue
cleaned = suffix.lower().strip()
if not cleaned:
continue
if not cleaned.startswith("."):
cleaned = f".{cleaned.lstrip('.')}"
normalized.append(cleaned)
return normalized
def find_job_file(job: Job, suffixes: Iterable[str]) -> Optional[Path]:
ordered_suffixes = normalize_suffixes(suffixes)
if not ordered_suffixes:
return None
files = iter_job_result_paths(job)
for suffix in ordered_suffixes:
for candidate in files:
if candidate.suffix.lower() == suffix:
return candidate
directories = iter_job_artifact_dirs(job)
for suffix in ordered_suffixes:
pattern = f"*{suffix}"
for directory in directories:
try:
match = next((path for path in directory.rglob(pattern) if path.is_file()), None)
except OSError:
match = None
if match:
return match
return None
def locate_job_epub(job: Job) -> Optional[Path]:
path = find_job_file(job, [".epub"])
if path:
return path
return None
def locate_job_m4b(job: Job) -> Optional[Path]:
return find_job_file(job, [".m4b"])
def locate_job_audio(job: Job, preferred_suffixes: Optional[Iterable[str]] = None) -> Optional[Path]:
suffix_order: List[str] = []
if preferred_suffixes:
suffix_order.extend(preferred_suffixes)
suffix_order.extend([".m4b", ".mp3", ".flac", ".opus", ".ogg", ".m4a", ".wav"])
path = find_job_file(job, suffix_order)
if path:
return path
files = iter_job_result_paths(job)
return files[0] if files else None
def job_download_flags(job: Job) -> Dict[str, bool]:
if job.status != JobStatus.COMPLETED:
return {"audio": False, "m4b": False, "epub3": False}
return {
"audio": locate_job_audio(job) is not None,
"m4b": locate_job_m4b(job) is not None,
"epub3": locate_job_epub(job) is not None,
}
+989
View File
@@ -0,0 +1,989 @@
import re
import time
import uuid
from typing import Any, Dict, Iterable, List, Mapping, Optional, Tuple, cast
from flask import request, render_template, jsonify
from flask.typing import ResponseReturnValue
from abogen.web.service import PendingJob, JobStatus
from abogen.web.routes.utils.service import get_service
from abogen.web.routes.utils.settings import (
load_settings,
coerce_bool,
coerce_int,
_CHUNK_LEVEL_VALUES,
_DEFAULT_ANALYSIS_THRESHOLD,
_NORMALIZATION_BOOLEAN_KEYS,
_NORMALIZATION_STRING_KEYS,
SAVE_MODE_LABELS,
audiobookshelf_manual_available,
)
from abogen.web.routes.utils.voice import (
parse_voice_formula,
formula_from_profile,
resolve_voice_setting,
resolve_voice_choice,
prepare_speaker_metadata,
template_options,
)
from abogen.web.routes.utils.entity import sync_pronunciation_overrides
from abogen.web.routes.utils.epub import job_download_flags
from abogen.web.routes.utils.common import split_profile_spec
from abogen.utils import calculate_text_length
from abogen.voice_profiles import serialize_profiles
from abogen.chunking import ChunkLevel, build_chunks_for_chapters
from abogen.constants import VOICES_INTERNAL
from abogen.speaker_configs import get_config
from abogen.kokoro_text_normalization import normalize_roman_numeral_titles
from dataclasses import dataclass
from pathlib import Path
import mimetypes
@dataclass
class PendingBuildResult:
pending: PendingJob
selected_speaker_config: Optional[str]
config_languages: List[str]
speaker_config_payload: Optional[Dict[str, Any]]
_WIZARD_STEP_ORDER = ["book", "chapters", "entities"]
_WIZARD_STEP_META = {
"book": {
"index": 1,
"title": "Book parameters",
"hint": "Choose your source file or paste text, then set the defaults used for chapter analysis and speaker casting.",
},
"chapters": {
"index": 2,
"title": "Select chapters",
"hint": "Choose which chapters to convert. We'll analyse entities automatically when you continue.",
},
"entities": {
"index": 3,
"title": "Review entities",
"hint": "Assign pronunciations, voices, and manual overrides before queueing the conversion.",
},
}
_SUPPLEMENT_TITLE_PATTERNS: List[tuple[re.Pattern[str], float]] = [
(re.compile(r"\btitle\s+page\b"), 3.0),
(re.compile(r"\bcopyright\b"), 2.4),
(re.compile(r"\btable\s+of\s+contents\b"), 2.8),
(re.compile(r"\bcontents\b"), 2.0),
(re.compile(r"\backnowledg(e)?ments?\b"), 2.0),
(re.compile(r"\bdedication\b"), 2.0),
(re.compile(r"\babout\s+the\s+author(s)?\b"), 2.4),
(re.compile(r"\balso\s+by\b"), 2.0),
(re.compile(r"\bpraise\s+for\b"), 2.0),
(re.compile(r"\bcolophon\b"), 2.2),
(re.compile(r"\bpublication\s+data\b"), 2.2),
(re.compile(r"\btranscriber'?s?\s+note\b"), 2.2),
(re.compile(r"\bglossary\b"), 2.0),
(re.compile(r"\bindex\b"), 2.0),
(re.compile(r"\bbibliograph(y|ies)\b"), 2.0),
(re.compile(r"\breferences\b"), 1.8),
(re.compile(r"\bappendix\b"), 1.9),
]
_CONTENT_TITLE_PATTERNS: List[re.Pattern[str]] = [
re.compile(r"\bchapter\b"),
re.compile(r"\bbook\b"),
re.compile(r"\bpart\b"),
re.compile(r"\bsection\b"),
re.compile(r"\bscene\b"),
re.compile(r"\bprologue\b"),
re.compile(r"\bepilogue\b"),
re.compile(r"\bintroduction\b"),
re.compile(r"\bstory\b"),
]
_SUPPLEMENT_TEXT_KEYWORDS: List[tuple[str, float]] = [
("copyright", 1.2),
("all rights reserved", 1.1),
("isbn", 0.9),
("library of congress", 1.0),
("table of contents", 1.0),
("dedicated to", 0.8),
("acknowledg", 0.8),
("printed in", 0.6),
("permission", 0.6),
("publisher", 0.5),
("praise for", 0.9),
("also by", 0.9),
("glossary", 0.8),
("index", 0.8),
("newsletter", 3.2),
("mailing list", 2.6),
("sign-up", 2.2),
]
def supplement_score(title: str, text: str, index: int) -> float:
normalized_title = (title or "").lower()
score = 0.0
for pattern, weight in _SUPPLEMENT_TITLE_PATTERNS:
if pattern.search(normalized_title):
score += weight
for pattern in _CONTENT_TITLE_PATTERNS:
if pattern.search(normalized_title):
score -= 2.0
stripped_text = (text or "").strip()
length = len(stripped_text)
if length <= 150:
score += 0.9
elif length <= 400:
score += 0.6
elif length <= 800:
score += 0.35
lowercase_text = stripped_text.lower()
for keyword, weight in _SUPPLEMENT_TEXT_KEYWORDS:
if keyword in lowercase_text:
score += weight
if index == 0 and score > 0:
score += 0.25
return score
def should_preselect_chapter(
title: str,
text: str,
index: int,
total_count: int,
) -> bool:
if total_count <= 1:
return True
score = supplement_score(title, text, index)
return score < 1.9
def ensure_at_least_one_chapter_enabled(chapters: List[Dict[str, Any]]) -> None:
if not chapters:
return
if any(chapter.get("enabled") for chapter in chapters):
return
best_index = max(range(len(chapters)), key=lambda idx: chapters[idx].get("characters", 0))
chapters[best_index]["enabled"] = True
def apply_prepare_form(
pending: PendingJob, form: Mapping[str, Any]
) -> tuple[
ChunkLevel,
List[Dict[str, Any]],
List[Dict[str, Any]],
List[str],
int,
str,
bool,
bool,
]:
raw_chunk_level = (form.get("chunk_level") or pending.chunk_level or "paragraph").strip().lower()
if raw_chunk_level not in _CHUNK_LEVEL_VALUES:
raw_chunk_level = pending.chunk_level if pending.chunk_level in _CHUNK_LEVEL_VALUES else "paragraph"
pending.chunk_level = raw_chunk_level
chunk_level_literal = cast(ChunkLevel, pending.chunk_level)
pending.speaker_mode = "single"
pending.generate_epub3 = coerce_bool(form.get("generate_epub3"), False)
threshold_default = getattr(pending, "speaker_analysis_threshold", _DEFAULT_ANALYSIS_THRESHOLD)
raw_threshold = form.get("speaker_analysis_threshold")
if raw_threshold is not None:
pending.speaker_analysis_threshold = coerce_int(
raw_threshold,
threshold_default,
minimum=1,
maximum=25,
)
else:
pending.speaker_analysis_threshold = threshold_default
if not pending.speakers:
narrator: Dict[str, Any] = {
"id": "narrator",
"label": "Narrator",
"voice": pending.voice,
}
if pending.voice_profile:
narrator["voice_profile"] = pending.voice_profile
pending.speakers = {"narrator": narrator}
else:
existing_narrator = pending.speakers.get("narrator")
if isinstance(existing_narrator, dict):
existing_narrator.setdefault("id", "narrator")
existing_narrator["label"] = existing_narrator.get("label", "Narrator")
existing_narrator["voice"] = pending.voice
if pending.voice_profile:
existing_narrator["voice_profile"] = pending.voice_profile
pending.speakers["narrator"] = existing_narrator
selected_config = (form.get("applied_speaker_config") or "").strip()
apply_config_requested = str(form.get("apply_speaker_config", "")).strip() in {"1", "true", "on"}
persist_config_requested = str(form.get("save_speaker_config", "")).strip() in {"1", "true", "on"}
pending.applied_speaker_config = selected_config or None
errors: List[str] = []
if isinstance(pending.speakers, dict):
for speaker_id, payload in list(pending.speakers.items()):
if not isinstance(payload, dict):
continue
field_key = f"speaker-{speaker_id}-pronunciation"
raw_value = form.get(field_key, "")
pronunciation = raw_value.strip()
if pronunciation:
payload["pronunciation"] = pronunciation
else:
payload.pop("pronunciation", None)
voice_value = (form.get(f"speaker-{speaker_id}-voice") or "").strip()
formula_key = f"speaker-{speaker_id}-formula"
formula_value = (form.get(formula_key) or "").strip()
has_formula = False
if formula_value:
try:
parse_voice_formula(formula_value)
except ValueError as exc:
label = payload.get("label") or speaker_id.replace("_", " ").title()
errors.append(f"Invalid custom mix for {label}: {exc}")
else:
payload["voice_formula"] = formula_value
payload["resolved_voice"] = formula_value
payload.pop("voice_profile", None)
has_formula = True
else:
payload.pop("voice_formula", None)
if voice_value == "__custom_mix":
voice_value = ""
if voice_value:
payload["voice"] = voice_value
if not has_formula:
payload["resolved_voice"] = voice_value
else:
payload.pop("voice", None)
if not has_formula:
payload.pop("resolved_voice", None)
lang_key = f"speaker-{speaker_id}-languages"
languages: List[str] = []
getter = getattr(form, "getlist", None)
if callable(getter):
values = cast(Iterable[str], getter(lang_key))
languages = [code.strip() for code in values if code]
else:
raw_langs = form.get(lang_key)
if isinstance(raw_langs, str):
languages = [item.strip() for item in raw_langs.split(",") if item.strip()]
payload["config_languages"] = languages
profiles = serialize_profiles()
raw_delay = form.get("chapter_intro_delay")
if raw_delay is not None:
raw_normalized = raw_delay.strip()
if raw_normalized:
try:
pending.chapter_intro_delay = max(0.0, float(raw_normalized))
except ValueError:
errors.append("Enter a valid number for the chapter intro delay.")
else:
pending.chapter_intro_delay = 0.0
intro_values: List[str] = []
getter = getattr(form, "getlist", None)
if callable(getter):
raw_intro_values = getter("read_title_intro")
if raw_intro_values:
intro_values = list(cast(Iterable[str], raw_intro_values))
else:
raw_intro = form.get("read_title_intro")
if raw_intro is not None:
intro_values = [raw_intro]
if intro_values:
pending.read_title_intro = coerce_bool(intro_values[-1], pending.read_title_intro)
elif hasattr(form, "__contains__") and "read_title_intro" in form:
pending.read_title_intro = False
outro_values: List[str] = []
if callable(getter):
raw_outro_values = getter("read_closing_outro")
if raw_outro_values:
outro_values = list(cast(Iterable[str], raw_outro_values))
else:
raw_outro = form.get("read_closing_outro")
if raw_outro is not None:
outro_values = [raw_outro]
if outro_values:
pending.read_closing_outro = coerce_bool(
outro_values[-1], getattr(pending, "read_closing_outro", True)
)
elif hasattr(form, "__contains__") and "read_closing_outro" in form:
pending.read_closing_outro = False
caps_values: List[str] = []
if callable(getter):
raw_caps_values = getter("normalize_chapter_opening_caps")
if raw_caps_values:
caps_values = list(cast(Iterable[str], raw_caps_values))
else:
raw_caps = form.get("normalize_chapter_opening_caps")
if raw_caps is not None:
caps_values = [raw_caps]
if caps_values:
pending.normalize_chapter_opening_caps = coerce_bool(
caps_values[-1], getattr(pending, "normalize_chapter_opening_caps", True)
)
elif hasattr(form, "__contains__") and "normalize_chapter_opening_caps" in form:
pending.normalize_chapter_opening_caps = False
overrides: List[Dict[str, Any]] = []
selected_total = 0
for index, chapter in enumerate(pending.chapters):
enabled = form.get(f"chapter-{index}-enabled") == "on"
title_input = (form.get(f"chapter-{index}-title") or "").strip()
title = title_input or chapter.get("title") or f"Chapter {index + 1}"
voice_selection = form.get(f"chapter-{index}-voice", "__default")
formula_input = (form.get(f"chapter-{index}-formula") or "").strip()
entry: Dict[str, Any] = {
"id": chapter.get("id") or f"{index:04d}",
"index": index,
"order": index,
"source_title": chapter.get("title") or title,
"title": title,
"text": chapter.get("text", ""),
"enabled": enabled,
}
entry["characters"] = calculate_text_length(entry["text"])
if enabled:
if voice_selection.startswith("voice:"):
entry["voice"] = voice_selection.split(":", 1)[1]
entry["resolved_voice"] = entry["voice"]
elif voice_selection.startswith("profile:"):
profile_name = voice_selection.split(":", 1)[1]
entry["voice_profile"] = profile_name
profile_entry = profiles.get(profile_name) or {}
formula_value = formula_from_profile(profile_entry)
if formula_value:
entry["voice_formula"] = formula_value
entry["resolved_voice"] = formula_value
else:
errors.append(f"Profile '{profile_name}' has no configured voices.")
elif voice_selection == "formula":
if not formula_input:
errors.append(f"Provide a custom formula for chapter {index + 1}.")
else:
try:
parse_voice_formula(formula_input)
except ValueError as exc:
errors.append(str(exc))
else:
entry["voice_formula"] = formula_input
entry["resolved_voice"] = formula_input
selected_total += entry["characters"]
overrides.append(entry)
pending.chapters[index] = dict(entry)
enabled_overrides = [entry for entry in overrides if entry.get("enabled")]
sync_pronunciation_overrides(pending)
return (
chunk_level_literal,
overrides,
enabled_overrides,
errors,
selected_total,
selected_config,
apply_config_requested,
persist_config_requested,
)
def apply_book_step_form(
pending: PendingJob,
form: Mapping[str, Any],
*,
settings: Mapping[str, Any],
profiles: Mapping[str, Any],
) -> None:
language_fallback = pending.language or settings.get("language", "en")
raw_language = (form.get("language") or language_fallback or "en").strip()
if raw_language:
pending.language = raw_language
subtitle_mode = (form.get("subtitle_mode") or pending.subtitle_mode or "Disabled").strip()
if subtitle_mode:
pending.subtitle_mode = subtitle_mode
pending.generate_epub3 = coerce_bool(form.get("generate_epub3"), bool(pending.generate_epub3))
chunk_level_default = str(settings.get("chunk_level", "paragraph")).strip().lower()
raw_chunk_level = (form.get("chunk_level") or pending.chunk_level or chunk_level_default).strip().lower()
if raw_chunk_level not in _CHUNK_LEVEL_VALUES:
raw_chunk_level = chunk_level_default if chunk_level_default in _CHUNK_LEVEL_VALUES else (pending.chunk_level or "paragraph")
pending.chunk_level = raw_chunk_level
threshold_default = pending.speaker_analysis_threshold or settings.get("speaker_analysis_threshold", _DEFAULT_ANALYSIS_THRESHOLD)
raw_threshold = form.get("speaker_analysis_threshold")
if raw_threshold is not None:
pending.speaker_analysis_threshold = coerce_int(
raw_threshold,
threshold_default,
minimum=1,
maximum=25,
)
raw_delay = form.get("chapter_intro_delay")
if raw_delay is not None:
try:
pending.chapter_intro_delay = max(0.0, float(str(raw_delay).strip() or 0.0))
except ValueError:
pass
intro_default = pending.read_title_intro if isinstance(pending.read_title_intro, bool) else bool(settings.get("read_title_intro", False))
intro_values: List[str] = []
getter = getattr(form, "getlist", None)
if callable(getter):
raw_intro_values = getter("read_title_intro")
if raw_intro_values:
intro_values = list(cast(Iterable[str], raw_intro_values))
else:
raw_intro_flag = form.get("read_title_intro")
if raw_intro_flag is not None:
intro_values = [raw_intro_flag]
if intro_values:
pending.read_title_intro = coerce_bool(intro_values[-1], intro_default)
elif hasattr(form, "__contains__") and "read_title_intro" in form:
pending.read_title_intro = False
else:
pending.read_title_intro = intro_default
outro_default = (
pending.read_closing_outro
if isinstance(getattr(pending, "read_closing_outro", None), bool)
else bool(settings.get("read_closing_outro", True))
)
outro_values: List[str] = []
if callable(getter):
raw_outro_values = getter("read_closing_outro")
if raw_outro_values:
outro_values = list(cast(Iterable[str], raw_outro_values))
else:
raw_outro_flag = form.get("read_closing_outro")
if raw_outro_flag is not None:
outro_values = [raw_outro_flag]
if outro_values:
pending.read_closing_outro = coerce_bool(outro_values[-1], outro_default)
elif hasattr(form, "__contains__") and "read_closing_outro" in form:
pending.read_closing_outro = False
else:
pending.read_closing_outro = outro_default
caps_default = (
pending.normalize_chapter_opening_caps
if isinstance(getattr(pending, "normalize_chapter_opening_caps", None), bool)
else bool(settings.get("normalize_chapter_opening_caps", True))
)
caps_values: List[str] = []
getter = getattr(form, "getlist", None)
if callable(getter):
raw_caps_values = getter("normalize_chapter_opening_caps")
if raw_caps_values:
caps_values = list(cast(Iterable[str], raw_caps_values))
else:
raw_caps_flag = form.get("normalize_chapter_opening_caps")
if raw_caps_flag is not None:
caps_values = [raw_caps_flag]
if caps_values:
pending.normalize_chapter_opening_caps = coerce_bool(caps_values[-1], caps_default)
elif hasattr(form, "__contains__") and "normalize_chapter_opening_caps" in form:
pending.normalize_chapter_opening_caps = False
else:
pending.normalize_chapter_opening_caps = caps_default
def _extract_checkbox(name: str, default: bool) -> bool:
values: List[str] = []
getter = getattr(form, "getlist", None)
if callable(getter):
raw_values = getter(name)
if raw_values:
values = list(cast(Iterable[str], raw_values))
else:
raw_flag = form.get(name)
if raw_flag is not None:
values = [raw_flag]
if values:
return coerce_bool(values[-1], default)
if hasattr(form, "__contains__") and name in form:
return False
return default
overrides_existing = getattr(pending, "normalization_overrides", None)
overrides: Dict[str, Any] = dict(overrides_existing or {})
for key in _NORMALIZATION_BOOLEAN_KEYS:
default_toggle = overrides.get(key, bool(settings.get(key, True)))
overrides[key] = _extract_checkbox(key, default_toggle)
for key in _NORMALIZATION_STRING_KEYS:
default_val = overrides.get(key, str(settings.get(key, "")))
val = form.get(key)
if val is not None:
overrides[key] = str(val)
else:
overrides[key] = default_val
pending.normalization_overrides = overrides
speed_value = form.get("speed")
if speed_value is not None:
try:
pending.speed = float(speed_value)
except ValueError:
pass
profile_selection = (form.get("voice_profile") or pending.voice_profile or "__standard").strip()
custom_formula_raw = (form.get("voice_formula") or "").strip()
narrator_voice_raw = (form.get("voice") or pending.voice or settings.get("default_voice") or "").strip()
profiles_map = dict(profiles) if isinstance(profiles, Mapping) else dict(profiles or {})
resolved_default_voice, inferred_profile, _ = resolve_voice_setting(
narrator_voice_raw,
profiles=profiles_map,
)
if profile_selection in {"__standard", "", None} and inferred_profile:
profile_selection = inferred_profile
if profile_selection == "__formula":
profile_name = ""
custom_formula = custom_formula_raw
elif profile_selection in {"__standard", "", None}:
profile_name = ""
custom_formula = ""
else:
profile_name = profile_selection
custom_formula = ""
base_voice_spec = resolved_default_voice or narrator_voice_raw
if not base_voice_spec and VOICES_INTERNAL:
base_voice_spec = VOICES_INTERNAL[0]
voice_choice, resolved_language, selected_profile = resolve_voice_choice(
pending.language,
base_voice_spec,
profile_name,
custom_formula,
profiles_map,
)
if resolved_language:
pending.language = resolved_language
if profile_selection == "__formula" and custom_formula_raw:
pending.voice = custom_formula_raw
pending.voice_profile = None
elif profile_selection not in {"__standard", "", None, "__formula"}:
pending.voice_profile = selected_profile or profile_selection
pending.voice = voice_choice
else:
pending.voice_profile = None
fallback_voice = base_voice_spec or narrator_voice_raw
pending.voice = voice_choice or fallback_voice
pending.applied_speaker_config = (form.get("speaker_config") or "").strip() or None
def persist_cover_image(extraction_result: Any, stored_path: Path) -> tuple[Optional[Path], Optional[str]]:
cover_bytes = getattr(extraction_result, "cover_image", None)
if not cover_bytes:
return None, None
mime = getattr(extraction_result, "cover_mime", None)
extension = mimetypes.guess_extension(mime or "") or ".png"
base_stem = Path(stored_path).stem or "cover"
candidate = stored_path.parent / f"{base_stem}_cover{extension}"
counter = 1
while candidate.exists():
candidate = stored_path.parent / f"{base_stem}_cover_{counter}{extension}"
counter += 1
try:
candidate.write_bytes(cover_bytes)
except OSError:
return None, None
return candidate, mime
def build_pending_job_from_extraction(
*,
stored_path: Path,
original_name: str,
extraction: Any,
form: Mapping[str, Any],
settings: Mapping[str, Any],
profiles: Mapping[str, Any],
metadata_overrides: Optional[Mapping[str, Any]] = None,
) -> PendingBuildResult:
profiles_map = dict(profiles)
cover_path, cover_mime = persist_cover_image(extraction, stored_path)
if getattr(extraction, "chapters", None):
original_titles = [chapter.title for chapter in extraction.chapters]
normalized_titles = normalize_roman_numeral_titles(original_titles)
if normalized_titles != original_titles:
for chapter, new_title in zip(extraction.chapters, normalized_titles):
chapter.title = new_title
metadata_tags = dict(getattr(extraction, "metadata", {}) or {})
if metadata_overrides:
normalized_keys = {str(existing_key).casefold(): str(existing_key) for existing_key in metadata_tags.keys()}
for key, value in metadata_overrides.items():
if value is None:
continue
key_text = str(key or "").strip()
if not key_text:
continue
value_text = str(value).strip()
if not value_text:
continue
lookup = key_text.casefold()
existing_key = normalized_keys.get(lookup)
if existing_key:
existing_value = str(metadata_tags.get(existing_key) or "").strip()
if existing_value:
continue
target_key = existing_key
else:
target_key = key_text
normalized_keys[lookup] = target_key
metadata_tags[target_key] = value_text
total_chars = getattr(extraction, "total_characters", None) or calculate_text_length(
getattr(extraction, "combined_text", "")
)
chapters_source = getattr(extraction, "chapters", []) or []
total_chapter_count = len(chapters_source)
chapters_payload: List[Dict[str, Any]] = []
for index, chapter in enumerate(chapters_source):
enabled = should_preselect_chapter(chapter.title, chapter.text, index, total_chapter_count)
chapters_payload.append(
{
"id": f"{index:04d}",
"index": index,
"title": chapter.title,
"text": chapter.text,
"characters": calculate_text_length(chapter.text),
"enabled": enabled,
}
)
if not chapters_payload:
chapters_payload.append(
{
"id": "0000",
"index": 0,
"title": original_name,
"text": "",
"characters": 0,
"enabled": True,
}
)
ensure_at_least_one_chapter_enabled(chapters_payload)
language = str(form.get("language") or "a").strip() or "a"
profiles_map = dict(profiles) if isinstance(profiles, Mapping) else dict(profiles or {})
default_voice_setting = settings.get("default_voice") or ""
resolved_default_voice, inferred_profile, inferred_language = resolve_voice_setting(
default_voice_setting,
profiles=profiles_map,
)
base_voice_input = str(form.get("voice") or "").strip()
profile_selection = (form.get("voice_profile") or "__standard").strip()
custom_formula_raw = str(form.get("voice_formula") or "").strip()
if profile_selection in {"__standard", ""} and inferred_profile:
profile_selection = inferred_profile
base_voice = base_voice_input or resolved_default_voice or str(default_voice_setting).strip()
if not base_voice and VOICES_INTERNAL:
base_voice = VOICES_INTERNAL[0]
selected_speaker_config = (form.get("speaker_config") or "").strip()
speaker_config_payload = get_config(selected_speaker_config) if selected_speaker_config else None
if profile_selection == "__formula":
profile_name = ""
custom_formula = custom_formula_raw
elif profile_selection in {"__standard", ""}:
profile_name = ""
custom_formula = ""
else:
profile_name = profile_selection
custom_formula = ""
voice, language, selected_profile = resolve_voice_choice(
language,
base_voice,
profile_name,
custom_formula,
profiles_map,
)
try:
speed = float(form.get("speed", 1.0))
except (TypeError, ValueError):
speed = 1.0
subtitle_mode = str(form.get("subtitle_mode") or "Disabled")
output_format = settings["output_format"]
subtitle_format = settings["subtitle_format"]
save_mode_key = settings["save_mode"]
save_mode = SAVE_MODE_LABELS.get(save_mode_key, SAVE_MODE_LABELS["save_next_to_input"])
replace_single_newlines = settings["replace_single_newlines"]
use_gpu = settings["use_gpu"]
save_chapters_separately = settings["save_chapters_separately"]
merge_chapters_at_end = settings["merge_chapters_at_end"] or not save_chapters_separately
save_as_project = settings["save_as_project"]
separate_chapters_format = settings["separate_chapters_format"]
silence_between_chapters = settings["silence_between_chapters"]
chapter_intro_delay = settings["chapter_intro_delay"]
read_title_intro = settings["read_title_intro"]
read_closing_outro = settings.get("read_closing_outro", True)
normalize_chapter_opening_caps = settings["normalize_chapter_opening_caps"]
max_subtitle_words = settings["max_subtitle_words"]
auto_prefix_chapter_titles = settings["auto_prefix_chapter_titles"]
chunk_level_default = str(settings.get("chunk_level", "paragraph")).strip().lower()
raw_chunk_level = str(form.get("chunk_level") or chunk_level_default).strip().lower()
if raw_chunk_level not in _CHUNK_LEVEL_VALUES:
raw_chunk_level = chunk_level_default if chunk_level_default in _CHUNK_LEVEL_VALUES else "paragraph"
chunk_level_value = raw_chunk_level
chunk_level_literal = cast(ChunkLevel, chunk_level_value)
speaker_mode_value = "single"
generate_epub3_default = bool(settings.get("generate_epub3", False))
generate_epub3 = coerce_bool(form.get("generate_epub3"), generate_epub3_default)
selected_chapter_sources = [entry for entry in chapters_payload if entry.get("enabled")]
raw_chunks = build_chunks_for_chapters(selected_chapter_sources, level=chunk_level_literal)
analysis_chunks = build_chunks_for_chapters(selected_chapter_sources, level="sentence")
analysis_threshold = coerce_int(
settings.get("speaker_analysis_threshold"),
_DEFAULT_ANALYSIS_THRESHOLD,
minimum=1,
maximum=25,
)
initial_analysis = False
(
processed_chunks,
speakers,
analysis_payload,
config_languages,
_,
) = prepare_speaker_metadata(
chapters=selected_chapter_sources,
chunks=raw_chunks,
analysis_chunks=analysis_chunks,
voice=voice,
voice_profile=selected_profile or None,
threshold=analysis_threshold,
run_analysis=initial_analysis,
speaker_config=speaker_config_payload,
apply_config=bool(speaker_config_payload),
)
pending = PendingJob(
id=uuid.uuid4().hex,
original_filename=original_name,
stored_path=stored_path,
language=language,
voice=voice,
speed=speed,
use_gpu=use_gpu,
subtitle_mode=subtitle_mode,
output_format=output_format,
save_mode=save_mode,
output_folder=None,
replace_single_newlines=replace_single_newlines,
subtitle_format=subtitle_format,
total_characters=total_chars,
save_chapters_separately=save_chapters_separately,
merge_chapters_at_end=merge_chapters_at_end,
separate_chapters_format=separate_chapters_format,
silence_between_chapters=silence_between_chapters,
save_as_project=save_as_project,
voice_profile=selected_profile or None,
max_subtitle_words=max_subtitle_words,
metadata_tags=metadata_tags,
chapters=chapters_payload,
normalization_overrides={
**{key: bool(settings.get(key, True)) for key in _NORMALIZATION_BOOLEAN_KEYS},
**{key: str(settings.get(key, "")) for key in _NORMALIZATION_STRING_KEYS},
},
created_at=time.time(),
cover_image_path=cover_path,
cover_image_mime=cover_mime,
chapter_intro_delay=chapter_intro_delay,
read_title_intro=bool(read_title_intro),
read_closing_outro=bool(read_closing_outro),
normalize_chapter_opening_caps=bool(normalize_chapter_opening_caps),
auto_prefix_chapter_titles=bool(auto_prefix_chapter_titles),
chunk_level=chunk_level_value,
speaker_mode=speaker_mode_value,
generate_epub3=generate_epub3,
chunks=processed_chunks,
speakers=speakers,
speaker_analysis=analysis_payload,
speaker_analysis_threshold=analysis_threshold,
analysis_requested=initial_analysis,
)
return PendingBuildResult(
pending=pending,
selected_speaker_config=selected_speaker_config or None,
config_languages=list(config_languages or []),
speaker_config_payload=speaker_config_payload,
)
def render_jobs_panel() -> str:
jobs = get_service().list_jobs()
active_statuses = {JobStatus.PENDING, JobStatus.RUNNING, JobStatus.PAUSED}
active_jobs = [job for job in jobs if job.status in active_statuses]
active_jobs.sort(key=lambda job: ((job.queue_position or 10_000), -job.created_at))
finished_jobs = [job for job in jobs if job.status not in active_statuses]
download_flags = {job.id: job_download_flags(job) for job in jobs}
return render_template(
"partials/jobs.html",
active_jobs=active_jobs,
finished_jobs=finished_jobs[:5],
total_finished=len(finished_jobs),
JobStatus=JobStatus,
download_flags=download_flags,
audiobookshelf_manual_available=audiobookshelf_manual_available(),
)
def normalize_wizard_step(step: Optional[str], pending: Optional[PendingJob] = None) -> str:
if pending is None:
default_step = "book"
else:
default_step = "chapters"
if not step:
chosen = default_step
else:
normalized = step.strip().lower()
if normalized in {"", "upload", "settings"}:
chosen = default_step
elif normalized == "speakers":
chosen = "entities"
elif normalized in _WIZARD_STEP_ORDER:
chosen = normalized
else:
chosen = default_step
return chosen
def wants_wizard_json() -> bool:
format_hint = request.args.get("format", "").strip().lower()
if format_hint == "json":
return True
accept_header = (request.headers.get("Accept") or "").lower()
if "application/json" in accept_header:
return True
requested_with = (request.headers.get("X-Requested-With") or "").lower()
if requested_with in {"xmlhttprequest", "fetch"}:
return True
wizard_header = (request.headers.get("X-Abogen-Wizard") or "").lower()
return wizard_header == "json"
def render_wizard_partial(
pending: Optional[PendingJob],
step: str,
*,
error: Optional[str] = None,
notice: Optional[str] = None,
) -> str:
templates = {
"book": "partials/new_job_step_book.html",
"chapters": "partials/new_job_step_chapters.html",
"entities": "partials/new_job_step_entities.html",
}
template_name = templates[step]
context: Dict[str, Any] = {
"pending": pending,
"readonly": False,
"options": template_options(),
"settings": load_settings(),
"error": error,
"notice": notice,
}
return render_template(template_name, **context)
def wizard_step_payload(
pending: Optional[PendingJob],
step: str,
html: str,
*,
error: Optional[str] = None,
notice: Optional[str] = None,
) -> Dict[str, Any]:
meta = _WIZARD_STEP_META.get(step, {})
try:
active_index = _WIZARD_STEP_ORDER.index(step)
except ValueError:
active_index = 0
max_recorded_index = active_index
if pending is not None:
stored_index = int(getattr(pending, "wizard_max_step_index", -1))
if stored_index < 0:
stored_index = -1
max_recorded_index = max(active_index, stored_index)
max_allowed = len(_WIZARD_STEP_ORDER) - 1
if max_recorded_index > max_allowed:
max_recorded_index = max_allowed
if stored_index != max_recorded_index:
pending.wizard_max_step_index = max_recorded_index
get_service().store_pending_job(pending)
else:
max_allowed = len(_WIZARD_STEP_ORDER) - 1
if max_recorded_index > max_allowed:
max_recorded_index = max_allowed
completed = [slug for idx, slug in enumerate(_WIZARD_STEP_ORDER) if idx <= max_recorded_index]
return {
"step": step,
"step_index": int(meta.get("index", active_index + 1)),
"total_steps": len(_WIZARD_STEP_ORDER),
"title": meta.get("title", ""),
"hint": meta.get("hint", ""),
"html": html,
"completed_steps": completed,
"pending_id": pending.id if pending else "",
"filename": pending.original_filename if pending and pending.original_filename else "",
"error": error or "",
"notice": notice or "",
}
def wizard_json_response(
pending: Optional[PendingJob],
step: str,
*,
error: Optional[str] = None,
notice: Optional[str] = None,
status: int = 200,
) -> ResponseReturnValue:
html = render_wizard_partial(pending, step, error=error, notice=notice)
payload = wizard_step_payload(pending, step, html, error=error, notice=notice)
return jsonify(payload), status
+104
View File
@@ -0,0 +1,104 @@
import io
import threading
from typing import Any, Dict, List, Optional, Tuple
import numpy as np
import soundfile as sf
from flask import current_app, send_file
from flask.typing import ResponseReturnValue
from abogen.utils import load_numpy_kpipeline
from abogen.voice_formulas import get_new_voice
from abogen.web.conversion_runner import SPLIT_PATTERN, SAMPLE_RATE, _select_device, _to_float32
from abogen.kokoro_text_normalization import normalize_for_pipeline
_preview_pipelines: Dict[Tuple[str, str], Any] = {}
_preview_pipeline_lock = threading.Lock()
def get_preview_pipeline(language: str, device: str) -> Any:
key = (language, device)
with _preview_pipeline_lock:
pipeline = _preview_pipelines.get(key)
if pipeline is not None:
return pipeline
_, KPipeline = load_numpy_kpipeline()
pipeline = KPipeline(lang_code=language, repo_id="hexgrad/Kokoro-82M", device=device)
_preview_pipelines[key] = pipeline
return pipeline
def synthesize_preview(
text: str,
voice_spec: str,
language: str,
speed: float,
use_gpu: bool,
max_seconds: float = 8.0,
) -> ResponseReturnValue:
if not text.strip():
raise ValueError("Preview text is required")
device = "cpu"
if use_gpu:
try:
device = _select_device()
except Exception:
device = "cpu"
use_gpu = False
pipeline = get_preview_pipeline(language, device)
if pipeline is None:
raise RuntimeError("Preview pipeline is unavailable")
voice_choice: Any = voice_spec
if voice_spec and "*" in voice_spec:
voice_choice = get_new_voice(pipeline, voice_spec, use_gpu)
try:
normalized_text = normalize_for_pipeline(text)
except Exception:
current_app.logger.exception("Preview normalization failed; using raw text")
normalized_text = text
segments = pipeline(
normalized_text,
voice=voice_choice,
speed=speed,
split_pattern=SPLIT_PATTERN,
)
audio_chunks: List[np.ndarray] = []
accumulated = 0
max_samples = int(max(1.0, max_seconds) * SAMPLE_RATE)
for segment in segments:
graphemes = getattr(segment, "graphemes", "").strip()
if not graphemes:
continue
audio = _to_float32(getattr(segment, "audio", None))
if audio.size == 0:
continue
remaining = max_samples - accumulated
if remaining <= 0:
break
if audio.shape[0] > remaining:
audio = audio[:remaining]
audio_chunks.append(audio)
accumulated += audio.shape[0]
if accumulated >= max_samples:
break
if not audio_chunks:
raise RuntimeError("Preview could not be generated")
audio_data = np.concatenate(audio_chunks)
buffer = io.BytesIO()
sf.write(buffer, audio_data, SAMPLE_RATE, format="WAV")
buffer.seek(0)
response = send_file(
buffer,
mimetype="audio/wav",
as_attachment=False,
download_name="speaker_preview.wav",
)
response.headers["Cache-Control"] = "no-store"
return response
+64
View File
@@ -0,0 +1,64 @@
from typing import cast
from flask import current_app, abort
from abogen.web.service import ConversionService, PendingJob
def get_service() -> ConversionService:
return current_app.extensions["conversion_service"]
def require_pending_job(pending_id: str) -> PendingJob:
pending = get_service().get_pending_job(pending_id)
if not pending:
abort(404)
return cast(PendingJob, pending)
def remove_pending_job(pending_id: str) -> None:
get_service().pop_pending_job(pending_id)
def submit_job(pending: PendingJob) -> str:
service = get_service()
service.pop_pending_job(pending.id)
job = service.enqueue(
original_filename=pending.original_filename,
stored_path=pending.stored_path,
language=pending.language,
voice=pending.voice,
speed=pending.speed,
use_gpu=pending.use_gpu,
subtitle_mode=pending.subtitle_mode,
output_format=pending.output_format,
save_mode=pending.save_mode,
output_folder=pending.output_folder,
replace_single_newlines=pending.replace_single_newlines,
subtitle_format=pending.subtitle_format,
total_characters=pending.total_characters,
chapters=pending.chapters,
save_chapters_separately=pending.save_chapters_separately,
merge_chapters_at_end=pending.merge_chapters_at_end,
separate_chapters_format=pending.separate_chapters_format,
silence_between_chapters=pending.silence_between_chapters,
save_as_project=pending.save_as_project,
voice_profile=pending.voice_profile,
max_subtitle_words=pending.max_subtitle_words,
metadata_tags=pending.metadata_tags,
cover_image_path=pending.cover_image_path,
cover_image_mime=pending.cover_image_mime,
chapter_intro_delay=pending.chapter_intro_delay,
read_title_intro=pending.read_title_intro,
read_closing_outro=pending.read_closing_outro,
auto_prefix_chapter_titles=pending.auto_prefix_chapter_titles,
normalize_chapter_opening_caps=pending.normalize_chapter_opening_caps,
chunk_level=pending.chunk_level,
chunks=pending.chunks,
speakers=pending.speakers,
speaker_mode=pending.speaker_mode,
generate_epub3=pending.generate_epub3,
speaker_analysis=pending.speaker_analysis,
speaker_analysis_threshold=pending.speaker_analysis_threshold,
analysis_requested=pending.analysis_requested,
entity_summary=getattr(pending, "entity_summary", None),
manual_overrides=getattr(pending, "manual_overrides", None),
pronunciation_overrides=getattr(pending, "pronunciation_overrides", None),
normalization_overrides=pending.normalization_overrides,
)
return job.id
+641
View File
@@ -0,0 +1,641 @@
import os
import re
from typing import Any, Dict, Mapping, Optional
from abogen.constants import (
LANGUAGE_DESCRIPTIONS,
SUBTITLE_FORMATS,
SUPPORTED_SOUND_FORMATS,
VOICES_INTERNAL,
)
from abogen.normalization_settings import (
DEFAULT_LLM_PROMPT,
environment_llm_defaults,
)
from abogen.utils import load_config, save_config
from abogen.integrations.calibre_opds import CalibreOPDSClient
from abogen.integrations.audiobookshelf import AudiobookshelfConfig
from abogen.web.routes.utils.common import split_profile_spec
SAVE_MODE_LABELS = {
"save_next_to_input": "Save next to input file",
"save_to_desktop": "Save to Desktop",
"choose_output_folder": "Choose output folder",
"default_output": "Use default save location",
}
LEGACY_SAVE_MODE_MAP = {label: key for key, label in SAVE_MODE_LABELS.items()}
_CHUNK_LEVEL_OPTIONS = [
{"value": "paragraph", "label": "Paragraphs"},
{"value": "sentence", "label": "Sentences"},
]
_CHUNK_LEVEL_VALUES = {option["value"] for option in _CHUNK_LEVEL_OPTIONS}
_DEFAULT_ANALYSIS_THRESHOLD = 3
_APOSTROPHE_MODE_OPTIONS = [
{"value": "off", "label": "Off"},
{"value": "spacy", "label": "spaCy (built-in)"},
{"value": "llm", "label": "LLM assisted"},
]
_NORMALIZATION_BOOLEAN_KEYS = {
"normalization_numbers",
"normalization_titles",
"normalization_terminal",
"normalization_phoneme_hints",
"normalization_caps_quotes",
"normalization_apostrophes_contractions",
"normalization_apostrophes_plural_possessives",
"normalization_apostrophes_sibilant_possessives",
"normalization_apostrophes_decades",
"normalization_apostrophes_leading_elisions",
"normalization_contraction_aux_be",
"normalization_contraction_aux_have",
"normalization_contraction_modal_will",
"normalization_contraction_modal_would",
"normalization_contraction_negation_not",
"normalization_contraction_let_us",
}
_NORMALIZATION_STRING_KEYS = {
"normalization_numbers_year_style",
"normalization_apostrophe_mode",
}
BOOLEAN_SETTINGS = {
"replace_single_newlines",
"use_gpu",
"save_chapters_separately",
"merge_chapters_at_end",
"save_as_project",
"generate_epub3",
"enable_entity_recognition",
"read_title_intro",
"read_closing_outro",
"auto_prefix_chapter_titles",
"normalize_chapter_opening_caps",
"normalization_numbers",
"normalization_titles",
"normalization_terminal",
"normalization_phoneme_hints",
"normalization_caps_quotes",
"normalization_apostrophes_contractions",
"normalization_apostrophes_plural_possessives",
"normalization_apostrophes_sibilant_possessives",
"normalization_apostrophes_decades",
"normalization_apostrophes_leading_elisions",
"normalization_contraction_aux_be",
"normalization_contraction_aux_have",
"normalization_contraction_modal_will",
"normalization_contraction_modal_would",
"normalization_contraction_negation_not",
"normalization_contraction_let_us",
}
FLOAT_SETTINGS = {"silence_between_chapters", "chapter_intro_delay", "llm_timeout"}
INT_SETTINGS = {"max_subtitle_words", "speaker_analysis_threshold"}
def integration_defaults() -> Dict[str, Dict[str, Any]]:
return {
"calibre_opds": {
"enabled": False,
"base_url": "",
"username": "",
"password": "",
"verify_ssl": True,
},
"audiobookshelf": {
"enabled": False,
"base_url": "",
"api_token": "",
"library_id": "",
"collection_id": "",
"folder_id": "",
"verify_ssl": True,
"send_cover": True,
"send_chapters": True,
"send_subtitles": False,
"auto_send": False,
"timeout": 30.0,
},
}
def has_output_override() -> bool:
return bool(os.environ.get("ABOGEN_OUTPUT_DIR") or os.environ.get("ABOGEN_OUTPUT_ROOT"))
def settings_defaults() -> Dict[str, Any]:
llm_env_defaults = environment_llm_defaults()
return {
"output_format": "wav",
"subtitle_format": "srt",
"save_mode": "default_output" if has_output_override() else "save_next_to_input",
"default_voice": VOICES_INTERNAL[0] if VOICES_INTERNAL else "",
"replace_single_newlines": False,
"use_gpu": True,
"save_chapters_separately": False,
"merge_chapters_at_end": True,
"save_as_project": False,
"separate_chapters_format": "wav",
"silence_between_chapters": 2.0,
"chapter_intro_delay": 0.5,
"read_title_intro": False,
"read_closing_outro": True,
"normalize_chapter_opening_caps": True,
"max_subtitle_words": 50,
"chunk_level": "paragraph",
"enable_entity_recognition": True,
"generate_epub3": False,
"auto_prefix_chapter_titles": True,
"speaker_analysis_threshold": _DEFAULT_ANALYSIS_THRESHOLD,
"speaker_pronunciation_sentence": "This is {{name}} speaking.",
"speaker_random_languages": [],
"llm_base_url": llm_env_defaults.get("llm_base_url", ""),
"llm_api_key": llm_env_defaults.get("llm_api_key", ""),
"llm_model": llm_env_defaults.get("llm_model", ""),
"llm_timeout": llm_env_defaults.get("llm_timeout", 30.0),
"llm_prompt": llm_env_defaults.get("llm_prompt", DEFAULT_LLM_PROMPT),
"llm_context_mode": llm_env_defaults.get("llm_context_mode", "sentence"),
"normalization_numbers": True,
"normalization_titles": True,
"normalization_terminal": True,
"normalization_phoneme_hints": True,
"normalization_caps_quotes": True,
"normalization_apostrophes_contractions": True,
"normalization_apostrophes_plural_possessives": True,
"normalization_apostrophes_sibilant_possessives": True,
"normalization_apostrophes_decades": True,
"normalization_apostrophes_leading_elisions": True,
"normalization_apostrophe_mode": "spacy",
}
def llm_ready(settings: Mapping[str, Any]) -> bool:
base_url = str(settings.get("llm_base_url") or "").strip()
return bool(base_url)
_PROMPT_TOKEN_RE = re.compile(r"{{\s*([a-zA-Z0-9_]+)\s*}}")
def render_prompt_template(template: str, context: Mapping[str, str]) -> str:
if not template:
return ""
def _replace(match: re.Match[str]) -> str:
key = match.group(1)
return context.get(key, "")
return _PROMPT_TOKEN_RE.sub(_replace, template)
def coerce_bool(value: Any, default: bool) -> bool:
if isinstance(value, bool):
return value
if isinstance(value, str):
return value.lower() in {"true", "1", "yes", "on"}
if value is None:
return default
return bool(value)
def coerce_float(value: Any, default: float) -> float:
try:
return max(0.0, float(value))
except (TypeError, ValueError):
return default
def coerce_int(value: Any, default: int, *, minimum: int = 1, maximum: int = 200) -> int:
try:
parsed = int(value)
except (TypeError, ValueError):
return default
return max(minimum, min(parsed, maximum))
def normalize_save_mode(value: Any, default: str) -> str:
if isinstance(value, str):
if value in SAVE_MODE_LABELS:
return value
if value in LEGACY_SAVE_MODE_MAP:
return LEGACY_SAVE_MODE_MAP[value]
return default
def normalize_setting_value(key: str, value: Any, defaults: Dict[str, Any]) -> Any:
if key in BOOLEAN_SETTINGS:
return coerce_bool(value, defaults[key])
if key in FLOAT_SETTINGS:
return coerce_float(value, defaults[key])
if key in INT_SETTINGS:
return coerce_int(value, defaults[key])
if key == "save_mode":
return normalize_save_mode(value, defaults[key])
if key == "output_format":
return value if value in SUPPORTED_SOUND_FORMATS else defaults[key]
if key == "subtitle_format":
valid = {item[0] for item in SUBTITLE_FORMATS}
return value if value in valid else defaults[key]
if key == "separate_chapters_format":
if isinstance(value, str):
normalized = value.lower()
if normalized in {"wav", "flac", "mp3", "opus"}:
return normalized
return defaults[key]
if key == "default_voice":
if isinstance(value, str):
text = value.strip()
if not text:
return defaults[key]
spec, profile_name = split_profile_spec(text)
if profile_name:
return f"profile:{profile_name}"
return spec
return defaults[key]
if key == "chunk_level":
if isinstance(value, str) and value in _CHUNK_LEVEL_VALUES:
return value
return defaults[key]
if key == "normalization_apostrophe_mode":
if isinstance(value, str):
normalized_mode = value.strip().lower()
if normalized_mode in {"off", "spacy", "llm"}:
return normalized_mode
return defaults[key]
if key == "llm_context_mode":
if isinstance(value, str):
normalized_scope = value.strip().lower()
if normalized_scope == "sentence":
return normalized_scope
return defaults[key]
if key == "llm_prompt":
candidate = str(value or "").strip()
return candidate if candidate else defaults[key]
if key in {"llm_base_url", "llm_api_key", "llm_model"}:
return str(value or "").strip()
if key == "speaker_random_languages":
if isinstance(value, (list, tuple, set)):
return [code for code in value if isinstance(code, str) and code in LANGUAGE_DESCRIPTIONS]
if isinstance(value, str):
parts = [item.strip().lower() for item in value.split(",") if item.strip()]
return [code for code in parts if code in LANGUAGE_DESCRIPTIONS]
return defaults.get(key, [])
return value if value is not None else defaults.get(key)
def load_settings() -> Dict[str, Any]:
defaults = settings_defaults()
cfg = load_config() or {}
settings: Dict[str, Any] = {}
for key, default in defaults.items():
raw_value = cfg.get(key, default)
settings[key] = normalize_setting_value(key, raw_value, defaults)
return settings
def load_integration_settings() -> Dict[str, Dict[str, Any]]:
defaults = integration_defaults()
cfg = load_config() or {}
integrations: Dict[str, Dict[str, Any]] = {}
for key, default in defaults.items():
stored = cfg.get(key)
merged: Dict[str, Any] = dict(default)
if isinstance(stored, Mapping):
for field, default_value in default.items():
value = stored.get(field, default_value)
if isinstance(default_value, bool):
merged[field] = coerce_bool(value, default_value)
elif isinstance(default_value, float):
try:
merged[field] = float(value)
except (TypeError, ValueError):
merged[field] = default_value
elif isinstance(default_value, int):
try:
merged[field] = int(value)
except (TypeError, ValueError):
merged[field] = default_value
else:
merged[field] = str(value or "")
if key == "calibre_opds":
merged["has_password"] = bool(isinstance(stored, Mapping) and stored.get("password"))
merged["password"] = ""
elif key == "audiobookshelf":
merged["has_api_token"] = bool(isinstance(stored, Mapping) and stored.get("api_token"))
merged["api_token"] = ""
integrations[key] = merged
return integrations
def stored_integration_config(name: str) -> Dict[str, Any]:
cfg = load_config() or {}
entry = cfg.get(name)
if isinstance(entry, Mapping):
return dict(entry)
return {}
def calibre_settings_from_payload(payload: Mapping[str, Any]) -> Dict[str, Any]:
defaults = integration_defaults()["calibre_opds"]
stored = stored_integration_config("calibre_opds")
base_url = str(
payload.get("base_url")
or payload.get("calibre_opds_base_url")
or stored.get("base_url")
or ""
).strip()
username = str(
payload.get("username")
or payload.get("calibre_opds_username")
or stored.get("username")
or ""
).strip()
password_input = str(
payload.get("password")
or payload.get("calibre_opds_password")
or ""
).strip()
use_saved_password = coerce_bool(
payload.get("use_saved_password")
or payload.get("calibre_opds_use_saved_password"),
False,
)
clear_saved_password = coerce_bool(
payload.get("clear_saved_password")
or payload.get("calibre_opds_password_clear"),
False,
)
password = ""
if password_input:
password = password_input
elif use_saved_password and not clear_saved_password:
password = str(stored.get("password") or "")
verify_ssl = coerce_bool(
payload.get("verify_ssl")
or payload.get("calibre_opds_verify_ssl"),
defaults["verify_ssl"],
)
enabled = coerce_bool(
payload.get("enabled")
or payload.get("calibre_opds_enabled"),
coerce_bool(stored.get("enabled"), False),
)
return {
"enabled": enabled,
"base_url": base_url,
"username": username,
"password": password,
"verify_ssl": verify_ssl,
}
def audiobookshelf_settings_from_payload(payload: Mapping[str, Any]) -> Dict[str, Any]:
defaults = integration_defaults()["audiobookshelf"]
stored = stored_integration_config("audiobookshelf")
base_url = str(
payload.get("base_url")
or payload.get("audiobookshelf_base_url")
or stored.get("base_url")
or ""
).strip()
library_id = str(
payload.get("library_id")
or payload.get("audiobookshelf_library_id")
or stored.get("library_id")
or ""
).strip()
collection_id = str(
payload.get("collection_id")
or payload.get("audiobookshelf_collection_id")
or stored.get("collection_id")
or ""
).strip()
folder_id = str(
payload.get("folder_id")
or payload.get("audiobookshelf_folder_id")
or stored.get("folder_id")
or ""
).strip()
token_input = str(
payload.get("api_token")
or payload.get("audiobookshelf_api_token")
or ""
).strip()
use_saved_token = coerce_bool(
payload.get("use_saved_token")
or payload.get("audiobookshelf_use_saved_token"),
False,
)
clear_saved_token = coerce_bool(
payload.get("clear_saved_token")
or payload.get("audiobookshelf_api_token_clear"),
False,
)
if token_input:
api_token = token_input
elif use_saved_token and not clear_saved_token:
api_token = str(stored.get("api_token") or "")
else:
api_token = ""
verify_ssl = coerce_bool(
payload.get("verify_ssl")
or payload.get("audiobookshelf_verify_ssl"),
defaults["verify_ssl"],
)
send_cover = coerce_bool(
payload.get("send_cover")
or payload.get("audiobookshelf_send_cover"),
defaults["send_cover"],
)
send_chapters = coerce_bool(
payload.get("send_chapters")
or payload.get("audiobookshelf_send_chapters"),
defaults["send_chapters"],
)
send_subtitles = coerce_bool(
payload.get("send_subtitles")
or payload.get("audiobookshelf_send_subtitles"),
defaults["send_subtitles"],
)
auto_send = coerce_bool(
payload.get("auto_send")
or payload.get("audiobookshelf_auto_send"),
defaults["auto_send"],
)
timeout_raw = (
payload.get("timeout")
or payload.get("audiobookshelf_timeout")
or stored.get("timeout")
or defaults["timeout"]
)
try:
timeout = float(timeout_raw)
except (TypeError, ValueError):
timeout = defaults["timeout"]
enabled = coerce_bool(
payload.get("enabled")
or payload.get("audiobookshelf_enabled"),
coerce_bool(stored.get("enabled"), False),
)
return {
"enabled": enabled,
"base_url": base_url,
"library_id": library_id,
"collection_id": collection_id,
"folder_id": folder_id,
"api_token": api_token,
"verify_ssl": verify_ssl,
"send_cover": send_cover,
"send_chapters": send_chapters,
"send_subtitles": send_subtitles,
"auto_send": auto_send,
"timeout": timeout,
}
def build_audiobookshelf_config(settings: Mapping[str, Any]) -> Optional[AudiobookshelfConfig]:
base_url = str(settings.get("base_url") or "").strip()
api_token = str(settings.get("api_token") or "").strip()
library_id = str(settings.get("library_id") or "").strip()
if not (base_url and api_token and library_id):
return None
try:
timeout = float(settings.get("timeout", 3600.0))
except (TypeError, ValueError):
timeout = 3600.0
return AudiobookshelfConfig(
base_url=base_url,
api_token=api_token,
library_id=library_id,
collection_id=(str(settings.get("collection_id") or "").strip() or None),
folder_id=(str(settings.get("folder_id") or "").strip() or None),
verify_ssl=coerce_bool(settings.get("verify_ssl"), True),
send_cover=coerce_bool(settings.get("send_cover"), True),
send_chapters=coerce_bool(settings.get("send_chapters"), True),
send_subtitles=coerce_bool(settings.get("send_subtitles"), False),
timeout=timeout,
)
def calibre_integration_enabled(
integrations: Optional[Mapping[str, Any]] = None,
) -> bool:
if integrations is None:
integrations = load_integration_settings()
payload = integrations.get("calibre_opds") if isinstance(integrations, Mapping) else None
if not isinstance(payload, Mapping):
return False
base_url = str(payload.get("base_url") or "").strip()
enabled_flag = coerce_bool(payload.get("enabled"), False)
return bool(enabled_flag and base_url)
def audiobookshelf_manual_available() -> bool:
settings = stored_integration_config("audiobookshelf")
if not settings:
return False
if not coerce_bool(settings.get("enabled"), False):
return False
config = build_audiobookshelf_config(settings)
return config is not None
def build_calibre_client(settings: Mapping[str, Any]) -> CalibreOPDSClient:
base_url = str(settings.get("base_url") or "").strip()
if not base_url:
raise ValueError("Calibre OPDS base URL is required")
username = str(settings.get("username") or "").strip() or None
password = str(settings.get("password") or "").strip() or None
verify_ssl = coerce_bool(settings.get("verify_ssl"), True)
timeout_raw = settings.get("timeout", 15.0)
try:
timeout = float(timeout_raw)
except (TypeError, ValueError):
timeout = 15.0
return CalibreOPDSClient(
base_url,
username=username,
password=password,
timeout=timeout,
verify=verify_ssl,
)
def apply_integration_form(cfg: Dict[str, Any], form: Mapping[str, Any]) -> None:
defaults = integration_defaults()
current_calibre = dict(cfg.get("calibre_opds") or {})
calibre_enabled = coerce_bool(form.get("calibre_opds_enabled"), False)
calibre_base = str(form.get("calibre_opds_base_url") or current_calibre.get("base_url") or "").strip()
calibre_username = str(form.get("calibre_opds_username") or current_calibre.get("username") or "").strip()
calibre_password_input = str(form.get("calibre_opds_password") or "")
calibre_clear = coerce_bool(form.get("calibre_opds_password_clear"), False)
if calibre_password_input:
calibre_password = calibre_password_input
elif calibre_clear:
calibre_password = ""
else:
calibre_password = str(current_calibre.get("password") or "")
calibre_verify = coerce_bool(form.get("calibre_opds_verify_ssl"), defaults["calibre_opds"]["verify_ssl"])
cfg["calibre_opds"] = {
"enabled": calibre_enabled,
"base_url": calibre_base,
"username": calibre_username,
"password": calibre_password,
"verify_ssl": calibre_verify,
}
current_abs = dict(cfg.get("audiobookshelf") or {})
abs_enabled = coerce_bool(form.get("audiobookshelf_enabled"), False)
abs_base = str(form.get("audiobookshelf_base_url") or current_abs.get("base_url") or "").strip()
abs_library = str(form.get("audiobookshelf_library_id") or current_abs.get("library_id") or "").strip()
abs_collection = str(form.get("audiobookshelf_collection_id") or current_abs.get("collection_id") or "").strip()
abs_folder = str(form.get("audiobookshelf_folder_id") or current_abs.get("folder_id") or "").strip()
abs_token_input = str(form.get("audiobookshelf_api_token") or "")
abs_token_clear = coerce_bool(form.get("audiobookshelf_api_token_clear"), False)
if abs_token_input:
abs_token = abs_token_input
elif abs_token_clear:
abs_token = ""
else:
abs_token = str(current_abs.get("api_token") or "")
abs_verify = coerce_bool(form.get("audiobookshelf_verify_ssl"), defaults["audiobookshelf"]["verify_ssl"])
abs_send_cover = coerce_bool(form.get("audiobookshelf_send_cover"), defaults["audiobookshelf"]["send_cover"])
abs_send_chapters = coerce_bool(form.get("audiobookshelf_send_chapters"), defaults["audiobookshelf"]["send_chapters"])
abs_send_subtitles = coerce_bool(form.get("audiobookshelf_send_subtitles"), defaults["audiobookshelf"]["send_subtitles"])
abs_auto_send = coerce_bool(form.get("audiobookshelf_auto_send"), defaults["audiobookshelf"]["auto_send"])
timeout_raw = form.get("audiobookshelf_timeout", current_abs.get("timeout", defaults["audiobookshelf"]["timeout"]))
try:
abs_timeout = float(timeout_raw)
except (TypeError, ValueError):
abs_timeout = defaults["audiobookshelf"]["timeout"]
cfg["audiobookshelf"] = {
"enabled": abs_enabled,
"base_url": abs_base,
"api_token": abs_token,
"library_id": abs_library,
"collection_id": abs_collection,
"folder_id": abs_folder,
"verify_ssl": abs_verify,
"send_cover": abs_send_cover,
"send_chapters": abs_send_chapters,
"send_subtitles": abs_send_subtitles,
"auto_send": abs_auto_send,
"timeout": abs_timeout,
}
def save_settings(settings: Dict[str, Any]) -> None:
save_config(settings)
+786
View File
@@ -0,0 +1,786 @@
import threading
from typing import Any, Dict, Iterable, List, Mapping, Optional, Tuple, cast
import numpy as np
from abogen.speaker_configs import slugify_label
from abogen.speaker_analysis import analyze_speakers
from abogen.web.routes.utils.settings import load_settings, settings_defaults, _DEFAULT_ANALYSIS_THRESHOLD, _CHUNK_LEVEL_OPTIONS, _APOSTROPHE_MODE_OPTIONS
from abogen.web.routes.utils.common import split_profile_spec
from abogen.voice_profiles import (
load_profiles,
serialize_profiles,
)
from abogen.voice_formulas import get_new_voice, parse_formula_terms
from abogen.constants import (
LANGUAGE_DESCRIPTIONS,
SUBTITLE_FORMATS,
SUPPORTED_SOUND_FORMATS,
SUPPORTED_LANGUAGES_FOR_SUBTITLE_GENERATION,
SAMPLE_VOICE_TEXTS,
VOICES_INTERNAL,
)
from abogen.speaker_configs import list_configs
from abogen.utils import load_numpy_kpipeline
from abogen.web.conversion_runner import _select_device, _to_float32, SAMPLE_RATE, SPLIT_PATTERN
_preview_pipeline_lock = threading.RLock()
_preview_pipelines: Dict[Tuple[str, str], Any] = {}
def build_narrator_roster(
voice: str,
voice_profile: Optional[str],
existing: Optional[Mapping[str, Any]] = None,
) -> Dict[str, Any]:
roster: Dict[str, Any] = {
"narrator": {
"id": "narrator",
"label": "Narrator",
"voice": voice,
}
}
if voice_profile:
roster["narrator"]["voice_profile"] = voice_profile
existing_entry: Optional[Mapping[str, Any]] = None
if existing is not None:
existing_entry = existing.get("narrator") if isinstance(existing, Mapping) else None
if isinstance(existing_entry, Mapping):
roster_entry = roster["narrator"]
for key in ("label", "voice", "voice_profile", "voice_formula", "pronunciation"):
value = existing_entry.get(key)
if value is not None and value != "":
roster_entry[key] = value
return roster
def build_speaker_roster(
analysis: Dict[str, Any],
base_voice: str,
voice_profile: Optional[str],
existing: Optional[Mapping[str, Any]] = None,
order: Optional[Iterable[str]] = None,
) -> Dict[str, Any]:
roster = build_narrator_roster(base_voice, voice_profile, existing)
existing_map: Dict[str, Any] = dict(existing) if isinstance(existing, Mapping) else {}
speakers = analysis.get("speakers", {}) if isinstance(analysis, dict) else {}
ordered_ids: Iterable[str]
if order is not None:
ordered_ids = [sid for sid in order if sid in speakers]
else:
ordered_ids = speakers.keys()
for speaker_id in ordered_ids:
payload = speakers.get(speaker_id, {})
if speaker_id == "narrator":
continue
if isinstance(payload, Mapping) and payload.get("suppressed"):
continue
previous = existing_map.get(speaker_id)
roster[speaker_id] = {
"id": speaker_id,
"label": payload.get("label") or speaker_id.replace("_", " ").title(),
"analysis_confidence": payload.get("confidence"),
"analysis_count": payload.get("count"),
"gender": payload.get("gender", "unknown"),
}
detected_gender = payload.get("detected_gender")
if detected_gender:
roster[speaker_id]["detected_gender"] = detected_gender
samples = payload.get("sample_quotes")
if isinstance(samples, list):
roster[speaker_id]["sample_quotes"] = samples
if isinstance(previous, Mapping):
for key in ("voice", "voice_profile", "voice_formula", "resolved_voice", "pronunciation"):
value = previous.get(key)
if value is not None and value != "":
roster[speaker_id][key] = value
if "sample_quotes" not in roster[speaker_id]:
prev_samples = previous.get("sample_quotes")
if isinstance(prev_samples, list):
roster[speaker_id]["sample_quotes"] = prev_samples
if "detected_gender" not in roster[speaker_id]:
prev_detected = previous.get("detected_gender")
if isinstance(prev_detected, str) and prev_detected:
roster[speaker_id]["detected_gender"] = prev_detected
return roster
def match_configured_speaker(
config_speakers: Mapping[str, Any],
roster_id: str,
roster_label: str,
) -> Optional[Mapping[str, Any]]:
if not config_speakers:
return None
entry = config_speakers.get(roster_id)
if entry:
return cast(Mapping[str, Any], entry)
slug = slugify_label(roster_label)
if slug != roster_id and slug in config_speakers:
return cast(Mapping[str, Any], config_speakers[slug])
lower_label = roster_label.strip().lower()
for record in config_speakers.values():
if not isinstance(record, Mapping):
continue
if str(record.get("label", "")).strip().lower() == lower_label:
return record
return None
def apply_speaker_config_to_roster(
roster: Mapping[str, Any],
config: Optional[Mapping[str, Any]],
*,
persist_changes: bool = False,
fallback_languages: Optional[Iterable[str]] = None,
) -> Tuple[Dict[str, Any], List[str], Optional[Dict[str, Any]]]:
if not isinstance(roster, Mapping):
effective_languages = [code for code in (fallback_languages or []) if isinstance(code, str) and code]
return {}, effective_languages, None
updated_roster: Dict[str, Any] = {key: dict(value) for key, value in roster.items() if isinstance(value, Mapping)}
if not config:
effective_languages = [code for code in (fallback_languages or []) if isinstance(code, str) and code]
return updated_roster, effective_languages, None
speakers_map = config.get("speakers")
if not isinstance(speakers_map, Mapping):
effective_languages = [code for code in (fallback_languages or []) if isinstance(code, str) and code]
return updated_roster, effective_languages, None
config_languages = config.get("languages")
if isinstance(config_languages, list):
allowed_languages = [code for code in config_languages if isinstance(code, str) and code]
else:
allowed_languages = []
if not allowed_languages and fallback_languages:
allowed_languages = [code for code in fallback_languages if isinstance(code, str) and code]
default_voice = config.get("default_voice") if isinstance(config.get("default_voice"), str) else ""
used_voices = {entry.get("resolved_voice") or entry.get("voice") for entry in updated_roster.values()} - {None}
narrator_voice = ""
narrator_entry = updated_roster.get("narrator") if isinstance(updated_roster, Mapping) else None
if isinstance(narrator_entry, Mapping):
narrator_voice = str(
narrator_entry.get("resolved_voice")
or narrator_entry.get("default_voice")
or ""
).strip()
if narrator_voice:
used_voices.add(narrator_voice)
config_changed = False
new_config_payload: Dict[str, Any] = {
"language": config.get("language", "a"),
"languages": allowed_languages,
"default_voice": default_voice,
"speakers": dict(speakers_map),
"version": config.get("version", 1),
"notes": config.get("notes", ""),
}
speakers_payload = new_config_payload["speakers"]
for speaker_id, roster_entry in updated_roster.items():
if speaker_id == "narrator":
continue
label = str(roster_entry.get("label") or speaker_id)
config_entry = match_configured_speaker(speakers_map, speaker_id, label)
if config_entry is None:
continue
voice_id = str(config_entry.get("voice") or "").strip()
voice_profile = str(config_entry.get("voice_profile") or "").strip()
voice_formula = str(config_entry.get("voice_formula") or "").strip()
resolved_voice = str(config_entry.get("resolved_voice") or "").strip()
languages = config_entry.get("languages") if isinstance(config_entry.get("languages"), list) else []
chosen_voice = resolved_voice or voice_formula or voice_id or roster_entry.get("voice")
usable_languages = languages or allowed_languages
if chosen_voice:
roster_entry["resolved_voice"] = chosen_voice
roster_entry["voice"] = chosen_voice if not voice_profile and not voice_formula else roster_entry.get("voice", chosen_voice)
if voice_profile:
roster_entry["voice_profile"] = voice_profile
if voice_formula:
roster_entry["voice_formula"] = voice_formula
roster_entry["resolved_voice"] = voice_formula
if not voice_formula and not voice_profile and resolved_voice:
roster_entry["resolved_voice"] = resolved_voice
roster_entry["config_languages"] = usable_languages or []
if chosen_voice:
used_voices.add(chosen_voice)
# persist updates back to config payload if required
if persist_changes:
slug = config_entry.get("id") or slugify_label(label)
speakers_payload[slug] = {
"id": slug,
"label": label,
"gender": config_entry.get("gender", "unknown"),
"voice": voice_id,
"voice_profile": voice_profile,
"voice_formula": voice_formula,
"resolved_voice": roster_entry.get("resolved_voice", resolved_voice or voice_id),
"languages": usable_languages,
}
new_config = new_config_payload if (persist_changes and config_changed) else None
return updated_roster, allowed_languages, new_config
def filter_voice_catalog(
catalog: Iterable[Mapping[str, Any]],
*,
gender: str,
allowed_languages: Optional[Iterable[str]] = None,
) -> List[str]:
allowed_set = {code.lower() for code in (allowed_languages or []) if isinstance(code, str) and code}
gender_normalized = (gender or "unknown").lower()
gender_code = ""
if gender_normalized == "male":
gender_code = "m"
elif gender_normalized == "female":
gender_code = "f"
matches: List[str] = []
seen: set[str] = set()
def _consider(entry: Mapping[str, Any]) -> None:
voice_id = entry.get("id")
if not isinstance(voice_id, str) or not voice_id:
return
if voice_id in seen:
return
seen.add(voice_id)
matches.append(voice_id)
primary: List[Mapping[str, Any]] = []
fallback: List[Mapping[str, Any]] = []
for entry in catalog:
if not isinstance(entry, Mapping):
continue
voice_lang = str(entry.get("language", "")).lower()
voice_gender_code = str(entry.get("gender_code", "")).lower()
if allowed_set and voice_lang not in allowed_set:
continue
if gender_code and voice_gender_code != gender_code:
fallback.append(entry)
continue
primary.append(entry)
for entry in primary:
_consider(entry)
if not matches:
for entry in fallback:
_consider(entry)
if not matches:
for entry in catalog:
if isinstance(entry, Mapping):
_consider(entry)
return matches
def build_voice_catalog() -> List[Dict[str, str]]:
catalog: List[Dict[str, str]] = []
gender_map = {"f": "Female", "m": "Male"}
for voice_id in VOICES_INTERNAL:
prefix, _, rest = voice_id.partition("_")
language_code = prefix[0] if prefix else "a"
gender_code = prefix[1] if len(prefix) > 1 else ""
catalog.append(
{
"id": voice_id,
"language": language_code,
"language_label": LANGUAGE_DESCRIPTIONS.get(language_code, language_code.upper()),
"gender": gender_map.get(gender_code, "Unknown"),
"gender_code": gender_code,
"display_name": rest.replace("_", " ").title() if rest else voice_id,
}
)
return catalog
def inject_recommended_voices(
roster: Mapping[str, Any],
*,
fallback_languages: Optional[Iterable[str]] = None,
) -> None:
voice_catalog = build_voice_catalog()
fallback_list = [code for code in (fallback_languages or []) if isinstance(code, str) and code]
for speaker_id, payload in roster.items():
if not isinstance(payload, dict):
continue
languages = payload.get("config_languages")
if isinstance(languages, list) and languages:
language_list = languages
else:
language_list = fallback_list
gender = str(payload.get("gender", "unknown"))
payload["recommended_voices"] = filter_voice_catalog(
voice_catalog,
gender=gender,
allowed_languages=language_list,
)
def extract_speaker_config_form(form: Mapping[str, Any]) -> Tuple[str, Dict[str, Any], List[str]]:
getter = getattr(form, "getlist", None)
def _get_list(name: str) -> List[str]:
if callable(getter):
values = cast(Iterable[Any], getter(name))
return [str(value).strip() for value in values if value]
raw_value = form.get(name)
if isinstance(raw_value, str):
return [item.strip() for item in raw_value.split(",") if item.strip()]
return []
name = (form.get("config_name") or "").strip()
language = str(form.get("config_language") or "a").strip() or "a"
allowed_languages = []
default_voice = (form.get("config_default_voice") or "").strip()
notes = (form.get("config_notes") or "").strip()
try:
parsed = int(form.get("config_version") or 1)
version = max(1, min(parsed, 9999))
except (TypeError, ValueError):
version = 1
speaker_rows = _get_list("speaker_rows")
speakers: Dict[str, Dict[str, Any]] = {}
for row_key in speaker_rows:
prefix = f"speaker-{row_key}-"
label = (form.get(prefix + "label") or "").strip()
if not label:
continue
raw_gender = (form.get(prefix + "gender") or "unknown").strip().lower()
gender = raw_gender if raw_gender in {"male", "female", "unknown"} else "unknown"
voice = (form.get(prefix + "voice") or "").strip()
voice_profile = (form.get(prefix + "profile") or "").strip()
voice_formula = (form.get(prefix + "formula") or "").strip()
speaker_id = (form.get(prefix + "id") or "").strip() or slugify_label(label)
speakers[speaker_id] = {
"id": speaker_id,
"label": label,
"gender": gender,
"voice": voice,
"voice_profile": voice_profile,
"voice_formula": voice_formula,
"resolved_voice": voice_formula or voice,
"languages": [],
}
payload = {
"language": language,
"languages": allowed_languages,
"default_voice": default_voice,
"speakers": speakers,
"notes": notes,
"version": version,
}
errors: List[str] = []
if not name:
errors.append("Configuration name is required.")
if not speakers:
errors.append("Add at least one speaker to the configuration.")
return name, payload, errors
def prepare_speaker_metadata(
*,
chapters: List[Dict[str, Any]],
chunks: List[Dict[str, Any]],
analysis_chunks: Optional[List[Dict[str, Any]]] = None,
voice: str,
voice_profile: Optional[str],
threshold: int,
existing_roster: Optional[Mapping[str, Any]] = None,
run_analysis: bool = True,
speaker_config: Optional[Mapping[str, Any]] = None,
apply_config: bool = False,
persist_config: bool = False,
) -> tuple[List[Dict[str, Any]], Dict[str, Any], Dict[str, Any], List[str], Optional[Dict[str, Any]]]:
chunk_list = [dict(chunk) for chunk in chunks]
analysis_source = [dict(chunk) for chunk in (analysis_chunks or chunks)]
threshold_value = max(1, int(threshold))
analysis_enabled = run_analysis
settings_state = load_settings()
global_random_languages = [
code
for code in settings_state.get("speaker_random_languages", [])
if isinstance(code, str) and code
]
if not analysis_enabled:
for chunk in chunk_list:
chunk["speaker_id"] = "narrator"
chunk["speaker_label"] = "Narrator"
analysis_payload = {
"version": "1.0",
"narrator": "narrator",
"assignments": {str(chunk.get("id")): "narrator" for chunk in chunk_list},
"speakers": {
"narrator": {
"id": "narrator",
"label": "Narrator",
"count": len(chunk_list),
"confidence": "low",
"sample_quotes": [],
"suppressed": False,
}
},
"suppressed": [],
"stats": {
"total_chunks": len(chunk_list),
"explicit_chunks": 0,
"active_speakers": 0,
"unique_speakers": 1,
"suppressed": 0,
},
}
roster = build_narrator_roster(voice, voice_profile, existing_roster)
narrator_pron = roster["narrator"].get("pronunciation")
if narrator_pron:
analysis_payload["speakers"]["narrator"]["pronunciation"] = narrator_pron
return chunk_list, roster, analysis_payload, [], None
analysis_result = analyze_speakers(
chapters,
analysis_source,
threshold=threshold_value,
max_speakers=0,
)
analysis_payload = analysis_result.to_dict()
speakers_payload = analysis_payload.get("speakers", {})
ordered_ids = [
sid
for sid, meta in sorted(
(
(sid, meta)
for sid, meta in speakers_payload.items()
if sid != "narrator" and isinstance(meta, Mapping) and not meta.get("suppressed")
),
key=lambda item: item[1].get("count", 0),
reverse=True,
)
]
analysis_payload["ordered_speakers"] = ordered_ids
assignments = analysis_payload.get("assignments", {})
suppressed_ids = analysis_payload.get("suppressed", [])
suppressed_details: List[Dict[str, Any]] = []
speakers_payload = analysis_payload.get("speakers", {})
if isinstance(suppressed_ids, Iterable):
for suppressed_id in suppressed_ids:
speaker_meta = speakers_payload.get(suppressed_id) if isinstance(speakers_payload, dict) else None
if isinstance(speaker_meta, dict):
suppressed_details.append(
{
"id": suppressed_id,
"label": speaker_meta.get("label")
or str(suppressed_id).replace("_", " ").title(),
"pronunciation": speaker_meta.get("pronunciation"),
}
)
else:
suppressed_details.append(
{
"id": suppressed_id,
"label": str(suppressed_id).replace("_", " ").title(),
"pronunciation": None,
}
)
analysis_payload["suppressed_details"] = suppressed_details
roster = build_speaker_roster(
analysis_payload,
voice,
voice_profile,
existing=existing_roster,
order=analysis_payload.get("ordered_speakers"),
)
applied_languages: List[str] = []
updated_config: Optional[Dict[str, Any]] = None
if apply_config and speaker_config:
roster, applied_languages, updated_config = apply_speaker_config_to_roster(
roster,
speaker_config,
persist_changes=persist_config,
fallback_languages=global_random_languages,
)
speakers_payload = analysis_payload.get("speakers")
if isinstance(speakers_payload, dict):
for roster_id, roster_payload in roster.items():
speaker_meta = speakers_payload.get(roster_id)
if isinstance(speaker_meta, dict):
for key in ("voice", "voice_profile", "voice_formula", "resolved_voice"):
value = roster_payload.get(key)
if value:
speaker_meta[key] = value
effective_languages: List[str] = []
if applied_languages:
effective_languages = applied_languages
elif isinstance(analysis_payload.get("config_languages"), list):
effective_languages = [
code for code in analysis_payload.get("config_languages", []) if isinstance(code, str) and code
]
elif global_random_languages:
effective_languages = list(global_random_languages)
if effective_languages:
analysis_payload["config_languages"] = effective_languages
speakers_payload = analysis_payload.get("speakers")
if isinstance(speakers_payload, dict):
for roster_id, roster_payload in roster.items():
if roster_id in speakers_payload and isinstance(roster_payload, dict):
pronunciation_value = roster_payload.get("pronunciation")
if pronunciation_value:
speakers_payload[roster_id]["pronunciation"] = pronunciation_value
fallback_languages = effective_languages or []
inject_recommended_voices(roster, fallback_languages=fallback_languages)
for chunk in chunk_list:
chunk_id = str(chunk.get("id"))
speaker_id = assignments.get(chunk_id, "narrator")
chunk["speaker_id"] = speaker_id
speaker_meta = roster.get(speaker_id)
chunk["speaker_label"] = speaker_meta.get("label") if isinstance(speaker_meta, dict) else speaker_id
return chunk_list, roster, analysis_payload, applied_languages, updated_config
def formula_from_profile(entry: Dict[str, Any]) -> Optional[str]:
voices = entry.get("voices") or []
if not voices:
return None
total = sum(weight for _, weight in voices)
if total <= 0:
return None
def _format_weight(value: float) -> str:
normalized = value / total if total else 0.0
return (f"{normalized:.4f}").rstrip("0").rstrip(".") or "0"
parts = [f"{name}*{_format_weight(weight)}" for name, weight in voices if weight > 0]
return "+".join(parts) if parts else None
def template_options() -> Dict[str, Any]:
current_settings = load_settings()
profiles = serialize_profiles()
ordered_profiles = sorted(profiles.items())
profile_options = []
for name, entry in ordered_profiles:
profile_options.append(
{
"name": name,
"language": (entry or {}).get("language", ""),
"formula": formula_from_profile(entry or {}) or "",
}
)
voice_catalog = build_voice_catalog()
return {
"languages": LANGUAGE_DESCRIPTIONS,
"voices": VOICES_INTERNAL,
"subtitle_formats": SUBTITLE_FORMATS,
"supported_langs_for_subs": SUPPORTED_LANGUAGES_FOR_SUBTITLE_GENERATION,
"output_formats": SUPPORTED_SOUND_FORMATS,
"voice_profiles": ordered_profiles,
"voice_profile_options": profile_options,
"separate_formats": ["wav", "flac", "mp3", "opus"],
"voice_catalog": voice_catalog,
"voice_catalog_map": {entry["id"]: entry for entry in voice_catalog},
"sample_voice_texts": SAMPLE_VOICE_TEXTS,
"voice_profiles_data": profiles,
"speaker_configs": list_configs(),
"chunk_levels": _CHUNK_LEVEL_OPTIONS,
"speaker_analysis_threshold": current_settings.get(
"speaker_analysis_threshold", _DEFAULT_ANALYSIS_THRESHOLD
),
"speaker_pronunciation_sentence": current_settings.get(
"speaker_pronunciation_sentence", settings_defaults()["speaker_pronunciation_sentence"]
),
"apostrophe_modes": _APOSTROPHE_MODE_OPTIONS,
}
def resolve_profile_voice(
profile_name: Optional[str],
*,
profiles: Optional[Mapping[str, Any]] = None,
) -> tuple[str, Optional[str]]:
if not profile_name:
return "", None
source = profiles if isinstance(profiles, Mapping) else None
if source is None:
source = load_profiles()
entry = source.get(profile_name) if isinstance(source, Mapping) else None
if not isinstance(entry, Mapping):
return "", None
formula = formula_from_profile(dict(entry)) or ""
language = entry.get("language") if isinstance(entry.get("language"), str) else None
if isinstance(language, str):
language = language.strip().lower() or None
return formula, language
def resolve_voice_setting(
value: Any,
*,
profiles: Optional[Mapping[str, Any]] = None,
) -> tuple[str, Optional[str], Optional[str]]:
base_spec, profile_name = split_profile_spec(value)
if profile_name:
formula, language = resolve_profile_voice(profile_name, profiles=profiles)
return formula or "", profile_name, language
return base_spec, None, None
def resolve_voice_choice(
language: str,
base_voice: str,
profile_name: str,
custom_formula: str,
profiles: Dict[str, Any],
) -> tuple[str, str, Optional[str]]:
resolved_voice = base_voice
resolved_language = language
selected_profile = None
if profile_name:
entry = profiles.get(profile_name)
formula = formula_from_profile(entry or {}) if entry else None
if formula:
resolved_voice = formula
selected_profile = profile_name
profile_language = (entry or {}).get("language")
if profile_language:
resolved_language = profile_language
if custom_formula:
resolved_voice = custom_formula
selected_profile = None
return resolved_voice, resolved_language, selected_profile
def parse_voice_formula(formula: str) -> List[tuple[str, float]]:
voices = parse_formula_terms(formula)
total = sum(weight for _, weight in voices)
if total <= 0:
raise ValueError("Voice weights must sum to a positive value")
return voices
def sanitize_voice_entries(entries: Iterable[Any]) -> List[Dict[str, Any]]:
sanitized: List[Dict[str, Any]] = []
for entry in entries or []:
if isinstance(entry, dict):
voice_id = entry.get("id") or entry.get("voice")
if not voice_id:
continue
enabled = entry.get("enabled", True)
if not enabled:
continue
sanitized.append({"voice": voice_id, "weight": entry.get("weight")})
elif isinstance(entry, (list, tuple)) and len(entry) >= 2:
sanitized.append({"voice": entry[0], "weight": entry[1]})
return sanitized
def pairs_to_formula(pairs: Iterable[Tuple[str, float]]) -> Optional[str]:
voices = [(voice, float(weight)) for voice, weight in pairs if float(weight) > 0]
if not voices:
return None
total = sum(weight for _, weight in voices)
if total <= 0:
return None
def _format_value(value: float) -> str:
normalized = value / total if total else 0.0
return (f"{normalized:.4f}").rstrip("0").rstrip(".") or "0"
parts = [f"{voice}*{_format_value(weight)}" for voice, weight in voices]
return "+".join(parts)
def profiles_payload() -> Dict[str, Any]:
return {"profiles": serialize_profiles()}
def get_preview_pipeline(language: str, device: str):
key = (language, device)
with _preview_pipeline_lock:
pipeline = _preview_pipelines.get(key)
if pipeline is not None:
return pipeline
_, KPipeline = load_numpy_kpipeline()
pipeline = KPipeline(lang_code=language, repo_id="hexgrad/Kokoro-82M", device=device)
_preview_pipelines[key] = pipeline
return pipeline
def synthesize_audio_from_normalized(
*,
normalized_text: str,
voice_spec: str,
language: str,
speed: float,
use_gpu: bool,
max_seconds: float,
) -> np.ndarray:
if not normalized_text.strip():
raise ValueError("Preview text is required")
device = "cpu"
if use_gpu:
try:
device = _select_device()
except Exception:
device = "cpu"
use_gpu = False
pipeline = get_preview_pipeline(language, device)
if pipeline is None:
raise RuntimeError("Preview pipeline is unavailable")
voice_choice: Any = voice_spec
if voice_spec and "*" in voice_spec:
voice_choice = get_new_voice(pipeline, voice_spec, use_gpu)
segments = pipeline(
normalized_text,
voice=voice_choice,
speed=speed,
split_pattern=SPLIT_PATTERN,
)
audio_chunks: List[np.ndarray] = []
accumulated = 0
max_samples = int(max(1.0, max_seconds) * SAMPLE_RATE)
for segment in segments:
graphemes = getattr(segment, "graphemes", "").strip()
if not graphemes:
continue
audio = _to_float32(getattr(segment, "audio", None))
if audio.size == 0:
continue
remaining = max_samples - accumulated
if remaining <= 0:
break
if audio.shape[0] > remaining:
audio = audio[:remaining]
audio_chunks.append(audio)
accumulated += audio.shape[0]
if accumulated >= max_samples:
break
if not audio_chunks:
raise RuntimeError("Preview could not be generated")
return np.concatenate(audio_chunks)
+81
View File
@@ -0,0 +1,81 @@
from typing import Any, Dict, List, Optional
from flask import Blueprint, render_template, request, jsonify, abort, flash, redirect, url_for
from flask.typing import ResponseReturnValue
from abogen.web.routes.utils.voice import (
template_options,
resolve_voice_setting,
resolve_voice_choice,
parse_voice_formula,
)
from abogen.web.routes.utils.settings import load_settings, coerce_bool
from abogen.web.routes.utils.preview import synthesize_preview
from abogen.speaker_configs import (
list_configs,
get_config,
load_configs,
save_configs,
delete_config,
)
from abogen.constants import VOICES_INTERNAL
voices_bp = Blueprint("voices", __name__)
@voices_bp.get("/")
def voices_list() -> ResponseReturnValue:
# This might not be a standalone page in the original app, but useful to have.
# Or maybe it redirects to settings or something.
# For now, I'll just redirect to settings as voices are managed there usually.
return redirect(url_for("settings.settings_page"))
@voices_bp.post("/test")
def test_voice() -> ResponseReturnValue:
text = (request.form.get("text") or "").strip()
voice = (request.form.get("voice") or "").strip()
speed = float(request.form.get("speed", 1.0))
# This seems to be the form-based preview
settings = load_settings()
use_gpu = coerce_bool(settings.get("use_gpu"), True)
try:
return synthesize_preview(
text=text,
voice_spec=voice,
language="a", # Default language
speed=speed,
use_gpu=use_gpu,
)
except Exception as e:
abort(400, str(e))
@voices_bp.get("/configs")
def speaker_configs() -> ResponseReturnValue:
return jsonify({"configs": list_configs()})
@voices_bp.post("/configs/save")
def save_speaker_config() -> ResponseReturnValue:
payload = request.get_json(force=True)
name = (payload.get("name") or "").strip()
config = payload.get("config")
if not name:
abort(400, "Config name is required")
if not config:
abort(400, "Config data is required")
configs = load_configs()
configs[name] = config
save_configs(configs)
return jsonify({"status": "saved", "configs": list_configs()})
@voices_bp.post("/configs/delete")
def delete_speaker_config() -> ResponseReturnValue:
payload = request.get_json(force=True)
name = (payload.get("name") or "").strip()
if not name:
abort(400, "Config name is required")
delete_config(name)
return jsonify({"status": "deleted", "configs": list_configs()})
+4 -4
View File
@@ -335,7 +335,7 @@ def _extract_year(raw: Optional[str]) -> Optional[int]:
return None
def _build_audiobookshelf_metadata(job: Job) -> Dict[str, Any]:
def build_audiobookshelf_metadata(job: Job) -> Dict[str, Any]:
tags = _normalize_metadata_casefold(job.metadata_tags)
filename = Path(job.original_filename or "").stem or job.original_filename or "Audiobook"
title = _first_nonempty(
@@ -474,7 +474,7 @@ def _normalize_series_sequence(raw: Any) -> Optional[str]:
return cleaned or "0"
def _load_audiobookshelf_chapters(job: Job) -> Optional[List[Dict[str, Any]]]:
def load_audiobookshelf_chapters(job: Job) -> Optional[List[Dict[str, Any]]]:
metadata_ref = job.result.artifacts.get("metadata")
if not metadata_ref:
return None
@@ -1085,8 +1085,8 @@ class ConversionService:
cover_path = cover_candidate
subtitles = _existing_paths(job.result.subtitle_paths) if config.send_subtitles else None
chapters = _load_audiobookshelf_chapters(job) if config.send_chapters else None
metadata = _build_audiobookshelf_metadata(job)
chapters = load_audiobookshelf_chapters(job) if config.send_chapters else None
metadata = build_audiobookshelf_metadata(job)
client = AudiobookshelfClient(config)