diff options
| -rw-r--r-- | README.md | 2 | ||||
| -rw-r--r-- | converter/converter.py | 63 | ||||
| -rw-r--r-- | tests/test_converter.py | 127 |
3 files changed, 189 insertions, 3 deletions
@@ -66,6 +66,8 @@ qwen-tts-demo Qwen/Qwen3-TTS-12Hz-1.7B-Base --ip 127.0.0.1 --port 7861 Put your book files (epub, etc.) in the `input/` folder. Then run the script. The output goes to `output/`. +If a book's output files already exist in `output/`, you are asked at startup whether to reconvert and overwrite them; answer `n` to skip that book. All questions are asked before conversion begins, so the run is unattended once started. + ### Custom voice ```bash diff --git a/converter/converter.py b/converter/converter.py index c0e97e4..a58fc33 100644 --- a/converter/converter.py +++ b/converter/converter.py @@ -1,5 +1,6 @@ """Orchestrates book-to-audiobook conversion.""" +import glob import logging import re import sys @@ -39,6 +40,48 @@ def setup_directories() -> None: Path(directory).mkdir(parents=True, exist_ok=True) +def find_existing_outputs(output_name: str, output_format: str) -> List[Path]: + """Return existing output files that a conversion would overwrite. + + Multi-section books (e.g. EPUB chapters) and speed-adjusted copies are + named ``{name}_suffix.{ext}``; exact chapter file names are only known + after text extraction, so any file matching that pattern counts. + """ + folder = config.AUDIOBOOKS_FOLDER + existing: List[Path] = [] + primary = folder / f"{output_name}.{output_format}" + if primary.exists(): + existing.append(primary) + existing.extend(sorted( + folder.glob(f"{glob.escape(output_name)}_*.{output_format}"))) + return existing + + +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. + """ + if len(existing) == 1: + message = f"{existing[0].name} already exists. Convert anyway and overwrite it?" + else: + message = (f"{len(existing)} output files for '{output_name}' already exist " + f"(e.g. {existing[0].name}). Convert anyway and overwrite them?") + while True: + try: + answer = input(f"{message} (y/n): ").strip().lower() + except EOFError: + print("\n[WARNING] No interactive input available; keeping existing output") + return False + if answer in ("y", "yes"): + return True + if answer in ("n", "no"): + return False + print("Please answer 'y' or 'n'.") + + class AudiobookConverter: """Audiobook converter using the Qwen TTS API.""" @@ -336,12 +379,28 @@ class AudiobookConverter: # Avoid output collisions when two books share a stem (e.g. dune.txt + dune.epub). stem_counts: Dict[str, int] = Counter(book_file.stem for book_file in book_files) - # Convert each book - results = {} + # Ask every overwrite question up front, before any conversion + # starts, so the rest of the run is unattended. + planned: List[Tuple[Path, str]] = [] 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('.')}" + existing = find_existing_outputs(output_name, self.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)) + + if not planned: + print("[INFO] Nothing to convert (all books skipped)") + return True + + print(f"[INFO] Converting {len(planned)} of {len(book_files)} book(s)") + + # Convert each book + results = {} + for book_file, output_name in planned: try: success = self.convert_book(book_file, output_name=output_name) results[book_file.name] = success diff --git a/tests/test_converter.py b/tests/test_converter.py index fce0439..91ce699 100644 --- a/tests/test_converter.py +++ b/tests/test_converter.py @@ -1,8 +1,12 @@ """Tests for the audiobook converter orchestration helpers.""" +import tempfile import unittest +from pathlib import Path +from unittest.mock import patch -from converter.converter import AudiobookConverter +from converter import config +from converter.converter import AudiobookConverter, find_existing_outputs, prompt_overwrite class SanitizeFilenameTests(unittest.TestCase): @@ -31,5 +35,126 @@ class ConfigurationValidationTests(unittest.TestCase): AudiobookConverter(output_format="wma") +class FindExistingOutputsTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.folder = Path(self._tmp.name) + self._original = config.AUDIOBOOKS_FOLDER + config.AUDIOBOOKS_FOLDER = self.folder + + def tearDown(self): + config.AUDIOBOOKS_FOLDER = self._original + self._tmp.cleanup() + + def _touch(self, name): + path = self.folder / name + path.write_bytes(b"x") + return path + + def test_no_existing_output(self): + self.assertEqual(find_existing_outputs("dune", "mp3"), []) + + def test_primary_output_detected(self): + self._touch("dune.mp3") + self.assertEqual([p.name for p in find_existing_outputs("dune", "mp3")], + ["dune.mp3"]) + + def test_chapter_and_speed_copies_detected(self): + for name in ("dune_01_Dune.mp3", "dune_02_Barony.mp3", "dune_1.5x.mp3"): + self._touch(name) + self._touch("dune2_01.mp3") # different book stem; must not match + found = [p.name for p in find_existing_outputs("dune", "mp3")] + self.assertEqual(len(found), 3) + + def test_other_extensions_ignored(self): + self._touch("dune.mp3") + self.assertEqual(find_existing_outputs("dune", "m4b"), []) + + def test_glob_metacharacters_in_stem(self): + self._touch("book [1].mp3") + self._touch("book [1]_1.5x.mp3") + found = [p.name for p in find_existing_outputs("book [1]", "mp3")] + self.assertEqual(sorted(found), ["book [1].mp3", "book [1]_1.5x.mp3"]) + + +class PromptOverwriteTests(unittest.TestCase): + def test_single_file_yes(self): + with patch("builtins.input", return_value="y"): + self.assertTrue(prompt_overwrite([Path("dune.mp3")], "dune")) + + def test_single_file_no(self): + with patch("builtins.input", return_value="n"): + self.assertFalse(prompt_overwrite([Path("dune.mp3")], "dune")) + + def test_accepts_full_words(self): + with patch("builtins.input", return_value="yes"): + self.assertTrue(prompt_overwrite([Path("dune.mp3")], "dune")) + with patch("builtins.input", return_value="No"): + 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: + self.assertFalse(prompt_overwrite([Path("dune.mp3")], "dune")) + self.assertEqual(mock_input.call_count, 3) + + def test_eof_keeps_existing_output(self): + with patch("builtins.input", side_effect=EOFError): + self.assertFalse(prompt_overwrite([Path("dune.mp3")], "dune")) + + def test_multiple_files_prompt_names_them(self): + files = [Path("dune_01_Dune.mp3"), Path("dune_02_Barony.mp3")] + with patch("builtins.input", return_value="y") as mock_input: + self.assertTrue(prompt_overwrite(files, "dune")) + prompt_text = mock_input.call_args[0][0] + self.assertIn("2 output files for 'dune'", prompt_text) + self.assertIn("dune_01_Dune.mp3", prompt_text) + self.assertIn("overwrite them", prompt_text) + + +class RunOverwritePromptTests(unittest.TestCase): + """The full run() flow: prompts collected before any conversion starts.""" + + def setUp(self): + self._books_tmp = tempfile.TemporaryDirectory() + self._output_tmp = tempfile.TemporaryDirectory() + self._original_folders = (config.BOOKS_FOLDER, config.AUDIOBOOKS_FOLDER) + config.BOOKS_FOLDER = Path(self._books_tmp.name) + config.AUDIOBOOKS_FOLDER = Path(self._output_tmp.name) + (config.BOOKS_FOLDER / "book.txt").write_text("hello world", encoding="utf-8") + self.converter = AudiobookConverter.__new__(AudiobookConverter) + self.converter.voice_mode = config.VOICE_MODE_CUSTOM + self.converter.voice_clone_ref_audio = None + self.converter.speed = 1.0 + self.converter.single_file = False + self.converter.output_format = "mp3" + self.converted = [] + self.converter.convert_book = ( + lambda file_path, output_name=None: + not self.converted.append((file_path.name, output_name)) or True) + + def tearDown(self): + config.BOOKS_FOLDER, config.AUDIOBOOKS_FOLDER = self._original_folders + self._books_tmp.cleanup() + self._output_tmp.cleanup() + + def test_declined_book_is_skipped(self): + (config.AUDIOBOOKS_FOLDER / "book.mp3").write_bytes(b"existing") + with patch("builtins.input", return_value="n"): + self.assertTrue(self.converter.run()) + self.assertEqual(self.converted, []) + self.assertTrue((config.AUDIOBOOKS_FOLDER / "book.mp3").exists()) + + def test_accepted_book_is_converted(self): + (config.AUDIOBOOKS_FOLDER / "book.mp3").write_bytes(b"existing") + with patch("builtins.input", return_value="y"): + self.assertTrue(self.converter.run()) + self.assertEqual(self.converted, [("book.txt", "book")]) + + def test_new_book_converted_without_prompt(self): + with patch("builtins.input", side_effect=AssertionError("should not prompt")): + self.assertTrue(self.converter.run()) + self.assertEqual(self.converted, [("book.txt", "book")]) + + if __name__ == "__main__": unittest.main() |