diff options
Diffstat (limited to 'converter/converter.py')
| -rw-r--r-- | converter/converter.py | 90 |
1 files changed, 79 insertions, 11 deletions
diff --git a/converter/converter.py b/converter/converter.py index 65efb68..ce2c351 100644 --- a/converter/converter.py +++ b/converter/converter.py @@ -1,6 +1,7 @@ """Orchestrates book-to-audiobook conversion.""" import logging +import re import sys import time import traceback @@ -42,12 +43,16 @@ class AudiobookConverter: def __init__(self, voice_mode: str = "custom_voice", voice_clone_ref_audio: Optional[str] = None, voice_clone_ref_text: Optional[str] = None, skip_transcription: bool = False, - speed: float = 1.0): + speed: float = 1.0, single_file: bool = False, output_format: str = "mp3"): if speed <= 0: raise ValueError(f"Speed must be a positive number, got {speed}") + if output_format not in config.AUDIO_FORMATS: + raise ValueError(f"Unsupported output format: {output_format}") self.voice_mode = voice_mode self.voice_clone_ref_audio = voice_clone_ref_audio self.speed = speed + self.single_file = single_file + self.output_format = output_format self._validate_configuration() self.tts = QwenTTSClient( voice_mode=voice_mode, @@ -70,8 +75,15 @@ class AudiobookConverter: f"Reference audio file not found: {self.voice_clone_ref_audio}" ) + @staticmethod + def _sanitize_filename(name: str) -> str: + """Make a chapter title safe to use as part of a file name.""" + cleaned = re.sub(r'[\\/:*?"<>|]', " ", name) + cleaned = re.sub(r"\s+", " ", cleaned).strip().strip(".") + return cleaned[:80] or "chapter" + def convert_book(self, file_path: Path, output_name: Optional[str] = None) -> bool: - """Convert a single book to an audiobook.""" + """Convert a single book to one or more audiobook files.""" logger.info("Converting: %s", file_path.name) start_time = time.time() @@ -80,13 +92,70 @@ class AudiobookConverter: # affect this run audio.cleanup_chunks() - # Extract text logger.info("Extracting text...") - text = extractors.extract_text(file_path) - if not text.strip(): + sections = extractors.extract_sections(file_path) + if not sections or all(not s.text.strip() for s in sections): logger.error("No text extracted") return False + stem = output_name or file_path.stem + embed_chapters = (self.output_format == "m4b" and self.single_file + and len(sections) > 1) + if embed_chapters: + return self._convert_single_m4b_with_chapters(sections, stem, start_time) + + if self.single_file or len(sections) == 1: + text = "\n\n".join(section.text for section in sections) + output_path = config.AUDIOBOOKS_FOLDER / f"{stem}.{self.output_format}" + return self._convert_text(text, output_path, start_time) + + success = True + for index, section in enumerate(sections, 1): + chapter_name = f"{stem}_{index:02d}_{self._sanitize_filename(section.title)}" + output_path = config.AUDIOBOOKS_FOLDER / f"{chapter_name}.{self.output_format}" + success = self._convert_text(section.text, output_path, start_time) and success + return success + + except Exception as exc: + logger.error("Conversion failed: %s", exc) + logger.error(traceback.format_exc()) + return False + finally: + # Always cleanup, even on failure or interrupt + audio.cleanup_chunks() + + def _convert_single_m4b_with_chapters(self, sections, stem: str, start_time: float) -> bool: + """Convert each chapter to audio, then assemble a single m4b with + embedded chapter markers.""" + chapter_files = [] + titles = [] + for index, section in enumerate(sections, 1): + chapter_path = config.CHUNKS_FOLDER / f"chapter_{index:04d}.{self.output_format}" + if not self._convert_text(section.text, chapter_path, start_time, speed=1.0): + logger.warning("Skipping chapter %d (%s) due to conversion failure", + index, section.title) + continue + chapter_files.append(chapter_path) + titles.append(section.title or f"Chapter {index}") + + if not chapter_files: + logger.error("No chapters were successfully converted") + return False + + output_path = config.AUDIOBOOKS_FOLDER / f"{stem}.{self.output_format}" + return audio.combine_chapters_to_m4b(chapter_files, titles, output_path, speed=self.speed) + + def _convert_text(self, text: str, output_path: Path, start_time: float, + speed: Optional[float] = None) -> bool: + """Chunk, synthesize, and assemble ``text`` into ``output_path``.""" + if speed is None: + speed = self.speed + + try: + if not text.strip(): + logger.error("No text to convert for %s", output_path.name) + return False + logger.info("Extracted %d characters (%d words)", len(text), len(text.split())) # Split into chunks @@ -143,8 +212,8 @@ class AudiobookConverter: successful_chunks, total_chunks) # Combine chunks (only the successful ones) - output_path = config.AUDIOBOOKS_FOLDER / f"{output_name or file_path.stem}.{config.AUDIO_FORMAT}" - success = audio.combine_chunks(total_chunks, output_path, results, speed=self.speed) + success = audio.combine_chunks(total_chunks, output_path, results, + speed=speed, output_format=self.output_format) if success: duration = time.time() - start_time @@ -161,9 +230,6 @@ class AudiobookConverter: logger.error("Conversion failed: %s", exc) logger.error(traceback.format_exc()) return False - finally: - # Always cleanup, even on failure or interrupt - audio.cleanup_chunks() def run(self) -> bool: """Main conversion process. Returns True if all books converted.""" @@ -183,7 +249,9 @@ class AudiobookConverter: elif self.voice_mode == "voice_clone": print(f"Reference audio: {Path(self.voice_clone_ref_audio).name}") print(f"Language: {config.VOICE_CLONE_LANGUAGE}") - print(f"Output format: {config.AUDIO_FORMAT}") + print(f"Output format: {self.output_format}") + if self.single_file: + print("Chapter mode: single file (--single-file)") if abs(self.speed - 1.0) >= 1e-6: print(f"Playback speed: {self.speed:g}x") print("=" * 70) |