diff options
Diffstat (limited to 'converter')
| -rw-r--r-- | converter/__init__.py | 1 | ||||
| -rw-r--r-- | converter/audio.py | 117 | ||||
| -rw-r--r-- | converter/chunking.py | 59 | ||||
| -rw-r--r-- | converter/config.py | 80 | ||||
| -rw-r--r-- | converter/converter.py | 244 | ||||
| -rw-r--r-- | converter/extractors.py | 185 | ||||
| -rw-r--r-- | converter/tts.py | 311 |
7 files changed, 997 insertions, 0 deletions
diff --git a/converter/__init__.py b/converter/__init__.py new file mode 100644 index 0000000..80735d1 --- /dev/null +++ b/converter/__init__.py @@ -0,0 +1 @@ +"""Qwen-based audiobook converter package.""" diff --git a/converter/audio.py b/converter/audio.py new file mode 100644 index 0000000..ab55e9b --- /dev/null +++ b/converter/audio.py @@ -0,0 +1,117 @@ +"""Audio assembly: combining chunks, speed adjustment, cleanup.""" + +import logging +import traceback +from pathlib import Path +from typing import Dict, List, Optional + +from . import config + +logger = logging.getLogger(__name__) + + +def speed_export_params(speed: float) -> List[str]: + """Return ffmpeg filter args for pitch-preserving speed adjustment. + + Uses ffmpeg's atempo filter, which accepts 0.5..2.0 per filter. Values + outside that range are handled by chaining multiple atempo filters. + """ + if speed <= 0: + raise ValueError(f"Speed must be a positive number, got {speed}") + if abs(speed - 1.0) < 1e-6: + return [] + remaining = float(speed) + chain = [] + while remaining > 2.0: + chain.append("atempo=2.0") + remaining /= 2.0 + while remaining < 0.5: + chain.append("atempo=0.5") + remaining /= 0.5 + chain.append(f"atempo={remaining:g}") + return ["-filter:a", ",".join(chain)] + + +def combine_chunks(total_chunks: int, output_path: Path, + results: Optional[Dict[int, bool]] = None, speed: float = 1.0) -> bool: + """Combine audio chunks into the final audiobook. + + ``results`` maps chunk numbers to success flags; failed chunks are + skipped. When ``speed`` differs from 1.0, an additional speed-adjusted + copy is written next to the normal-speed file. + """ + try: + from pydub import AudioSegment + except ImportError: + logger.error("pydub is required to combine audio chunks (pip install pydub)") + return False + + try: + combined = AudioSegment.empty() + successful = 0 + missing_chunks = [] + + for i in range(1, total_chunks + 1): + # Skip chunks that failed if we have results tracking + if results is not None and not results.get(i, False): + missing_chunks.append(i) + continue + + chunk_file = config.CHUNKS_FOLDER / f"chunk_{i:04d}.wav" + if chunk_file.exists(): + try: + combined += AudioSegment.from_wav(str(chunk_file)) + successful += 1 + if successful % 10 == 0: + logger.info("Combined %d chunks", successful) + except Exception as exc: + logger.warning("Failed to load chunk %d: %s", i, exc) + missing_chunks.append(i) + else: + logger.warning("Chunk file not found: %s", chunk_file) + missing_chunks.append(i) + + if successful == 0: + raise RuntimeError("No valid chunks found") + + if missing_chunks: + logger.warning("Missing chunks: %s", missing_chunks) + + combined.export(str(output_path), format=config.AUDIO_FORMAT, bitrate=config.AUDIO_BITRATE) + logger.info("Audiobook saved: %s (%d/%d chunks)", output_path, successful, total_chunks) + print(f"[INFO] Saved audiobook: {output_path.name} ({successful}/{total_chunks} chunks)") + + export_params = speed_export_params(speed) + if export_params: + speed_path = output_path.with_name(f"{output_path.stem}_{speed:g}{output_path.suffix}") + combined.export(str(speed_path), format=config.AUDIO_FORMAT, + bitrate=config.AUDIO_BITRATE, parameters=export_params) + logger.info("Saved speed-adjusted audiobook (%gx): %s", speed, speed_path) + print(f"[INFO] Saved speed-adjusted audiobook: {speed_path.name} ({speed:g}x)") + + if missing_chunks: + print(f"[WARNING] Missing chunks: {missing_chunks}") + return True + + except Exception as exc: + logger.error("Failed to combine chunks: %s", exc) + logger.error(traceback.format_exc()) + return False + + +def cleanup_chunks() -> None: + """Remove temporary chunk files from the scratch folder.""" + try: + chunk_count = 0 + for chunk_file in config.CHUNKS_FOLDER.glob("chunk_*.wav"): + try: + chunk_file.unlink() + chunk_count += 1 + except Exception as exc: + logger.warning("Failed to delete %s: %s", chunk_file, exc) + + if chunk_count > 0: + logger.info("Cleaned up %d chunk files", chunk_count) + print(f"[INFO] Cleaned up {chunk_count} chunk files") + except Exception as exc: + logger.warning("Cleanup failed: %s", exc) diff --git a/converter/chunking.py b/converter/chunking.py new file mode 100644 index 0000000..f649310 --- /dev/null +++ b/converter/chunking.py @@ -0,0 +1,59 @@ +"""Split extracted book text into TTS-sized chunks.""" + +import re +from typing import List + +from . import config + + +def split_into_chunks(text: str, max_words: int = config.CHUNK_SIZE_WORDS) -> List[str]: + """Split text into chunks of at most ``max_words`` words. + + Splits on sentence boundaries. Sentences longer than the limit are split + further at clause punctuation (which is kept attached for TTS prosody). + A single sentence with no clause punctuation longer than the limit is + kept intact as one oversized chunk. + """ + if not text.strip(): + return [] + + sentences = re.split(r"(?<=[.!?])\s+", text) + chunks = [] + current_chunk = "" + current_words = 0 + + for sentence in sentences: + sentence_words = len(sentence.split()) + + if sentence_words > max_words: + if current_chunk: + chunks.append(current_chunk.strip()) + current_chunk = "" + current_words = 0 + + # Split long sentences at clause boundaries, keeping punctuation. + parts = re.split(r"(?<=[,;:])\s*", sentence) + for part in parts: + part_words = len(part.split()) + if current_words + part_words <= max_words: + current_chunk += part + " " + current_words += part_words + else: + if current_chunk: + chunks.append(current_chunk.strip()) + current_chunk = part + " " + current_words = part_words + else: + if current_words + sentence_words <= max_words: + current_chunk += sentence + " " + current_words += sentence_words + else: + if current_chunk: + chunks.append(current_chunk.strip()) + current_chunk = sentence + " " + current_words = sentence_words + + if current_chunk.strip(): + chunks.append(current_chunk.strip()) + + return [chunk for chunk in chunks if chunk.strip()] diff --git a/converter/config.py b/converter/config.py new file mode 100644 index 0000000..1b203d2 --- /dev/null +++ b/converter/config.py @@ -0,0 +1,80 @@ +"""Configuration for the audiobook converter. + +Edit these values to change the default voice and processing behavior. +All paths are resolved relative to the project root, so the converter can +be run from any working directory. +""" + +from pathlib import Path + +# Project root (directory containing audiobook_converter.py) +BASE_DIR = Path(__file__).resolve().parent.parent + +# ============================================================================= +# QWEN API CONFIGURATION +# ============================================================================= + +QWEN_API_URL = "http://127.0.0.1:7860" # CustomVoice demo endpoint +API_TIMEOUT = 300 # Seconds before an API call times out +MAX_RETRIES = 3 # Retry failed chunks + +# ============================================================================= +# CUSTOM VOICE SETTINGS (pre-built speakers, always uses the 1.7B model) +# ============================================================================= + +CUSTOM_VOICE_SPEAKER = "Vivian" +CUSTOM_VOICE_LANGUAGE = "English" +CUSTOM_VOICE_INSTRUCT = "Speak naturally and clearly, as if reading a dramatic book to an adult audience." +CUSTOM_VOICE_MODEL_SIZE = "1.7B" +CUSTOM_VOICE_SEED = -1 +CUSTOM_VOICE_MODEL_ID = "Qwen/Qwen3-TTS-12Hz-1.7B-CustomVoice" + +# Map canonical speaker names to the display names used by the qwen-tts Gradio demo. +SPEAKER_DISPLAY_NAMES = { + "ryan": "Ryan", + "serena": "Serena", + "vivian": "Vivian", + "uncle_fu": "Uncle Fu", + "aiden": "Aiden", + "ono_anna": "Ono Anna", + "sohee": "Sohee", + "eric": "Eric", + "dylan": "Dylan", +} + +# ============================================================================= +# VOICE CLONE SETTINGS (clone a voice from a reference audio file) +# ============================================================================= +# Voice clone requires the Base-model demo (Qwen3-TTS-12Hz-1.7B-Base), which +# exposes /run_voice_clone. The CustomVoice demo only exposes /run_instruct, so +# run the Base demo on a separate port and point this at it. + +VOICE_CLONE_LANGUAGE = "English" +VOICE_CLONE_USE_XVECTOR_ONLY = False +VOICE_CLONE_MODEL_SIZE = "1.7B" +VOICE_CLONE_MAX_CHUNK_CHARS = 200 +VOICE_CLONE_CHUNK_GAP = 0 +VOICE_CLONE_SEED = -1 +VOICE_CLONE_API_URL = "http://127.0.0.1:7861" + +# ============================================================================= +# PROCESSING SETTINGS +# ============================================================================= + +BOOKS_FOLDER = BASE_DIR / "book_to_convert" # Input folder +AUDIOBOOKS_FOLDER = BASE_DIR / "audiobooks" # Output folder +CHUNKS_FOLDER = BASE_DIR / "chunks" # Scratch space for per-chunk audio (cleaned per book) +LOGS_FOLDER = BASE_DIR / "logs" + +CHUNK_SIZE_WORDS = 1500 # Words per TTS chunk +MIN_DELAY_BETWEEN_CHUNKS = 1 # Seconds between API calls +HEARTBEAT_INTERVAL_SECONDS = 30 # Print "still working" this often during a chunk + +# ============================================================================= +# AUDIO OUTPUT SETTINGS +# ============================================================================= + +AUDIO_FORMAT = "mp3" +AUDIO_BITRATE = "128k" + +SUPPORTED_FORMATS = [".txt", ".pdf", ".epub"] diff --git a/converter/converter.py b/converter/converter.py new file mode 100644 index 0000000..eea84ed --- /dev/null +++ b/converter/converter.py @@ -0,0 +1,244 @@ +"""Orchestrates book-to-audiobook conversion.""" + +import logging +import sys +import time +import traceback +from datetime import datetime +from pathlib import Path +from typing import Optional + +from . import audio, chunking, config, extractors +from .tts import QwenTTSClient + +logger = logging.getLogger(__name__) + + +def setup_logging() -> None: + """Configure logging to both a dated file and the console.""" + config.LOGS_FOLDER.mkdir(parents=True, exist_ok=True) + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s - %(levelname)s - %(message)s", + handlers=[ + logging.FileHandler( + config.LOGS_FOLDER / f"audiobook_{datetime.now():%Y%m%d}.log", + encoding="utf-8", + ), + logging.StreamHandler(sys.stdout), + ], + ) + + +def setup_directories() -> None: + """Create necessary directories.""" + for directory in (config.BOOKS_FOLDER, config.AUDIOBOOKS_FOLDER, + config.CHUNKS_FOLDER, config.LOGS_FOLDER): + Path(directory).mkdir(parents=True, exist_ok=True) + + +class AudiobookConverter: + """Audiobook converter using the Qwen TTS API.""" + + def __init__(self, voice_mode: str = "custom_voice", voice_clone_ref_audio: Optional[str] = None, + voice_clone_ref_text: Optional[str] = None, skip_transcription: bool = False, + speed: float = 1.0): + if speed <= 0: + raise ValueError(f"Speed must be a positive number, got {speed}") + self.voice_mode = voice_mode + self.voice_clone_ref_audio = voice_clone_ref_audio + self.speed = speed + self._validate_configuration() + self.tts = QwenTTSClient( + voice_mode=voice_mode, + voice_clone_ref_audio=voice_clone_ref_audio, + voice_clone_ref_text=voice_clone_ref_text, + skip_transcription=skip_transcription, + ) + + def _validate_configuration(self) -> None: + """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) + + 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) + + def convert_book(self, file_path: Path) -> bool: + """Convert a single book to an audiobook.""" + logger.info("Converting: %s", file_path.name) + start_time = time.time() + + try: + # Start from a clean scratch folder so a previous crash can never + # affect this run + audio.cleanup_chunks() + + # Extract text + logger.info("Extracting text...") + text = extractors.extract_text(file_path) + if not text.strip(): + logger.error("No text extracted") + return False + + logger.info("Extracted %d characters (%d words)", len(text), len(text.split())) + + # Split into chunks + chunks = chunking.split_into_chunks(text) + total_chunks = len(chunks) + if total_chunks == 0: + logger.error("No chunks created") + return False + + # Log chunk info + chunk_sizes = [len(chunk.split()) for chunk in chunks] + avg_chunk_size = sum(chunk_sizes) / len(chunk_sizes) + logger.info("Split into %d chunks (avg %.0f words per chunk)", total_chunks, avg_chunk_size) + 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) + + 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: + logger.warning("Only %d/%d chunks succeeded. Proceeding with partial audiobook.", + successful_chunks, total_chunks) + + # Combine chunks (only the successful ones) + output_path = config.AUDIOBOOKS_FOLDER / f"{file_path.stem}.{config.AUDIO_FORMAT}" + success = audio.combine_chunks(total_chunks, output_path, results, speed=self.speed) + + if success: + duration = time.time() - start_time + minutes = int(duration // 60) + seconds = int(duration % 60) + logger.info("Conversion completed in %dm %ds: %s", minutes, seconds, output_path) + print(f"[SUCCESS] Conversion completed in {minutes}m {seconds}s") + 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 + + def run(self) -> None: + """Main conversion process.""" + api_url = config.VOICE_CLONE_API_URL if self.voice_mode == "voice_clone" else config.QWEN_API_URL + + 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"Qwen API endpoint: {api_url}") + print(f"Voice mode: {self.voice_mode}") + print("Model size: 1.7B (always)") + if self.voice_mode == "custom_voice": + print(f"Speaker: {config.CUSTOM_VOICE_SPEAKER}") + print(f"Language: {config.CUSTOM_VOICE_LANGUAGE}") + elif self.voice_mode == "voice_clone": + print(f"Reference audio: {Path(self.voice_clone_ref_audio).name}") + print(f"Language: {config.VOICE_CLONE_LANGUAGE}") + print(f"Output format: {config.AUDIO_FORMAT}") + if abs(self.speed - 1.0) >= 1e-6: + print(f"Playback speed: {self.speed:g}x") + print("=" * 70) + + # 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 + ) + + if not book_files: + print(f"[INFO] No supported files found in {config.BOOKS_FOLDER}") + print(f"Supported formats: {', '.join(config.SUPPORTED_FORMATS)}") + + # Create sample file + sample_file = config.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 + + print(f"[INFO] Found {len(book_files)} books to convert") + + # Convert each book + results = {} + for book_file in book_files: + try: + success = self.convert_book(book_file) + results[book_file.name] = success + except KeyboardInterrupt: + print("\n[WARNING] Conversion interrupted by user") + break + except Exception as exc: + logger.error("Unexpected error: %s", exc) + results[book_file.name] = False + + # Print summary + successful = sum(results.values()) + total = len(results) + + print("\n" + "=" * 70) + print("CONVERSION SUMMARY") + print("=" * 70) + print(f"Total: {total} | Success: {successful} | Failed: {total - successful}") + print("=" * 70) + + for filename, success in results.items(): + status = "[OK]" if success else "[FAIL]" + print(f"{status} {filename}") + + if successful > 0: + print(f"\n[INFO] Audiobooks saved to: {config.AUDIOBOOKS_FOLDER}/") diff --git a/converter/extractors.py b/converter/extractors.py new file mode 100644 index 0000000..e49b48a --- /dev/null +++ b/converter/extractors.py @@ -0,0 +1,185 @@ +"""Text extraction from book files (TXT, PDF, EPUB) and text/HTML cleaning.""" + +import logging +import re +import zipfile +from html import unescape +from pathlib import Path + +try: + from bs4 import BeautifulSoup + BS4_AVAILABLE = True +except ImportError: + BS4_AVAILABLE = False + +logger = logging.getLogger(__name__) + + +def extract_text(file_path: Path) -> str: + """Extract text from a book file based on its extension.""" + extension = file_path.suffix.lower() + if extension == ".txt": + return _extract_txt(file_path) + if extension == ".pdf": + return _extract_pdf(file_path) + if extension == ".epub": + return extract_epub(file_path) + raise ValueError(f"Unsupported file format: {extension}") + + +def clean_text(text: str) -> str: + """Normalize whitespace and strip standalone page numbers. + + Page numbers are removed only when they appear as a short number alone on + its own line (before whitespace collapsing), so inline numbers like + "42 years", "1,000" or "3.5" are preserved. + """ + if not text: + return "" + # Standalone page numbers (digits alone on a line) must go BEFORE the + # newline-collapsing step below. + text = re.sub(r"(?m)^\s*\d{1,4}\s*$", " ", text) + text = re.sub(r"\s+", " ", text) + return text.strip() + + +def clean_html(html_content: str) -> str: + """Strip markup, scripts and styles from HTML content.""" + if not html_content: + return "" + + if BS4_AVAILABLE: + try: + soup = BeautifulSoup(html_content, "html.parser") + for tag in soup(["script", "style"]): + tag.decompose() + text = soup.get_text() + lines = (line.strip() for line in text.splitlines()) + chunks = (phrase.strip() for line in lines for phrase in line.split(" ")) + return " ".join(chunk for chunk in chunks if chunk) + except Exception as exc: + logger.debug("BeautifulSoup cleaning failed, falling back to regex: %s", exc) + + # Fallback regex cleaning + html_content = re.sub(r"<style[^>]*>.*?</style>", "", html_content, flags=re.DOTALL | re.IGNORECASE) + html_content = re.sub(r"<script[^>]*>.*?</script>", "", html_content, flags=re.DOTALL | re.IGNORECASE) + html_content = re.sub(r"<[^>]+>", " ", html_content) + html_content = unescape(html_content) + html_content = re.sub(r"\s+", " ", html_content) + return html_content.strip() + + +def extract_epub(file_path: Path) -> str: + """Extract text from EPUB, trying several methods in order.""" + methods = [ + _extract_epub_ebooklib, + _extract_epub_zipfile, + _extract_epub_manual, + ] + + for method in methods: + try: + text = method(file_path) + if text and text.strip(): + logger.info("EPUB extraction successful (%s): %d characters", method.__name__, len(text)) + return text + except Exception as exc: + logger.warning("EPUB method %s failed: %s", method.__name__, exc) + + raise RuntimeError("All EPUB extraction methods failed") + + +def _extract_epub_ebooklib(file_path: Path) -> str: + """Extract using ebooklib, following the spine (reading) order.""" + import ebooklib + from ebooklib import epub + + book = epub.read_epub(str(file_path)) + text_parts = [] + + for entry in book.spine: + item_id = entry[0] if isinstance(entry, (tuple, list)) else entry + try: + item = book.get_item_with_id(item_id) + if item and item.get_type() == ebooklib.ITEM_DOCUMENT: + content = item.get_body_content() + if content: + if isinstance(content, bytes): + content = content.decode("utf-8", errors="ignore") + cleaned = clean_html(str(content)) + if cleaned.strip(): + text_parts.append(cleaned) + except Exception as exc: + logger.debug("Skipping EPUB spine item %r: %s", item_id, exc) + + return "\n\n".join(text_parts) + + +def _extract_epub_zipfile(file_path: Path) -> str: + """Extract by parsing HTML members of the EPUB zip directly.""" + text_parts = [] + with zipfile.ZipFile(file_path, "r") as epub_zip: + for file_name in sorted(epub_zip.namelist()): + if file_name.lower().endswith((".html", ".xhtml", ".htm")): + try: + content = epub_zip.read(file_name).decode("utf-8", errors="ignore") + cleaned = clean_html(content) + if cleaned.strip(): + text_parts.append(cleaned) + except Exception as exc: + logger.debug("Skipping EPUB member %r: %s", file_name, exc) + return "\n\n".join(text_parts) + + +def _extract_epub_manual(file_path: Path) -> str: + """Last-resort extraction from any markup-looking EPUB member.""" + skipped_extensions = (".jpg", ".jpeg", ".png", ".gif", ".css", ".js") + text_parts = [] + with zipfile.ZipFile(file_path, "r") as epub_zip: + for file_name in sorted(epub_zip.namelist()): + if file_name.lower().endswith(skipped_extensions): + continue + try: + content = epub_zip.read(file_name).decode("utf-8", errors="ignore") + if "<" in content and len(content.strip()) > 100: + cleaned = clean_html(content) + if cleaned: + text_parts.append(cleaned) + except Exception as exc: + logger.debug("Skipping EPUB member %r: %s", file_name, exc) + return "\n\n".join(text_parts) + + +def _extract_txt(file_path: Path) -> str: + """Extract from TXT, trying common encodings (latin-1 is the catch-all).""" + for encoding in ("utf-8", "utf-16", "cp1252", "latin-1"): + try: + with open(file_path, "r", encoding=encoding) as f: + return clean_text(f.read()) + except UnicodeError: + continue + raise ValueError(f"Could not decode text file: {file_path}") + + +def _extract_pdf(file_path: Path) -> str: + """Extract from PDF.""" + from pypdf import PdfReader + + text = "" + with open(file_path, "rb") as file: + pdf_reader = PdfReader(file) + total_pages = len(pdf_reader.pages) + logger.info("PDF has %d pages", total_pages) + + for page_num, page in enumerate(pdf_reader.pages, 1): + try: + page_text = page.extract_text() or "" + if page_text.strip(): + text += f"\n\n{page_text}" + if page_num % 10 == 0: + logger.debug("Extracted %d/%d pages", page_num, total_pages) + except Exception as exc: + logger.warning("Failed to extract page %d: %s", page_num, exc) + + logger.info("Extracted text from %d pages, %d characters total", total_pages, len(text)) + return clean_text(text) diff --git a/converter/tts.py b/converter/tts.py new file mode 100644 index 0000000..ac47ecb --- /dev/null +++ b/converter/tts.py @@ -0,0 +1,311 @@ +"""Client wrapper for the Qwen3-TTS Gradio demos (custom voice / voice clone).""" + +import contextlib +import io +import logging +import shutil +import sys +import threading +import time +from pathlib import Path +from typing import Any, Dict, Optional, Tuple + +from . import config + +logger = logging.getLogger(__name__) + + +class QwenTTSClient: + """Generates audio chunks through a Qwen3-TTS Gradio server.""" + + def __init__(self, voice_mode: str = "custom_voice", voice_clone_ref_audio: Optional[str] = None, + voice_clone_ref_text: Optional[str] = None, skip_transcription: bool = False): + self.voice_mode = voice_mode + self.voice_clone_ref_audio = voice_clone_ref_audio + self.voice_clone_ref_text = (voice_clone_ref_text or "").strip() + self.skip_transcription = skip_transcription + self.client = None + self.api_info: Dict[str, Any] = {} + self.clone_client = None + self.clone_api_info: Dict[str, Any] = {} + self._ref_audio_filedata: Optional[Dict[str, Any]] = None + self._connect() + + # ------------------------------------------------------------------ + # Connection + # ------------------------------------------------------------------ + + def _connect(self) -> None: + try: + if self.voice_mode == "voice_clone": + # Voice clone uses the Base-model demo, which is a separate server + # from the CustomVoice demo (that one only exposes /run_instruct). + self._init_client(config.VOICE_CLONE_API_URL, clone=True) + print(f"[OK] Connected to Voice Clone API at {config.VOICE_CLONE_API_URL}") + self._resolve_reference_text() + else: + self._init_client(config.QWEN_API_URL, clone=False) + print("[OK] Connected to Qwen API") + except Exception as exc: + api_url = config.VOICE_CLONE_API_URL if self.voice_mode == "voice_clone" else config.QWEN_API_URL + print("[ERROR] Qwen API initialization failed!") + print(f"API endpoint: {api_url}") + print("Make sure:") + print("1. Qwen Gradio server is running") + print("2. The server is accessible at the configured URL") + print("3. The endpoint URL is correct") + print("4. Your installed Qwen3-TTS version matches this converter's API expectations") + print(" (voice clone requires the Base-model demo: Qwen/Qwen3-TTS-12Hz-1.7B-Base)") + print(f"Error: {exc}") + sys.exit(1) + + def _resolve_reference_text(self) -> None: + """Resolve the reference transcript: explicit text, then local + transcription, then x-vector-only mode.""" + if not self.voice_clone_ref_text and self.voice_clone_ref_audio: + if self.skip_transcription: + print("[INFO] Skipping reference audio transcription (--no-transcription).") + else: + print("[INFO] Transcribing reference audio for voice cloning...") + self.voice_clone_ref_text = self.transcribe_audio(self.voice_clone_ref_audio) or "" + if not self.voice_clone_ref_text: + print("[WARNING] No reference text available; using x-vector-only clone mode (lower quality).") + print(' Pass --voice-sample-text "..." for higher-quality in-context cloning.') + else: + print(f"[OK] Reference text: {self.voice_clone_ref_text[:100]}...") + + def _init_client(self, url: str, clone: bool = False) -> None: + """Initialize a Gradio client and store its API metadata.""" + from gradio_client import Client + + logger.info("Connecting to Qwen API at %s...", url) + old_stdout = sys.stdout + sys.stdout = io.TextIOWrapper(io.BytesIO(), encoding="utf-8", errors="replace") + try: + try: + client = Client(url, httpx_kwargs={"timeout": config.API_TIMEOUT}) + except TypeError: + # Older gradio_client versions don't support httpx_kwargs. + client = Client(url) + finally: + sys.stdout = old_stdout + if clone: + self.clone_client = client + self.clone_api_info = self._load_api_info(client) + else: + self.client = client + self.api_info = self._load_api_info(client) + logger.info("Connected to Qwen API") + + @staticmethod + def _load_api_info(client) -> Dict[str, Any]: + """Load available API metadata from the Gradio app.""" + try: + return client.view_api(return_format="dict") + except Exception as exc: + logger.warning("Unable to read API metadata: %s", exc) + return {} + + def _resolve_api_name(self, *candidates: str, api_info: Optional[Dict[str, Any]] = None) -> str: + """Return the first available api_name from candidate list.""" + info = api_info if api_info is not None else self.api_info + named_endpoints = info.get("named_endpoints", {}) + for candidate in candidates: + if candidate in named_endpoints: + return candidate + return candidates[0] + + def _endpoint_accepts_param(self, api_name: str, param_name: str, + api_info: Optional[Dict[str, Any]] = None) -> bool: + """Check whether endpoint input schema includes the given parameter.""" + info = api_info if api_info is not None else self.api_info + endpoint = info.get("named_endpoints", {}).get(api_name, {}) + parameters = endpoint.get("parameters", []) + return any(parameter.get("parameter_name") == param_name for parameter in parameters) + + # ------------------------------------------------------------------ + # Reference audio transcription (voice clone) + # ------------------------------------------------------------------ + + def transcribe_audio(self, audio_path: str) -> Optional[str]: + """Transcribe reference audio locally using an optional Whisper backend. + + The current qwen-tts demo does not expose a transcription endpoint, so + transcription is done client-side when a Whisper package is available. + Returns None if no backend is installed. + """ + for backend in ("faster_whisper", "whisper"): + try: + if backend == "faster_whisper": + from faster_whisper import WhisperModel + model = WhisperModel("base", device="cpu", compute_type="int8") + segments, _ = model.transcribe(audio_path) + text = " ".join(seg.text.strip() for seg in segments).strip() + else: + import whisper + model = whisper.load_model("base") + result = model.transcribe(audio_path) + text = (result.get("text") or "").strip() + if text: + logger.info("Transcription complete via %s: %s...", backend, text[:100]) + return text + except ImportError: + continue + except Exception as exc: + logger.warning("%s transcription failed: %s", backend, exc) + logger.warning("No Whisper backend available; transcription skipped.") + return None + + # ------------------------------------------------------------------ + # Chunk generation + # ------------------------------------------------------------------ + + def generate_chunk(self, text: str, chunk_num: int) -> Optional[str]: + """Generate one audio chunk; returns its path in the chunks folder.""" + try: + if self.voice_mode == "custom_voice": + with self._chunk_heartbeat(chunk_num): + result = self._generate_custom_voice(text) + elif self.voice_mode == "voice_clone": + with self._chunk_heartbeat(chunk_num): + result = self._generate_voice_clone(text) + else: + raise ValueError(f"Unknown voice mode: {self.voice_mode}") + + if not result or len(result) < 2: + raise RuntimeError("Qwen API returned invalid result") + + audio_path = result[0] # First element is the audio file path + if not audio_path or not Path(audio_path).exists(): + raise RuntimeError(f"Generated audio file not found: {audio_path}") + + output_path = config.CHUNKS_FOLDER / f"chunk_{chunk_num:04d}.wav" + shutil.copy2(audio_path, output_path) + + logger.debug("Chunk %d generated successfully", chunk_num) + return str(output_path) + + except Exception as exc: + logger.error("Qwen chunk processing failed for chunk %d: %s", chunk_num, exc) + return None + + def process_chunk_with_retry(self, chunk_num: int, text: str) -> bool: + """Process a chunk with retry logic and rate limiting.""" + # Small delay between chunks to avoid rate limiting (only if not first chunk) + if chunk_num > 1: + time.sleep(config.MIN_DELAY_BETWEEN_CHUNKS) + + for attempt in range(config.MAX_RETRIES): + try: + result = self.generate_chunk(text, chunk_num) + if result and Path(result).exists(): + return True + logger.warning("Chunk %d attempt %d failed", chunk_num, attempt + 1) + except Exception as exc: + logger.warning("Chunk %d attempt %d error: %s", chunk_num, attempt + 1, exc) + + if attempt < config.MAX_RETRIES - 1: + sleep_time = 5 + (2 ** attempt) + logger.info("Waiting %ds before retry...", sleep_time) + time.sleep(sleep_time) + + logger.error("Chunk %d failed after %d attempts", chunk_num, config.MAX_RETRIES) + return False + + @contextlib.contextmanager + def _chunk_heartbeat(self, chunk_num: int): + """Print a periodic "still working" message while a chunk generates.""" + stop = threading.Event() + + def _beat(): + start = time.time() + while not stop.wait(config.HEARTBEAT_INTERVAL_SECONDS): + elapsed = time.time() - start + print(f"[...] Chunk {chunk_num} still generating — " + f"{int(elapsed // 60)}m {int(elapsed % 60)}s elapsed", flush=True) + + thread = threading.Thread(target=_beat, daemon=True) + thread.start() + try: + yield + finally: + stop.set() + + # ------------------------------------------------------------------ + # API payloads + # ------------------------------------------------------------------ + + def _generate_custom_voice(self, text: str) -> Tuple: + """Generate audio using CustomVoice mode.""" + custom_api = self._resolve_api_name("/run_instruct", "/run_custom_voice", "/generate_custom_voice") + if custom_api == "/run_instruct": + payload = dict( + text=text, + lang_disp=config.CUSTOM_VOICE_LANGUAGE, + spk_disp=config.SPEAKER_DISPLAY_NAMES.get( + config.CUSTOM_VOICE_SPEAKER.lower(), config.CUSTOM_VOICE_SPEAKER), + instruct=config.CUSTOM_VOICE_INSTRUCT, + ) + else: + payload = dict( + text=text, + language=config.CUSTOM_VOICE_LANGUAGE, + speaker=config.CUSTOM_VOICE_SPEAKER, + instruct=config.CUSTOM_VOICE_INSTRUCT, + ) + if self._endpoint_accepts_param(custom_api, "model_id_cv"): + payload["model_id_cv"] = config.CUSTOM_VOICE_MODEL_ID + elif self._endpoint_accepts_param(custom_api, "model_size"): + payload["model_size"] = config.CUSTOM_VOICE_MODEL_SIZE + + if self._endpoint_accepts_param(custom_api, "seed"): + payload["seed"] = config.CUSTOM_VOICE_SEED + + return self.client.predict(**payload, api_name=custom_api) + + def _ref_audio_payload(self) -> Dict[str, Any]: + """Gradio file payload for the reference audio (built once, reused).""" + if self._ref_audio_filedata is None: + from gradio_client import handle_file + self._ref_audio_filedata = handle_file(self.voice_clone_ref_audio) + return self._ref_audio_filedata + + def _generate_voice_clone(self, text: str) -> Tuple: + """Generate audio using Voice Clone mode.""" + if not Path(self.voice_clone_ref_audio).exists(): + raise FileNotFoundError(f"Reference audio not found: {self.voice_clone_ref_audio}") + + if self.clone_client is None: + raise RuntimeError("Voice Clone client is not initialized. Is the Base-model demo running?") + + clone_api = self._resolve_api_name("/run_voice_clone", "/generate_voice_clone", + api_info=self.clone_api_info) + use_xvector = config.VOICE_CLONE_USE_XVECTOR_ONLY or not self.voice_clone_ref_text + + if clone_api == "/run_voice_clone": + payload = dict( + ref_aud=self._ref_audio_payload(), + ref_txt=self.voice_clone_ref_text, + use_xvec=use_xvector, + text=text, + lang_disp=config.VOICE_CLONE_LANGUAGE, + ) + else: + payload = dict( + ref_audio=self._ref_audio_payload(), + ref_text=self.voice_clone_ref_text, + target_text=text, + language=config.VOICE_CLONE_LANGUAGE, + use_xvector_only=use_xvector, + ) + optional_params = { + "model_size": config.VOICE_CLONE_MODEL_SIZE, + "max_chunk_chars": config.VOICE_CLONE_MAX_CHUNK_CHARS, + "chunk_gap": config.VOICE_CLONE_CHUNK_GAP, + "seed": config.VOICE_CLONE_SEED, + } + for name, value in optional_params.items(): + if self._endpoint_accepts_param(clone_api, name, api_info=self.clone_api_info): + payload[name] = value + + return self.clone_client.predict(**payload, api_name=clone_api) |