aboutsummaryrefslogtreecommitdiff
path: root/converter/converter.py
diff options
context:
space:
mode:
Diffstat (limited to 'converter/converter.py')
-rw-r--r--converter/converter.py46
1 files changed, 27 insertions, 19 deletions
diff --git a/converter/converter.py b/converter/converter.py
index eea84ed..65efb68 100644
--- a/converter/converter.py
+++ b/converter/converter.py
@@ -6,7 +6,7 @@ import time
import traceback
from datetime import datetime
from pathlib import Path
-from typing import Optional
+from typing import Dict, Optional
from . import audio, chunking, config, extractors
from .tts import QwenTTSClient
@@ -60,17 +60,17 @@ class AudiobookConverter:
"""Validate configuration settings."""
if self.voice_mode == "voice_clone":
if not self.voice_clone_ref_audio:
- print("[ERROR] Configuration Error!")
- print("Voice Clone mode requires a reference audio file.")
- print("Use --voice-sample <path> to specify the reference audio.")
- sys.exit(1)
+ raise ValueError(
+ "Voice Clone mode requires a reference audio file. "
+ "Use --voice-sample <path> to specify it."
+ )
if not Path(self.voice_clone_ref_audio).exists():
- print("[ERROR] Configuration Error!")
- print(f"Reference audio file not found: {self.voice_clone_ref_audio}")
- sys.exit(1)
+ raise ValueError(
+ f"Reference audio file not found: {self.voice_clone_ref_audio}"
+ )
- def convert_book(self, file_path: Path) -> bool:
+ def convert_book(self, file_path: Path, output_name: Optional[str] = None) -> bool:
"""Convert a single book to an audiobook."""
logger.info("Converting: %s", file_path.name)
start_time = time.time()
@@ -136,7 +136,6 @@ class AudiobookConverter:
if successful_chunks == 0:
logger.error("No chunks were successfully processed")
- audio.cleanup_chunks() # Cleanup even on failure
return False
if successful_chunks < total_chunks:
@@ -144,7 +143,7 @@ class AudiobookConverter:
successful_chunks, total_chunks)
# Combine chunks (only the successful ones)
- output_path = config.AUDIOBOOKS_FOLDER / f"{file_path.stem}.{config.AUDIO_FORMAT}"
+ 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)
if success:
@@ -156,19 +155,18 @@ class AudiobookConverter:
else:
logger.error("Failed to combine chunks into final audiobook")
- # Always cleanup, even on failure
- audio.cleanup_chunks()
return success
except Exception as exc:
logger.error("Conversion failed: %s", exc)
logger.error(traceback.format_exc())
- # Cleanup on exception
- audio.cleanup_chunks()
return False
+ finally:
+ # Always cleanup, even on failure or interrupt
+ audio.cleanup_chunks()
- def run(self) -> None:
- """Main conversion process."""
+ def run(self) -> bool:
+ """Main conversion process. Returns True if all books converted."""
api_url = config.VOICE_CLONE_API_URL if self.voice_mode == "voice_clone" else config.QWEN_API_URL
print("=" * 70)
@@ -209,15 +207,23 @@ class AudiobookConverter:
encoding="utf-8",
)
print(f"[INFO] Created sample file: {sample_file}")
- return
+ return True
print(f"[INFO] Found {len(book_files)} books to convert")
+ # Avoid output collisions when two books share a stem (e.g. dune.txt + dune.epub).
+ stem_counts: Dict[str, int] = {}
+ for book_file in book_files:
+ stem_counts[book_file.stem] = stem_counts.get(book_file.stem, 0) + 1
+
# Convert each book
results = {}
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('.')}"
try:
- success = self.convert_book(book_file)
+ success = self.convert_book(book_file, output_name=output_name)
results[book_file.name] = success
except KeyboardInterrupt:
print("\n[WARNING] Conversion interrupted by user")
@@ -242,3 +248,5 @@ class AudiobookConverter:
if successful > 0:
print(f"\n[INFO] Audiobooks saved to: {config.AUDIOBOOKS_FOLDER}/")
+
+ return total > 0 and successful == total