From dbb52ad99ace09f62638fde6a932d6de27498547 Mon Sep 17 00:00:00 2001 From: gummiflip Date: Sun, 21 Jun 2026 16:38:09 +0200 Subject: [PATCH 1/2] feat(tts): finish audio export with atomic temp-file pipeline --- app/i18n.py | 16 +++ app/tts_window.py | 293 +++++++++++++++++++++++++++++++++++++++-- tests/test_features.py | 181 +++++++++++++++++++++++++ tests/test_i18n.py | 1 + 4 files changed, 480 insertions(+), 11 deletions(-) diff --git a/app/i18n.py b/app/i18n.py index e8e867c..c313823 100644 --- a/app/i18n.py +++ b/app/i18n.py @@ -94,7 +94,10 @@ "tts.button.pause": "Pause", "tts.button.resume": "Fortsetzen", "tts.button.speak": "Vorlesen", + "tts.button.export": "Exportieren", "tts.button.stop": "Stopp", + "tts.export.dialog_title": "Audiodatei exportieren", + "tts.export.default_filename_prefix": "Blitztext-Audio_", "tts.error.piper_not_found": "Piper nicht gefunden. Installieren: pip install piper-tts und Stimmen nach ~/.local/share/piper-voices legen.", "tts.error.openai_not_available": "OpenAI Cloud-TTS ist nicht verfuegbar. Bitte OPENAI_API_KEY in ~/.config/blitztext-linux/secrets.env setzen.", "tts.consent.title": "OpenAI Cloud-TTS aktivieren?", @@ -108,6 +111,11 @@ "tts.status.synthesis": "Synthese…", "tts.status.openai_not_confirmed": "OpenAI Cloud-TTS wurde nicht bestaetigt.", "tts.status.cloud_synthesis": "Cloud-Synthese…", + "tts.status.exporting": "Exportiere Audiodatei…", + "tts.status.export_done": "Audiodatei exportiert.", + "tts.status.export_ffmpeg_missing": "ffmpeg nicht gefunden. Audio-Export ist nicht verfuegbar.", + "tts.status.export_format_unsupported": "Exportformat nicht unterstuetzt.", + "tts.status.missing_wav_output": "WAV-Ausgabe fehlt.", "tts.status.cancelled": "Abgebrochen.", "tts.status.error": "Fehler: {message}", "tts.status.done": "Fertig.", @@ -245,7 +253,10 @@ "tts.button.pause": "Pause", "tts.button.resume": "Resume", "tts.button.speak": "Read Aloud", + "tts.button.export": "Export", "tts.button.stop": "Stop", + "tts.export.dialog_title": "Export audio file", + "tts.export.default_filename_prefix": "blitztext-audio_", "tts.error.piper_not_found": "Piper not found. Install with: pip install piper-tts and place voices in ~/.local/share/piper-voices.", "tts.error.openai_not_available": "OpenAI Cloud-TTS is not available. Please set OPENAI_API_KEY in ~/.config/blitztext-linux/secrets.env.", "tts.consent.title": "Enable OpenAI Cloud-TTS?", @@ -259,6 +270,11 @@ "tts.status.synthesis": "Synthesizing…", "tts.status.openai_not_confirmed": "OpenAI Cloud-TTS was not confirmed.", "tts.status.cloud_synthesis": "Cloud synthesis…", + "tts.status.exporting": "Exporting audio file…", + "tts.status.export_done": "Audio file exported.", + "tts.status.export_ffmpeg_missing": "ffmpeg not found. Audio export is not available.", + "tts.status.export_format_unsupported": "Export format is not supported.", + "tts.status.missing_wav_output": "WAV output is missing.", "tts.status.cancelled": "Cancelled.", "tts.status.error": "Error: {message}", "tts.status.done": "Done.", diff --git a/app/tts_window.py b/app/tts_window.py index 0ed80cf..ffa463e 100644 --- a/app/tts_window.py +++ b/app/tts_window.py @@ -5,6 +5,7 @@ import shutil import signal import tempfile +from datetime import datetime from pathlib import Path from typing import Optional @@ -13,6 +14,7 @@ from PyQt6.QtWidgets import ( QComboBox, QDialog, + QFileDialog, QHBoxLayout, QLabel, QLineEdit, @@ -27,7 +29,7 @@ PIPER_VENV_PATH = str(Path(__file__).resolve().parents[1] / ".venv" / "bin" / "piper") VOICES_DIR = Path.home() / ".local" / "share" / "piper-voices" -TTS_WAV = os.path.join(os.environ.get("XDG_RUNTIME_DIR", tempfile.gettempdir()), "blitztext-tts.wav") +TTS_RUNTIME_DIR = os.environ.get("XDG_RUNTIME_DIR", tempfile.gettempdir()) OPENAI_TTS_MODEL_DEFAULT = "gpt-4o-mini-tts" OPENAI_TTS_VOICE_DEFAULT = "marin" OPENAI_TTS_VOICES = [ @@ -57,6 +59,10 @@ def _openai_tts_install_hint() -> str: return t("tts.error.openai_not_available") +def _find_ffmpeg() -> Optional[str]: + return shutil.which("ffmpeg") + + def _openai_tts_consent_text() -> str: return t("tts.consent.message") @@ -103,6 +109,73 @@ def _playback_command(wav_path: str) -> tuple[str, list[str]]: return "aplay", ["-D", "pipewire", wav_path] +def _build_ffmpeg_export_command(input_wav: str, output_path: str, ffmpeg_path: Optional[str] = None) -> tuple[str, list[str]]: + program = ffmpeg_path or _find_ffmpeg() or "ffmpeg" + suffix = Path(output_path).suffix.lower() + if suffix in {".ogg", ".opus"}: + args = [ + "-y", + "-i", input_wav, + "-vn", + "-c:a", "libopus", + "-b:a", "32k", + output_path, + ] + elif suffix == ".mp3": + args = [ + "-y", + "-i", input_wav, + "-vn", + "-c:a", "libmp3lame", + "-q:a", "4", + output_path, + ] + else: + raise ValueError(f"unsupported export format: {suffix or '(none)'}") + return program, args + + +def _safe_unlink(path: Optional[str]) -> None: + if not path: + return + try: + Path(path).unlink(missing_ok=True) + except OSError: + pass + + +def _make_tts_wav_path() -> str: + fd, path = tempfile.mkstemp(prefix="blitztext-tts-", suffix=".wav", dir=TTS_RUNTIME_DIR) + os.close(fd) + _safe_unlink(path) + return path + + +def _make_export_temp_path(output_path: str) -> str: + target = Path(output_path) + fd, temp_path = tempfile.mkstemp( + prefix=f".{target.stem}.blitztext-", + suffix=target.suffix, + dir=str(target.parent), + ) + os.close(fd) + _safe_unlink(temp_path) + return temp_path + + +def _normalize_export_path(selected_path: str, selected_filter: str) -> str: + path = Path(selected_path) + if path.suffix: + return str(path) + + normalized_filter = selected_filter.lower() + if "mp3" in normalized_filter and "ogg" not in normalized_filter and "opus" not in normalized_filter: + return str(path.with_suffix(".mp3")) + if "opus" in normalized_filter and "ogg" not in normalized_filter: + return str(path.with_suffix(".opus")) + return str(path.with_suffix(".ogg")) + + class CloudTtsServiceError(Exception): """Raised when OpenAI Cloud-TTS cannot be synthesized.""" @@ -147,10 +220,12 @@ def _check_ready(self) -> None: if not self._openai_installed or self.client is None: raise CloudTtsServiceError(self._missing_openai_package_message()) - def synthesize(self, text: str, output_path: str = TTS_WAV) -> str: + def synthesize(self, text: str, output_path: Optional[str] = None) -> str: self._check_ready() if not text or not text.strip(): raise ValueError("text must not be empty") + if output_path is None: + output_path = _make_tts_wav_path() response = self.client.audio.speech.create( model=self.config.tts_openai_model, @@ -224,6 +299,7 @@ def __init__(self, config, parent: Optional[QWidget] = None) -> None: self.resize(430, 340) self._piper_proc: Optional[QProcess] = None self._aplay_proc: Optional[QProcess] = None + self._export_proc: Optional[QProcess] = None self._cloud_thread: Optional[QThread] = None self._cloud_worker: Optional[_CloudTtsWorker] = None # Noch laufende, vom Dialog geloeste Cloud-Threads bis 'finished' referenziert @@ -231,6 +307,9 @@ def __init__(self, config, parent: Optional[QWidget] = None) -> None: self._detached_cloud_threads: list[QThread] = [] self._piper_path = _find_piper() self._last_tts_text = "" + self._pending_export_path: Optional[str] = None + self._active_wav_path: Optional[str] = None + self._export_temp_path: Optional[str] = None self._is_paused = False self._setup_ui() self._populate_voices() @@ -304,6 +383,10 @@ def _setup_ui(self) -> None: self._btn_replay.setEnabled(False) btn_row.addWidget(self._btn_replay) + self._btn_export = QPushButton(t("tts.button.export")) + self._btn_export.clicked.connect(self._on_export_clicked) + btn_row.addWidget(self._btn_export) + self._btn_pause = QPushButton(t("tts.button.pause")) self._btn_pause.clicked.connect(self._on_pause_clicked) self._btn_pause.setEnabled(False) @@ -371,9 +454,11 @@ def _refresh_status_hint(self) -> None: def _update_speak_button_state(self) -> None: provider = self._current_provider() if provider == "openai": - self._btn_speak.setEnabled(CloudTtsService(self._config).is_available() or self._cloud_is_running()) + can_synthesize = CloudTtsService(self._config).is_available() or self._cloud_is_running() else: - self._btn_speak.setEnabled(bool(self._piper_path) or self._cloud_is_running()) + can_synthesize = bool(self._piper_path) or self._cloud_is_running() + self._btn_speak.setEnabled(can_synthesize or self._is_speaking()) + self._btn_export.setEnabled(can_synthesize and not self._is_speaking()) def _current_provider(self) -> str: provider = self._provider_combo.currentData() @@ -391,6 +476,24 @@ def _current_voice(self) -> str: def _current_tts_text(self) -> str: return self._text_edit.toPlainText().strip() + def _cleanup_active_wav(self) -> None: + _safe_unlink(self._active_wav_path) + self._active_wav_path = None + + def _cleanup_export_temp(self) -> None: + _safe_unlink(self._export_temp_path) + self._export_temp_path = None + + def _prepare_new_tts_job(self) -> str: + self._cleanup_active_wav() + self._cleanup_export_temp() + wav_path = _make_tts_wav_path() + self._active_wav_path = wav_path + return wav_path + + def _current_wav_path(self) -> Optional[str]: + return self._active_wav_path + def _cloud_is_running(self) -> bool: return self._cloud_thread is not None and self._cloud_thread.isRunning() @@ -486,7 +589,7 @@ def _on_speed_changed(self, _idx: int) -> None: pass def _is_speaking(self) -> bool: - for proc in (self._piper_proc, self._aplay_proc): + for proc in (self._piper_proc, self._aplay_proc, self._export_proc): if proc is not None and proc.state() != QProcess.ProcessState.NotRunning: return True return self._cloud_is_running() @@ -496,6 +599,7 @@ def _on_speak_clicked(self) -> None: if self._is_speaking(): self._stop_tts() return + self._pending_export_path = None self._start_tts() @pyqtSlot() @@ -504,8 +608,36 @@ def _on_replay_clicked(self) -> None: return if self._is_speaking(): self._stop_tts() + self._pending_export_path = None self._start_tts(text=self._last_tts_text) + @pyqtSlot() + def _on_export_clicked(self) -> None: + if self._is_speaking(): + return + text = self._current_tts_text() + if not text: + self._status_label.setText(t("tts.status.no_text")) + self._status_label.setStyleSheet("color: #f44336;") + return + if not _find_ffmpeg(): + self._status_label.setText(t("tts.status.export_ffmpeg_missing")) + self._status_label.setStyleSheet("color: #f44336;") + QTimer.singleShot(2500, self._clear_status) + return + default_name = t("tts.export.default_filename_prefix") + datetime.now().strftime("%Y-%m-%d_%H-%M-%S") + ".ogg" + suggested_path = str(Path.home() / default_name) + selected_path, selected_filter = QFileDialog.getSaveFileName( + self, + t("tts.export.dialog_title"), + suggested_path, + "Audio (*.ogg *.opus *.mp3);;Ogg (*.ogg);;Opus (*.opus);;MP3 (*.mp3)", + ) + if not selected_path: + return + self._pending_export_path = _normalize_export_path(selected_path, selected_filter) + self._start_tts(text=text) + @pyqtSlot() def _on_pause_clicked(self) -> None: if not self._aplay_proc or self._aplay_proc.state() == QProcess.ProcessState.NotRunning: @@ -555,12 +687,13 @@ def _start_piper_tts(self, text: str) -> None: self._status_label.setStyleSheet("color: #f44336;") return + wav_path = self._prepare_new_tts_job() proc = QProcess(self) proc.setProgram(self._piper_path) proc.setArguments([ "--model", model_path, "--length_scale", str(float(self._config.tts_speed)), - "--output_file", TTS_WAV, + "--output_file", wav_path, ]) proc.finished.connect(self._on_piper_finished) proc.errorOccurred.connect(self._on_tts_error) @@ -588,8 +721,9 @@ def _start_cloud_tts(self, text: str) -> None: self._update_speak_button_state() return + wav_path = self._prepare_new_tts_job() thread = QThread(self) - worker = _CloudTtsWorker(service, text, TTS_WAV) + worker = _CloudTtsWorker(service, text, wav_path) worker.moveToThread(thread) thread.started.connect(worker.run) worker.finished.connect(self._on_cloud_finished) @@ -609,8 +743,79 @@ def _start_cloud_tts(self, text: str) -> None: thread.start() self._update_speak_button_state() + def _start_export_process(self, input_wav: str) -> None: + output_path = self._pending_export_path + ffmpeg_path = _find_ffmpeg() + if not output_path or not ffmpeg_path: + self._cleanup_export_temp() + self._cleanup_active_wav() + self._pending_export_path = None + self._status_label.setText(t("tts.status.export_ffmpeg_missing")) + self._status_label.setStyleSheet("color: #f44336;") + self._btn_speak.setText(t("tts.button.speak")) + self._btn_pause.setEnabled(False) + self._is_paused = False + self._btn_pause.setText(t("tts.button.pause")) + self._update_speak_button_state() + clear_status = getattr(self, "_clear_status", None) + if callable(clear_status): + QTimer.singleShot(2500, clear_status) + return + + try: + export_temp_path = _make_export_temp_path(output_path) + program, args = _build_ffmpeg_export_command(input_wav, export_temp_path, ffmpeg_path=ffmpeg_path) + except ValueError: + self._cleanup_export_temp() + self._cleanup_active_wav() + self._pending_export_path = None + self._status_label.setText(t("tts.status.export_format_unsupported")) + self._status_label.setStyleSheet("color: #f44336;") + self._btn_speak.setText(t("tts.button.speak")) + self._btn_pause.setEnabled(False) + self._is_paused = False + self._btn_pause.setText(t("tts.button.pause")) + self._update_speak_button_state() + clear_status = getattr(self, "_clear_status", None) + if callable(clear_status): + QTimer.singleShot(2500, clear_status) + return + except OSError as exc: + self._cleanup_export_temp() + self._cleanup_active_wav() + self._pending_export_path = None + self._status_label.setText(t("tts.status.error").format(message=str(exc))) + self._status_label.setStyleSheet("color: #f44336;") + self._btn_speak.setText(t("tts.button.speak")) + self._btn_pause.setEnabled(False) + self._is_paused = False + self._btn_pause.setText(t("tts.button.pause")) + self._update_speak_button_state() + clear_status = getattr(self, "_clear_status", None) + if callable(clear_status): + QTimer.singleShot(2500, clear_status) + return + + self._cleanup_export_temp() + self._export_temp_path = export_temp_path + export_proc = QProcess(self) + export_proc.setProgram(program) + export_proc.setArguments(args) + export_proc.finished.connect(self._on_export_finished) + export_proc.errorOccurred.connect(self._on_tts_error) + self._export_proc = export_proc + + self._status_label.setText(t("tts.status.exporting")) + self._status_label.setStyleSheet("") + self._btn_speak.setText(t("tts.button.stop")) + self._btn_pause.setEnabled(False) + self._is_paused = False + self._btn_pause.setText(t("tts.button.pause")) + self._update_speak_button_state() + export_proc.start() + def _stop_tts(self) -> None: - for attr in ("_piper_proc", "_aplay_proc"): + for attr in ("_piper_proc", "_aplay_proc", "_export_proc"): proc = getattr(self, attr, None) if proc is None: continue @@ -637,6 +842,9 @@ def _stop_tts(self) -> None: setattr(self, attr, None) self._detach_cloud_thread() + self._cleanup_export_temp() + self._cleanup_active_wav() + self._pending_export_path = None self._btn_speak.setText(t("tts.button.speak")) self._btn_pause.setEnabled(False) @@ -689,6 +897,9 @@ def _on_detached_thread_finished(self, thread: QThread) -> None: @pyqtSlot(str) def _on_cloud_finished(self, wav_path: str) -> None: self._cleanup_cloud_state() + if self._pending_export_path: + self._start_export_process(wav_path) + return self._status_label.setText(t("tts.status.playback")) program, args = _playback_command(wav_path) aplay = QProcess(self) @@ -706,6 +917,8 @@ def _on_cloud_finished(self, wav_path: str) -> None: @pyqtSlot(str) def _on_cloud_error(self, message: str) -> None: self._cleanup_cloud_state() + self._cleanup_export_temp() + self._cleanup_active_wav() self._status_label.setText(t("tts.status.error").format(message=message)) self._status_label.setStyleSheet("color: #f44336;") self._btn_speak.setText(t("tts.button.speak")) @@ -722,6 +935,7 @@ def _on_cloud_thread_finished(self) -> None: @pyqtSlot(int, QProcess.ExitStatus) def _on_piper_finished(self, exit_code: int, exit_status: QProcess.ExitStatus) -> None: proc = self._piper_proc + wav_path = self._current_wav_path() self._piper_proc = None if exit_status != QProcess.ExitStatus.NormalExit or exit_code != 0: @@ -729,6 +943,8 @@ def _on_piper_finished(self, exit_code: int, exit_status: QProcess.ExitStatus) - if proc is not None: stderr = bytes(proc.readAllStandardError()).decode("utf-8", "replace").strip() proc.deleteLater() + self._cleanup_export_temp() + self._cleanup_active_wav() msg = stderr or f"Exit {exit_code}" self._status_label.setText(t("tts.status.error").format(message=msg)) self._status_label.setStyleSheet("color: #f44336;") @@ -739,8 +955,20 @@ def _on_piper_finished(self, exit_code: int, exit_status: QProcess.ExitStatus) - if proc is not None: proc.deleteLater() + if self._pending_export_path and wav_path: + self._start_export_process(wav_path) + return + + if not wav_path: + self._status_label.setText(t("tts.status.error").format(message=t("tts.status.missing_wav_output"))) + self._status_label.setStyleSheet("color: #f44336;") + self._btn_speak.setText(t("tts.button.speak")) + self._update_speak_button_state() + QTimer.singleShot(2500, self._clear_status) + return + self._status_label.setText(t("tts.status.playback")) - program, args = _playback_command(TTS_WAV) + program, args = _playback_command(wav_path) aplay = QProcess(self) aplay.setProgram(program) aplay.setArguments(args) @@ -772,13 +1000,53 @@ def _on_aplay_finished(self, exit_code: int, exit_status: QProcess.ExitStatus) - self._status_label.setStyleSheet("color: #f44336;") if proc is not None: proc.deleteLater() + self._cleanup_export_temp() + self._cleanup_active_wav() + self._update_speak_button_state() + QTimer.singleShot(2500, self._clear_status) + + @pyqtSlot(int, QProcess.ExitStatus) + def _on_export_finished(self, exit_code: int, exit_status: QProcess.ExitStatus) -> None: + proc = self._export_proc + export_temp_path = self._export_temp_path + output_path = self._pending_export_path + self._export_proc = None + self._btn_speak.setText(t("tts.button.speak")) + self._btn_pause.setEnabled(False) + self._is_paused = False + self._btn_pause.setText(t("tts.button.pause")) + self._pending_export_path = None + if exit_status == QProcess.ExitStatus.NormalExit and exit_code == 0 and export_temp_path and output_path: + try: + os.replace(export_temp_path, output_path) + self._status_label.setText(t("tts.status.export_done")) + self._status_label.setStyleSheet("color: #4caf50;") + self._export_temp_path = None + except OSError as exc: + self._status_label.setText(t("tts.status.error").format(message=str(exc))) + self._status_label.setStyleSheet("color: #f44336;") + self._cleanup_export_temp() + else: + stderr = "" + if proc is not None: + stderr = bytes(proc.readAllStandardError()).decode("utf-8", "replace").strip() + msg = stderr or f"Exit {exit_code}" + self._status_label.setText(t("tts.status.error").format(message=msg)) + self._status_label.setStyleSheet("color: #f44336;") + self._cleanup_export_temp() + if proc is not None: + proc.deleteLater() + self._cleanup_active_wav() self._update_speak_button_state() QTimer.singleShot(2500, self._clear_status) @pyqtSlot(QProcess.ProcessError) def _on_tts_error(self, error: QProcess.ProcessError) -> None: if error == QProcess.ProcessError.FailedToStart: - self._status_label.setText(_piper_install_hint() if self._current_provider() == "piper" else _openai_tts_install_hint()) + if self._export_proc is not None or self._pending_export_path: + self._status_label.setText(t("tts.status.export_ffmpeg_missing")) + else: + self._status_label.setText(_piper_install_hint() if self._current_provider() == "piper" else _openai_tts_install_hint()) else: self._status_label.setText(t("tts.status.error").format(message=error.name)) self._status_label.setStyleSheet("color: #f44336;") @@ -786,11 +1054,14 @@ def _on_tts_error(self, error: QProcess.ProcessError) -> None: self._btn_pause.setEnabled(False) self._is_paused = False self._btn_pause.setText(t("tts.button.pause")) - for attr in ("_piper_proc", "_aplay_proc"): + self._pending_export_path = None + for attr in ("_piper_proc", "_aplay_proc", "_export_proc"): proc = getattr(self, attr, None) if proc is not None: proc.deleteLater() setattr(self, attr, None) + self._cleanup_export_temp() + self._cleanup_active_wav() self._update_speak_button_state() def _clear_status(self) -> None: diff --git a/tests/test_features.py b/tests/test_features.py index eed8053..a2907eb 100644 --- a/tests/test_features.py +++ b/tests/test_features.py @@ -311,6 +311,48 @@ def test_openai_cloud_service_propagates_timeout_error(self, tmp_path, monkeypat with pytest.raises(TimeoutError): service.synthesize("Hallo Welt", output_path=str(tmp_path / "x.wav")) + def test_build_ffmpeg_export_command_prefers_opus_for_ogg(self): + program, args = tts_window._build_ffmpeg_export_command("/tmp/in.wav", "/tmp/out.ogg", ffmpeg_path="/usr/bin/ffmpeg") + assert program == "/usr/bin/ffmpeg" + assert args == [ + "-y", + "-i", "/tmp/in.wav", + "-vn", + "-c:a", "libopus", + "-b:a", "32k", + "/tmp/out.ogg", + ] + + def test_build_ffmpeg_export_command_supports_opus_extension(self): + program, args = tts_window._build_ffmpeg_export_command("/tmp/in.wav", "/tmp/out.opus", ffmpeg_path="/usr/bin/ffmpeg") + assert program == "/usr/bin/ffmpeg" + assert args == [ + "-y", + "-i", "/tmp/in.wav", + "-vn", + "-c:a", "libopus", + "-b:a", "32k", + "/tmp/out.opus", + ] + + def test_build_ffmpeg_export_command_uses_mp3_codec_for_mp3(self): + program, args = tts_window._build_ffmpeg_export_command("/tmp/in.wav", "/tmp/out.mp3", ffmpeg_path="/usr/bin/ffmpeg") + assert program == "/usr/bin/ffmpeg" + assert args == [ + "-y", + "-i", "/tmp/in.wav", + "-vn", + "-c:a", "libmp3lame", + "-q:a", "4", + "/tmp/out.mp3", + ] + + def test_normalize_export_path_uses_filter_when_suffix_missing(self): + assert tts_window._normalize_export_path("/tmp/audio", "Opus (*.opus)") == "/tmp/audio.opus" + assert tts_window._normalize_export_path("/tmp/audio", "MP3 (*.mp3)") == "/tmp/audio.mp3" + assert tts_window._normalize_export_path("/tmp/audio", "Audio (*.ogg *.opus *.mp3)") == "/tmp/audio.ogg" + assert tts_window._normalize_export_path("/tmp/audio.ogg", "MP3 (*.mp3)") == "/tmp/audio.ogg" + def test_scrub_secret_removes_api_key(self): msg = tts_window._scrub_secret("Fehler mit sk-secret123 im Text", "sk-secret123") assert "sk-secret123" not in msg @@ -396,6 +438,7 @@ def test_start_cloud_tts_deletes_thread_on_normal_finish(self, tmp_path): _on_cloud_finished=MagicMock(), _on_cloud_error=MagicMock(), _on_cloud_thread_finished=MagicMock(), + _prepare_new_tts_job=MagicMock(return_value=str(tmp_path / "cloud.wav")), ) service = MagicMock() service.is_available.return_value = True @@ -412,6 +455,144 @@ def test_start_cloud_tts_deletes_thread_on_normal_finish(self, tmp_path): thread.finished.connect.assert_any_call(fake._on_cloud_thread_finished) thread.start.assert_called_once() + def test_start_export_process_reports_missing_ffmpeg(self, tmp_path): + input_wav = tmp_path / "input.wav" + input_wav.write_bytes(b"RIFF\x00WAVE") + fake = SimpleNamespace( + _status_label=MagicMock(), + _btn_speak=MagicMock(), + _btn_pause=MagicMock(), + _update_speak_button_state=MagicMock(), + _pending_export_path=str(tmp_path / "out.ogg"), + _cleanup_export_temp=MagicMock(), + _cleanup_active_wav=MagicMock(), + ) + with patch.object(tts_window, "_find_ffmpeg", return_value=None): + tts_window.TtsWindow._start_export_process(fake, str(input_wav)) + fake._status_label.setText.assert_called_once_with(tts_window.t("tts.status.export_ffmpeg_missing")) + fake._btn_speak.setText.assert_called_once() + + def test_on_tts_error_uses_export_message_for_failed_export_start(self): + fake = SimpleNamespace( + _status_label=MagicMock(), + _btn_speak=MagicMock(), + _btn_pause=MagicMock(), + _update_speak_button_state=MagicMock(), + _export_proc=MagicMock(), + _piper_proc=None, + _aplay_proc=None, + _pending_export_path="/tmp/out.ogg", + _cleanup_export_temp=MagicMock(), + _cleanup_active_wav=MagicMock(), + ) + fake._current_provider = lambda: "piper" + tts_window.TtsWindow._on_tts_error(fake, tts_window.QProcess.ProcessError.FailedToStart) + fake._status_label.setText.assert_called_once_with(tts_window.t("tts.status.export_ffmpeg_missing")) + assert fake._export_proc is None + assert fake._pending_export_path is None + + def test_cloud_finished_starts_export_when_path_pending(self): + fake = SimpleNamespace( + _cleanup_cloud_state=MagicMock(), + _pending_export_path="/tmp/out.ogg", + _start_export_process=MagicMock(), + ) + tts_window.TtsWindow._on_cloud_finished(fake, "/tmp/source.wav") + fake._cleanup_cloud_state.assert_called_once() + fake._start_export_process.assert_called_once_with("/tmp/source.wav") + + def test_on_export_finished_replaces_temp_file_and_cleans_wav(self, tmp_path): + final_path = tmp_path / "final.ogg" + temp_path = tmp_path / "temp.ogg" + wav_path = tmp_path / "source.wav" + temp_path.write_bytes(b"ogg") + wav_path.write_bytes(b"wav") + fake = SimpleNamespace( + _export_proc=MagicMock(), + _export_temp_path=str(temp_path), + _pending_export_path=str(final_path), + _btn_speak=MagicMock(), + _btn_pause=MagicMock(), + _update_speak_button_state=MagicMock(), + _status_label=MagicMock(), + _cleanup_active_wav=MagicMock(), + _cleanup_export_temp=MagicMock(), + _clear_status=MagicMock(), + _is_paused=False, + ) + with patch.object(tts_window.os, "replace") as replace_mock: + tts_window.TtsWindow._on_export_finished(fake, 0, tts_window.QProcess.ExitStatus.NormalExit) + replace_mock.assert_called_once_with(str(temp_path), str(final_path)) + assert fake._export_temp_path is None + fake._cleanup_active_wav.assert_called_once() + fake._status_label.setText.assert_called_once_with(tts_window.t("tts.status.export_done")) + + def test_on_export_finished_failure_cleans_temp_and_wav(self, tmp_path): + temp_path = tmp_path / "temp.ogg" + temp_path.write_bytes(b"ogg") + fake = SimpleNamespace( + _export_proc=MagicMock(), + _export_temp_path=str(temp_path), + _pending_export_path=str(tmp_path / "final.ogg"), + _btn_speak=MagicMock(), + _btn_pause=MagicMock(), + _update_speak_button_state=MagicMock(), + _status_label=MagicMock(), + _cleanup_active_wav=MagicMock(), + _clear_status=MagicMock(), + _is_paused=False, + ) + fake._cleanup_export_temp = lambda: (tts_window._safe_unlink(fake._export_temp_path), setattr(fake, "_export_temp_path", None)) + fake._export_proc.readAllStandardError.return_value = b"ffmpeg failed" + tts_window.TtsWindow._on_export_finished(fake, 1, tts_window.QProcess.ExitStatus.NormalExit) + assert fake._export_temp_path is None + assert not temp_path.exists() + fake._cleanup_active_wav.assert_called_once() + fake._status_label.setText.assert_called_once_with(tts_window.t("tts.status.error").format(message="ffmpeg failed")) + + def test_start_piper_tts_uses_generated_wav_path(self, tmp_path): + wav_path = str(tmp_path / "job.wav") + fake = SimpleNamespace( + _piper_path="/usr/bin/piper", + _config=SimpleNamespace(tts_speed=1.0), + _current_voice=lambda: "/tmp/voice.onnx", + _prepare_new_tts_job=MagicMock(return_value=wav_path), + _status_label=MagicMock(), + _btn_speak=MagicMock(), + _update_speak_button_state=MagicMock(), + _on_piper_finished=MagicMock(), + _on_tts_error=MagicMock(), + ) + proc = MagicMock() + with patch.object(tts_window, "QProcess", return_value=proc): + tts_window.TtsWindow._start_piper_tts(fake, "Hallo Welt") + proc.setArguments.assert_called_once_with([ + "--model", "/tmp/voice.onnx", + "--length_scale", "1.0", + "--output_file", wav_path, + ]) + + def test_stop_tts_cleans_temp_paths(self): + fake = SimpleNamespace( + _piper_proc=None, + _aplay_proc=None, + _export_proc=None, + _detach_cloud_thread=MagicMock(), + _cleanup_export_temp=MagicMock(), + _cleanup_active_wav=MagicMock(), + _btn_speak=MagicMock(), + _btn_pause=MagicMock(), + _status_label=MagicMock(), + _update_speak_button_state=MagicMock(), + _pending_export_path="/tmp/out.ogg", + _is_paused=False, + _clear_status=MagicMock(), + ) + tts_window.TtsWindow._stop_tts(fake) + fake._cleanup_export_temp.assert_called_once() + fake._cleanup_active_wav.assert_called_once() + assert fake._pending_export_path is None + class TestDetachCloudThread: """Detach-Timeout-Pfad GUI-frei via Fake-self und gemockten QThreads.""" diff --git a/tests/test_i18n.py b/tests/test_i18n.py index fa9be46..3acb88b 100644 --- a/tests/test_i18n.py +++ b/tests/test_i18n.py @@ -77,6 +77,7 @@ def test_tts_and_history_namespaces_seeded(self): expected_keys = { "tts.window_title", "tts.button.speak", + "tts.button.export", "tts.error.piper_not_found", "tts.error.openai_not_available", "tts.consent.message", From a5933f39c3069c059c371d9cafd24ea31043acb0 Mon Sep 17 00:00:00 2001 From: gummiflip Date: Mon, 22 Jun 2026 02:32:19 +0200 Subject: [PATCH 2/2] fix(tts): address audio export review findings --- app/tts_window.py | 123 +++++++++++-------- tests/test_features.py | 265 +++++++++++++++++++++++++++++++++++++++-- 2 files changed, 330 insertions(+), 58 deletions(-) diff --git a/app/tts_window.py b/app/tts_window.py index ffa463e..552a9a1 100644 --- a/app/tts_window.py +++ b/app/tts_window.py @@ -5,6 +5,7 @@ import shutil import signal import tempfile +from uuid import uuid4 from datetime import datetime from pathlib import Path from typing import Optional @@ -144,11 +145,13 @@ def _safe_unlink(path: Optional[str]) -> None: pass +def _make_tts_runtime_job_dir() -> Path: + return Path(tempfile.mkdtemp(prefix=f"blitztext-tts-{uuid4().hex}-", dir=TTS_RUNTIME_DIR)) + + def _make_tts_wav_path() -> str: - fd, path = tempfile.mkstemp(prefix="blitztext-tts-", suffix=".wav", dir=TTS_RUNTIME_DIR) - os.close(fd) - _safe_unlink(path) - return path + job_dir = _make_tts_runtime_job_dir() + return str(job_dir / "speech.wav") def _make_export_temp_path(output_path: str) -> str: @@ -159,7 +162,6 @@ def _make_export_temp_path(output_path: str) -> str: dir=str(target.parent), ) os.close(fd) - _safe_unlink(temp_path) return temp_path @@ -311,6 +313,7 @@ def __init__(self, config, parent: Optional[QWidget] = None) -> None: self._active_wav_path: Optional[str] = None self._export_temp_path: Optional[str] = None self._is_paused = False + self._status_clear_generation = 0 self._setup_ui() self._populate_voices() self._refresh_status_hint() @@ -433,6 +436,7 @@ def _populate_voices(self) -> None: self._update_speak_button_state() def _refresh_status_hint(self) -> None: + self._invalidate_status_clear() provider = self._current_provider() if provider == "openai": service = CloudTtsService(self._config) @@ -477,7 +481,13 @@ def _current_tts_text(self) -> str: return self._text_edit.toPlainText().strip() def _cleanup_active_wav(self) -> None: + wav_path = Path(self._active_wav_path) if self._active_wav_path else None _safe_unlink(self._active_wav_path) + if wav_path is not None: + try: + wav_path.parent.rmdir() + except OSError: + pass self._active_wav_path = None def _cleanup_export_temp(self) -> None: @@ -491,6 +501,32 @@ def _prepare_new_tts_job(self) -> str: self._active_wav_path = wav_path return wav_path + def _invalidate_status_clear(self) -> None: + self._status_clear_generation += 1 + + def _schedule_status_clear(self, delay_ms: int) -> None: + self._status_clear_generation += 1 + generation = self._status_clear_generation + QTimer.singleShot(delay_ms, lambda: self._clear_status_if_current(generation)) + + def _clear_status_if_current(self, generation: int) -> None: + if generation == self._status_clear_generation: + self._clear_status() + + def _abort_export(self, status_text: str) -> None: + self._invalidate_status_clear() + self._cleanup_export_temp() + self._cleanup_active_wav() + self._pending_export_path = None + self._status_label.setText(status_text) + self._status_label.setStyleSheet("color: #f44336;") + self._btn_speak.setText(t("tts.button.speak")) + self._btn_pause.setEnabled(False) + self._is_paused = False + self._btn_pause.setText(t("tts.button.pause")) + self._update_speak_button_state() + self._schedule_status_clear(2500) + def _current_wav_path(self) -> Optional[str]: return self._active_wav_path @@ -617,13 +653,15 @@ def _on_export_clicked(self) -> None: return text = self._current_tts_text() if not text: + self._invalidate_status_clear() self._status_label.setText(t("tts.status.no_text")) self._status_label.setStyleSheet("color: #f44336;") return if not _find_ffmpeg(): + self._invalidate_status_clear() self._status_label.setText(t("tts.status.export_ffmpeg_missing")) self._status_label.setStyleSheet("color: #f44336;") - QTimer.singleShot(2500, self._clear_status) + self._schedule_status_clear(2500) return default_name = t("tts.export.default_filename_prefix") + datetime.now().strftime("%Y-%m-%d_%H-%M-%S") + ".ogg" suggested_path = str(Path.home() / default_name) @@ -663,6 +701,7 @@ def _start_tts(self, text: Optional[str] = None) -> None: if text is None: text = self._current_tts_text() if not text: + self._invalidate_status_clear() self._status_label.setText(t("tts.status.no_text")) self._status_label.setStyleSheet("color: #f44336;") return @@ -677,12 +716,14 @@ def _start_tts(self, text: Optional[str] = None) -> None: def _start_piper_tts(self, text: str) -> None: if not self._piper_path: + self._invalidate_status_clear() self._status_label.setText(_piper_install_hint()) self._status_label.setStyleSheet("color: #f44336;") return model_path = self._current_voice() if not model_path: + self._invalidate_status_clear() self._status_label.setText(t("tts.status.no_voice_path")) self._status_label.setStyleSheet("color: #f44336;") return @@ -710,12 +751,14 @@ def _start_piper_tts(self, text: str) -> None: def _start_cloud_tts(self, text: str) -> None: if not self._config.tts_openai_consent: + self._invalidate_status_clear() self._status_label.setText(t("tts.status.openai_not_confirmed")) self._status_label.setStyleSheet("color: #f44336;") self._update_speak_button_state() return service = CloudTtsService(self._config) if not service.is_available(): + self._invalidate_status_clear() self._status_label.setText(_openai_tts_install_hint()) self._status_label.setStyleSheet("color: #f44336;") self._update_speak_button_state() @@ -747,53 +790,20 @@ def _start_export_process(self, input_wav: str) -> None: output_path = self._pending_export_path ffmpeg_path = _find_ffmpeg() if not output_path or not ffmpeg_path: - self._cleanup_export_temp() - self._cleanup_active_wav() - self._pending_export_path = None - self._status_label.setText(t("tts.status.export_ffmpeg_missing")) - self._status_label.setStyleSheet("color: #f44336;") - self._btn_speak.setText(t("tts.button.speak")) - self._btn_pause.setEnabled(False) - self._is_paused = False - self._btn_pause.setText(t("tts.button.pause")) - self._update_speak_button_state() - clear_status = getattr(self, "_clear_status", None) - if callable(clear_status): - QTimer.singleShot(2500, clear_status) + self._abort_export(t("tts.status.export_ffmpeg_missing")) return + export_temp_path: Optional[str] = None try: export_temp_path = _make_export_temp_path(output_path) program, args = _build_ffmpeg_export_command(input_wav, export_temp_path, ffmpeg_path=ffmpeg_path) except ValueError: - self._cleanup_export_temp() - self._cleanup_active_wav() - self._pending_export_path = None - self._status_label.setText(t("tts.status.export_format_unsupported")) - self._status_label.setStyleSheet("color: #f44336;") - self._btn_speak.setText(t("tts.button.speak")) - self._btn_pause.setEnabled(False) - self._is_paused = False - self._btn_pause.setText(t("tts.button.pause")) - self._update_speak_button_state() - clear_status = getattr(self, "_clear_status", None) - if callable(clear_status): - QTimer.singleShot(2500, clear_status) + _safe_unlink(export_temp_path) + self._abort_export(t("tts.status.export_format_unsupported")) return except OSError as exc: - self._cleanup_export_temp() - self._cleanup_active_wav() - self._pending_export_path = None - self._status_label.setText(t("tts.status.error").format(message=str(exc))) - self._status_label.setStyleSheet("color: #f44336;") - self._btn_speak.setText(t("tts.button.speak")) - self._btn_pause.setEnabled(False) - self._is_paused = False - self._btn_pause.setText(t("tts.button.pause")) - self._update_speak_button_state() - clear_status = getattr(self, "_clear_status", None) - if callable(clear_status): - QTimer.singleShot(2500, clear_status) + _safe_unlink(export_temp_path) + self._abort_export(t("tts.status.error").format(message=str(exc))) return self._cleanup_export_temp() @@ -853,7 +863,7 @@ def _stop_tts(self) -> None: self._status_label.setText(t("tts.status.cancelled")) self._status_label.setStyleSheet("color: #ff9800;") self._update_speak_button_state() - QTimer.singleShot(2000, self._clear_status) + self._schedule_status_clear(2000) def _cleanup_cloud_state(self) -> None: self._cloud_worker = None @@ -919,6 +929,7 @@ def _on_cloud_error(self, message: str) -> None: self._cleanup_cloud_state() self._cleanup_export_temp() self._cleanup_active_wav() + self._invalidate_status_clear() self._status_label.setText(t("tts.status.error").format(message=message)) self._status_label.setStyleSheet("color: #f44336;") self._btn_speak.setText(t("tts.button.speak")) @@ -926,7 +937,7 @@ def _on_cloud_error(self, message: str) -> None: self._is_paused = False self._btn_pause.setText(t("tts.button.pause")) self._update_speak_button_state() - QTimer.singleShot(2500, self._clear_status) + self._schedule_status_clear(2500) def _on_cloud_thread_finished(self) -> None: self._cleanup_cloud_state() @@ -946,11 +957,12 @@ def _on_piper_finished(self, exit_code: int, exit_status: QProcess.ExitStatus) - self._cleanup_export_temp() self._cleanup_active_wav() msg = stderr or f"Exit {exit_code}" + self._invalidate_status_clear() self._status_label.setText(t("tts.status.error").format(message=msg)) self._status_label.setStyleSheet("color: #f44336;") self._btn_speak.setText(t("tts.button.speak")) self._update_speak_button_state() - QTimer.singleShot(2500, self._clear_status) + self._schedule_status_clear(2500) return if proc is not None: proc.deleteLater() @@ -960,11 +972,12 @@ def _on_piper_finished(self, exit_code: int, exit_status: QProcess.ExitStatus) - return if not wav_path: + self._invalidate_status_clear() self._status_label.setText(t("tts.status.error").format(message=t("tts.status.missing_wav_output"))) self._status_label.setStyleSheet("color: #f44336;") self._btn_speak.setText(t("tts.button.speak")) self._update_speak_button_state() - QTimer.singleShot(2500, self._clear_status) + self._schedule_status_clear(2500) return self._status_label.setText(t("tts.status.playback")) @@ -996,6 +1009,7 @@ def _on_aplay_finished(self, exit_code: int, exit_status: QProcess.ExitStatus) - if proc is not None: stderr = bytes(proc.readAllStandardError()).decode("utf-8", "replace").strip() msg = stderr or f"Exit {exit_code}" + self._invalidate_status_clear() self._status_label.setText(t("tts.status.error").format(message=msg)) self._status_label.setStyleSheet("color: #f44336;") if proc is not None: @@ -1003,13 +1017,15 @@ def _on_aplay_finished(self, exit_code: int, exit_status: QProcess.ExitStatus) - self._cleanup_export_temp() self._cleanup_active_wav() self._update_speak_button_state() - QTimer.singleShot(2500, self._clear_status) + if exit_status == QProcess.ExitStatus.NormalExit and exit_code == 0: + self._schedule_status_clear(2500) @pyqtSlot(int, QProcess.ExitStatus) def _on_export_finished(self, exit_code: int, exit_status: QProcess.ExitStatus) -> None: proc = self._export_proc export_temp_path = self._export_temp_path output_path = self._pending_export_path + clear_status_later = False self._export_proc = None self._btn_speak.setText(t("tts.button.speak")) self._btn_pause.setEnabled(False) @@ -1022,7 +1038,9 @@ def _on_export_finished(self, exit_code: int, exit_status: QProcess.ExitStatus) self._status_label.setText(t("tts.status.export_done")) self._status_label.setStyleSheet("color: #4caf50;") self._export_temp_path = None + clear_status_later = True except OSError as exc: + self._invalidate_status_clear() self._status_label.setText(t("tts.status.error").format(message=str(exc))) self._status_label.setStyleSheet("color: #f44336;") self._cleanup_export_temp() @@ -1031,6 +1049,7 @@ def _on_export_finished(self, exit_code: int, exit_status: QProcess.ExitStatus) if proc is not None: stderr = bytes(proc.readAllStandardError()).decode("utf-8", "replace").strip() msg = stderr or f"Exit {exit_code}" + self._invalidate_status_clear() self._status_label.setText(t("tts.status.error").format(message=msg)) self._status_label.setStyleSheet("color: #f44336;") self._cleanup_export_temp() @@ -1038,10 +1057,12 @@ def _on_export_finished(self, exit_code: int, exit_status: QProcess.ExitStatus) proc.deleteLater() self._cleanup_active_wav() self._update_speak_button_state() - QTimer.singleShot(2500, self._clear_status) + if clear_status_later: + self._schedule_status_clear(2500) @pyqtSlot(QProcess.ProcessError) def _on_tts_error(self, error: QProcess.ProcessError) -> None: + self._invalidate_status_clear() if error == QProcess.ProcessError.FailedToStart: if self._export_proc is not None or self._pending_export_path: self._status_label.setText(t("tts.status.export_ffmpeg_missing")) diff --git a/tests/test_features.py b/tests/test_features.py index a2907eb..89f854a 100644 --- a/tests/test_features.py +++ b/tests/test_features.py @@ -347,6 +347,39 @@ def test_build_ffmpeg_export_command_uses_mp3_codec_for_mp3(self): "/tmp/out.mp3", ] + def test_make_tts_wav_path_returns_unique_paths_without_creating_files(self, tmp_path): + with patch.object(tts_window, "TTS_RUNTIME_DIR", str(tmp_path)): + first = Path(tts_window._make_tts_wav_path()) + second = Path(tts_window._make_tts_wav_path()) + + assert first != second + assert first.parent.parent == tmp_path + assert second.parent.parent == tmp_path + assert first.parent.name.startswith("blitztext-tts-") + assert second.parent.name.startswith("blitztext-tts-") + assert (first.parent.stat().st_mode & 0o777) == 0o700 + assert (second.parent.stat().st_mode & 0o777) == 0o700 + assert first.parent != second.parent + assert first.suffix == ".wav" + assert second.suffix == ".wav" + assert first.name == "speech.wav" + assert second.name == "speech.wav" + assert not first.exists() + assert not second.exists() + + def test_cleanup_active_wav_removes_private_job_dir(self, tmp_path): + job_dir = tmp_path / "blitztext-tts" / "job-1" + job_dir.mkdir(parents=True) + wav_path = job_dir / "speech.wav" + wav_path.write_bytes(b"wav") + fake = SimpleNamespace(_active_wav_path=str(wav_path)) + + tts_window.TtsWindow._cleanup_active_wav(fake) + + assert fake._active_wav_path is None + assert not wav_path.exists() + assert not job_dir.exists() + def test_normalize_export_path_uses_filter_when_suffix_missing(self): assert tts_window._normalize_export_path("/tmp/audio", "Opus (*.opus)") == "/tmp/audio.opus" assert tts_window._normalize_export_path("/tmp/audio", "MP3 (*.mp3)") == "/tmp/audio.mp3" @@ -373,6 +406,33 @@ def test_consent_defaults_false_and_persists(self, tmp_path): class TestCloudTtsConsentGate: """Consent-Logik GUI-frei via Fake-self (Muster wie test_settings_dialog).""" + def test_schedule_status_clear_uses_generation_guard(self): + fake = SimpleNamespace( + _status_clear_generation=0, + _clear_status_if_current=MagicMock(), + ) + with patch.object(tts_window.QTimer, "singleShot") as single_shot: + tts_window.TtsWindow._schedule_status_clear(fake, 2500) + + assert fake._status_clear_generation == 1 + single_shot.assert_called_once() + delay_ms, callback = single_shot.call_args.args + assert delay_ms == 2500 + callback() + fake._clear_status_if_current.assert_called_once_with(1) + + def test_clear_status_if_current_ignores_stale_generations(self): + fake = SimpleNamespace( + _status_clear_generation=2, + _clear_status=MagicMock(), + ) + + tts_window.TtsWindow._clear_status_if_current(fake, 1) + fake._clear_status.assert_not_called() + + tts_window.TtsWindow._clear_status_if_current(fake, 2) + fake._clear_status.assert_called_once() + def _fake_window(self, tmp_path): cfg = Config.load(tmp_path / "config.json") return SimpleNamespace(_config=cfg) @@ -419,11 +479,51 @@ def test_start_cloud_tts_blocks_without_consent(self, tmp_path): _config=cfg, _status_label=MagicMock(), _update_speak_button_state=MagicMock(), + _invalidate_status_clear=MagicMock(), ) with patch.object(tts_window, "CloudTtsService") as service: tts_window.TtsWindow._start_cloud_tts(fake, "Hallo Welt") service.assert_not_called() fake._status_label.setText.assert_called_once() + fake._invalidate_status_clear.assert_called_once() + + def test_start_cloud_tts_unavailable_service_invalidates_status_clear(self, tmp_path): + cfg = Config.load(tmp_path / "config.json") + cfg.tts_provider = "openai" + cfg.tts_openai_consent = True + fake = SimpleNamespace( + _config=cfg, + _status_label=MagicMock(), + _update_speak_button_state=MagicMock(), + _invalidate_status_clear=MagicMock(), + ) + service = MagicMock() + service.is_available.return_value = False + with patch.object(tts_window, "CloudTtsService", return_value=service): + tts_window.TtsWindow._start_cloud_tts(fake, "Hallo Welt") + fake._invalidate_status_clear.assert_called_once() + fake._status_label.setText.assert_called_once_with(tts_window._openai_tts_install_hint()) + + def test_start_piper_tts_missing_binary_invalidates_status_clear(self): + fake = SimpleNamespace( + _piper_path=None, + _status_label=MagicMock(), + _invalidate_status_clear=MagicMock(), + ) + tts_window.TtsWindow._start_piper_tts(fake, "Hallo Welt") + fake._invalidate_status_clear.assert_called_once() + fake._status_label.setText.assert_called_once_with(tts_window._piper_install_hint()) + + def test_start_piper_tts_missing_voice_invalidates_status_clear(self): + fake = SimpleNamespace( + _piper_path="/usr/bin/piper", + _current_voice=lambda: None, + _status_label=MagicMock(), + _invalidate_status_clear=MagicMock(), + ) + tts_window.TtsWindow._start_piper_tts(fake, "Hallo Welt") + fake._invalidate_status_clear.assert_called_once() + fake._status_label.setText.assert_called_once_with(tts_window.t("tts.status.no_voice_path")) def test_start_cloud_tts_deletes_thread_on_normal_finish(self, tmp_path): cfg = Config.load(tmp_path / "config.json") @@ -459,18 +559,95 @@ def test_start_export_process_reports_missing_ffmpeg(self, tmp_path): input_wav = tmp_path / "input.wav" input_wav.write_bytes(b"RIFF\x00WAVE") fake = SimpleNamespace( - _status_label=MagicMock(), - _btn_speak=MagicMock(), - _btn_pause=MagicMock(), - _update_speak_button_state=MagicMock(), _pending_export_path=str(tmp_path / "out.ogg"), - _cleanup_export_temp=MagicMock(), - _cleanup_active_wav=MagicMock(), + _abort_export=MagicMock(), ) with patch.object(tts_window, "_find_ffmpeg", return_value=None): tts_window.TtsWindow._start_export_process(fake, str(input_wav)) + fake._abort_export.assert_called_once_with(tts_window.t("tts.status.export_ffmpeg_missing")) + + def test_on_export_clicked_missing_ffmpeg_invalidates_and_schedules_clear(self): + fake = SimpleNamespace( + _is_speaking=lambda: False, + _current_tts_text=lambda: "Hallo", + _invalidate_status_clear=MagicMock(), + _schedule_status_clear=MagicMock(), + _status_label=MagicMock(), + ) + + with patch.object(tts_window, "_find_ffmpeg", return_value=None): + tts_window.TtsWindow._on_export_clicked(fake) + + fake._invalidate_status_clear.assert_called_once() + fake._schedule_status_clear.assert_called_once_with(2500) fake._status_label.setText.assert_called_once_with(tts_window.t("tts.status.export_ffmpeg_missing")) - fake._btn_speak.setText.assert_called_once() + + def test_start_tts_without_text_invalidates_status_clear(self): + fake = SimpleNamespace( + _current_tts_text=lambda: "", + _invalidate_status_clear=MagicMock(), + _status_label=MagicMock(), + ) + + tts_window.TtsWindow._start_tts(fake) + + fake._invalidate_status_clear.assert_called_once() + fake._status_label.setText.assert_called_once_with(tts_window.t("tts.status.no_text")) + + def test_start_export_process_uses_abort_export_for_unsupported_format(self, tmp_path): + input_wav = tmp_path / "input.wav" + input_wav.write_bytes(b"RIFF\x00WAVE") + export_temp_path = tmp_path / ".out.blitztext-temp.wav" + export_temp_path.write_bytes(b"temp") + fake = SimpleNamespace( + _pending_export_path=str(tmp_path / "out.wav"), + _abort_export=MagicMock(), + _cleanup_export_temp=MagicMock(), + ) + with patch.object(tts_window, "_find_ffmpeg", return_value="/usr/bin/ffmpeg"), \ + patch.object(tts_window, "_make_export_temp_path", return_value=str(export_temp_path)): + tts_window.TtsWindow._start_export_process(fake, str(input_wav)) + fake._abort_export.assert_called_once_with(tts_window.t("tts.status.export_format_unsupported")) + assert not export_temp_path.exists() + + def test_start_export_process_uses_abort_export_for_oserror(self, tmp_path): + input_wav = tmp_path / "input.wav" + input_wav.write_bytes(b"RIFF\x00WAVE") + fake = SimpleNamespace( + _pending_export_path=str(tmp_path / "out.ogg"), + _abort_export=MagicMock(), + ) + with patch.object(tts_window, "_find_ffmpeg", return_value="/usr/bin/ffmpeg"), \ + patch.object(tts_window, "_make_export_temp_path", side_effect=OSError("disk full")): + tts_window.TtsWindow._start_export_process(fake, str(input_wav)) + fake._abort_export.assert_called_once_with(tts_window.t("tts.status.error").format(message="disk full")) + + def test_abort_export_resets_cleanup_and_ui_state(self): + fake = SimpleNamespace( + _cleanup_export_temp=MagicMock(), + _cleanup_active_wav=MagicMock(), + _pending_export_path="/tmp/out.ogg", + _status_label=MagicMock(), + _btn_speak=MagicMock(), + _btn_pause=MagicMock(), + _is_paused=True, + _update_speak_button_state=MagicMock(), + _invalidate_status_clear=MagicMock(), + _schedule_status_clear=MagicMock(), + ) + tts_window.TtsWindow._abort_export(fake, "Export fehlgeschlagen") + fake._invalidate_status_clear.assert_called_once() + fake._schedule_status_clear.assert_called_once_with(2500) + fake._cleanup_export_temp.assert_called_once() + fake._cleanup_active_wav.assert_called_once() + assert fake._pending_export_path is None + fake._status_label.setText.assert_called_once_with("Export fehlgeschlagen") + fake._status_label.setStyleSheet.assert_called_once_with("color: #f44336;") + fake._btn_speak.setText.assert_called_once_with(tts_window.t("tts.button.speak")) + fake._btn_pause.setEnabled.assert_called_once_with(False) + assert fake._is_paused is False + fake._btn_pause.setText.assert_called_once_with(tts_window.t("tts.button.pause")) + fake._update_speak_button_state.assert_called_once() def test_on_tts_error_uses_export_message_for_failed_export_start(self): fake = SimpleNamespace( @@ -484,9 +661,11 @@ def test_on_tts_error_uses_export_message_for_failed_export_start(self): _pending_export_path="/tmp/out.ogg", _cleanup_export_temp=MagicMock(), _cleanup_active_wav=MagicMock(), + _invalidate_status_clear=MagicMock(), ) fake._current_provider = lambda: "piper" tts_window.TtsWindow._on_tts_error(fake, tts_window.QProcess.ProcessError.FailedToStart) + fake._invalidate_status_clear.assert_called_once() fake._status_label.setText.assert_called_once_with(tts_window.t("tts.status.export_ffmpeg_missing")) assert fake._export_proc is None assert fake._pending_export_path is None @@ -518,11 +697,13 @@ def test_on_export_finished_replaces_temp_file_and_cleans_wav(self, tmp_path): _cleanup_active_wav=MagicMock(), _cleanup_export_temp=MagicMock(), _clear_status=MagicMock(), + _schedule_status_clear=MagicMock(), _is_paused=False, ) with patch.object(tts_window.os, "replace") as replace_mock: tts_window.TtsWindow._on_export_finished(fake, 0, tts_window.QProcess.ExitStatus.NormalExit) replace_mock.assert_called_once_with(str(temp_path), str(final_path)) + fake._schedule_status_clear.assert_called_once_with(2500) assert fake._export_temp_path is None fake._cleanup_active_wav.assert_called_once() fake._status_label.setText.assert_called_once_with(tts_window.t("tts.status.export_done")) @@ -540,16 +721,84 @@ def test_on_export_finished_failure_cleans_temp_and_wav(self, tmp_path): _status_label=MagicMock(), _cleanup_active_wav=MagicMock(), _clear_status=MagicMock(), + _invalidate_status_clear=MagicMock(), + _schedule_status_clear=MagicMock(), _is_paused=False, ) fake._cleanup_export_temp = lambda: (tts_window._safe_unlink(fake._export_temp_path), setattr(fake, "_export_temp_path", None)) fake._export_proc.readAllStandardError.return_value = b"ffmpeg failed" tts_window.TtsWindow._on_export_finished(fake, 1, tts_window.QProcess.ExitStatus.NormalExit) + fake._invalidate_status_clear.assert_called_once() + fake._schedule_status_clear.assert_not_called() assert fake._export_temp_path is None assert not temp_path.exists() fake._cleanup_active_wav.assert_called_once() fake._status_label.setText.assert_called_once_with(tts_window.t("tts.status.error").format(message="ffmpeg failed")) + def test_on_export_finished_replace_error_keeps_error_status(self, tmp_path): + temp_path = tmp_path / "temp.ogg" + temp_path.write_bytes(b"ogg") + fake = SimpleNamespace( + _export_proc=MagicMock(), + _export_temp_path=str(temp_path), + _pending_export_path=str(tmp_path / "final.ogg"), + _btn_speak=MagicMock(), + _btn_pause=MagicMock(), + _update_speak_button_state=MagicMock(), + _status_label=MagicMock(), + _cleanup_active_wav=MagicMock(), + _invalidate_status_clear=MagicMock(), + _schedule_status_clear=MagicMock(), + _clear_status=MagicMock(), + _is_paused=False, + ) + fake._cleanup_export_temp = lambda: (tts_window._safe_unlink(fake._export_temp_path), setattr(fake, "_export_temp_path", None)) + with patch.object(tts_window.os, "replace", side_effect=OSError("rename failed")): + tts_window.TtsWindow._on_export_finished(fake, 0, tts_window.QProcess.ExitStatus.NormalExit) + fake._invalidate_status_clear.assert_called_once() + fake._schedule_status_clear.assert_not_called() + assert fake._export_temp_path is None + assert not temp_path.exists() + fake._status_label.setText.assert_called_once_with(tts_window.t("tts.status.error").format(message="rename failed")) + + def test_on_aplay_finished_failure_keeps_error_status(self): + proc = MagicMock() + proc.readAllStandardError.return_value = b"aplay failed" + fake = SimpleNamespace( + _aplay_proc=proc, + _btn_speak=MagicMock(), + _btn_pause=MagicMock(), + _status_label=MagicMock(), + _cleanup_export_temp=MagicMock(), + _cleanup_active_wav=MagicMock(), + _update_speak_button_state=MagicMock(), + _clear_status=MagicMock(), + _invalidate_status_clear=MagicMock(), + _schedule_status_clear=MagicMock(), + _is_paused=True, + ) + tts_window.TtsWindow._on_aplay_finished(fake, 1, tts_window.QProcess.ExitStatus.NormalExit) + fake._invalidate_status_clear.assert_called_once() + fake._schedule_status_clear.assert_not_called() + fake._status_label.setText.assert_called_once_with(tts_window.t("tts.status.error").format(message="aplay failed")) + + def test_on_aplay_finished_success_still_clears_status_later(self): + proc = MagicMock() + fake = SimpleNamespace( + _aplay_proc=proc, + _btn_speak=MagicMock(), + _btn_pause=MagicMock(), + _status_label=MagicMock(), + _cleanup_export_temp=MagicMock(), + _cleanup_active_wav=MagicMock(), + _update_speak_button_state=MagicMock(), + _clear_status=MagicMock(), + _schedule_status_clear=MagicMock(), + _is_paused=False, + ) + tts_window.TtsWindow._on_aplay_finished(fake, 0, tts_window.QProcess.ExitStatus.NormalExit) + fake._schedule_status_clear.assert_called_once_with(2500) + def test_start_piper_tts_uses_generated_wav_path(self, tmp_path): wav_path = str(tmp_path / "job.wav") fake = SimpleNamespace( @@ -587,11 +836,13 @@ def test_stop_tts_cleans_temp_paths(self): _pending_export_path="/tmp/out.ogg", _is_paused=False, _clear_status=MagicMock(), + _schedule_status_clear=MagicMock(), ) tts_window.TtsWindow._stop_tts(fake) fake._cleanup_export_temp.assert_called_once() fake._cleanup_active_wav.assert_called_once() assert fake._pending_export_path is None + fake._schedule_status_clear.assert_called_once_with(2000) class TestDetachCloudThread: