Ebook Voice Player Architecture Deep Dive: Kivy + edge-tts Cross-Platform E-Book TTS Reader

┌─────────────────────────────────────────────────────────────┐ │ Presentation Layer (Kivy UI) │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌────────┐ │ │ │ ReaderPanel │ │ VoicePanel │ │ControlPanel │ │ Menus │ │ │ └─────────────┘ └─────────────┘ └─────────────┘ └────────┘ │ ├─────────────────────────────────────────────────────────────┤ │ Business Logic Layer │ │ ┌──────────────────┐ ┌──────────────────┐ ┌──────────────┐ │ │ │ TTSService │ │ VoiceManager │ │ I18n System │ │ │ │ (Playback Orch.)│ │ (300+ Voices) │ │ (zh/en) │ │ │ └──────────────────┘ └──────────────────┘ └──────────────┘ │ ├─────────────────────────────────────────────────────────────┤ │ Core Engine Layer │ │ ┌──────────────────┐ ┌──────────────────┐ ┌──────────────┐ │ │ │ TTSGenerator │ │ TTSPlayer │ │ Reader │ │ │ │ (Async Synthesis)│ │ (Audio Playback)│ │ (Doc Parsing)│ │ │ └──────────────────┘ └──────────────────┘ └──────────────┘ │ ├─────────────────────────────────────────────────────────────┤ │ Infrastructure Layer │ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ │ │ edge-tts │ │ PyMuPDF/ │ │ Kivy │ │ │ │ (Cloud TTS) │ │ pdfplumber │ │ (Framework) │ │ │ └──────────────┘ └──────────────┘ └──────────────┘ │ └─────────────────────────────────────────────────────────────┘ 

class TTSService: """High-level playback orchestrator. Same state machine as the original Qt version.""" def __init__(self): self.events = Emitter() # Event bus  self.paragraphs = [] # Pending synthesis paragraphs  self.audio_queue = {} # Index -> local audio path  self.current_index = -1 # Current playing paragraph index  self.total = 0 # Total paragraph count  self.base_index = 0 # Base offset (for chapter jumps)  # State flags  self._playing = False self._paused = False self._stopped = False self._finished = False self._speed = 1.0 # Sub-components  self.generator = TTSGenerator() # Background synthesizer  self.player = TTSPlayer() # Audio player  def start(self, paragraphs, voice, base_index=0): """Start playback flow""" self._replace_generator() # Replace generator, cleanup old tasks  self.player.stop() self.audio_queue.clear() self.paragraphs = paragraphs self.total = len(paragraphs) self.base_index = base_index self.current_index = -1 self._playing = True self._paused = False self._stopped = False self._finished = False self._emit("status", "playing") self.generator.start_generation(paragraphs, voice, self._current_rate()) 

class TTSGenerator: """Background edge-tts synthesis. Runs an asyncio loop in a thread.""" def __init__(self, max_workers=3): self.events = Emitter() self.max_workers = max_workers # Concurrency limit  self._thread = None self._cancelled = False def _run(self): loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) sem = asyncio.Semaphore(self.max_workers) # Rate limiting  async def gen_one(idx, text): # Cache key: text + voice + rate  cache_key = hashlib.sha256( f"{text}|{self.voice}|{rate_str}".encode()).hexdigest()[:16] cache_file = cache_dir / f"c_{cache_key}.mp3" # Cache hit -> return immediately  if cache_file.exists() and is_valid_mp3(str(cache_file)): self._emit("paragraph", idx, str(cache_file)) return # Atomic write: write .part.mp3 first, verify, then rename  tmp_file = cache_dir / f"c_{cache_key}.part.mp3" communicate = edge_tts.Communicate(text, self.voice, rate=rate_str) await communicate.save(str(tmp_file)) # Validate file integrity  if is_valid_mp3(str(tmp_file)): os.replace(str(tmp_file), str(cache_file)) # Atomic replace  self._emit("paragraph", idx, str(cache_file)) # Concurrent execution of all paragraphs  tasks = [limited(i, p) for i, p in enumerate(self.paragraphs)] loop.run_until_complete(asyncio.gather(*tasks)) 

class TTSPlayer: def _load_and_play(self, path, seek_pos=0.0): snd = SoundLoader.load(path) self._expected_len = float(getattr(snd, "length", 0) or 0) # Windows often reports length=0, estimate from bitrate  if self._expected_len <= 0: self._expected_len = max(0.0, os.path.getsize(path) * 8.0 / 48000.0 - 0.2) snd.bind(on_stop=self._on_sound_stop) self._play_start = time.time() self._start_watchdog() # Start watchdog  snd.play() def _start_watchdog(self): def tick(dt): if self._stopped or self._paused or self.sound is None: return False if self.sound.state != "play": return False # Native on_stop will drive completion  elapsed = time.time() - self._play_start if elapsed >= self._expected_len + 1.0: # 1s tolerance  self._finish_playback() # Force advance  return False return True self._watchdog = Clock.schedule_interval(tick, 0.2) 

def read_file(filepath): path = Path(filepath) suffix = path.suffix.lower() if suffix == ".txt": return read_txt(filepath) elif suffix in (".docx", ".doc"): return read_docx(filepath) elif suffix == ".pdf": return read_pdf(filepath) elif suffix == ".epub": return read_epub(filepath) else: raise ValueError(f"Unsupported format: {suffix}") 

def read_txt(filepath, encoding=None): encodings = encoding or ["utf-8-sig", "utf-8", "gbk", "gb2312", "gb18030", "big5", "utf-16", "shift-jis", "euc-jp"] for enc in encodings: try: with open(filepath, "r", encoding=enc) as f: text = f.read() break except (UnicodeDecodeError, UnicodeError): continue else: with open(filepath, "r", encoding="utf-8", errors="replace") as f: text = f.read() # Strip BOM  if text.startswith("\ufeff"): text = text[1:] return [p.strip() for p in text.split("\n") if p.strip()] 

def read_pdf(filepath): try: import fitz # PyMuPDF - fast, accurate  doc = fitz.open(filepath) for page in doc: text = page.get_text().strip() ... except ImportError: try: import pdfplumber # Fallback - better table extraction  with pdfplumber.open(filepath) as pdf: for page in pdf.pages: text = page.extract_text() ... except ImportError: raise ImportError("Please install PyMuPDF or pdfplumber") 

def read_epub(filepath): import ebooklib from ebooklib import epub from bs4 import BeautifulSoup book = epub.read_epub(filepath) paragraphs = [] for item in book.get_items(): if item.get_type() == ebooklib.ITEM_DOCUMENT: soup = BeautifulSoup(item.get_content(), "html.parser") # Preserve semantic tag hierarchy  for p in soup.find_all(["p", "h1", "h2", "h3", "h4", "h5", "h6", "div", "span"]): text = p.get_text().strip() if text: paragraphs.append(text) return paragraphs 

def detect_chapters(paragraphs): primary_patterns = [ r"^第[一二三四五六七八九十百千万零〇\d]+[章卷部篇]", r"^第\s*\d+\s*[章卷部篇]", r"^Chapter\s+\d+", r"^CHAPTER\s+\d+", ] secondary_patterns = [ r"^第[一二三四五六七八九十百千万零〇\d]+[话节集]", r"^[Pp]art\s+\d+", r"^[Ss]ection\s+\d+", ] # Primary headings prioritized; secondary merged if within 3 paragraphs of primary  ... 

def is_phone_layout(w, h): """Narrow portrait = phone mode""" return w > 0 and w < h and w <= 820 def _update_layout_mode(self, *a): w, h = _logical_window_size() phone = is_phone_layout(w, h) if phone: # Phone: vertical stack + bottom tab switch  self._body.orientation = "vertical" self.reader.size_hint_x = 1 self.voice.size_hint_x = 1 self.voice.size_hint_y = 0 # Hidden, use popup instead  self.voice.opacity = 0 self._tabbar.opacity = 1 # Show bottom tab bar  else: # Desktop: horizontal split  self._body.orientation = "horizontal" self.reader.size_hint_x = 3 self.voice.size_hint_x = 1 self.voice.size_hint_y = 1 self.voice.opacity = 1 self._tabbar.opacity = 0 

def _logical_window_size(): """Android Window.size is physical pixels, convert to dp""" try: from jnius import autoclass DisplayMetrics = autoclass('android.util.DisplayMetrics') PythonActivity = autoclass('org.kivy.android.PythonActivity') metrics = DisplayMetrics() PythonActivity.mActivity.getWindowManager().getDefaultDisplay().getRealMetrics(metrics) factor = 160.0 / metrics.densityDpi return int(metrics.widthPixels * factor), int(metrics.heightPixels * factor) except Exception: return int(Window.width), int(Window.height) 

def voice_to_locale(voice_id): """Derive UI language from voice id: zh-* -> zh, others -> en""" if not voice_id: return LANG_EN prefix = voice_id.split("-", 1)[0].lower() return _VOICE_PREFIX_TO_LOCALE.get(prefix, LANG_EN) 

T = { "app_title": {"zh": "TTS 电子书语音朗读器", "en": "TTS E-Book Audio Reader"}, "menu_file": {"zh": "文件", "en": "File"}, "vp_title": {"zh": "语音选择", "en": "Voice Selection"}, # ... 200+ entries } def L(key): entry = T.get(key) if entry is None: return key return entry.get(_lang, entry.get(LANG_EN, key)) # Fallback chain  def set_language(lang): global _lang if lang == _lang: return False _lang = lang for cb in _callbacks: cb(lang) # Notify all UI components to refresh  return True