Dymas/abogen
1
0
Fork 0
mirror of https://github.com/denizsafak/abogen.git synced 2026-07-18 13:40:27 +02:00
Code Issues Packages Projects Releases Wiki Activity

feat: Add title and suffix abbreviation expansion, ensure terminal punctuation, and enhance speaker analysis functionality

Browse Source
This commit is contained in:
JB
2025-10-08 05:43:49 -07:00
parent b0cfd8d687
commit 3b07df9708
7 changed files with 419 additions and 153 deletions
Download Patch File Download Diff File Expand all files Collapse all files
abogen/kokoro_text_normalization.py
+95
View File
@@ -120,6 +120,34 @@ WORD_TOKEN_RE = re.compile(r"[A-Za-z0-9'’]+|[^A-Za-z0-9\s]")
APOSTROPHE_CHARS = "’`´ꞌʼ"
TERMINAL_PUNCTUATION = {".", "?", "!", "…", ";", ":"}
CLOSING_PUNCTUATION = '"\'”’)]}»›'
ELLIPSIS_SUFFIXES = ("...", "…")
_LINE_SPLIT_RE = re.compile(r"(\n+)")
TITLE_ABBREVIATIONS = {
"mr": "mister",
"mrs": "missus",
"ms": "miz",
"dr": "doctor",
"prof": "professor",
"rev": "reverend",
}
SUFFIX_ABBREVIATIONS = {
"jr": "junior",
"sr": "senior",
}
_TITLE_PATTERN = re.compile(
r"\b(?P<abbr>" + "|".join(sorted(TITLE_ABBREVIATIONS.keys(), key=len, reverse=True)) + r")\.",
re.IGNORECASE,
)
_SUFFIX_PATTERN = re.compile(
r"\b(?P<abbr>" + "|".join(sorted(SUFFIX_ABBREVIATIONS.keys(), key=len, reverse=True)) + r")\.",
re.IGNORECASE,
)
# ---------- Utility Functions ----------
def normalize_unicode_apostrophes(text: str) -> str:
@@ -132,6 +160,73 @@ def tokenize(text: str) -> List[str]:
# Simple tokenization preserving punctuation tokens
return WORD_TOKEN_RE.findall(text)
def _match_casing(template: str, replacement: str) -> str:
if template.isupper():
return replacement.upper()
if template[:1].isupper() and template[1:].islower():
return replacement.capitalize()
if template[:1].isupper():
# Mixed case (e.g., Mc), fall back to title case
return replacement.capitalize()
return replacement
def expand_titles_and_suffixes(text: str) -> str:
def _replace(match: re.Match[str], mapping: dict[str, str]) -> str:
abbr = match.group("abbr")
lookup = mapping.get(abbr.lower())
if not lookup:
return match.group(0)
return _match_casing(abbr, lookup)
text = _TITLE_PATTERN.sub(lambda m: _replace(m, TITLE_ABBREVIATIONS), text)
text = _SUFFIX_PATTERN.sub(lambda m: _replace(m, SUFFIX_ABBREVIATIONS), text)
return text
def ensure_terminal_punctuation(text: str) -> str:
def _amend(segment: str) -> str:
if not segment or not segment.strip():
return segment
stripped = segment.rstrip()
trailing_ws = segment[len(stripped) :]
match = re.match(rf"^(.*?)([{re.escape(CLOSING_PUNCTUATION)}]*)$", stripped)
if not match:
return segment
body, closers = match.groups()
if not body:
return segment
normalized_body = body.rstrip()
trailing_body_ws = body[len(normalized_body) :]
if normalized_body.endswith(ELLIPSIS_SUFFIXES):
return normalized_body + trailing_body_ws + closers + trailing_ws
last_char = normalized_body[-1]
if last_char in TERMINAL_PUNCTUATION:
return normalized_body + trailing_body_ws + closers + trailing_ws
return normalized_body + "." + trailing_body_ws + closers + trailing_ws
parts = _LINE_SPLIT_RE.split(text)
amended: List[str] = []
for part in parts:
if not part:
continue
if part.startswith("\n"):
amended.append(part)
else:
amended.append(_amend(part))
if not parts:
return _amend(text)
return "".join(amended)
def is_cultural_name(token: str, cfg: ApostropheConfig) -> bool:
if not cfg.protect_cultural_names:
return False
abogen/web/conversion_runner.py
+4
View File
@@ -21,6 +21,8 @@ from abogen.epub3.exporter import build_epub3_package
from abogen.kokoro_text_normalization import (
ApostropheConfig,
apply_phoneme_hints,
expand_titles_and_suffixes,
ensure_terminal_punctuation,
normalize_apostrophes,
)
from abogen.text_extractor import ExtractedChapter, extract_from_path
@@ -298,6 +300,8 @@ _APOSTROPHE_CONFIG = ApostropheConfig()
def _normalize_for_pipeline(text: str) -> str:
normalized, _details = normalize_apostrophes(text, _APOSTROPHE_CONFIG)
normalized = expand_titles_and_suffixes(normalized)
normalized = ensure_terminal_punctuation(normalized)
if _APOSTROPHE_CONFIG.add_phoneme_hints:
return apply_phoneme_hints(normalized, iz_marker=_APOSTROPHE_CONFIG.sibilant_iz_marker)
return normalized
abogen/web/routes.py
+252 -153
View File
@@ -13,7 +13,6 @@ from typing import Any, Dict, Iterable, List, Mapping, Optional, Tuple, cast
from flask import (
Blueprint,
Response,
abort,
current_app,
jsonify,
@@ -23,6 +22,7 @@ from flask import (
send_file,
url_for,
)
from flask.typing import ResponseReturnValue
from werkzeug.utils import secure_filename
import numpy as np
@@ -154,11 +154,13 @@ def _prepare_speaker_metadata(
voice_profile: Optional[str],
threshold: int,
existing_roster: Optional[Mapping[str, Any]] = None,
run_analysis: bool = True,
) -> tuple[List[Dict[str, Any]], Dict[str, Any], Dict[str, Any]]:
chunk_list = [dict(chunk) for chunk in chunks]
threshold_value = max(1, int(threshold))
analysis_enabled = speaker_mode == "multi" and run_analysis
if speaker_mode != "multi":
if not analysis_enabled:
for chunk in chunk_list:
chunk["speaker_id"] = "narrator"
chunk["speaker_label"] = "Narrator"
@@ -239,6 +241,132 @@ def _prepare_speaker_metadata(
return chunk_list, roster, analysis_payload
def _apply_prepare_form(
pending: PendingJob, form: Mapping[str, Any]
) -> tuple[ChunkLevel, List[Dict[str, Any]], List[Dict[str, Any]], List[str], int]:
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)
raw_speaker_mode = (form.get("speaker_mode") or pending.speaker_mode or "single").strip().lower()
if raw_speaker_mode not in _SPEAKER_MODE_VALUES:
raw_speaker_mode = "single"
pending.speaker_mode = raw_speaker_mode
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
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)
profiles = serialize_profiles()
errors: List[str] = []
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
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"] = len(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 += len(entry["text"] or "")
overrides.append(entry)
pending.chapters[index] = dict(entry)
enabled_overrides = [entry for entry in overrides if entry.get("enabled")]
return chunk_level_literal, overrides, enabled_overrides, errors, selected_total
_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),
@@ -689,7 +817,7 @@ def queue_page() -> str:
@web_bp.route("/settings", methods=["GET", "POST"])
def settings_page() -> Response | str:
def settings_page() -> ResponseReturnValue:
options = _template_options()
current_settings = _load_settings()
@@ -766,7 +894,7 @@ def voice_profiles_page() -> str:
@web_bp.post("/voices")
def save_voice_profile_route() -> Response:
def save_voice_profile_route() -> ResponseReturnValue:
name = request.form.get("name", "").strip()
language = request.form.get("language", "a").strip() or "a"
formula = request.form.get("formula", "").strip()
@@ -780,18 +908,18 @@ def save_voice_profile_route() -> Response:
@web_bp.post("/voices/<name>/delete")
def delete_voice_profile_route(name: str) -> Response:
def delete_voice_profile_route(name: str) -> ResponseReturnValue:
delete_profile(name)
return redirect(url_for("web.voice_profiles_page"))
@api_bp.get("/voice-profiles")
def api_list_voice_profiles() -> Response:
def api_list_voice_profiles() -> ResponseReturnValue:
return jsonify(_profiles_payload())
@api_bp.post("/voice-profiles")
def api_save_voice_profile() -> Response:
def api_save_voice_profile() -> ResponseReturnValue:
payload = request.get_json(force=True, silent=False)
name = (payload.get("name") or "").strip()
if not name:
@@ -819,13 +947,13 @@ def api_save_voice_profile() -> Response:
@api_bp.delete("/voice-profiles/<name>")
def api_delete_voice_profile(name: str) -> Response:
def api_delete_voice_profile(name: str) -> ResponseReturnValue:
remove_profile(name)
return jsonify({"ok": True, **_profiles_payload()})
@api_bp.post("/voice-profiles/<name>/duplicate")
def api_duplicate_voice_profile(name: str) -> Response:
def api_duplicate_voice_profile(name: str) -> ResponseReturnValue:
payload = request.get_json(silent=True) or {}
new_name = (payload.get("name") or payload.get("new_name") or "").strip()
if not new_name:
@@ -835,13 +963,18 @@ def api_duplicate_voice_profile(name: str) -> Response:
@api_bp.post("/voice-profiles/import")
def api_import_voice_profiles() -> Response:
def api_import_voice_profiles() -> ResponseReturnValue:
replace = False
data: Optional[Dict[str, Any]] = None
if "file" in request.files:
file_storage = request.files["file"]
try:
data = json.load(file_storage)
file_storage.stream.seek(0)
raw_bytes = file_storage.read()
text_payload = raw_bytes.decode("utf-8")
data = json.loads(text_payload)
except UnicodeDecodeError as exc:
abort(400, f"JSON file must be UTF-8 encoded: {exc}")
except Exception as exc: # pragma: no cover - defensive
abort(400, f"Invalid JSON file: {exc}")
replace = request.form.get("replace_existing") in {"true", "1", "on"}
@@ -863,7 +996,7 @@ def api_import_voice_profiles() -> Response:
@api_bp.get("/voice-profiles/export")
def api_export_voice_profiles() -> Response:
def api_export_voice_profiles() -> ResponseReturnValue:
names_param = request.args.get("names")
names = None
if names_param:
@@ -882,7 +1015,7 @@ def api_export_voice_profiles() -> Response:
@api_bp.post("/voice-profiles/preview")
def api_preview_voice_mix() -> Response:
def api_preview_voice_mix() -> ResponseReturnValue:
payload = request.get_json(force=True, silent=False)
language = (payload.get("language") or "a").strip() or "a"
text = (payload.get("text") or "").strip()
@@ -1010,7 +1143,7 @@ def api_preview_voice_mix() -> Response:
@api_bp.post("/speaker-preview")
def api_speaker_preview() -> Response:
def api_speaker_preview() -> ResponseReturnValue:
payload = request.get_json(force=True, silent=False)
text = (payload.get("text") or "").strip()
voice_spec = (payload.get("voice") or "").strip()
@@ -1106,7 +1239,7 @@ def api_speaker_preview() -> Response:
@web_bp.post("/jobs")
def enqueue_job() -> Response:
def enqueue_job() -> ResponseReturnValue:
service = _service()
uploads_dir = Path(current_app.config["UPLOAD_FOLDER"])
uploads_dir.mkdir(parents=True, exist_ok=True)
@@ -1252,6 +1385,7 @@ def enqueue_job() -> Response:
maximum=25,
)
initial_analysis = speaker_mode_value != "multi"
processed_chunks, speakers, analysis_payload = _prepare_speaker_metadata(
chapters=selected_chapter_sources,
chunks=raw_chunks,
@@ -1259,6 +1393,7 @@ def enqueue_job() -> Response:
voice=voice,
voice_profile=selected_profile or None,
threshold=analysis_threshold,
run_analysis=initial_analysis,
)
pending = PendingJob(
@@ -1296,6 +1431,7 @@ def enqueue_job() -> Response:
speakers=speakers,
speaker_analysis=analysis_payload,
speaker_analysis_threshold=analysis_threshold,
analysis_requested=False,
)
service.store_pending_job(pending)
@@ -1311,142 +1447,44 @@ def prepare_job(pending_id: str) -> str:
return _render_prepare_page(pending)
@web_bp.post("/jobs/prepare/<pending_id>")
def finalize_job(pending_id: str) -> Response:
@web_bp.post("/jobs/prepare/<pending_id>/analyze")
def analyze_pending_job(pending_id: str) -> ResponseReturnValue:
service = _service()
pending = service.get_pending_job(pending_id)
if not pending:
abort(404)
pending = cast(PendingJob, pending)
raw_chunk_level = (request.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)
chunk_level_literal, overrides, enabled_overrides, errors, selected_total = _apply_prepare_form(
pending, request.form
)
raw_speaker_mode = (request.form.get("speaker_mode") or pending.speaker_mode or "single").strip().lower()
if raw_speaker_mode not in _SPEAKER_MODE_VALUES:
raw_speaker_mode = "single"
pending.speaker_mode = raw_speaker_mode
if errors:
return _render_prepare_page(pending, error=" ".join(errors))
pending.generate_epub3 = _coerce_bool(request.form.get("generate_epub3"), False)
threshold_default = getattr(pending, "speaker_analysis_threshold", _DEFAULT_ANALYSIS_THRESHOLD)
raw_threshold = request.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
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 = request.form.get(field_key, "")
pronunciation = raw_value.strip()
if pronunciation:
payload["pronunciation"] = pronunciation
else:
payload.pop("pronunciation", None)
profiles = serialize_profiles()
delay_value = pending.chapter_intro_delay
raw_delay = request.form.get("chapter_intro_delay")
if raw_delay is not None:
raw_normalized = raw_delay.strip()
if raw_normalized:
try:
delay_value = max(0.0, float(raw_normalized))
except ValueError:
return _render_prepare_page(pending, error="Enter a valid number for the chapter intro delay.")
else:
delay_value = 0.0
pending.chapter_intro_delay = delay_value
overrides: List[Dict[str, Any]] = []
selected_total = 0
errors: List[str] = []
for index, chapter in enumerate(pending.chapters):
enabled = request.form.get(f"chapter-{index}-enabled") == "on"
title_input = (request.form.get(f"chapter-{index}-title") or "").strip()
title = title_input or chapter.get("title") or f"Chapter {index + 1}"
voice_selection = request.form.get(f"chapter-{index}-voice", "__default")
formula_input = (request.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"] = len(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 += len(entry["text"] or "")
overrides.append(entry)
pending.chapters[index] = dict(entry)
enabled_overrides = [entry for entry in overrides if entry.get("enabled")]
if not enabled_overrides:
if pending.speaker_mode != "multi":
setattr(pending, "analysis_requested", False)
pending.chunks = []
return _render_prepare_page(pending, error="Select at least one chapter to convert.")
pending.speaker_analysis = {}
return _render_prepare_page(
pending,
error="Switch to multi-speaker mode to analyze speakers.",
)
if not enabled_overrides:
setattr(pending, "analysis_requested", False)
pending.chunks = []
pending.speaker_analysis = {}
return _render_prepare_page(pending, error="Select at least one chapter to analyze.")
raw_chunks = build_chunks_for_chapters(enabled_overrides, level=chunk_level_literal)
existing_roster: Optional[Mapping[str, Any]]
if getattr(pending, "analysis_requested", False):
existing_roster = pending.speakers
else:
existing_roster = None
processed_chunks, roster, analysis_payload = _prepare_speaker_metadata(
chapters=enabled_overrides,
chunks=raw_chunks,
@@ -1454,15 +1492,69 @@ def finalize_job(pending_id: str) -> Response:
voice=pending.voice,
voice_profile=pending.voice_profile,
threshold=pending.speaker_analysis_threshold,
existing_roster=pending.speakers,
existing_roster=existing_roster,
run_analysis=True,
)
pending.chunks = processed_chunks
pending.speakers = roster
pending.speaker_analysis = analysis_payload
setattr(pending, "analysis_requested", True)
if selected_total:
pending.total_characters = selected_total
service.store_pending_job(pending)
return _render_prepare_page(pending, notice="Speaker analysis updated.")
@web_bp.post("/jobs/prepare/<pending_id>")
def finalize_job(pending_id: str) -> ResponseReturnValue:
service = _service()
pending = service.get_pending_job(pending_id)
if not pending:
abort(404)
pending = cast(PendingJob, pending)
chunk_level_literal, overrides, enabled_overrides, errors, selected_total = _apply_prepare_form(
pending, request.form
)
if errors:
return _render_prepare_page(pending, error=" ".join(errors))
if pending.speaker_mode != "multi":
setattr(pending, "analysis_requested", False)
if not enabled_overrides:
pending.chunks = []
return _render_prepare_page(pending, error="Select at least one chapter to convert.")
raw_chunks = build_chunks_for_chapters(enabled_overrides, level=chunk_level_literal)
analysis_active = pending.speaker_mode == "multi" and getattr(pending, "analysis_requested", False)
if analysis_active:
existing_roster: Optional[Mapping[str, Any]] = pending.speakers
else:
narrator_only: Dict[str, Any] = {}
if isinstance(pending.speakers, dict):
narrator_payload = pending.speakers.get("narrator")
if isinstance(narrator_payload, Mapping):
narrator_only["narrator"] = dict(narrator_payload)
existing_roster = narrator_only or None
processed_chunks, roster, analysis_payload = _prepare_speaker_metadata(
chapters=enabled_overrides,
chunks=raw_chunks,
speaker_mode=pending.speaker_mode,
voice=pending.voice,
voice_profile=pending.voice_profile,
threshold=pending.speaker_analysis_threshold,
existing_roster=existing_roster,
run_analysis=analysis_active,
)
pending.chunks = processed_chunks
pending.speakers = roster
pending.speaker_analysis = analysis_payload
if errors:
return _render_prepare_page(pending, error=" ".join(errors))
total_characters = selected_total or pending.total_characters
service.pop_pending_job(pending_id)
@@ -1500,13 +1592,14 @@ def finalize_job(pending_id: str) -> Response:
speaker_analysis=analysis_payload,
speaker_analysis_threshold=pending.speaker_analysis_threshold,
generate_epub3=pending.generate_epub3,
analysis_requested=getattr(pending, "analysis_requested", False),
)
return redirect(url_for("web.queue_page"))
@web_bp.post("/jobs/prepare/<pending_id>/cancel")
def cancel_pending_job(pending_id: str) -> Response:
def cancel_pending_job(pending_id: str) -> ResponseReturnValue:
pending = _service().pop_pending_job(pending_id)
if pending and pending.stored_path.exists():
try:
@@ -1536,13 +1629,19 @@ def _render_jobs_panel() -> str:
)
def _render_prepare_page(pending: PendingJob, *, error: Optional[str] = None) -> str:
def _render_prepare_page(
pending: PendingJob,
*,
error: Optional[str] = None,
notice: Optional[str] = None,
) -> str:
return render_template(
"prepare_job.html",
pending=pending,
options=_template_options(),
settings=_load_settings(),
error=error,
notice=notice,
)
@@ -1559,7 +1658,7 @@ def job_detail(job_id: str) -> str:
@web_bp.post("/jobs/<job_id>/pause")
def pause_job(job_id: str) -> Response:
def pause_job(job_id: str) -> ResponseReturnValue:
_service().pause(job_id)
if request.headers.get("HX-Request"):
return _render_jobs_panel()
@@ -1567,7 +1666,7 @@ def pause_job(job_id: str) -> Response:
@web_bp.post("/jobs/<job_id>/resume")
def resume_job(job_id: str) -> Response:
def resume_job(job_id: str) -> ResponseReturnValue:
_service().resume(job_id)
if request.headers.get("HX-Request"):
return _render_jobs_panel()
@@ -1575,7 +1674,7 @@ def resume_job(job_id: str) -> Response:
@web_bp.post("/jobs/<job_id>/cancel")
def cancel_job(job_id: str) -> Response:
def cancel_job(job_id: str) -> ResponseReturnValue:
_service().cancel(job_id)
if request.headers.get("HX-Request"):
return _render_jobs_panel()
@@ -1583,7 +1682,7 @@ def cancel_job(job_id: str) -> Response:
@web_bp.post("/jobs/<job_id>/delete")
def delete_job(job_id: str) -> Response:
def delete_job(job_id: str) -> ResponseReturnValue:
_service().delete(job_id)
if request.headers.get("HX-Request"):
return _render_jobs_panel()
@@ -1591,7 +1690,7 @@ def delete_job(job_id: str) -> Response:
@web_bp.post("/jobs/clear-finished")
def clear_finished_jobs() -> Response:
def clear_finished_jobs() -> ResponseReturnValue:
_service().clear_finished()
if request.headers.get("HX-Request"):
return _render_jobs_panel()
@@ -1599,7 +1698,7 @@ def clear_finished_jobs() -> Response:
@web_bp.get("/jobs/<job_id>/download")
def download_job(job_id: str) -> Response:
def download_job(job_id: str) -> ResponseReturnValue:
job = _service().get_job(job_id)
if job is None or job.status != JobStatus.COMPLETED:
abort(404)
@@ -1634,7 +1733,7 @@ def job_logs_partial(job_id: str) -> str:
@api_bp.get("/jobs/<job_id>")
def job_json(job_id: str) -> Response:
def job_json(job_id: str) -> ResponseReturnValue:
job = _service().get_job(job_id)
if job is None:
abort(404)
abogen/web/service.py
+7
View File
@@ -97,6 +97,7 @@ class Job:
generate_epub3: bool = False
speaker_analysis: Dict[str, Any] = field(default_factory=dict)
speaker_analysis_threshold: int = 3
analysis_requested: bool = False
def add_log(self, message: str, level: str = "info") -> None:
self.logs.append(JobLog(timestamp=time.time(), message=message, level=level))
@@ -154,6 +155,7 @@ class Job:
"generate_epub3": self.generate_epub3,
"speaker_analysis": dict(self.speaker_analysis),
"speaker_analysis_threshold": self.speaker_analysis_threshold,
"analysis_requested": self.analysis_requested,
}
@@ -193,6 +195,7 @@ class PendingJob:
generate_epub3: bool = False
speaker_analysis: Dict[str, Any] = field(default_factory=dict)
speaker_analysis_threshold: int = 3
analysis_requested: bool = False
class ConversionService:
@@ -263,6 +266,7 @@ class ConversionService:
generate_epub3: bool = False,
speaker_analysis: Optional[Mapping[str, Any]] = None,
speaker_analysis_threshold: int = 3,
analysis_requested: bool = False,
) -> Job:
job_id = uuid.uuid4().hex
normalized_metadata = self._normalize_metadata_tags(metadata_tags)
@@ -305,6 +309,7 @@ class ConversionService:
generate_epub3=bool(generate_epub3),
speaker_analysis=dict(speaker_analysis or {}),
speaker_analysis_threshold=int(speaker_analysis_threshold or 3),
analysis_requested=bool(analysis_requested),
)
with self._lock:
self._jobs[job_id] = job
@@ -581,6 +586,7 @@ class ConversionService:
"generate_epub3": job.generate_epub3,
"speaker_analysis": dict(job.speaker_analysis),
"speaker_analysis_threshold": job.speaker_analysis_threshold,
"analysis_requested": job.analysis_requested,
}
def _persist_state(self) -> None:
@@ -697,6 +703,7 @@ class ConversionService:
job.speaker_analysis_threshold = int(
payload.get("speaker_analysis_threshold", job.speaker_analysis_threshold or 3)
)
job.analysis_requested = bool(payload.get("analysis_requested", job.analysis_requested))
job.pause_event.set()
return job
abogen/web/static/prepare.js
+15
View File
@@ -61,4 +61,19 @@ document.addEventListener("DOMContentLoaded", () => {
toggleFormula(select);
}
});
const analyzeButton = form.querySelector('[data-role="analyze-button"]');
const speakerModeSelect = form.querySelector("#speaker_mode");
const updateAnalyzeVisibility = () => {
if (!analyzeButton || !speakerModeSelect) return;
const isMulti = speakerModeSelect.value === "multi";
analyzeButton.hidden = !isMulti;
analyzeButton.setAttribute("aria-hidden", isMulti ? "false" : "true");
analyzeButton.disabled = !isMulti;
};
if (analyzeButton && speakerModeSelect) {
speakerModeSelect.addEventListener("change", updateAnalyzeVisibility);
updateAnalyzeVisibility();
}
});
abogen/web/templates/prepare_job.html
+17
View File
@@ -71,6 +71,11 @@
<p>No additional speakers met the threshold yet. All dialogue will use the narrator voice.</p>
{% endif %}
</div>
{% elif pending.speaker_mode == 'multi' and not pending.analysis_requested %}
<div class="prepare-analysis">
<h2>Detected speakers</h2>
<p>Press <strong>Analyze speakers</strong> after selecting chapters to discover recurring voices.</p>
</div>
{% endif %}
{% set roster = pending.speakers or {} %}
{% if roster %}
@@ -130,6 +135,9 @@
{% if error %}
<div class="alert alert--error">{{ error }}</div>
{% endif %}
{% if notice %}
<div class="alert alert--info">{{ notice }}</div>
{% endif %}
<form method="post" action="{{ url_for('web.finalize_job', pending_id=pending.id) }}" class="prepare-form" id="prepare-form">
<div class="chapter-grid">
@@ -226,6 +234,15 @@
</div>
</div>
<div class="prepare-actions">
<button type="submit"
class="button button--ghost"
data-role="analyze-button"
formaction="{{ url_for('web.analyze_pending_job', pending_id=pending.id) }}"
formmethod="post"
formnovalidate
{% if pending.speaker_mode != 'multi' %}hidden aria-hidden="true"{% else %}aria-hidden="false"{% endif %}>
Analyze speakers
</button>
<button type="submit" class="button">Queue conversion</button>
<button type="submit" class="button button--ghost" form="cancel-form">Cancel</button>
</div>