diff options
Diffstat (limited to 'converter/converter.py')
| -rw-r--r-- | converter/converter.py | 87 |
1 files changed, 54 insertions, 33 deletions
diff --git a/converter/converter.py b/converter/converter.py index 30418e5..8371c02 100644 --- a/converter/converter.py +++ b/converter/converter.py @@ -14,10 +14,32 @@ from typing import Dict, List, Optional, Tuple from . import audio, chunking, config, cover, extractors from .audio import TrackMeta -from .tts import FasterTTSClient, QwenTTSClient, normalize_language, speaker_display_name +from .tts import ( + VOICE_MODE_CLONE, + VOICE_MODE_CUSTOM, + VOICE_MODES, + FasterTTSClient, + QwenTTSClient, + normalize_language, + speaker_display_name, +) logger = logging.getLogger(__name__) +# Folders, resolved from the project root so the converter runs from any +# working directory. +BASE_DIR = Path(__file__).resolve().parent.parent + +BOOKS_FOLDER = BASE_DIR / "input" +AUDIOBOOKS_FOLDER = BASE_DIR / "output" +CHUNKS_FOLDER = BASE_DIR / "chunks" # Per-chunk scratch audio, cleaned per book +LOGS_FOLDER = BASE_DIR / "logs" +DEBUG_FOLDER = BASE_DIR / "debug" # --debug dumps, kept across runs + +# Output containers and supported input formats. +AUDIO_FORMATS = ("mp3", "m4b", "ogg", "flac") +SUPPORTED_FORMATS = [".txt", ".pdf", ".epub"] + def _console_log_filter(record: logging.LogRecord) -> bool: """Keep httpx/httpcore request logs out of the console (file only).""" @@ -32,9 +54,9 @@ def setup_logging(debug: bool = False) -> None: (DEBUG with --debug) so progress prints are never mirrored as timestamped log lines; httpx/httpcore request logs stay file-only. """ - config.LOGS_FOLDER.mkdir(parents=True, exist_ok=True) + LOGS_FOLDER.mkdir(parents=True, exist_ok=True) file_handler = logging.FileHandler( - config.LOGS_FOLDER / f"audiobook_{datetime.now():%Y%m%d}.log", + LOGS_FOLDER / f"audiobook_{datetime.now():%Y%m%d}.log", encoding="utf-8", ) file_handler.setLevel(logging.DEBUG if debug else logging.INFO) @@ -52,8 +74,8 @@ def setup_logging(debug: bool = False) -> None: def setup_directories() -> None: """Create necessary directories.""" - for directory in (config.BOOKS_FOLDER, config.AUDIOBOOKS_FOLDER, - config.CHUNKS_FOLDER, config.LOGS_FOLDER): + for directory in (BOOKS_FOLDER, AUDIOBOOKS_FOLDER, + CHUNKS_FOLDER, LOGS_FOLDER): Path(directory).mkdir(parents=True, exist_ok=True) @@ -64,7 +86,7 @@ def find_existing_outputs(output_name: str, output_format: str) -> List[Path]: 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 + folder = AUDIOBOOKS_FOLDER existing: List[Path] = [] primary = folder / f"{output_name}.{output_format}" if primary.exists(): @@ -102,18 +124,17 @@ def prompt_overwrite(existing: List[Path], output_name: str) -> bool: class AudiobookConverter: """Audiobook converter using the Qwen TTS API.""" - def __init__(self, voice_mode: str = config.VOICE_MODE_CUSTOM, voice_clone_ref_audio: Optional[str] = None, + def __init__(self, voice_mode: str = 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 = config.AUDIO_FORMAT, language: Optional[str] = None, faster: bool = False, faster_voice: Optional[str] = None, debug: bool = False): if speed <= 0: raise ValueError(f"Speed must be a positive number, got {speed}") - if output_format not in config.AUDIO_FORMATS: + if output_format not in AUDIO_FORMATS: raise ValueError(f"Unsupported output format: {output_format}") if language is None: - language = (config.VOICE_CLONE_LANGUAGE if voice_mode == config.VOICE_MODE_CLONE - else config.CUSTOM_VOICE_LANGUAGE) + language = config.LANGUAGE self.language = normalize_language(language) self.voice_mode = voice_mode self.voice_clone_ref_audio = voice_clone_ref_audio @@ -139,12 +160,12 @@ class AudiobookConverter: def _validate_configuration(self) -> None: """Validate configuration settings.""" - if self.voice_mode not in config.VOICE_MODES: + if self.voice_mode not in VOICE_MODES: raise ValueError( f"Unknown voice mode: {self.voice_mode!r} " - f"(expected one of {config.VOICE_MODES})" + f"(expected one of {VOICE_MODES})" ) - if self.voice_mode == config.VOICE_MODE_CLONE and not self.faster: + if self.voice_mode == VOICE_MODE_CLONE and not self.faster: if not self.voice_clone_ref_audio: raise ValueError( "Voice Clone mode requires a reference audio file. " @@ -173,7 +194,7 @@ class AudiobookConverter: """ if self.faster: narrator = self.faster_voice or config.FASTER_TTS_VOICE - elif self.voice_mode == config.VOICE_MODE_CLONE: + elif self.voice_mode == VOICE_MODE_CLONE: narrator = Path(self.voice_clone_ref_audio).stem else: narrator = speaker_display_name() @@ -243,13 +264,13 @@ class AudiobookConverter: stem = output_name or f"{file_path.stem}_{self._narrator_tag()}" # --debug: chunk text/audio dumps land in a per-book folder - debug_dir = config.DEBUG_FOLDER / stem if self.debug else None + debug_dir = DEBUG_FOLDER / stem if self.debug else None # Cover art: generated once per book. Named with the chunk_ # prefix so cleanup_chunks() removes it with the other scratch # files at the end of the book. cover_path = cover.generate_cover( - book.title, config.CHUNKS_FOLDER / "chunk_cover.png") + book.title, CHUNKS_FOLDER / "chunk_cover.png") if cover_path: print(f"[INFO] Generated cover art for '{book.title}'") meta = TrackMeta(title=book.title, artist=book.author, album=book.title) @@ -261,20 +282,20 @@ class AudiobookConverter: return self._convert_m4b_with_chapters(sections, stem, start_time, meta=meta, cover=cover_path, debug_dir=debug_dir) - output_path = config.AUDIOBOOKS_FOLDER / f"{stem}.{self.output_format}" + output_path = AUDIOBOOKS_FOLDER / f"{stem}.{self.output_format}" return self._convert_text(sections[0].text, output_path, start_time, meta=meta, cover=cover_path, debug_dir=debug_dir) 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}" + output_path = AUDIOBOOKS_FOLDER / f"{stem}.{self.output_format}" return self._convert_text(text, output_path, start_time, meta=meta, cover=cover_path, debug_dir=debug_dir) 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}" + output_path = AUDIOBOOKS_FOLDER / f"{chapter_name}.{self.output_format}" track_meta = meta._replace( title=(section.title or "").strip() or f"Chapter {index}", track=index, total_tracks=len(sections)) @@ -309,7 +330,7 @@ class AudiobookConverter: titles = [] total_chapters = len(sections) for index, section in enumerate(sections, 1): - chapter_path = config.CHUNKS_FOLDER / f"chapter_{index:04d}.wav" + chapter_path = CHUNKS_FOLDER / f"chapter_{index:04d}.wav" title = (section.title or "").strip() or f"Chapter {index}" print(f"\n{'=' * 50}") print(f"CHAPTER {index}/{total_chapters}: {title}") @@ -329,7 +350,7 @@ class AudiobookConverter: logger.error("No chapters were successfully converted") return False - output_path = config.AUDIOBOOKS_FOLDER / f"{stem}.{self.output_format}" + output_path = AUDIOBOOKS_FOLDER / f"{stem}.{self.output_format}" if not audio.combine_chapters_to_m4b(chapter_files, titles, output_path, speed=self.speed, meta=meta, cover=cover): return False @@ -472,22 +493,22 @@ class AudiobookConverter: print("=" * 70) print("QWEN-BASED AUDIOBOOK CONVERTER") print("=" * 70) - print(f"Books folder: {config.BOOKS_FOLDER}") - print(f"Output folder: {config.AUDIOBOOKS_FOLDER}") + print(f"Books folder: {BOOKS_FOLDER}") + print(f"Output folder: {AUDIOBOOKS_FOLDER}") if self.faster: print(f"Faster TTS endpoint: {config.FASTER_TTS_API_URL}") print("Backend: faster (voice cloning, reference configured on server)") print(f"Voice: {self.faster_voice or config.FASTER_TTS_VOICE}") else: - api_url = (config.VOICE_CLONE_API_URL if self.voice_mode == config.VOICE_MODE_CLONE + api_url = (config.VOICE_CLONE_API_URL if self.voice_mode == VOICE_MODE_CLONE else config.QWEN_API_URL) print(f"Qwen API endpoint: {api_url}") print(f"Voice mode: {self.voice_mode}") print("Model size: 1.7B (always)") - if self.voice_mode == config.VOICE_MODE_CUSTOM: + if self.voice_mode == VOICE_MODE_CUSTOM: print(f"Speaker: {config.CUSTOM_VOICE_SPEAKER}") print(f"Language: {self.language}") - elif self.voice_mode == config.VOICE_MODE_CLONE: + elif self.voice_mode == VOICE_MODE_CLONE: print(f"Reference audio: {Path(self.voice_clone_ref_audio).name}") print(f"Language: {self.language}") print(f"Output format: {self.output_format}") @@ -496,7 +517,7 @@ class AudiobookConverter: if abs(self.speed - 1.0) >= 1e-6: print(f"Playback speed: {self.speed:g}x") if self.debug: - print(f"Debug dumps (per-chunk text + raw audio): {config.DEBUG_FOLDER}") + print(f"Debug dumps (per-chunk text + raw audio): {DEBUG_FOLDER}") print("=" * 70) def run(self) -> bool: @@ -506,16 +527,16 @@ class AudiobookConverter: # Check for books book_files = sorted( - f for f in config.BOOKS_FOLDER.iterdir() - if f.is_file() and f.suffix.lower() in config.SUPPORTED_FORMATS + 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 {config.BOOKS_FOLDER}") - print(f"Supported formats: {', '.join(config.SUPPORTED_FORMATS)}") + print(f"[INFO] No supported files found in {BOOKS_FOLDER}") + print(f"Supported formats: {', '.join(SUPPORTED_FORMATS)}") # Create sample file - sample_file = config.BOOKS_FOLDER / "sample.txt" + 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. " @@ -579,7 +600,7 @@ class AudiobookConverter: print(f"{status} {filename}") if successful > 0: - print(f"\n[INFO] Audiobooks saved to: {config.AUDIOBOOKS_FOLDER}/") + print(f"\n[INFO] Audiobooks saved to: {AUDIOBOOKS_FOLDER}/") elapsed = int(time.time() - run_start) hours, remainder = divmod(elapsed, 3600) |