diff options
| -rwxr-xr-x | audiobook.py | 22 | ||||
| -rw-r--r-- | converter/converter.py | 113 | ||||
| -rw-r--r-- | tests/test_converter.py | 57 |
3 files changed, 158 insertions, 34 deletions
diff --git a/audiobook.py b/audiobook.py index b881a12..0b3734c 100755 --- a/audiobook.py +++ b/audiobook.py @@ -227,6 +227,26 @@ Examples: else: voice_mode = VOICE_MODE_CLONE if args.clone else VOICE_MODE_CUSTOM + # Ask every overwrite question up front, before spending time connecting + # to a TTS server: a user who declines (or has nothing to convert) never + # waits on a slow server handshake. Nothing in this step needs the server. + book_files, planned = AudiobookConverter.preflight_overwrites( + backend=args.backend, + voice=args.voice, + voice_mode=voice_mode, + voice_clone_ref_audio=args.clone, + output_format=args.format, + ) + + if not book_files: + print("[INFO] Nothing to convert. Add a .txt, .pdf, or .epub file " + "to the input folder and run again.") + sys.exit(0) + + if not planned: + print("[INFO] Nothing to convert (all books skipped)") + sys.exit(0) + try: converter = AudiobookConverter( voice_mode=voice_mode, @@ -242,6 +262,8 @@ Examples: debug=args.debug, chunk=args.chunk, ) + converter._book_files = book_files + converter._planned = planned ok = converter.run() except KeyboardInterrupt: print("\n[WARNING] Shutdown requested by user") diff --git a/converter/converter.py b/converter/converter.py index f1064f1..84ac766 100644 --- a/converter/converter.py +++ b/converter/converter.py @@ -106,8 +106,9 @@ def prompt_overwrite(existing: List[Path], output_name: str) -> bool: """Ask whether to reconvert a book whose output files already exist. All overwrite questions are asked before any conversion starts so the - rest of the run is unattended. Returns False when no interactive input - is available (stdin closed), keeping existing files safe. + rest of the run is unattended. Pressing Enter defaults to yes (so a + user can just hit Enter through the prompts), but a closed stdin + (non-interactive run) declines and keeps existing files safe. """ if len(existing) == 1: message = f"{existing[0].name} already exists. Convert anyway and overwrite it?" @@ -116,15 +117,17 @@ def prompt_overwrite(existing: List[Path], output_name: str) -> bool: f"(e.g. {existing[0].name}). Convert anyway and overwrite them?") while True: try: - answer = input(f"{message} (y/n): ").strip().lower() + answer = input(f"{message} [Y/n]: ").strip().lower() except EOFError: print("\n[WARNING] No interactive input available; keeping existing output") return False + if not answer: + return True if answer in ("y", "yes"): return True if answer in ("n", "no"): return False - print("Please answer 'y' or 'n'.") + print("Please answer 'y' or 'n' (or press Enter for yes).") class AudiobookConverter: @@ -207,23 +210,36 @@ class AudiobookConverter: return cleaned[:80] or fallback def _narrator_tag(self) -> str: - """Narrator name used in output file names. + """Narrator name used in output file names (see compute_narrator_tag).""" + return self.compute_narrator_tag( + self.backend, self.voice, self.voice_mode, self.voice_clone_ref_audio) + + @staticmethod + def compute_narrator_tag(backend: str, voice: Optional[str], + voice_mode: str, + voice_clone_ref_audio: Optional[str]) -> str: + """Narrator name used in output file names, without a server connection. Custom voice mode uses the built-in speaker's display name; voice clone mode uses the reference audio file's stem; the faster and audiocpp backends use the server-side voice name (falling back to the built-in speaker for the audiocpp backend's speaker mode). Spaces become underscores (e.g. "Uncle Fu" -> "Uncle_Fu"). + + Pure (no I/O, no server) so the pre-flight overwrite check can + compute the exact output names a run would produce before spending + time connecting to a TTS server. """ - if self.backend == BACKEND_FASTER: - narrator = self.voice or config.FASTER_VOICE - elif self.backend == BACKEND_AUDIOCPP: - narrator = self.voice or speaker_display_name() - elif self.voice_mode == VOICE_MODE_CLONE: - narrator = Path(self.voice_clone_ref_audio).stem + if backend == BACKEND_FASTER: + narrator = voice or config.FASTER_VOICE + elif backend == BACKEND_AUDIOCPP: + narrator = voice or speaker_display_name() + elif voice_mode == VOICE_MODE_CLONE: + narrator = Path(voice_clone_ref_audio).stem else: narrator = speaker_display_name() - return self._sanitize_filename(narrator, fallback="narrator").replace(" ", "_") + return AudiobookConverter._sanitize_filename( + narrator, fallback="narrator").replace(" ", "_") # ------------------------------------------------------------------ # Debug dumps (--debug) @@ -580,30 +596,33 @@ class AudiobookConverter: print(f"Debug dumps (per-chunk text + raw audio): {DEBUG_FOLDER}") print("=" * 70) - def run(self) -> bool: - """Main conversion process. Returns True if all books converted.""" - run_start = time.time() - self._print_banner() + # ------------------------------------------------------------------ + # Pre-flight: overwrite checks before connecting to a TTS server + # ------------------------------------------------------------------ + @staticmethod + def preflight_overwrites(backend: str, voice: Optional[str], + voice_mode: str, + voice_clone_ref_audio: Optional[str], + output_format: str) -> Tuple[List[Path], + List[Tuple[Path, str]]]: + """Discover books and ask every overwrite question up front. + + Pure of the TTS server: it scans the books folder, computes the + output name each book would produce (including the narrator tag + and stem-collision suffix), and asks whether to overwrite any + existing output files. Returns ``(book_files, planned)`` where + ``planned`` is the subset the user agreed to (re)convert. + + Asking before connecting means a user who declines a prompt (or + has nothing to convert) never waits on a slow server handshake. + """ book_files = sorted( f for f in BOOKS_FOLDER.iterdir() if f.is_file() and f.suffix.lower() in SUPPORTED_FORMATS ) - if not book_files: - print(f"[INFO] No supported files found in {BOOKS_FOLDER}") - print(f"Supported formats: {', '.join(SUPPORTED_FORMATS)}") - - # Create sample file - sample_file = BOOKS_FOLDER / "sample.txt" - sample_file.write_text( - "This is a sample audiobook for testing the Qwen-based converter. " - "The system will send this text to the Qwen API for voice generation. " - "You can replace this file with your own books to convert.", - encoding="utf-8", - ) - print(f"[INFO] Created sample file: {sample_file}") - return True + return [], [] print(f"[INFO] Found {len(book_files)} books to convert") @@ -613,16 +632,46 @@ class AudiobookConverter: # Ask every overwrite question up front, before any conversion # starts, so the rest of the run is unattended. planned: List[Tuple[Path, str]] = [] + narrator_tag = AudiobookConverter.compute_narrator_tag( + backend, voice, voice_mode, voice_clone_ref_audio) for book_file in book_files: output_name = book_file.stem if stem_counts[book_file.stem] > 1: output_name = f"{book_file.stem}_{book_file.suffix.lstrip('.')}" - output_name = f"{output_name}_{self._narrator_tag()}" - existing = find_existing_outputs(output_name, self.output_format) + output_name = f"{output_name}_{narrator_tag}" + existing = find_existing_outputs(output_name, output_format) if existing and not prompt_overwrite(existing, output_name): print(f"[INFO] Skipping {book_file.name} (existing output kept)") continue planned.append((book_file, output_name)) + return book_files, planned + + # ------------------------------------------------------------------ + # Main conversion loop + # ------------------------------------------------------------------ + + def run(self) -> bool: + """Main conversion process. Returns True if all books converted.""" + run_start = time.time() + self._print_banner() + + # When main() has already done the pre-flight overwrite check, use + # its results so the prompts are not asked a second time; otherwise + # (e.g. a converter constructed directly) discover and ask here. + if getattr(self, "_planned", None) is not None: + book_files = self._book_files + planned = self._planned + else: + book_files, planned = AudiobookConverter.preflight_overwrites( + self.backend, self.voice, self.voice_mode, + self.voice_clone_ref_audio, self.output_format) + + if not book_files: + print(f"[INFO] No supported files found in {BOOKS_FOLDER}") + print(f"Supported formats: {', '.join(SUPPORTED_FORMATS)}") + print("[INFO] Nothing to convert. Add a .txt, .pdf, or .epub file " + f"to {BOOKS_FOLDER} and run again.") + return True if not planned: print("[INFO] Nothing to convert (all books skipped)") diff --git a/tests/test_converter.py b/tests/test_converter.py index 5ef9785..8402b30 100644 --- a/tests/test_converter.py +++ b/tests/test_converter.py @@ -358,9 +358,15 @@ class PromptOverwriteTests(unittest.TestCase): self.assertFalse(prompt_overwrite([Path("dune.mp3")], "dune")) def test_invalid_answer_reasked(self): - with patch("builtins.input", side_effect=["maybe", "", "n"]) as mock_input: + with patch("builtins.input", side_effect=["maybe", "n"]) as mock_input: self.assertFalse(prompt_overwrite([Path("dune.mp3")], "dune")) - self.assertEqual(mock_input.call_count, 3) + self.assertEqual(mock_input.call_count, 2) + + def test_empty_answer_defaults_yes(self): + # Pressing Enter (empty input) accepts the default of yes, matching + # the make_audiocpp_server_json tool's ask_bool(default=True) prompt. + with patch("builtins.input", return_value=""): + self.assertTrue(prompt_overwrite([Path("dune.mp3")], "dune")) def test_eof_keeps_existing_output(self): with patch("builtins.input", side_effect=EOFError): @@ -376,6 +382,53 @@ class PromptOverwriteTests(unittest.TestCase): self.assertIn("overwrite them", prompt_text) +class PreflightOverwritesTests(unittest.TestCase): + """The pre-flight overwrite check runs without a TTS server connection.""" + + def setUp(self): + self._books_tmp = tempfile.TemporaryDirectory() + self._output_tmp = tempfile.TemporaryDirectory() + self._original_folders = (converter_mod.BOOKS_FOLDER, converter_mod.AUDIOBOOKS_FOLDER) + converter_mod.BOOKS_FOLDER = Path(self._books_tmp.name) + converter_mod.AUDIOBOOKS_FOLDER = Path(self._output_tmp.name) + (converter_mod.BOOKS_FOLDER / "book.txt").write_text("hello world", encoding="utf-8") + + def tearDown(self): + converter_mod.BOOKS_FOLDER, converter_mod.AUDIOBOOKS_FOLDER = self._original_folders + self._books_tmp.cleanup() + self._output_tmp.cleanup() + + def test_no_books_returns_empty(self): + (converter_mod.BOOKS_FOLDER / "book.txt").unlink() + with patch("builtins.input", side_effect=AssertionError("should not prompt")): + book_files, planned = AudiobookConverter.preflight_overwrites( + tts.BACKEND_GRADIO, None, tts.VOICE_MODE_CUSTOM, None, "mp3") + self.assertEqual(book_files, []) + self.assertEqual(planned, []) + + def test_new_book_planned_without_prompt(self): + with patch("builtins.input", side_effect=AssertionError("should not prompt")): + book_files, planned = AudiobookConverter.preflight_overwrites( + tts.BACKEND_GRADIO, None, tts.VOICE_MODE_CUSTOM, None, "mp3") + self.assertEqual(len(book_files), 1) + self.assertEqual(planned, [(book_files[0], "book_Vivian")]) + + def test_existing_output_enter_defaults_yes(self): + (converter_mod.AUDIOBOOKS_FOLDER / "book_Vivian.mp3").write_bytes(b"existing") + with patch("builtins.input", return_value=""): + book_files, planned = AudiobookConverter.preflight_overwrites( + tts.BACKEND_GRADIO, None, tts.VOICE_MODE_CUSTOM, None, "mp3") + self.assertEqual(planned, [(book_files[0], "book_Vivian")]) + + def test_existing_output_declined_is_skipped(self): + (converter_mod.AUDIOBOOKS_FOLDER / "book_Vivian.mp3").write_bytes(b"existing") + with patch("builtins.input", return_value="n"): + book_files, planned = AudiobookConverter.preflight_overwrites( + tts.BACKEND_GRADIO, None, tts.VOICE_MODE_CUSTOM, None, "mp3") + self.assertEqual(len(book_files), 1) + self.assertEqual(planned, []) + + class RunOverwritePromptTests(unittest.TestCase): """The full run() flow: prompts collected before any conversion starts.""" |