diff options
Diffstat (limited to 'converter/converter.py')
| -rw-r--r-- | converter/converter.py | 106 |
1 files changed, 63 insertions, 43 deletions
diff --git a/converter/converter.py b/converter/converter.py index 0454892..d5a1096 100644 --- a/converter/converter.py +++ b/converter/converter.py @@ -5,9 +5,10 @@ import re import sys import time import traceback +from collections import Counter from datetime import datetime from pathlib import Path -from typing import Dict, Optional +from typing import Dict, List, Optional from . import audio, chunking, config, extractors from .tts import QwenTTSClient @@ -41,7 +42,7 @@ def setup_directories() -> None: class AudiobookConverter: """Audiobook converter using the Qwen TTS API.""" - def __init__(self, voice_mode: str = "custom_voice", voice_clone_ref_audio: Optional[str] = None, + def __init__(self, voice_mode: str = config.VOICE_MODE_CUSTOM, voice_clone_ref_audio: Optional[str] = None, voice_clone_ref_text: Optional[str] = None, skip_transcription: bool = False, speed: float = 1.0, single_file: bool = False, output_format: str = "mp3"): if speed <= 0: @@ -63,7 +64,12 @@ class AudiobookConverter: def _validate_configuration(self) -> None: """Validate configuration settings.""" - if self.voice_mode == "voice_clone": + if self.voice_mode not in config.VOICE_MODES: + raise ValueError( + f"Unknown voice mode: {self.voice_mode!r} " + f"(expected one of {config.VOICE_MODES})" + ) + if self.voice_mode == config.VOICE_MODE_CLONE: if not self.voice_clone_ref_audio: raise ValueError( "Voice Clone mode requires a reference audio file. " @@ -117,7 +123,7 @@ class AudiobookConverter: 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 + success = self._convert_text(section.text, output_path, time.time()) and success return success except Exception as exc: @@ -139,7 +145,7 @@ class AudiobookConverter: titles = [] for index, section in enumerate(sections, 1): chapter_path = config.CHUNKS_FOLDER / f"chapter_{index:04d}.wav" - if not self._convert_text(section.text, chapter_path, start_time, + if not self._convert_text(section.text, chapter_path, time.time(), speed=1.0, output_format="wav"): logger.warning("Skipping chapter %d (%s) due to conversion failure", index, section.title) @@ -152,7 +158,50 @@ class AudiobookConverter: 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) + if not audio.combine_chapters_to_m4b(chapter_files, titles, output_path, speed=self.speed): + return False + duration = time.time() - start_time + logger.info("Conversion completed in %dm %ds: %s", + int(duration // 60), int(duration % 60), output_path) + print(f"[SUCCESS] Conversion completed in {int(duration // 60)}m {int(duration % 60)}s") + return True + + def _synthesize_chunks(self, chunks: List[str]) -> Dict[int, Optional[Path]]: + """Synthesize chunks sequentially, preserving order and naming. + + Returns a mapping of chunk number to the generated audio path, with + None for chunks that failed after retries. + """ + total_chunks = len(chunks) + print(f"\n{'=' * 50}") + print(f"PROCESSING {total_chunks} CHUNKS") + print(f"{'=' * 50}") + + results: Dict[int, Optional[Path]] = {} + for chunk_num, chunk_text in enumerate(chunks, 1): + try: + result = self.tts.process_chunk_with_retry(chunk_num, chunk_text) + results[chunk_num] = result + + if result: + print(f"[OK] Chunk {chunk_num:3d}/{total_chunks} completed") + logger.info("+ Chunk %d/%d completed", chunk_num, total_chunks) + else: + print(f"[FAIL] Chunk {chunk_num:3d}/{total_chunks} FAILED") + logger.error("- Chunk %d/%d failed", chunk_num, total_chunks) + + except Exception as exc: + results[chunk_num] = None + print(f"[ERROR] Chunk {chunk_num:3d}/{total_chunks} ERROR: {exc}") + logger.error("- Chunk %d/%d error: %s", chunk_num, total_chunks, exc) + + successful_chunks = sum(1 for path in results.values() if path) + print(f"\n{'=' * 50}") + print("CHUNK PROCESSING COMPLETE") + print(f"Successful: {successful_chunks}/{total_chunks}") + print(f"{'=' * 50}") + logger.info("Qwen processing completed: %d/%d chunks", successful_chunks, total_chunks) + return results def _convert_text(self, text: str, output_path: Path, start_time: float, speed: Optional[float] = None, @@ -184,36 +233,8 @@ class AudiobookConverter: print(f"[INFO] Processing {total_chunks} chunks via Qwen API...") print(f"[INFO] Estimated time: ~{total_chunks * 4} minutes (4 min per chunk)") - print(f"\n{'=' * 50}") - print(f"PROCESSING {total_chunks} CHUNKS") - print(f"{'=' * 50}") - - # Process chunks sequentially to ensure correct order and naming: - # chunks are named 1, 2, 3, 4... in order. - results = {} # chunk_num -> success (bool) - for chunk_num, chunk_text in enumerate(chunks, 1): - try: - result = self.tts.process_chunk_with_retry(chunk_num, chunk_text) - results[chunk_num] = result - - if result: - print(f"[OK] Chunk {chunk_num:3d}/{total_chunks} completed") - logger.info("+ Chunk %d/%d completed", chunk_num, total_chunks) - else: - print(f"[FAIL] Chunk {chunk_num:3d}/{total_chunks} FAILED") - logger.error("- Chunk %d/%d failed", chunk_num, total_chunks) - - except Exception as exc: - results[chunk_num] = False - print(f"[ERROR] Chunk {chunk_num:3d}/{total_chunks} ERROR: {exc}") - logger.error("- Chunk %d/%d error: %s", chunk_num, total_chunks, exc) - - successful_chunks = sum(1 for v in results.values() if v) - print(f"\n{'=' * 50}") - print("CHUNK PROCESSING COMPLETE") - print(f"Successful: {successful_chunks}/{total_chunks}") - print(f"{'=' * 50}") - logger.info("Qwen processing completed: %d/%d chunks", successful_chunks, total_chunks) + results = self._synthesize_chunks(chunks) + successful_chunks = sum(1 for path in results.values() if path) if successful_chunks == 0: logger.error("No chunks were successfully processed") @@ -224,7 +245,7 @@ class AudiobookConverter: successful_chunks, total_chunks) # Combine chunks (only the successful ones) - success = audio.combine_chunks(total_chunks, output_path, results, + success = audio.combine_chunks(total_chunks, output_path, chunk_results=results, speed=speed, output_format=output_format) if success: @@ -245,7 +266,7 @@ class AudiobookConverter: 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 + api_url = config.VOICE_CLONE_API_URL if self.voice_mode == config.VOICE_MODE_CLONE else config.QWEN_API_URL print("=" * 70) print("QWEN-BASED AUDIOBOOK CONVERTER") @@ -255,10 +276,10 @@ class AudiobookConverter: print(f"Qwen API endpoint: {api_url}") print(f"Voice mode: {self.voice_mode}") print("Model size: 1.7B (always)") - if self.voice_mode == "custom_voice": + if self.voice_mode == config.VOICE_MODE_CUSTOM: print(f"Speaker: {config.CUSTOM_VOICE_SPEAKER}") print(f"Language: {config.CUSTOM_VOICE_LANGUAGE}") - elif self.voice_mode == "voice_clone": + elif self.voice_mode == config.VOICE_MODE_CLONE: print(f"Reference audio: {Path(self.voice_clone_ref_audio).name}") print(f"Language: {config.VOICE_CLONE_LANGUAGE}") print(f"Output format: {self.output_format}") @@ -292,9 +313,7 @@ class AudiobookConverter: 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 + stem_counts: Dict[str, int] = Counter(book_file.stem for book_file in book_files) # Convert each book results = {} @@ -307,6 +326,7 @@ class AudiobookConverter: results[book_file.name] = success except KeyboardInterrupt: print("\n[WARNING] Conversion interrupted by user") + results[book_file.name] = False break except Exception as exc: logger.error("Unexpected error: %s", exc) |