diff --git a/CHANGELOG.md b/CHANGELOG.md index ba5d726..a228d79 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,7 @@ +# Unreleased +- Added an EPUB 3 packaging pipeline that builds media-overlay EPUBs from generated audio and chunk metadata. +- Persisted chunk timing metadata in job artifacts and exercised the exporter with automated tests. + # 1.1.9 - Fixed the issue where spaces were deleted before punctuation marks while generating subtitles. - Fixed markdown TOC generation breaks when "Replace single newlines" is enabled. diff --git a/abogen/chunking.py b/abogen/chunking.py new file mode 100644 index 0000000..0b28280 --- /dev/null +++ b/abogen/chunking.py @@ -0,0 +1,166 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import Dict, Iterable, Iterator, List, Literal, Optional + +import re + +ChunkLevel = Literal["paragraph", "sentence"] + +_SENTENCE_SPLIT_REGEX = re.compile(r"(? Dict[str, object]: + return { + "id": self.id, + "chapter_index": self.chapter_index, + "chunk_index": self.chunk_index, + "level": self.level, + "text": self.text, + "speaker_id": self.speaker_id, + "voice": self.voice, + "voice_profile": self.voice_profile, + "voice_formula": self.voice_formula, + } + + +def _iter_paragraphs(text: str) -> Iterator[str]: + for raw_segment in _PARAGRAPH_SPLIT_REGEX.split(text.strip()): + normalized = raw_segment.strip() + if normalized: + yield normalized + + +def _iter_sentences(paragraph: str) -> Iterator[str]: + if not paragraph: + return + start = 0 + for match in _SENTENCE_SPLIT_REGEX.finditer(paragraph): + end = match.end() + candidate = paragraph[start:end].strip() + if candidate: + yield candidate + start = match.end() + tail = paragraph[start:].strip() + if tail: + yield tail + + +def _normalize_whitespace(value: str) -> str: + return _WHITESPACE_REGEX.sub(" ", value).strip() + + +def chunk_text( + *, + chapter_index: int, + chapter_title: str, + text: str, + level: ChunkLevel, + speaker_id: str = "narrator", + voice: Optional[str] = None, + voice_profile: Optional[str] = None, + voice_formula: Optional[str] = None, + chunk_prefix: Optional[str] = None, +) -> List[Dict[str, object]]: + """Split text into ordered chunk dictionaries.""" + + prefix = chunk_prefix or f"chap{chapter_index:04d}" + chunks: List[Dict[str, object]] = [] + + if level == "paragraph": + paragraphs = list(_iter_paragraphs(text)) or [text.strip()] + for para_index, paragraph in enumerate(paragraphs): + normalized = _normalize_whitespace(paragraph) + if not normalized: + continue + chunk_id = f"{prefix}_p{para_index:04d}" + chunks.append( + Chunk( + id=chunk_id, + chapter_index=chapter_index, + chunk_index=len(chunks), + level=level, + text=normalized, + speaker_id=speaker_id, + voice=voice, + voice_profile=voice_profile, + voice_formula=voice_formula, + ).as_dict() + ) + return chunks + + # Sentence level – flatten paragraphs into individual sentences + sentence_index = 0 + for para_index, paragraph in enumerate(list(_iter_paragraphs(text)) or [text.strip()]): + normalized_para = _normalize_whitespace(paragraph) + if not normalized_para: + continue + sentences = list(_iter_sentences(normalized_para)) or [normalized_para] + for sent_local_index, sentence in enumerate(sentences): + normalized_sentence = _normalize_whitespace(sentence) + if not normalized_sentence: + continue + chunk_id = f"{prefix}_p{para_index:04d}_s{sent_local_index:04d}" + chunks.append( + Chunk( + id=chunk_id, + chapter_index=chapter_index, + chunk_index=sentence_index, + level=level, + text=normalized_sentence, + speaker_id=speaker_id, + voice=voice, + voice_profile=voice_profile, + voice_formula=voice_formula, + ).as_dict() + ) + sentence_index += 1 + + return chunks + + +def build_chunks_for_chapters( + chapters: Iterable[Dict[str, object]], + *, + level: ChunkLevel, + speaker_id: str = "narrator", +) -> List[Dict[str, object]]: + """Generate chunk dictionaries for a sequence of chapter payloads.""" + all_chunks: List[Dict[str, object]] = [] + for chapter_index, entry in enumerate(chapters): + if not isinstance(entry, dict): # defensive + continue + text = str(entry.get("text", "") or "").strip() + if not text: + continue + voice = entry.get("voice") + voice_profile = entry.get("voice_profile") + voice_formula = entry.get("voice_formula") + prefix = entry.get("id") or f"chap{chapter_index:04d}" + chapter_chunks = chunk_text( + chapter_index=chapter_index, + chapter_title=str(entry.get("title") or f"Chapter {chapter_index + 1}"), + text=text, + level=level, + speaker_id=speaker_id, + voice=str(voice) if voice else None, + voice_profile=str(voice_profile) if voice_profile else None, + voice_formula=str(voice_formula) if voice_formula else None, + chunk_prefix=str(prefix), + ) + all_chunks.extend(chapter_chunks) + return all_chunks \ No newline at end of file diff --git a/abogen/epub3/__init__.py b/abogen/epub3/__init__.py new file mode 100644 index 0000000..7683764 --- /dev/null +++ b/abogen/epub3/__init__.py @@ -0,0 +1,3 @@ +from .exporter import EPUB3PackageBuilder, build_epub3_package + +__all__ = ["EPUB3PackageBuilder", "build_epub3_package"] diff --git a/abogen/epub3/exporter.py b/abogen/epub3/exporter.py new file mode 100644 index 0000000..5981ca3 --- /dev/null +++ b/abogen/epub3/exporter.py @@ -0,0 +1,778 @@ +from __future__ import annotations + +import html +import shutil +import uuid +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from tempfile import TemporaryDirectory +from typing import Any, Dict, Iterable, List, Optional, Sequence +import zipfile + +from abogen.text_extractor import ExtractedChapter, ExtractionResult + + +@dataclass(slots=True) +class ChunkOverlay: + id: str + text: str + start: Optional[float] + end: Optional[float] + speaker_id: str + voice: Optional[str] + + +@dataclass(slots=True) +class ChapterDocument: + index: int # zero-based + title: str + xhtml_name: str + smil_name: str + chunks: List[ChunkOverlay] + start: Optional[float] + end: Optional[float] + + +class EPUB3PackageBuilder: + """Constructs an EPUB 3 package with media overlays.""" + + def __init__( + self, + *, + output_path: Path, + book_id: str, + extraction: ExtractionResult, + metadata_tags: Dict[str, Any], + chapter_markers: Sequence[Dict[str, Any]], + chunk_markers: Sequence[Dict[str, Any]], + chunks: Iterable[Dict[str, Any]], + audio_path: Path, + speaker_mode: str = "single", + cover_image_path: Optional[Path] = None, + cover_image_mime: Optional[str] = None, + ) -> None: + self.output_path = output_path + self.book_id = book_id or str(uuid.uuid4()) + self.extraction = extraction + self.metadata_tags = _normalize_metadata(metadata_tags) + self.chapter_markers = list(chapter_markers or []) + self.chunk_markers = list(chunk_markers or []) + self.chunks = list(chunks or []) + self.audio_path = audio_path + self.speaker_mode = speaker_mode or "single" + self.cover_image_path = cover_image_path if cover_image_path and cover_image_path.exists() else None + self.cover_image_mime = cover_image_mime + + self._combined_metadata = _combine_metadata(extraction.metadata, self.metadata_tags) + self._title = self._combined_metadata.get("title") or self._fallback_title() + self._authors = _split_authors(self._combined_metadata) + self._language = self._determine_language() + self._publisher = self._combined_metadata.get("publisher") or "" + self._description = self._combined_metadata.get("comment") + self._duration = _calculate_total_duration(self.chunk_markers, self.chapter_markers) + self._modified = _utc_now_iso() + + def build(self) -> Path: + if not self.audio_path or not self.audio_path.exists(): + raise FileNotFoundError(f"Audio asset missing: {self.audio_path}") + + chapter_documents = self._build_chapter_documents() + + with TemporaryDirectory() as tmp_dir: + root = Path(tmp_dir) + oebps = root / "OEBPS" + text_dir = oebps / "text" + smil_dir = oebps / "smil" + audio_dir = oebps / "audio" + image_dir = oebps / "images" + stylesheet_dir = oebps / "styles" + + for directory in (oebps, text_dir, smil_dir, audio_dir, stylesheet_dir): + directory.mkdir(parents=True, exist_ok=True) + if self.cover_image_path: + image_dir.mkdir(parents=True, exist_ok=True) + + _write_mimetype(root) + _write_container_xml(root) + + audio_filename = self.audio_path.name + embedded_audio = audio_dir / audio_filename + shutil.copy2(self.audio_path, embedded_audio) + + if self.cover_image_path: + shutil.copy2(self.cover_image_path, image_dir / self.cover_image_path.name) + + stylesheet_path = stylesheet_dir / "style.css" + stylesheet_path.write_text(_DEFAULT_STYLESHEET, encoding="utf-8") + + for chapter in chapter_documents: + chapter_path = text_dir / chapter.xhtml_name + chapter_path.write_text( + self._render_chapter_xhtml(chapter), + encoding="utf-8", + ) + smil_path = smil_dir / chapter.smil_name + smil_path.write_text( + self._render_chapter_smil(chapter, f"audio/{audio_filename}"), + encoding="utf-8", + ) + + nav_path = oebps / "nav.xhtml" + nav_path.write_text(self._render_nav(chapter_documents), encoding="utf-8") + + opf_path = oebps / "content.opf" + opf_path.write_text( + self._render_opf( + chapter_documents, + audio_filename, + has_cover=self.cover_image_path is not None, + stylesheet_path=stylesheet_path.relative_to(oebps), + ), + encoding="utf-8", + ) + + self.output_path.parent.mkdir(parents=True, exist_ok=True) + with zipfile.ZipFile(self.output_path, "w", compression=zipfile.ZIP_DEFLATED) as archive: + # Ensure mimetype is the first entry and stored without compression + mimetype_path = root / "mimetype" + info = zipfile.ZipInfo("mimetype") + info.compress_type = zipfile.ZIP_STORED + archive.writestr(info, mimetype_path.read_bytes()) + + for file_path in sorted(root.rglob("*")): + if file_path == mimetype_path or file_path.is_dir(): + continue + archive.write(file_path, file_path.relative_to(root)) + + return self.output_path + + # ------------------------------------------------------------------ + def _build_chapter_documents(self) -> List[ChapterDocument]: + chunk_lookup = _build_chunk_lookup(self.chunks) + markers_by_chapter = _group_markers_by_chapter(self.chunk_markers) + chapter_meta = {int(entry.get("index", idx + 1)) - 1: dict(entry) for idx, entry in enumerate(self.chapter_markers)} + + documents: List[ChapterDocument] = [] + for chapter_index, chapter in enumerate(self.extraction.chapters): + markers = markers_by_chapter.get(chapter_index, []) + if not markers and chunk_lookup.by_chapter.get(chapter_index): + markers = [ + { + "id": item.get("id"), + "chapter_index": chapter_index, + "chunk_index": item.get("chunk_index"), + "start": None, + "end": None, + "speaker_id": item.get("speaker_id", "narrator"), + "voice": item.get("voice"), + } + for item in chunk_lookup.by_chapter.get(chapter_index, []) + ] + + if not markers: + markers = [ + { + "id": f"chap{chapter_index:04d}_auto0000", + "chapter_index": chapter_index, + "chunk_index": 0, + "start": chapter_meta.get(chapter_index, {}).get("start"), + "end": chapter_meta.get(chapter_index, {}).get("end"), + "speaker_id": "narrator", + "voice": None, + } + ] + + overlays = self._build_overlays_for_chapter( + chapter_index, + markers, + chunk_lookup, + ) + + xhtml_name = f"chapter_{chapter_index + 1:04d}.xhtml" + smil_name = f"chapter_{chapter_index + 1:04d}.smil" + + chapter_start = chapter_meta.get(chapter_index, {}).get("start") + chapter_end = chapter_meta.get(chapter_index, {}).get("end") + + documents.append( + ChapterDocument( + index=chapter_index, + title=chapter.title or f"Chapter {chapter_index + 1}", + xhtml_name=xhtml_name, + smil_name=smil_name, + chunks=overlays, + start=chapter_start, + end=chapter_end, + ) + ) + + return documents + + def _build_overlays_for_chapter( + self, + chapter_index: int, + markers: Sequence[Dict[str, Any]], + chunk_lookup: "ChunkLookup", + ) -> List[ChunkOverlay]: + overlays: List[ChunkOverlay] = [] + used_ids: set[str] = set() + + chapter_chunks = list(chunk_lookup.by_chapter.get(chapter_index, [])) + chapter_chunks.sort(key=lambda entry: _safe_int(entry.get("chunk_index"))) + + for position, marker in enumerate(markers): + chunk_id = marker.get("id") + chunk_entry = None + if chunk_id and chunk_id in chunk_lookup.by_id: + chunk_entry = chunk_lookup.by_id[chunk_id] + else: + candidate_index = _safe_int(marker.get("chunk_index")) + chunk_entry = _find_chunk_by_index(chapter_chunks, candidate_index) + if chunk_entry is None and chapter_chunks and position < len(chapter_chunks): + chunk_entry = chapter_chunks[position] + + if chunk_entry is None: + text = self.extraction.chapters[chapter_index].text + speaker_id = str(marker.get("speaker_id") or "narrator") + voice = marker.get("voice") + else: + text = str(chunk_entry.get("text") or "") + speaker_id = str(chunk_entry.get("speaker_id") or marker.get("speaker_id") or "narrator") + voice = chunk_entry.get("voice") or chunk_entry.get("resolved_voice") or marker.get("voice") + + normalized_id = _normalize_chunk_id(chunk_id) if chunk_id else None + if not normalized_id: + normalized_id = f"chap{chapter_index:04d}_chunk{position:04d}" + while normalized_id in used_ids: + normalized_id = f"{normalized_id}_dup" + used_ids.add(normalized_id) + + overlays.append( + ChunkOverlay( + id=normalized_id, + text=text or self.extraction.chapters[chapter_index].text, + start=_safe_float(marker.get("start")), + end=_safe_float(marker.get("end")), + speaker_id=speaker_id, + voice=str(voice) if voice else None, + ) + ) + + return overlays + + def _render_chapter_xhtml(self, chapter: ChapterDocument) -> str: + language = html.escape(self._language or "en") + title = html.escape(chapter.title) + chunk_html = "\n".join(_render_chunk_html(chunk) for chunk in chapter.chunks) + if not chunk_html: + chunk_html = "

" + + return ( + "\n" + "\n" + " \n" + " {title}\n" + " \n" + " \n" + " \n" + " \n" + "
\n" + "

{title}

\n" + " {chunks}\n" + "
\n" + " \n" + "\n" + ).format(lang=language, title=title, index=chapter.index + 1, chunks=chunk_html) + + def _render_chapter_smil(self, chapter: ChapterDocument, audio_href: str) -> str: + par_lines = [] + for chunk in chapter.chunks: + par_lines.append( + " \n" + " \n" + " ".format( + chunk_id=html.escape(chunk.id), + xhtml=html.escape(chapter.xhtml_name), + audio=html.escape(audio_href), + start=_format_smil_time(chunk.start), + end=_format_smil_time(chunk.end), + ) + ) + + return ( + "\n" + "\n" + " \n" + " \n" + " \n" + " \n" + " \n" + " \n" + " \n" + "{pars}\n" + " \n" + " \n" + "\n" + ).format( + title=html.escape(chapter.title), + book_id=html.escape(self.book_id), + index=chapter.index + 1, + xhtml=html.escape(chapter.xhtml_name), + pars="\n".join(par_lines) if par_lines else " ", + ) + + def _render_nav(self, chapters: Sequence[ChapterDocument]) -> str: + items = [] + for chapter in chapters: + href = f"text/{chapter.xhtml_name}" + items.append( + "
  • {title}
  • ".format( + href=html.escape(href), + title=html.escape(chapter.title), + ) + ) + + return ( + "\n" + "\n" + " \n" + " Navigation\n" + " \n" + " \n" + " \n" + " \n" + " \n" + "\n" + ).format( + lang=html.escape(self._language or "en"), + title=html.escape(self._title), + items="\n".join(items) if items else "
  • Chapter 1
  • ", + ) + + def _render_opf( + self, + chapters: Sequence[ChapterDocument], + audio_filename: str, + *, + has_cover: bool, + stylesheet_path: Path, + ) -> str: + manifest_items = [] + spine_refs = [] + for chapter in chapters: + item_id = f"chap{chapter.index + 1:04d}" + overlay_id = f"mo-{chapter.index + 1:04d}" + manifest_items.append( + " ".format( + item_id=item_id, + href=html.escape(chapter.xhtml_name), + overlay_id=overlay_id, + ) + ) + manifest_items.append( + " ".format( + overlay_id=overlay_id, + smil=html.escape(chapter.smil_name), + ) + ) + spine_refs.append(f" ") + + audio_item_id = "primary-audio" + manifest_items.append( + " ".format( + item_id=audio_item_id, + href=html.escape(audio_filename), + mime=_detect_audio_mime(audio_filename), + ) + ) + + manifest_items.append( + " " + ) + + manifest_items.append( + " ".format( + href=html.escape(str(stylesheet_path).replace("\\", "/")), + ) + ) + + if has_cover and self.cover_image_path: + cover_id = "cover-image" + manifest_items.append( + " ".format( + item_id=cover_id, + href=html.escape(self.cover_image_path.name), + mime=self.cover_image_mime or _detect_image_mime(self.cover_image_path.suffix), + ) + ) + + metadata_elements = _render_metadata_xml( + self._title, + self._authors, + self._language, + self.book_id, + duration=self._duration, + publisher=self._publisher, + description=self._description, + speaker_mode=self.speaker_mode, + modified=self._modified, + ) + + return ( + "\n" + "\n" + " \n" + "{metadata}\n" + " \n" + " \n" + "{manifest}\n" + " \n" + " \n" + "{spine}\n" + " \n" + "\n" + ).format( + metadata="\n".join(metadata_elements), + manifest="\n".join(manifest_items), + spine="\n".join(spine_refs) if spine_refs else " ", + ) + + def _fallback_title(self) -> str: + if self.extraction.chapters: + first_title = self.extraction.chapters[0].title + if first_title: + return first_title + return "Generated Audiobook" + + def _determine_language(self) -> str: + language = self._combined_metadata.get("language") + if language: + return language + return "en" + + +def build_epub3_package( + *, + output_path: Path, + book_id: str, + extraction: ExtractionResult, + metadata_tags: Dict[str, Any], + chapter_markers: Sequence[Dict[str, Any]], + chunk_markers: Sequence[Dict[str, Any]], + chunks: Iterable[Dict[str, Any]], + audio_path: Path, + speaker_mode: str = "single", + cover_image_path: Optional[Path] = None, + cover_image_mime: Optional[str] = None, +) -> Path: + builder = EPUB3PackageBuilder( + output_path=output_path, + book_id=book_id, + extraction=extraction, + metadata_tags=metadata_tags, + chapter_markers=chapter_markers, + chunk_markers=chunk_markers, + chunks=chunks, + audio_path=audio_path, + speaker_mode=speaker_mode, + cover_image_path=cover_image_path, + cover_image_mime=cover_image_mime, + ) + return builder.build() + + +# --------------------------------------------------------------------------- +# Helpers + + +@dataclass +class ChunkLookup: + by_id: Dict[str, Dict[str, Any]] + by_chapter: Dict[int, List[Dict[str, Any]]] + + +def _normalize_metadata(metadata: Optional[Dict[str, Any]]) -> Dict[str, str]: + normalized: Dict[str, str] = {} + for key, value in (metadata or {}).items(): + if value is None: + continue + normalized[str(key).lower()] = str(value) + return normalized + + +def _combine_metadata(*sources: Dict[str, Any]) -> Dict[str, str]: + combined: Dict[str, str] = {} + for source in sources: + for key, value in (source or {}).items(): + if value is None: + continue + combined[str(key).lower()] = str(value) + return combined + + +def _split_authors(metadata: Dict[str, str]) -> List[str]: + candidates = [] + for key in ("artist", "author", "authors", "album_artist", "creator"): + value = metadata.get(key) + if value: + candidates.extend(part.strip() for part in value.replace(";", ",").split(",")) + return [author for author in candidates if author] + + +def _calculate_total_duration( + chunk_markers: Sequence[Dict[str, Any]], + chapter_markers: Sequence[Dict[str, Any]], +) -> Optional[float]: + candidates: List[float] = [] + for marker in chunk_markers or []: + end_value = _safe_float(marker.get("end")) + if end_value is not None: + candidates.append(end_value) + for marker in chapter_markers or []: + end_value = _safe_float(marker.get("end")) + if end_value is not None: + candidates.append(end_value) + if not candidates: + return None + return max(candidates) + + +def _write_mimetype(root: Path) -> None: + (root / "mimetype").write_text("application/epub+zip", encoding="utf-8") + + +def _write_container_xml(root: Path) -> None: + meta_inf = root / "META-INF" + meta_inf.mkdir(parents=True, exist_ok=True) + container = meta_inf / "container.xml" + container.write_text( + ( + "\n" + "\n" + " \n" + " \n" + " \n" + "\n" + ), + encoding="utf-8", + ) + + +def _build_chunk_lookup(chunks: Iterable[Dict[str, Any]]) -> ChunkLookup: + by_id: Dict[str, Dict[str, Any]] = {} + by_chapter: Dict[int, List[Dict[str, Any]]] = {} + for entry in chunks or []: + if not isinstance(entry, dict): + continue + chunk_id = entry.get("id") + if chunk_id: + by_id[str(chunk_id)] = dict(entry) + chapter_index = _safe_int(entry.get("chapter_index")) + by_chapter.setdefault(chapter_index, []).append(dict(entry)) + return ChunkLookup(by_id=by_id, by_chapter=by_chapter) + + +def _group_markers_by_chapter(markers: Iterable[Dict[str, Any]]) -> Dict[int, List[Dict[str, Any]]]: + grouped: Dict[int, List[Dict[str, Any]]] = {} + for entry in markers or []: + if not isinstance(entry, dict): + continue + chapter_index = _safe_int(entry.get("chapter_index")) + grouped.setdefault(chapter_index, []).append(dict(entry)) + for chapter_index, items in grouped.items(): + items.sort(key=lambda payload: (_safe_int(payload.get("chunk_index")), _safe_float(payload.get("start")) or 0.0)) + return grouped + + +def _find_chunk_by_index( + chapter_chunks: Sequence[Dict[str, Any]], + chunk_index: Optional[int], +) -> Optional[Dict[str, Any]]: + if chunk_index is None: + return None + for entry in chapter_chunks: + if _safe_int(entry.get("chunk_index")) == chunk_index: + return entry + return None + + +def _normalize_chunk_id(chunk_id: Optional[Any]) -> Optional[str]: + if chunk_id is None: + return None + text = str(chunk_id).strip() + if not text: + return None + safe = "".join(ch if ch.isalnum() or ch in {"_", "-"} else "_" for ch in text) + return safe[:120] + + +def _render_chunk_html(chunk: ChunkOverlay) -> str: + escaped_id = html.escape(chunk.id) + speaker_attr = f" data-speaker=\"{html.escape(chunk.speaker_id)}\"" if chunk.speaker_id else "" + voice_attr = f" data-voice=\"{html.escape(chunk.voice)}\"" if chunk.voice else "" + paragraphs = _split_paragraphs(chunk.text) + if not paragraphs: + paragraphs = [" "] + return "
    \n{body}\n
    ".format( + id=escaped_id, + speaker=speaker_attr, + voice=voice_attr, + body="\n".join(f"

    {para}

    " for para in paragraphs), + ) + + +def _split_paragraphs(text: str) -> List[str]: + if not text: + return [] + segments = [segment.strip() for segment in text.replace("\r", "").split("\n\n")] + paragraphs: List[str] = [] + for segment in segments: + if not segment: + continue + lines = [html.escape(line.strip()) for line in segment.split("\n") if line.strip()] + if not lines: + continue + if len(lines) == 1: + paragraphs.append(lines[0]) + else: + paragraphs.append("
    ".join(lines)) + return paragraphs + + +def _format_smil_time(value: Optional[float]) -> str: + if value is None or value < 0: + value = 0.0 + total_ms = int(round(value * 1000)) + hours, remainder = divmod(total_ms, 3600_000) + minutes, remainder = divmod(remainder, 60_000) + seconds, milliseconds = divmod(remainder, 1000) + return f"{hours:02d}:{minutes:02d}:{seconds:02d}.{milliseconds:03d}" + + +def _safe_int(value: Any, default: int = 0) -> int: + try: + return int(value) + except (TypeError, ValueError): + return default + + +def _safe_float(value: Any) -> Optional[float]: + if value is None: + return None + try: + return float(value) + except (TypeError, ValueError): + return None + + +def _render_metadata_xml( + title: str, + authors: Sequence[str], + language: str, + book_id: str, + *, + duration: Optional[float], + publisher: Optional[str], + description: Optional[str], + speaker_mode: Optional[str], + modified: Optional[str], +) -> List[str]: + elements = [ + f" {html.escape(book_id)}", + f" {html.escape(title)}", + f" {html.escape(language or 'en')}", + ] + + for author in authors or ["Unknown"]: + elements.append(f" {html.escape(author)}") + + if publisher: + elements.append(f" {html.escape(publisher)}") + + if description: + elements.append(f" {html.escape(description)}") + + if duration is not None: + elements.append(f" {_format_iso_duration(duration)}") + + if speaker_mode: + elements.append( + " {}".format( + html.escape(str(speaker_mode)) + ) + ) + + if modified: + elements.append(f" {html.escape(modified)}") + return elements + + +def _format_iso_duration(value: float) -> str: + total_seconds = int(value) + remainder = value - total_seconds + hours, remainder_seconds = divmod(total_seconds, 3600) + minutes, seconds = divmod(remainder_seconds, 60) + seconds_with_fraction = seconds + remainder + if seconds_with_fraction.is_integer(): + seconds_text = f"{int(seconds_with_fraction)}" + else: + seconds_text = f"{seconds_with_fraction:.3f}".rstrip("0").rstrip(".") + return f"PT{hours}H{minutes}M{seconds_text}S" + + +def _detect_audio_mime(audio_filename: str) -> str: + suffix = Path(audio_filename).suffix.lower() + return { + ".mp3": "audio/mpeg", + ".m4a": "audio/mp4", + ".m4b": "audio/mp4", + ".aac": "audio/aac", + ".wav": "audio/wav", + ".flac": "audio/flac", + ".ogg": "audio/ogg", + ".opus": "audio/ogg", + }.get(suffix, "audio/mpeg") + + +def _detect_image_mime(suffix: str) -> str: + normalized = suffix.lower() + return { + ".jpg": "image/jpeg", + ".jpeg": "image/jpeg", + ".png": "image/png", + ".gif": "image/gif", + ".webp": "image/webp", + }.get(normalized, "image/jpeg") + + +def _utc_now_iso() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +_DEFAULT_STYLESHEET = """ +body { + font-family: 'Georgia', serif; + line-height: 1.6; + margin: 1.5em; +} + +h1 { + font-size: 1.5em; + margin-bottom: 0.5em; +} + +div.chunk { + margin-bottom: 1em; +} + +p { + margin: 0.5em 0; +} +""" diff --git a/abogen/speaker_analysis.py b/abogen/speaker_analysis.py new file mode 100644 index 0000000..a170d04 --- /dev/null +++ b/abogen/speaker_analysis.py @@ -0,0 +1,297 @@ +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple + +_DIALOGUE_VERBS = ( + "said", + "asked", + "replied", + "whispered", + "shouted", + "cried", + "muttered", + "answered", + "hissed", + "called", + "added", + "continued", + "insisted", + "remarked", + "yelled", + "breathed", + "murmured", + "exclaimed", + "explained", + "noted", +) + +_VERB_PATTERN = "(?:" + "|".join(_DIALOGUE_VERBS) + ")" +_NAME_FRAGMENT = r"[A-Z][A-Za-z'\-]+" +_NAME_PATTERN = rf"{_NAME_FRAGMENT}(?:\s+{_NAME_FRAGMENT})*" + +_COLON_PATTERN = re.compile(rf"^\s*({_NAME_PATTERN})\s*:\s*(.+)$") +_NAME_BEFORE_VERB = re.compile(rf"({_NAME_PATTERN})\s+{_VERB_PATTERN}\b", re.IGNORECASE) +_VERB_BEFORE_NAME = re.compile(rf"{_VERB_PATTERN}\s+({_NAME_PATTERN})", re.IGNORECASE) +_PRONOUN_PATTERN = re.compile(r"\b(?:he|she|they)\b", re.IGNORECASE) +_QUOTE_PATTERN = re.compile(r'"([^"\\]*(?:\\.[^"\\]*)*)"') + +_CONFIDENCE_RANK = {"low": 1, "medium": 2, "high": 3} + + +@dataclass(slots=True) +class SpeakerGuess: + speaker_id: str + label: str + count: int = 0 + confidence: str = "low" + sample_quotes: List[str] = field(default_factory=list) + suppressed: bool = False + + def register_occurrence(self, confidence: str, quote: Optional[str]) -> None: + self.count += 1 + if _CONFIDENCE_RANK.get(confidence, 0) > _CONFIDENCE_RANK.get(self.confidence, 0): + self.confidence = confidence + if quote: + normalized = quote.strip() + if normalized and normalized not in self.sample_quotes: + self.sample_quotes.append(normalized[:240]) + if len(self.sample_quotes) > 3: + self.sample_quotes = self.sample_quotes[:3] + + def as_dict(self) -> Dict[str, Any]: + return { + "id": self.speaker_id, + "label": self.label, + "count": self.count, + "confidence": self.confidence, + "sample_quotes": list(self.sample_quotes), + "suppressed": self.suppressed, + } + + +@dataclass(slots=True) +class SpeakerAnalysis: + assignments: Dict[str, str] + speakers: Dict[str, SpeakerGuess] + suppressed: List[str] + narrator: str = "narrator" + version: str = "1.0" + stats: Dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> Dict[str, Any]: + return { + "version": self.version, + "narrator": self.narrator, + "assignments": dict(self.assignments), + "speakers": {speaker_id: guess.as_dict() for speaker_id, guess in self.speakers.items()}, + "suppressed": list(self.suppressed), + "stats": dict(self.stats), + } + + +def analyze_speakers( + chapters: Sequence[Dict[str, Any]] | Iterable[Dict[str, Any]], + chunks: Sequence[Dict[str, Any]] | Iterable[Dict[str, Any]], + *, + threshold: int = 3, + max_speakers: int = 8, +) -> SpeakerAnalysis: + narrator_id = "narrator" + speaker_guesses: Dict[str, SpeakerGuess] = { + narrator_id: SpeakerGuess(speaker_id=narrator_id, label="Narrator", confidence="low") + } + label_index: Dict[str, str] = {"Narrator": narrator_id} + assignments: Dict[str, str] = {} + suppressed: List[str] = [] + + ordered_chunks = sorted( + (dict(chunk) for chunk in chunks), + key=lambda entry: ( + _safe_int(entry.get("chapter_index")), + _safe_int(entry.get("chunk_index")), + ), + ) + last_explicit: Optional[str] = None + explicit_assignments = 0 + unique_speakers: set[str] = set() + + for chunk in ordered_chunks: + chunk_id = str(chunk.get("id") or "") + text = str(chunk.get("text") or "") + speaker_id, confidence, quote = _infer_chunk_speaker(text, last_explicit) + if speaker_id is None: + speaker_id = last_explicit or narrator_id + confidence = "medium" if last_explicit else "low" + quote = quote or _extract_quote(text) + if speaker_id != narrator_id: + last_explicit = speaker_id + explicit_assignments += 1 + assignments[chunk_id] = speaker_id + unique_speakers.add(speaker_id) + + label = _normalize_label(speaker_id) + record_id = label_index.get(label) + if record_id is None: + record_id = _dedupe_slug(_slugify(label), speaker_guesses) + label_index[label] = record_id + speaker_guesses[record_id] = SpeakerGuess(speaker_id=record_id, label=label) + guess = speaker_guesses[record_id] + guess.register_occurrence(confidence, quote) + if record_id != speaker_id: + # Maintain mapping to canonical ID in assignments. + assignments[chunk_id] = record_id + if speaker_id == last_explicit: + last_explicit = record_id + + active_speakers = [sid for sid in speaker_guesses if sid != narrator_id] + # Apply minimum occurrence threshold. + for speaker_id in list(active_speakers): + guess = speaker_guesses[speaker_id] + if guess.count < max(1, threshold): + guess.suppressed = True + suppressed.append(speaker_id) + _reassign(assignments, speaker_id, narrator_id) + active_speakers.remove(speaker_id) + + # Apply maximum active speaker cap. + if max_speakers and len(active_speakers) > max_speakers: + active_speakers.sort(key=lambda sid: (-speaker_guesses[sid].count, sid)) + for speaker_id in active_speakers[max_speakers:]: + guess = speaker_guesses[speaker_id] + guess.suppressed = True + suppressed.append(speaker_id) + _reassign(assignments, speaker_id, narrator_id) + active_speakers = active_speakers[:max_speakers] + + narrator_guess = speaker_guesses[narrator_id] + narrator_guess.count = sum(1 for value in assignments.values() if value == narrator_id) + narrator_guess.confidence = "low" + + stats = { + "total_chunks": len(ordered_chunks), + "explicit_chunks": explicit_assignments, + "active_speakers": len(active_speakers), + "unique_speakers": len(unique_speakers), + "suppressed": len(suppressed), + } + + return SpeakerAnalysis( + assignments=assignments, + speakers=speaker_guesses, + suppressed=suppressed, + narrator=narrator_id, + stats=stats, + ) + + +def _infer_chunk_speaker(text: str, last_explicit: Optional[str]) -> Tuple[Optional[str], str, Optional[str]]: + normalized = text.strip() + if not normalized: + return None, "low", None + + colon_match = _COLON_PATTERN.match(normalized) + if colon_match: + raw_label = colon_match.group(1) + quote = colon_match.group(2).strip() + return raw_label, "high", quote + + quote = _extract_quote(normalized) + if not quote: + return None, "low", None + + before, after = _split_around_quote(normalized, quote) + + candidate = _match_name_near_quote(before, after) + if candidate: + return candidate, "high", quote + + if last_explicit: + pronoun_after = _PRONOUN_PATTERN.search(after) + pronoun_before = _PRONOUN_PATTERN.search(before) + if pronoun_after or pronoun_before: + return last_explicit, "medium", quote + + return None, "low", quote + + +def _split_around_quote(text: str, quote: str) -> Tuple[str, str]: + quote_index = text.find(quote) + if quote_index == -1: + return text, "" + before = text[:quote_index] + after = text[quote_index + len(quote) :] + return before, after + + +def _match_name_near_quote(before: str, after: str) -> Optional[str]: + trailing = before[-120:] + leading = after[:120] + + match = _NAME_BEFORE_VERB.search(trailing) + if match: + name = match.group(1) + if _looks_like_name(name): + return name + + match = re.search(rf"({_NAME_PATTERN})\s*,?\s*{_VERB_PATTERN}", leading, flags=re.IGNORECASE) + if match: + name = match.group(1) + if _looks_like_name(name): + return name + + match = _VERB_BEFORE_NAME.search(leading) + if match: + name = match.group(1) + if _looks_like_name(name): + return name + + return None + + +def _looks_like_name(value: str) -> bool: + parts = value.strip().split() + if not parts: + return False + return all(part[0].isupper() for part in parts) + + +def _extract_quote(text: str) -> Optional[str]: + match = _QUOTE_PATTERN.search(text) + if not match: + return None + return match.group(0) + + +def _slugify(label: str) -> str: + slug = re.sub(r"[^a-z0-9]+", "_", label.lower()).strip("_") + return slug or "speaker" + + +def _dedupe_slug(slug: str, existing: Dict[str, SpeakerGuess]) -> str: + candidate = slug + index = 2 + while candidate in existing: + candidate = f"{slug}_{index}" + index += 1 + return candidate + + +def _normalize_label(label: str) -> str: + words = re.split(r"\s+", label.strip()) + return " ".join(word.capitalize() for word in words if word) + + +def _safe_int(value: Any, default: int = 0) -> int: + try: + return int(value) + except (TypeError, ValueError): + return default + + +def _reassign(assignments: Dict[str, str], old: str, new: str) -> None: + for key, value in list(assignments.items()): + if value == old: + assignments[key] = new +```}, \ No newline at end of file diff --git a/abogen/web/conversion_runner.py b/abogen/web/conversion_runner.py index c42a2f2..b822e17 100644 --- a/abogen/web/conversion_runner.py +++ b/abogen/web/conversion_runner.py @@ -7,15 +7,17 @@ import re import subprocess import sys import tempfile +from collections import defaultdict from contextlib import ExitStack from dataclasses import dataclass from pathlib import Path -from typing import Any, Callable, Dict, List, Optional, cast +from typing import Any, Callable, Dict, Iterable, List, Optional, cast import numpy as np import soundfile as sf import static_ffmpeg +from abogen.epub3.exporter import build_epub3_package from abogen.kokoro_text_normalization import ( ApostropheConfig, apply_phoneme_hints, @@ -320,6 +322,62 @@ def _chapter_voice_spec(job: Job, override: Optional[Dict[str, Any]]) -> str: return job.voice or "" +def _chunk_voice_spec(job: Any, chunk: Dict[str, Any], fallback: str) -> str: + for key in ("resolved_voice", "voice_formula", "voice"): + value = chunk.get(key) + if value: + return str(value) + + speaker_id = chunk.get("speaker_id") + speakers = getattr(job, "speakers", None) + if isinstance(speakers, dict) and speaker_id in speakers: + speaker_entry = speakers.get(speaker_id) or {} + if isinstance(speaker_entry, dict): + for key in ("resolved_voice", "voice_formula", "voice"): + value = speaker_entry.get(key) + if value: + return str(value) + profile_formula = speaker_entry.get("voice_formula") + if profile_formula: + return str(profile_formula) + + profile_name = chunk.get("voice_profile") + if profile_name: + if isinstance(speakers, dict): + speaker_entry = speakers.get(profile_name) + if isinstance(speaker_entry, dict): + for key in ("resolved_voice", "voice_formula", "voice"): + value = speaker_entry.get(key) + if value: + return str(value) + + return fallback or getattr(job, "voice", "") or "" + + +def _group_chunks_by_chapter(chunks: Iterable[Dict[str, Any]]) -> Dict[int, List[Dict[str, Any]]]: + grouped: Dict[int, List[Dict[str, Any]]] = defaultdict(list) + for entry in chunks or []: + if not isinstance(entry, dict): + continue + try: + chapter_index = int(entry.get("chapter_index", 0)) + except (TypeError, ValueError): + chapter_index = 0 + grouped[chapter_index].append(dict(entry)) + + for chapter_index, items in grouped.items(): + items.sort(key=lambda payload: _safe_int(payload.get("chunk_index"))) + + return grouped + + +def _safe_int(value: Any, default: int = 0) -> int: + try: + return int(value) + except (TypeError, ValueError): + return default + + def _escape_ffmetadata_value(value: str) -> str: escaped = str(value).replace("\\", "\\\\").replace("\n", "\\n") escaped = escaped.replace("=", "\\=").replace(";", "\\;").replace("#", "\\#") @@ -559,7 +617,8 @@ def run_conversion_job(job: Job) -> None: subtitle_writer: Optional[SubtitleWriter] = None chapter_paths: list[Path] = [] chapter_markers: List[Dict[str, Any]] = [] - metadata_payload: Dict[str, Any] = {"metadata": {}, "chapters": []} + chunk_markers: List[Dict[str, Any]] = [] + metadata_payload: Dict[str, Any] = {} audio_output_path: Optional[Path] = None try: pipeline = _load_pipeline(job) @@ -598,6 +657,7 @@ def run_conversion_job(job: Job) -> None: metadata_overrides: Dict[str, Any] = dict(job.metadata_tags or {}) active_chapter_configs: List[Dict[str, Any]] = [] + chunk_groups: Dict[int, List[Dict[str, Any]]] = {} if job.chapters: selected_chapters, chapter_metadata, diagnostics = _apply_chapter_overrides( extraction.chapters, @@ -615,8 +675,12 @@ def run_conversion_job(job: Job) -> None: active_chapter_configs = [ entry for entry in job.chapters if _coerce_truthy(entry.get("enabled", True)) ][: len(selected_chapters)] + if job.chunks: + chunk_groups = _group_chunks_by_chapter(job.chunks) else: raise ValueError("No chapters were enabled in the requested job.") + elif job.chunks: + chunk_groups = _group_chunks_by_chapter(job.chunks) job.metadata_tags = _merge_metadata(extraction.metadata, metadata_overrides) @@ -661,6 +725,10 @@ def run_conversion_job(job: Job) -> None: subtitle_index = 1 current_time = 0.0 total_chapters = len(extraction.chapters) + if chunk_groups: + chunk_groups = { + idx: items for idx, items in chunk_groups.items() if 0 <= idx < total_chapters + } job.add_log(f"Detected {total_chapters} chapter{'s' if total_chapters != 1 else ''}") def emit_text( @@ -800,11 +868,93 @@ def run_conversion_job(job: Job) -> None: chapter_sink=chapter_sink, ) - segments_emitted += emit_text( - chapter.text, - voice_choice=voice_choice, - chapter_sink=chapter_sink, - ) + chunks_for_chapter = chunk_groups.get(idx - 1, []) if chunk_groups else [] + body_segments = 0 + if chunks_for_chapter: + job.add_log( + f"Emitting {len(chunks_for_chapter)} {job.chunk_level} chunks for chapter {idx}.", + level="debug", + ) + for chunk_entry in chunks_for_chapter: + chunk_text = str(chunk_entry.get("text") or "").strip() + if not chunk_text: + continue + + chunk_voice_spec = _chunk_voice_spec( + job, + chunk_entry, + chapter_voice_spec or base_voice_spec, + ) + if not chunk_voice_spec: + chunk_voice_spec = chapter_voice_spec or base_voice_spec + + if chunk_voice_spec == chapter_voice_spec: + chunk_voice_choice = voice_choice + else: + chunk_voice_choice = voice_cache.get(chunk_voice_spec) + if chunk_voice_choice is None: + chunk_voice_choice = _resolve_voice( + pipeline, + chunk_voice_spec, + job.use_gpu, + ) + voice_cache[chunk_voice_spec] = chunk_voice_choice + + chunk_start = current_time + emitted = emit_text( + chunk_text, + voice_choice=chunk_voice_choice, + chapter_sink=chapter_sink, + preview_prefix=f"Chunk {chunk_entry.get('id') or chunk_entry.get('chunk_index')}", + ) + if emitted <= 0: + continue + + body_segments += emitted + segments_emitted += emitted + chunk_markers.append( + { + "id": chunk_entry.get("id"), + "chapter_index": idx - 1, + "chunk_index": _safe_int( + chunk_entry.get("chunk_index"), len(chunk_markers) + ), + "start": chunk_start, + "end": current_time, + "speaker_id": chunk_entry.get("speaker_id", "narrator"), + "voice": chunk_voice_spec, + "level": chunk_entry.get("level", job.chunk_level), + "characters": len(chunk_text), + } + ) + + if body_segments == 0: + chapter_body_start = current_time + emitted = emit_text( + chapter.text, + voice_choice=voice_choice, + chapter_sink=chapter_sink, + ) + if emitted > 0: + segments_emitted += emitted + chunk_markers.append( + { + "id": None, + "chapter_index": idx - 1, + "chunk_index": 0, + "start": chapter_body_start, + "end": current_time, + "speaker_id": "narrator", + "voice": chapter_voice_spec, + "level": job.chunk_level, + "characters": len(chapter.text or ""), + } + ) + elif chunks_for_chapter: + job.add_log( + "No audio generated for supplied chunks; chapter text also empty.", + level="warning", + ) chapter_end_time = current_time @@ -849,6 +999,11 @@ def run_conversion_job(job: Job) -> None: metadata_payload = { "metadata": dict(job.metadata_tags or {}), "chapters": chapter_markers, + "chunks": chunk_markers, + "chunk_level": job.chunk_level, + "speaker_mode": job.speaker_mode, + "speakers": dict(getattr(job, "speakers", {}) or {}), + "generate_epub3": job.generate_epub3, } if metadata_dir: @@ -857,6 +1012,37 @@ def run_conversion_job(job: Job) -> None: metadata_file.write_text(json.dumps(metadata_payload, indent=2), encoding="utf-8") job.result.artifacts["metadata"] = metadata_file + if job.generate_epub3: + audio_asset = job.result.audio_path + if not audio_asset and chapter_paths: + audio_asset = chapter_paths[0] + + if audio_asset: + try: + epub_root = project_root if job.save_as_project else base_output_dir + epub_output_path = _build_output_path(epub_root, job.original_filename, "epub") + job.add_log("Generating EPUB 3 package with synchronized narration…") + epub_path = build_epub3_package( + output_path=epub_output_path, + book_id=job.id, + extraction=extraction, + metadata_tags=metadata_payload.get("metadata") or {}, + chapter_markers=chapter_markers, + chunk_markers=chunk_markers, + chunks=job.chunks, + audio_path=audio_asset, + speaker_mode=job.speaker_mode, + cover_image_path=job.cover_image_path, + cover_image_mime=job.cover_image_mime, + ) + job.result.epub_path = epub_path + job.result.artifacts["epub3"] = epub_path + job.add_log(f"EPUB 3 package created at {epub_path}") + except Exception as exc: + job.add_log(f"Failed to generate EPUB 3 package: {exc}", level="error") + else: + job.add_log("Skipped EPUB 3 generation: audio output unavailable.", level="warning") + if job.save_as_project: job.result.artifacts["project_root"] = project_root diff --git a/abogen/web/routes.py b/abogen/web/routes.py index 7dae1a4..e4b6063 100644 --- a/abogen/web/routes.py +++ b/abogen/web/routes.py @@ -9,7 +9,7 @@ import threading import time import uuid from pathlib import Path -from typing import Any, Dict, Iterable, List, Optional, Tuple, cast +from typing import Any, Dict, Iterable, List, Mapping, Optional, Tuple, cast from flask import ( Blueprint, @@ -35,6 +35,7 @@ from abogen.constants import ( SUPPORTED_SOUND_FORMATS, VOICES_INTERNAL, ) +from abogen.chunking import ChunkLevel, build_chunks_for_chapters from abogen.utils import ( calculate_text_length, clean_text, @@ -57,6 +58,7 @@ from abogen.voice_profiles import ( ) from abogen.voice_formulas import get_new_voice +from abogen.speaker_analysis import analyze_speakers from abogen.text_extractor import extract_from_path from .conversion_runner import SPLIT_PATTERN, SAMPLE_RATE, _select_device, _to_float32 from .service import ConversionService, Job, JobStatus, PendingJob @@ -69,6 +71,174 @@ _preview_pipeline_lock = threading.RLock() _preview_pipelines: Dict[Tuple[str, str], Any] = {} +_CHUNK_LEVEL_OPTIONS = [ + {"value": "paragraph", "label": "Paragraphs"}, + {"value": "sentence", "label": "Sentences"}, +] + +_SPEAKER_MODE_OPTIONS = [ + {"value": "single", "label": "Single Speaker"}, + {"value": "multi", "label": "Multi-Speaker"}, +] + +_CHUNK_LEVEL_VALUES = {option["value"] for option in _CHUNK_LEVEL_OPTIONS} +_SPEAKER_MODE_VALUES = {option["value"] for option in _SPEAKER_MODE_OPTIONS} + + +_DEFAULT_ANALYSIS_THRESHOLD = 3 +_MAX_ANALYSIS_SPEAKERS = 6 + + +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, +) -> 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 {} + for speaker_id, payload in speakers.items(): + if speaker_id == "narrator": + continue + if 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(), + "voice": base_voice, + "analysis_confidence": payload.get("confidence"), + "analysis_count": payload.get("count"), + } + 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 + return roster + + +def _prepare_speaker_metadata( + *, + chapters: List[Dict[str, Any]], + chunks: List[Dict[str, Any]], + speaker_mode: str, + voice: str, + voice_profile: Optional[str], + threshold: int, + existing_roster: Optional[Mapping[str, Any]] = None, +) -> 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)) + + if speaker_mode != "multi": + 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 + + analysis_result = analyze_speakers( + chapters, chunk_list, threshold=threshold_value, max_speakers=_MAX_ANALYSIS_SPEAKERS + ) + analysis_payload = analysis_result.to_dict() + 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) + 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 + + 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 + + _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), @@ -196,6 +366,7 @@ def _build_voice_catalog() -> List[Dict[str, str]]: def _template_options() -> Dict[str, Any]: + current_settings = _load_settings() profiles = serialize_profiles() ordered_profiles = sorted(profiles.items()) profile_options = [] @@ -219,6 +390,14 @@ def _template_options() -> Dict[str, Any]: "voice_catalog": _build_voice_catalog(), "sample_voice_texts": SAMPLE_VOICE_TEXTS, "voice_profiles_data": profiles, + "chunk_levels": _CHUNK_LEVEL_OPTIONS, + "speaker_modes": _SPEAKER_MODE_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"] + ), } @@ -237,10 +416,11 @@ BOOLEAN_SETTINGS = { "save_chapters_separately", "merge_chapters_at_end", "save_as_project", + "generate_epub3", } FLOAT_SETTINGS = {"silence_between_chapters", "chapter_intro_delay"} -INT_SETTINGS = {"max_subtitle_words"} +INT_SETTINGS = {"max_subtitle_words", "speaker_analysis_threshold"} def _has_output_override() -> bool: @@ -262,6 +442,11 @@ def _settings_defaults() -> Dict[str, Any]: "silence_between_chapters": 2.0, "chapter_intro_delay": 0.5, "max_subtitle_words": 50, + "chunk_level": "paragraph", + "speaker_mode": "single", + "generate_epub3": False, + "speaker_analysis_threshold": _DEFAULT_ANALYSIS_THRESHOLD, + "speaker_pronunciation_sentence": "This is {{name}} speaking.", } @@ -323,6 +508,14 @@ def _normalize_setting_value(key: str, value: Any, defaults: Dict[str, Any]) -> if isinstance(value, str) and value in VOICES_INTERNAL: return value return defaults[key] + if key == "chunk_level": + if isinstance(value, str) and value in _CHUNK_LEVEL_VALUES: + return value + return defaults[key] + if key == "speaker_mode": + if isinstance(value, str) and value in _SPEAKER_MODE_VALUES: + return value + return defaults[key] return value if value is not None else defaults.get(key) @@ -519,6 +712,12 @@ def settings_page() -> Response | str: ) for key in sorted(BOOLEAN_SETTINGS): updated[key] = _coerce_bool(form.get(key), False) + updated["chunk_level"] = _normalize_setting_value( + "chunk_level", form.get("chunk_level"), defaults + ) + updated["speaker_mode"] = _normalize_setting_value( + "speaker_mode", form.get("speaker_mode"), defaults + ) updated["separate_chapters_format"] = _normalize_setting_value( "separate_chapters_format", form.get("separate_chapters_format"), defaults ) @@ -531,6 +730,16 @@ def settings_page() -> Response | str: updated["max_subtitle_words"] = _coerce_int( form.get("max_subtitle_words"), defaults["max_subtitle_words"] ) + updated["speaker_analysis_threshold"] = _coerce_int( + form.get("speaker_analysis_threshold"), + defaults["speaker_analysis_threshold"], + minimum=1, + maximum=25, + ) + sentence_value = (form.get("speaker_pronunciation_sentence") or "").strip() + if not sentence_value: + sentence_value = defaults["speaker_pronunciation_sentence"] + updated["speaker_pronunciation_sentence"] = sentence_value cfg = load_config() or {} cfg.update(updated) @@ -800,6 +1009,102 @@ def api_preview_voice_mix() -> Response: return response +@api_bp.post("/speaker-preview") +def api_speaker_preview() -> Response: + payload = request.get_json(force=True, silent=False) + text = (payload.get("text") or "").strip() + voice_spec = (payload.get("voice") or "").strip() + language = (payload.get("language") or "a").strip() or "a" + speed_input = payload.get("speed", 1.0) + try: + speed = float(speed_input) + except (TypeError, ValueError): + speed = 1.0 + max_seconds_input = payload.get("max_seconds", 8.0) + try: + max_seconds = max(1.0, min(15.0, float(max_seconds_input))) + except (TypeError, ValueError): + max_seconds = 8.0 + + if not text: + abort(400, "Preview text is required") + if not voice_spec: + abort(400, "Voice selection is required") + + settings = _load_settings() + use_gpu_default = settings.get("use_gpu", True) + if "use_gpu" in payload: + use_gpu = _coerce_bool(payload.get("use_gpu"), use_gpu_default) + else: + use_gpu = use_gpu_default + + device = "cpu" + if use_gpu: + try: + device = _select_device() + except Exception: # pragma: no cover - fallback + device = "cpu" + use_gpu = False + + try: + pipeline = _get_preview_pipeline(language, device) + except Exception as exc: # pragma: no cover - defensive guard + abort(500, f"Failed to initialise preview pipeline: {exc}") + if pipeline is None: # pragma: no cover - defensive double-check + abort(500, "Preview pipeline initialisation failed") + + voice_choice: Any = voice_spec + if "*" in voice_spec: + try: + voice_choice = get_new_voice(pipeline, voice_spec, use_gpu) + except ValueError as exc: + abort(400, str(exc)) + + segments = pipeline( + text, + voice=voice_choice, + speed=speed, + split_pattern=SPLIT_PATTERN, + ) + + audio_chunks: List[np.ndarray] = [] + accumulated = 0 + max_samples = int(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: + abort(500, "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 + + @web_bp.post("/jobs") def enqueue_job() -> Response: service = _service() @@ -921,6 +1226,41 @@ def enqueue_job() -> Response: chapter_intro_delay = settings["chapter_intro_delay"] max_subtitle_words = settings["max_subtitle_words"] + chunk_level_default = str(settings.get("chunk_level", "paragraph")).strip().lower() + raw_chunk_level = (request.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_default = str(settings.get("speaker_mode", "single")).strip().lower() + raw_speaker_mode = (request.form.get("speaker_mode") or speaker_mode_default).strip().lower() + if raw_speaker_mode not in _SPEAKER_MODE_VALUES: + raw_speaker_mode = "single" + speaker_mode_value = raw_speaker_mode + + generate_epub3_default = bool(settings.get("generate_epub3", False)) + generate_epub3 = _coerce_bool(request.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_threshold = _coerce_int( + settings.get("speaker_analysis_threshold"), + _DEFAULT_ANALYSIS_THRESHOLD, + minimum=1, + maximum=25, + ) + + processed_chunks, speakers, analysis_payload = _prepare_speaker_metadata( + chapters=selected_chapter_sources, + chunks=raw_chunks, + speaker_mode=speaker_mode_value, + voice=voice, + voice_profile=selected_profile or None, + threshold=analysis_threshold, + ) + pending = PendingJob( id=uuid.uuid4().hex, original_filename=original_name, @@ -941,7 +1281,7 @@ def enqueue_job() -> Response: separate_chapters_format=separate_chapters_format, silence_between_chapters=silence_between_chapters, save_as_project=save_as_project, - voice_profile=selected_profile, + voice_profile=selected_profile or None, max_subtitle_words=max_subtitle_words, metadata_tags=metadata_tags, chapters=chapters_payload, @@ -949,6 +1289,13 @@ def enqueue_job() -> Response: cover_image_path=cover_path, cover_image_mime=cover_mime, chapter_intro_delay=chapter_intro_delay, + 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, ) service.store_pending_job(pending) @@ -972,6 +1319,62 @@ def finalize_job(pending_id: str) -> Response: 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) + + 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 + + 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") @@ -1038,9 +1441,25 @@ def finalize_job(pending_id: str) -> Response: overrides.append(entry) pending.chapters[index] = dict(entry) - if not any(item.get("enabled") for item in overrides): + enabled_overrides = [entry for entry in overrides if entry.get("enabled")] + 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) + 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=pending.speakers, + ) + pending.chunks = processed_chunks + pending.speakers = roster + pending.speaker_analysis = analysis_payload + if errors: return _render_prepare_page(pending, error=" ".join(errors)) @@ -1074,6 +1493,13 @@ def finalize_job(pending_id: str) -> Response: cover_image_path=pending.cover_image_path, cover_image_mime=pending.cover_image_mime, chapter_intro_delay=pending.chapter_intro_delay, + chunk_level=pending.chunk_level, + chunks=processed_chunks, + speakers=roster, + speaker_mode=pending.speaker_mode, + speaker_analysis=analysis_payload, + speaker_analysis_threshold=pending.speaker_analysis_threshold, + generate_epub3=pending.generate_epub3, ) return redirect(url_for("web.queue_page")) diff --git a/abogen/web/service.py b/abogen/web/service.py index cda9ced..6f6c64d 100644 --- a/abogen/web/service.py +++ b/abogen/web/service.py @@ -20,7 +20,7 @@ def _create_set_event() -> threading.Event: return event -STATE_VERSION = 3 +STATE_VERSION = 5 class JobStatus(str, Enum): @@ -44,6 +44,7 @@ class JobResult: audio_path: Optional[Path] = None subtitle_paths: List[Path] = field(default_factory=list) artifacts: Dict[str, Path] = field(default_factory=dict) + epub_path: Optional[Path] = None @dataclass @@ -89,6 +90,13 @@ class Job: pause_event: threading.Event = field(default_factory=_create_set_event, repr=False, compare=False) cover_image_path: Optional[Path] = None cover_image_mime: Optional[str] = None + chunk_level: str = "paragraph" + chunks: List[Dict[str, Any]] = field(default_factory=list) + speakers: Dict[str, Any] = field(default_factory=dict) + speaker_mode: str = "single" + generate_epub3: bool = False + speaker_analysis: Dict[str, Any] = field(default_factory=dict) + speaker_analysis_threshold: int = 3 def add_log(self, message: str, level: str = "info") -> None: self.logs.append(JobLog(timestamp=time.time(), message=message, level=level)) @@ -139,6 +147,13 @@ class Job: } for entry in self.chapters ], + "chunk_level": self.chunk_level, + "chunks": [dict(chunk) for chunk in self.chunks], + "speakers": dict(self.speakers), + "speaker_mode": self.speaker_mode, + "generate_epub3": self.generate_epub3, + "speaker_analysis": dict(self.speaker_analysis), + "speaker_analysis_threshold": self.speaker_analysis_threshold, } @@ -171,6 +186,13 @@ class PendingJob: cover_image_path: Optional[Path] = None cover_image_mime: Optional[str] = None chapter_intro_delay: float = 0.5 + chunk_level: str = "paragraph" + chunks: List[Dict[str, Any]] = field(default_factory=list) + speakers: Dict[str, Any] = field(default_factory=dict) + speaker_mode: str = "single" + generate_epub3: bool = False + speaker_analysis: Dict[str, Any] = field(default_factory=dict) + speaker_analysis_threshold: int = 3 class ConversionService: @@ -234,10 +256,18 @@ class ConversionService: cover_image_path: Optional[Path] = None, cover_image_mime: Optional[str] = None, chapter_intro_delay: float = 0.5, + chunk_level: str = "paragraph", + chunks: Optional[Iterable[Any]] = None, + speakers: Optional[Mapping[str, Any]] = None, + speaker_mode: str = "single", + generate_epub3: bool = False, + speaker_analysis: Optional[Mapping[str, Any]] = None, + speaker_analysis_threshold: int = 3, ) -> Job: job_id = uuid.uuid4().hex normalized_metadata = self._normalize_metadata_tags(metadata_tags) normalized_chapters = self._normalize_chapters(chapters) + normalized_chunks = self._normalize_chunks(chunks) if total_characters <= 0 and normalized_chapters: total_characters = sum(len(str(entry.get("text", ""))) for entry in normalized_chapters) job = Job( @@ -268,6 +298,13 @@ class ConversionService: cover_image_path=cover_image_path, cover_image_mime=cover_image_mime, chapter_intro_delay=chapter_intro_delay, + chunk_level=chunk_level, + chunks=normalized_chunks, + speakers=dict(speakers or {}), + speaker_mode=speaker_mode, + generate_epub3=bool(generate_epub3), + speaker_analysis=dict(speaker_analysis or {}), + speaker_analysis_threshold=int(speaker_analysis_threshold or 3), ) with self._lock: self._jobs[job_id] = job @@ -490,6 +527,7 @@ class ConversionService: result_audio = str(job.result.audio_path) if job.result.audio_path else None result_subtitles = [str(path) for path in job.result.subtitle_paths] result_artifacts = {key: str(path) for key, path in job.result.artifacts.items()} + result_epub = str(job.result.epub_path) if job.result.epub_path else None return { "id": job.id, "original_filename": job.original_filename, @@ -525,6 +563,7 @@ class ConversionService: "audio_path": result_audio, "subtitle_paths": result_subtitles, "artifacts": result_artifacts, + "epub_path": result_epub, }, "chapters": [dict(entry) for entry in job.chapters], "queue_position": job.queue_position, @@ -535,6 +574,13 @@ class ConversionService: "cover_image_path": str(job.cover_image_path) if job.cover_image_path else None, "cover_image_mime": job.cover_image_mime, "chapter_intro_delay": job.chapter_intro_delay, + "chunk_level": job.chunk_level, + "chunks": [dict(entry) for entry in job.chunks], + "speakers": dict(job.speakers), + "speaker_mode": job.speaker_mode, + "generate_epub3": job.generate_epub3, + "speaker_analysis": dict(job.speaker_analysis), + "speaker_analysis_threshold": job.speaker_analysis_threshold, } def _persist_state(self) -> None: @@ -631,6 +677,8 @@ class ConversionService: job.result.artifacts = { key: Path(value) for key, value in result_payload.get("artifacts", {}).items() } + epub_path_raw = result_payload.get("epub_path") + job.result.epub_path = Path(epub_path_raw) if epub_path_raw else None job.chapters = payload.get("chapters", []) job.queue_position = payload.get("queue_position") job.cancel_requested = bool(payload.get("cancel_requested", False)) @@ -640,6 +688,15 @@ class ConversionService: cover_path_raw = payload.get("cover_image_path") job.cover_image_path = Path(cover_path_raw) if cover_path_raw else None job.cover_image_mime = payload.get("cover_image_mime") + job.chunk_level = str(payload.get("chunk_level", job.chunk_level or "paragraph")) + job.chunks = self._normalize_chunks(payload.get("chunks")) + job.speakers = dict(payload.get("speakers", {})) + job.speaker_mode = str(payload.get("speaker_mode", job.speaker_mode or "single")) + job.generate_epub3 = bool(payload.get("generate_epub3", job.generate_epub3)) + job.speaker_analysis = payload.get("speaker_analysis", {}) + job.speaker_analysis_threshold = int( + payload.get("speaker_analysis_threshold", job.speaker_analysis_threshold or 3) + ) job.pause_event.set() return job @@ -837,6 +894,59 @@ class ConversionService: return normalized + @classmethod + def _normalize_chunks(cls, chunks: Optional[Iterable[Any]]) -> List[Dict[str, Any]]: + if not chunks: + return [] + + normalized: List[Dict[str, Any]] = [] + for order, raw in enumerate(chunks): + if raw is None: + continue + if isinstance(raw, dict): + entry = dict(raw) + else: + continue + + chunk: Dict[str, Any] = {} + + identifier = entry.get("id") or entry.get("chunk_id") + if identifier is not None: + chunk["id"] = str(identifier) + + try: + chunk_index = int(entry.get("chunk_index", order)) + except (TypeError, ValueError): + chunk_index = order + chunk["chunk_index"] = chunk_index + + try: + chapter_index = int(entry.get("chapter_index", 0)) + except (TypeError, ValueError): + chapter_index = 0 + chunk["chapter_index"] = chapter_index + + level_raw = str(entry.get("level", "paragraph")).lower() + if level_raw not in {"paragraph", "sentence"}: + level_raw = "paragraph" + chunk["level"] = level_raw + + text_value = entry.get("text") + if text_value is not None: + chunk["text"] = str(text_value) + else: + chunk["text"] = "" + + speaker_value = entry.get("speaker_id", entry.get("speaker")) + chunk["speaker_id"] = str(speaker_value) if speaker_value else "narrator" + + for key in ("voice", "voice_profile", "voice_formula", "audio_path", "start", "end"): + if key in entry and entry[key] is not None: + chunk[key] = entry[key] + + normalized.append(chunk) + return normalized + def default_storage_root() -> Path: base = Path.cwd() diff --git a/abogen/web/static/speakers.js b/abogen/web/static/speakers.js new file mode 100644 index 0000000..4062446 --- /dev/null +++ b/abogen/web/static/speakers.js @@ -0,0 +1,97 @@ +const audioElement = new Audio(); +let activeButton = null; +let activeUrl = null; + +const setLoadingState = (button, isLoading) => { + if (!button) return; + button.disabled = isLoading; + if (isLoading) { + button.setAttribute("data-loading", "true"); + } else { + button.removeAttribute("data-loading"); + } +}; + +const stopCurrentPlayback = () => { + if (audioElement && !audioElement.paused) { + audioElement.pause(); + } + if (activeUrl) { + URL.revokeObjectURL(activeUrl); + activeUrl = null; + } + if (activeButton) { + setLoadingState(activeButton, false); + activeButton = null; + } +}; + +audioElement.addEventListener("ended", () => { + stopCurrentPlayback(); +}); + +audioElement.addEventListener("pause", () => { + if (audioElement.currentTime === 0 || audioElement.currentTime >= audioElement.duration) { + stopCurrentPlayback(); + } +}); + +const playPreview = async (button) => { + const text = (button.dataset.previewText || "").trim(); + const voice = (button.dataset.voice || "").trim(); + const language = (button.dataset.language || "a").trim() || "a"; + const speedRaw = button.dataset.speed || "1"; + const useGpu = (button.dataset.useGpu || "true") !== "false"; + const speed = Number.parseFloat(speedRaw); + + if (!text) { + console.warn("Skipping speaker preview: no text provided"); + return; + } + if (!voice) { + console.warn("Skipping speaker preview: no voice provided"); + return; + } + + const payload = { + text, + voice, + language, + speed: Number.isFinite(speed) ? speed : 1.0, + use_gpu: useGpu, + max_seconds: 8, + }; + + stopCurrentPlayback(); + activeButton = button; + setLoadingState(button, true); + + try { + const response = await fetch("/api/speaker-preview", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(payload), + }); + if (!response.ok) { + const message = await response.text(); + throw new Error(message || `Preview failed with status ${response.status}`); + } + const blob = await response.blob(); + activeUrl = URL.createObjectURL(blob); + audioElement.src = activeUrl; + await audioElement.play(); + } catch (error) { + console.error("Failed to play speaker preview", error); + stopCurrentPlayback(); + } finally { + setLoadingState(button, false); + } +}; + +document.addEventListener("click", (event) => { + const trigger = event.target.closest('[data-role="speaker-preview"]'); + if (!trigger) return; + event.preventDefault(); + if (trigger.disabled) return; + playPreview(trigger); +}); diff --git a/abogen/web/static/styles.css b/abogen/web/static/styles.css index b9e2f61..05146e9 100644 --- a/abogen/web/static/styles.css +++ b/abogen/web/static/styles.css @@ -673,6 +673,112 @@ body { font-weight: 500; } +.prepare-speakers { + margin-top: 2rem; + border-top: 1px solid var(--panel-border); + padding-top: 1.5rem; + display: flex; + flex-direction: column; + gap: 1rem; +} + +.speaker-list { + list-style: none; + margin: 0; + padding: 0; + display: flex; + flex-direction: column; + gap: 1rem; +} + +.speaker-list__item { + background: rgba(148, 163, 184, 0.05); + border: 1px solid var(--panel-border); + border-radius: 18px; + padding: 1rem 1.25rem; + display: flex; + flex-direction: column; + gap: 0.65rem; +} + +.speaker-line { + display: flex; + align-items: center; + justify-content: space-between; + gap: 0.75rem; +} + +.speaker-list__name { + font-weight: 600; + font-size: 1rem; +} + +.speaker-list__preview { + font-size: 1.1rem; + line-height: 1; + width: 2.4rem; + height: 2.4rem; + display: flex; + align-items: center; + justify-content: center; + border-radius: 999px; + border: 1px solid var(--panel-border); + color: var(--accent); + background: rgba(56, 189, 248, 0.08); + transition: transform 0.15s ease, box-shadow 0.15s ease, border-color 0.15s ease; +} + +.speaker-list__preview:hover { + border-color: var(--accent); + box-shadow: 0 0 0 3px rgba(56, 189, 248, 0.15); + transform: translateY(-1px); +} + +.speaker-list__preview .spinner { + display: none; +} + +.speaker-list__preview[data-loading="true"] .spinner { + display: inline-block; +} + +.speaker-list__preview[data-loading="true"] .icon-button__glyph { + display: none; +} + +.speaker-list__preview[data-loading="true"] { + cursor: progress; + box-shadow: none; + color: transparent; +} + +.speaker-list__field { + display: flex; + flex-direction: column; + gap: 0.35rem; +} + +.speaker-list__field input { + border-radius: 12px; + border: 1px solid var(--panel-border); + background: rgba(15, 23, 42, 0.6); + padding: 0.6rem 0.8rem; + color: var(--text); + font-size: 0.95rem; +} + +.speaker-list__field input:focus { + outline: none; + border-color: var(--accent); + box-shadow: 0 0 0 2px rgba(56, 189, 248, 0.2); +} + +.speaker-list__stats { + margin: 0; + color: var(--muted); + font-size: 0.85rem; +} + .prepare-metadata h2 { font-size: 1rem; margin: 0 0 0.6rem; @@ -1518,6 +1624,20 @@ input[data-state="locked"] { box-shadow: none; } +.icon-button--borderless { + background: transparent; + border-color: transparent; +} + +.icon-button--borderless:hover { + background: rgba(148, 163, 184, 0.1); + border-color: rgba(148, 163, 184, 0.3); +} + +.icon-button--borderless:focus-visible { + border-color: var(--accent); +} + .icon-button--primary { background: linear-gradient(135deg, var(--accent), var(--accent-strong)); border: none; @@ -1553,6 +1673,32 @@ input[data-state="locked"] { box-shadow: none; } +.spinner { + display: inline-block; + width: 1.1rem; + height: 1.1rem; + border-radius: 50%; + border: 2px solid rgba(148, 163, 184, 0.28); + border-top-color: var(--accent); + animation: spin 0.8s linear infinite; +} + +.spinner--sm { + width: 0.85rem; + height: 0.85rem; +} + +.spinner--lg { + width: 1.5rem; + height: 1.5rem; + border-width: 3px; +} + +.spinner--muted { + border-color: rgba(148, 163, 184, 0.2); + border-top-color: rgba(148, 163, 184, 0.6); +} + .button[data-role="preview-button"] { position: relative; } @@ -1577,7 +1723,8 @@ input[data-state="locked"] { } @media (prefers-reduced-motion: reduce) { - .button[data-role="preview-button"][data-loading="true"]::after { + .button[data-role="preview-button"][data-loading="true"]::after, + .spinner { animation-duration: 1.6s; } } diff --git a/abogen/web/templates/index.html b/abogen/web/templates/index.html index b99a274..d7cf990 100644 --- a/abogen/web/templates/index.html +++ b/abogen/web/templates/index.html @@ -61,6 +61,30 @@ {% endfor %} +
    + + +

    Controls how chapters are split into TTS-ready chunks.

    +
    +
    + + +
    +
    + +

    Creates a synchronized EPUB alongside audio output.

    +
    diff --git a/abogen/web/templates/job_detail.html b/abogen/web/templates/job_detail.html index 7ae11c9..189ace9 100644 --- a/abogen/web/templates/job_detail.html +++ b/abogen/web/templates/job_detail.html @@ -26,6 +26,10 @@
  • Chapter intro delay: {{ '%.1f'|format(job.chapter_intro_delay) }}s
  • Max words per subtitle: {{ job.max_subtitle_words }}
  • Project folder: {{ 'Yes' if job.save_as_project else 'No' }}
  • +
  • Chunk granularity: {{ job.chunk_level|replace('_', ' ')|title }}
  • +
  • Speaker mode: {{ job.speaker_mode|replace('_', ' ')|title }}
  • +
  • Speaker analysis threshold: {{ job.speaker_analysis_threshold }}
  • +
  • Generate EPUB 3: {{ 'Yes' if job.generate_epub3 else 'No' }}
  • @@ -45,7 +49,97 @@
    +{% set analysis = job.speaker_analysis or {} %} +{% if analysis %} +{% set preview_template = options.speaker_pronunciation_sentence or "This is {{name}} speaking." %} +
    +
    Speaker analysis
    +
    +
    +

    Summary

    + {% set stats = analysis.get('stats', {}) %} +
      +
    • Total chunks: {{ stats.get('total_chunks', '—') }}
    • +
    • Explicit dialogue chunks: {{ stats.get('explicit_chunks', '—') }}
    • +
    • Active speakers: {{ stats.get('active_speakers', '—') }}
    • +
    • Unique speakers observed: {{ stats.get('unique_speakers', '—') }}
    • +
    • Suppressed speakers: {{ stats.get('suppressed', 0) }}
    • +
    +
    +
    +

    Detected speakers

    + {% set speakers = analysis.get('speakers', {}) %} + {% set narrator_id = analysis.get('narrator', 'narrator') %} + {% if speakers %} +
      + {% for speaker_id, payload in speakers.items() if speaker_id != narrator_id and not payload.get('suppressed') %} + {% set spoken_name = payload.get('pronunciation') or payload.get('label') or speaker_id|replace('_', ' ')|title %} + {% set preview_text = preview_template | replace("{{name}}", spoken_name) %} +
    • +
      + {{ payload.get('label', speaker_id|replace('_', ' ')|title) }} + +
      +
      + {{ payload.get('count', 0) }} chunks + Confidence: {{ payload.get('confidence', 'low')|title }} + {% if payload.get('pronunciation') %} + Pronunciation: {{ payload.get('pronunciation') }} + {% endif %} +
      + {% set quotes = payload.get('sample_quotes', []) %} + {% if quotes %} +
      + Sample quotes +
        + {% for quote in quotes %} +
      • {{ quote }}
      • + {% endfor %} +
      +
      + {% endif %} +
    • + {% endfor %} +
    + {% else %} +

    No additional speakers detected.

    + {% endif %} + {% set suppressed = analysis.get('suppressed_details') or analysis.get('suppressed', []) %} + {% if suppressed %} +

    + Suppressed speakers: + {% if suppressed[0] is string %} + {{ suppressed | join(', ') }} + {% else %} + {{ suppressed | map(attribute='label') | join(', ') }} + {% endif %} +

    + {% endif %} +
    +
    +
    +{% endif %} +
    {% include "partials/logs.html" %}
    {% endblock %} + +{% block scripts %} + {{ super() }} + +{% endblock %} diff --git a/abogen/web/templates/prepare_job.html b/abogen/web/templates/prepare_job.html index 7d57da6..9aa99aa 100644 --- a/abogen/web/templates/prepare_job.html +++ b/abogen/web/templates/prepare_job.html @@ -33,7 +33,88 @@
    Chapter intro delay
    {{ '%.1f'|format(pending.chapter_intro_delay) }} seconds
    +
    +
    Chunk granularity
    +
    {{ pending.chunk_level|replace('_', ' ')|title }}
    +
    +
    +
    Speaker mode
    +
    {{ pending.speaker_mode|replace('_', ' ')|title }}
    +
    +
    +
    Speaker analysis threshold
    +
    {{ pending.speaker_analysis_threshold }} {{ 'mention' if pending.speaker_analysis_threshold == 1 else 'mentions' }}
    +
    +
    +
    EPUB 3 package
    +
    {% if pending.generate_epub3 %}Enabled{% else %}Disabled{% endif %}
    +
    + {% set analysis = pending.speaker_analysis or {} %} + {% set analysis_speakers = analysis.get('speakers', {}) %} + {% if analysis_speakers %} + {% set active = namespace(items=[]) %} + {% for sid, payload in analysis_speakers.items() %} + {% if sid != 'narrator' and not payload.get('suppressed') %} + {% set _ = active.items.append(payload) %} + {% endif %} + {% endfor %} +
    +

    Detected speakers

    + {% if active.items %} +
      + {% for speaker in active.items|sort(attribute='label') %} +
    • {{ speaker.label }} · {{ speaker.count }} lines · confidence {{ speaker.confidence|title }}
    • + {% endfor %} +
    + {% else %} +

    No additional speakers met the threshold yet. All dialogue will use the narrator voice.

    + {% endif %} +
    + {% endif %} + {% set roster = pending.speakers or {} %} + {% if roster %} + {% set preview_template = options.speaker_pronunciation_sentence or "This is {{name}} speaking." %} +
    +

    Speaker pronunciation guide

    +

    Add a phonetic spelling (IPA or plain text) so pronunciations sound right. Leave blank to use the written label.

    +
      + {% for speaker_id, speaker in roster.items() %} + {% set spoken_name = speaker.pronunciation or speaker.label %} + {% set preview_text = preview_template | replace("{{name}}", spoken_name) %} +
    • +
      + {{ speaker.label }} + +
      + + {% if speaker.get('analysis_count') %} +

      {{ speaker.analysis_count }} detected lines · confidence {{ speaker.analysis_confidence|default('low')|title }}

      + {% endif %} +
    • + {% endfor %} +
    +
    + {% endif %} {% if pending.metadata_tags %}
    +
    + + +

    Paragraphs work well for long-form narration; sentences give finer subtitle sync.

    +
    +
    + + +
    +
    + + +

    Only speakers that appear at least this many times will keep unique voices in multi-speaker mode.

    +

    Set to 0 to disable the pause after speaking each chapter title.

    +
    + +
    @@ -127,5 +236,6 @@ {% block scripts %} {{ super() }} + {% endblock %} diff --git a/abogen/web/templates/settings.html b/abogen/web/templates/settings.html index f792090..c775efc 100644 --- a/abogen/web/templates/settings.html +++ b/abogen/web/templates/settings.html @@ -73,6 +73,10 @@ Save as Project With Metadata +
    @@ -83,6 +87,32 @@
    + + +
    +
    + + +
    +
    + + +

    Speakers detected fewer times than this fallback to the narrator voice.

    +
    +
    + + +

    Sentence template used when previewing name pronunciation. Include {{ '{{name}}' }} where the speaker name should be inserted.

    +
    +
    diff --git a/docs/epub3_upgrade_plan.md b/docs/epub3_upgrade_plan.md new file mode 100644 index 0000000..d343a5a --- /dev/null +++ b/docs/epub3_upgrade_plan.md @@ -0,0 +1,208 @@ +# EPUB 3 Upgrade Plan + +## Overview +Elevate Abogen to produce rich EPUB 3 packages with synchronized narration, configurable TTS chunking, and groundwork for multi-speaker voice assignment. This document records the objectives, architectural adjustments, data model changes, UI flows, and implementation phases required to deliver the upgrade. + +## Goals +- Generate EPUB 3 output that preserves source metadata and embeds audio narration via media overlays. +- Allow users to choose the chunking granularity (paragraph vs. sentence) used for TTS synthesis and media-overlay alignment. +- Introduce speaker assignments for every chunk, starting with a single narrator but paving the way for multi-speaker control. +- Prototype practical, lightweight strategies for detecting likely speakers and estimating their dialogue frequency. + +## Non-goals / Out-of-scope +- Full multi-speaker editing UI (beyond gating the option). +- Automatic voice-casting or LLM-based dialogue attribution. +- Desktop GUI resurrection (web UI remains primary). + +## Current Architecture Snapshot +| Area | Notes | +| --- | --- | +| Text ingestion | `abogen/text_extractor.py` outputs `ExtractionResult` with chapter-level text. +| Job prep UI | `web/routes.py` builds `PendingJob` objects and renders chapter selection. +| Audio pipeline | `web/conversion_runner.py` creates per-job audio artifacts; chunking is effectively paragraph-level. +| Metadata | `ExtractionResult.metadata` feeds into FF metadata and output tagging, but not yet into EPUB packaging. + +## Feature 1 – EPUB 3 Output with Narration +### Requirements +- Preserve original EPUB metadata (Dublin Core entries, TOC, cover art). +- Package synthesized audio and SMIL media overlays aligned to chosen chunk granularity. +- Provide EPUB as an additional selectable output alongside current audio/subtitle formats. + +### Proposed Components +1. **`abogen/epub3/exporter.py`** (new module) + - Responsibilities: build XHTML spine with IDs, generate overlay SMIL files, write OPF manifest/spine, assemble zip package. + - Status: **Implemented** — `build_epub3_package` emits EPUB 3 archives with media overlays driven by chunk metadata. + - Dependencies: reuse `ebooklib` for reading source metadata; use `zipfile` for packaging; optional `lxml` for DOM manipulation. +2. **`EPUB3PackageBuilder` class** + - Inputs: extraction payload, chunk collection (with IDs, speaker mapping, timing metadata), audio asset paths, source metadata. + - Outputs: path to generated EPUB. +3. **Metadata preservation** + - Copy from source `ExtractionResult.metadata` and EPUB navigation if available. + - Ensure custom fields (e.g., chapter count) survive. +4. **Media overlay generation** + - Create one SMIL per content doc or per chapter, depending on chunk count. + - `` nodes reference chunk IDs and audio clip times. +5. **Configuration surface** + - Add “EPUB 3 (audio + text)” to output format selector (or a dedicated toggle under project settings). + +### Data Flow +``` +extract_from_path -> Chapter payload + |-> chunker (sentence/paragraph) + |-> chunk IDs + audio segments (timestamps from runner) +Conversion runner -> audio files + timing index +EPUB3PackageBuilder -> manifest, spine, SMIL, zip +``` + +### Open Questions +- Should we embed audio inside the EPUB or link externally? (Plan: embed to comply with spec.) +- How to handle very large audio assets? Consider splitting per chapter to keep file sizes manageable. + +## Feature 2 – Configurable Chunking +### Requirements +- Users select chunking level (paragraph or sentence) before audio generation. +- Pipeline produces stable, unique IDs for each chunk regardless of level. +- Provide chunk metadata (text, speaker, offsets) to both TTS and EPUB exporter. + +### Proposed Architecture +1. **Chunk Model** + ```python + @dataclass + class Chunk: + id: str + chapter_index: int + order: int + level: Literal["paragraph", "sentence"] + text: str + speaker_id: str + approx_characters: int + ``` +2. **Chunker Service (`abogen/chunking.py`)** + - Accepts chapter text and desired level. + - Uses spaCy (already bundled via `en-core-web-sm`) for sentence segmentation; fallback to regex when model unavailable. + - Emits `Chunk` objects with deterministic IDs (e.g., `chap{chapter_index:04d}_para{paragraph_idx:03d}_sent{sentence_idx:03d}`). +3. **Integration points** + - `web/routes.py` -> apply chunker when building `PendingJob` instead of storing raw paragraphs only. + - `PendingJob` / `Job` dataclasses -> include `chunks` list and `chunk_level` enum. + - `conversion_runner` -> iterate over `chunks` when synthesizing audio, producing per-chunk audio and capturing actual duration for overlay. +4. **Settings persistence** + - Extend config with `chunking_level` default; expose in UI (radio buttons or select). + +### Testing +- Unit tests for chunk splitting across languages, punctuation, abbreviations. +- Property-based tests ensuring concatenated chunks reproduce original text (except whitespace normalization). + +## Feature 3 – Speaker Assignment Foundations +### Requirements +- Every chunk must carry a `speaker_id` (default `narrator`). +- UI offers new option: “Single Speaker” (proceeds) vs. “Multi-Speaker (Coming Soon)” (blocks and shows message). +- Data model anticipates future multi-speaker support. + +### Implementation Outline +1. **Data Model Changes** + - `Chunk.speaker_id` default `"narrator"`. + - `PendingJob` & `Job` store `speakers` metadata (dictionary of speaker descriptors). + - `JobResult` optionally includes `chunk_speakers.json` artifact for downstream use. +2. **UI Adjustments** + - On upload form (`index.html` / JS), add selector for speaker mode. + - If “Multi-Speaker” chosen, display tooltip/modal: “Coming soon; please choose Single Speaker to continue.” disable submission. + - In `prepare_job.html`, display speaker info column (read-only for now). +3. **Serialization** + - Update JSON API routes to include speaker data. + - Update queue/job detail templates to show chunk level & speaker summary. + +### Testing +- Add web route tests ensuring multi-speaker path blocks progression. +- Verify job persistence includes `speaker_id` fields. + +## Feature 4 – Speaker Detection Strategies +### Objectives +Build groundwork for lightweight, deterministic speaker inference to inform future multi-speaker mode. + +### User Stories +1. **As a producer**, I can run an automated analysis on a book to see the list of likely speakers and how often they talk, so I can decide where multiple voices make sense. + - _Acceptance_: System outputs a JSON report containing speaker IDs/names, occurrence counts, representative excerpts, and confidence tier. Report stored with job artifacts and downloadable from job detail page. +2. **As a producer**, I can set a minimum occurrence threshold so that infrequent speakers automatically fall back to the narrator voice. + - _Acceptance_: Analysis respects configurable threshold; speakers below it are tagged as `default_narrator` in the report. +3. **As a developer/operator**, I can trigger the analysis via CLI or background task without blocking the main conversion pipeline. + - _Acceptance_: Command `abogen analyze-speakers ` (or background queue hook) runs in isolation, returns exit code 0 on success, emits metrics/logs for CI. + +### Strategy Ideas +1. **Quotation-bound heuristic** + - Split paragraphs on dialogue quotes. + - Use verb cues ("said", "asked") to associate names preceding/following quotes. +2. **Name detection via NER** + - Use spaCy’s entity recognition to spot `PERSON` entities inside dialogue spans. + - Maintain frequency counts per name. +3. **Speaker dictionary** + - Pre-build mapping of common narrator cues ("he said", "Mary replied") to propagate speaker assignment across adjacent sentences. +4. **Pronoun fallback with gender hints** + - Map pronouns to most recent speaker mention; degrade gracefully when ambiguous. +5. **Thresholding mechanism** + - After counting occurrences, expose a threshold slider (future UI) to decide when to allocate unique voices vs. default narrator. +6. **Diagnostics** + - Provide summary report: top N speaker candidates, counts, unresolved dialogue segments. + +### Implementation Staging +1. **Phase 1 – Analysis Engine (Backend)** + - Build `speaker_analysis.py` module implementing heuristics, returning structured results. + - Add CLI entry point `abogen-speaker-analyze` for standalone runs. + - Persist analysis artifacts (`speakers.json`, `speaker_excerpts.csv`) alongside job data when invoked post-extraction. + - Tests: unit tests for heuristic functions; snapshot tests for sample novels. +2. **Phase 2 – Configuration & Thresholding** + - Extend settings UI with optional “speaker analysis threshold” control (numeric). + - Update analysis module to accept threshold; mark low-frequency speakers as narrator. + - Emit summary digest (top speakers, narrator fallback count) in job logs. +3. **Phase 3 – UI Surfacing** + - Display analysis summary on job detail page (charts/table). + - Offer download link for raw JSON/CSV artifacts. + - Provide warning banner when analysis confidence is low (e.g., high unmatched dialogue percentage). +4. **Phase 4 – Integration Hooks** + - Wire analysis output into chunk speaker assignments (without yet enabling multi-speaker playback). + - Store mapping in `Job.speakers` metadata for future voice routing. + +### Technical Notes +- Reuse spaCy `en_core_web_sm` for entity recognition; allow pluggable models per language. +- Maintain rolling context window to resolve pronouns (e.g., last two named speakers). +- Provide instrumentation (timings, counts) to assess heuristic accuracy on sample corpora. +- Design analysis output schema versioning (`speaker_analysis_version`) to support iterative improvements. + +## UI & Configuration Updates +| Screen | Update | +| --- | --- | +| Upload form (`index.html`) | Add chunking level selector and speaker mode buttons. | +| Prepare job (`prepare_job.html`) | Display chunk level, IDs, speaker column; allow future editing hooks. | +| Settings modal | Persist defaults for chunking level and speaker mode. | + +## Data Model Checklist +- [x] Update `PendingJob` and `Job` dataclasses with `chunk_level`, `chunks`, `speakers` metadata. +- [x] Ensure serialization persists these fields in queue state file. +- [x] Persist chunk timing metadata from TTS (start/end timestamps). + +## Testing Strategy +- Unit tests for chunker and speaker heuristics. +- Integration tests: enqueue job with sentence-level chunking, assert chunk IDs and speaker metadata. +- Regression tests: ensure existing paragraph-level jobs still succeed. +- Acceptance tests for EPUB exporter: validate manifest, spine, and SMIL structure against schema (use `epubcheck` in CI if feasible). + +## Migration & Compat +- Bump state version in `ConversionService` when augmenting job schema; include migration logic for legacy queues. +- Provide CLI flag to reprocess older jobs without speaker metadata. +- Document new dependencies (e.g., `lxml`, optional spaCy models for languages beyond English). + +## Implementation Phases +1. **Foundation** – Introduce chunk model, chunker service, speaker defaults. +2. **Pipeline integration** – Update job lifecycle and TTS runner to work with chunks. +3. **EPUB exporter** – Build packaging module, connect to pipeline. +4. **UI polish** – Expose settings, guard multi-speaker path, surface diagnostics. +5. **Speaker analysis tool** – Prototype heuristics and reporting. + +## Open Questions +- How to handle non-EPUB inputs (PDF/TXT) when exporting EPUB 3? (Possible: generate synthetic XHTML with normalized chapters.) +- Storage impact of embedding per-chunk audio – do we need compression or streaming strategies? +- Internationalization: sentence segmentation quality varies; need language-specific models. + +## Next Steps +- Review plan with stakeholders for scope confirmation. +- Break down Phase 1 into actionable tickets (chunker, data model migration, UI toggle). +- Estimate resource requirements for EPUB packaging and testing (including epubcheck integration). diff --git a/tests/test_chunk_helpers.py b/tests/test_chunk_helpers.py new file mode 100644 index 0000000..70755cc --- /dev/null +++ b/tests/test_chunk_helpers.py @@ -0,0 +1,39 @@ +from __future__ import annotations + +from types import SimpleNamespace + +from abogen.web.conversion_runner import _chunk_voice_spec, _group_chunks_by_chapter + + +def test_group_chunks_by_chapter_orders_and_groups() -> None: + chunks = [ + {"chapter_index": "0", "chunk_index": "5", "text": "tail"}, + {"chapter_index": 0, "chunk_index": 1, "text": "body"}, + {"chapter_index": 1, "chunk_index": 0, "text": "next"}, + ] + + grouped = _group_chunks_by_chapter(chunks) + + assert [entry["text"] for entry in grouped[0]] == ["body", "tail"] + assert grouped[1][0]["text"] == "next" + + +def test_chunk_voice_spec_prefers_chunk_overrides() -> None: + job = SimpleNamespace(voice="base_voice", speakers={}) + chunk = {"voice": "override_voice", "speaker_id": "narrator"} + + assert _chunk_voice_spec(job, chunk, "fallback") == "override_voice" + + +def test_chunk_voice_spec_falls_back_to_speaker_voice() -> None: + job = SimpleNamespace(voice="base_voice", speakers={"narrator": {"voice": "speaker_voice"}}) + chunk = {"speaker_id": "narrator"} + + assert _chunk_voice_spec(job, chunk, "fallback") == "speaker_voice" + + +def test_chunk_voice_spec_uses_fallback_when_no_overrides() -> None: + job = SimpleNamespace(voice="base_voice", speakers={}) + chunk = {"speaker_id": "unknown"} + + assert _chunk_voice_spec(job, chunk, "fallback") == "fallback" diff --git a/tests/test_epub_exporter.py b/tests/test_epub_exporter.py new file mode 100644 index 0000000..ad9bfc2 --- /dev/null +++ b/tests/test_epub_exporter.py @@ -0,0 +1,107 @@ +from __future__ import annotations + +import zipfile + +from abogen.epub3.exporter import build_epub3_package +from abogen.text_extractor import ExtractedChapter, ExtractionResult + + +def _make_sample_extraction() -> ExtractionResult: + return ExtractionResult( + chapters=[ + ExtractedChapter(title="Chapter 1", text="Hello world."), + ExtractedChapter(title="Chapter 2", text="Another passage."), + ], + metadata={"title": "Sample Book", "artist": "Test Author", "language": "en"}, + ) + + +def test_build_epub3_package_creates_expected_structure(tmp_path) -> None: + extraction = _make_sample_extraction() + chunks = [ + { + "id": "chap0000_p0000", + "chapter_index": 0, + "chunk_index": 0, + "text": "Hello world.", + "speaker_id": "narrator", + }, + { + "id": "chap0001_p0000", + "chapter_index": 1, + "chunk_index": 0, + "text": "Another passage.", + "speaker_id": "narrator", + }, + ] + chunk_markers = [ + {"id": "chap0000_p0000", "chapter_index": 0, "chunk_index": 0, "start": 0.0, "end": 1.2}, + {"id": "chap0001_p0000", "chapter_index": 1, "chunk_index": 0, "start": 1.2, "end": 2.4}, + ] + chapter_markers = [ + {"index": 1, "title": "Chapter 1", "start": 0.0, "end": 1.2}, + {"index": 2, "title": "Chapter 2", "start": 1.2, "end": 2.4}, + ] + metadata_tags = {"title": "Sample Book", "artist": "Test Author", "language": "en"} + + audio_path = tmp_path / "sample.mp3" + audio_path.write_bytes(b"ID3 test audio") + + output_path = tmp_path / "output.epub" + result_path = build_epub3_package( + output_path=output_path, + book_id="job-123", + extraction=extraction, + metadata_tags=metadata_tags, + chapter_markers=chapter_markers, + chunk_markers=chunk_markers, + chunks=chunks, + audio_path=audio_path, + speaker_mode="single", + ) + + assert result_path == output_path + assert output_path.exists() + + with zipfile.ZipFile(output_path) as archive: + names = set(archive.namelist()) + assert "mimetype" in names + assert archive.read("mimetype") == b"application/epub+zip" + assert "META-INF/container.xml" in names + assert "OEBPS/content.opf" in names + assert "OEBPS/nav.xhtml" in names + assert "OEBPS/audio/sample.mp3" in names + chapter_doc = archive.read("OEBPS/text/chapter_0001.xhtml").decode("utf-8") + assert "Hello world." in chapter_doc + smil_doc = archive.read("OEBPS/smil/chapter_0001.smil").decode("utf-8") + assert "clipBegin=\"00:00:00.000\"" in smil_doc + opf_doc = archive.read("OEBPS/content.opf").decode("utf-8") + assert "media-overlay" in opf_doc + assert "media:duration" in opf_doc + assert "abogen:speakerMode" in opf_doc + + +def test_build_epub3_package_handles_missing_markers(tmp_path) -> None: + extraction = _make_sample_extraction() + metadata_tags = {"title": "Sample Book", "artist": "Test Author", "language": "en"} + audio_path = tmp_path / "audio.mp3" + audio_path.write_bytes(b"ID3 audio") + output_path = tmp_path / "output.epub" + + result_path = build_epub3_package( + output_path=output_path, + book_id="job-456", + extraction=extraction, + metadata_tags=metadata_tags, + chapter_markers=[], + chunk_markers=[], + chunks=[], + audio_path=audio_path, + speaker_mode="single", + ) + + with zipfile.ZipFile(result_path) as archive: + nav_doc = archive.read("OEBPS/nav.xhtml").decode("utf-8") + assert "Chapter 1" in nav_doc + chapter_doc = archive.read("OEBPS/text/chapter_0001.xhtml").decode("utf-8") + assert "Hello world." in chapter_doc \ No newline at end of file diff --git a/tests/test_service.py b/tests/test_service.py index a744617..4c449b2 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -54,4 +54,8 @@ def test_service_processes_job(tmp_path): assert runner_invocations, "conversion runner was never called" assert job.status is JobStatus.COMPLETED assert job.progress == 1.0 - assert job.result.audio_path == outputs / f"{job.id}.wav" \ No newline at end of file + assert job.result.audio_path == outputs / f"{job.id}.wav" + assert job.chunk_level == "paragraph" + assert job.speaker_mode == "single" + assert job.chunks == [] + assert not job.generate_epub3 \ No newline at end of file