deblan-mirror/buzz
1
0
Fork 0
mirror of https://github.com/chidiwilliams/buzz.git synced 2026-03-14 14:45:46 +01:00
Code Issues Projects Releases Packages Wiki Activity

1329 improve folder watch (#1402)

Browse source
This commit is contained in:
Raivis Dejus 2026-02-27 19:49:38 +02:00 • committed by GitHub
commit 3869ac08db
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
24 changed files with 561 additions and 195 deletions
Download patch file Download diff file Expand all files Collapse all files

36
tests/transcriber/whisper_file_transcriber_test.py
View file

@ -357,6 +357,42 @@ class TestWhisperFileTranscriber:
transcriber.stop()
time.sleep(3)
def test_transcribe_from_folder_watch_source_deletes_file(self, qtbot):
file_path = tempfile.mktemp(suffix=".mp3")
shutil.copy(test_audio_path, file_path)
file_transcription_options = FileTranscriptionOptions(
file_paths=[file_path],
output_formats={OutputFormat.TXT},
)
transcription_options = TranscriptionOptions()
model_path = get_model_path(transcription_options.model)
output_directory = tempfile.mkdtemp()
transcriber = WhisperFileTranscriber(
task=FileTranscriptionTask(
model_path=model_path,
transcription_options=transcription_options,
file_transcription_options=file_transcription_options,
file_path=file_path,
original_file_path=file_path,
output_directory=output_directory,
source=FileTranscriptionTask.Source.FOLDER_WATCH,
delete_source_file=True,
)
)
with qtbot.wait_signal(transcriber.completed, timeout=10 * 6000):
transcriber.run()
assert not os.path.isfile(file_path)
assert not os.path.isfile(
os.path.join(output_directory, os.path.basename(file_path))
)
assert len(glob.glob("*.txt", root_dir=output_directory)) > 0
transcriber.stop()
time.sleep(3)
@pytest.mark.skip()
def test_transcribe_stop(self):
output_file_path = os.path.join(tempfile.gettempdir(), "whisper.txt")

43
tests/widgets/preferences_dialog/folder_watch_preferences_widget_test.py
View file

@ -48,8 +48,12 @@ class TestFolderWatchPreferencesWidget:
assert not checkbox.isChecked()
assert input_folder_line_edit.text() == ""
assert output_folder_line_edit.text() == ""
assert not input_folder_line_edit.isEnabled()
assert not output_folder_line_edit.isEnabled()
checkbox.setChecked(True)
assert input_folder_line_edit.isEnabled()
assert output_folder_line_edit.isEnabled()
input_folder_line_edit.setText("test/input/folder")
output_folder_line_edit.setText("test/output/folder")
@ -57,3 +61,42 @@ class TestFolderWatchPreferencesWidget:
assert last_config_changed_call[0][0].enabled
assert last_config_changed_call[0][0].input_directory == "test/input/folder"
assert last_config_changed_call[0][0].output_directory == "test/output/folder"
def test_delete_processed_files_checkbox(self, qtbot):
widget = FolderWatchPreferencesWidget(
config=FolderWatchPreferences(
enabled=False,
input_directory="",
output_directory="",
file_transcription_options=FileTranscriptionPreferences(
language=None,
task=Task.TRANSCRIBE,
model=TranscriptionModel.default(),
word_level_timings=False,
extract_speech=False,
temperature=DEFAULT_WHISPER_TEMPERATURE,
initial_prompt="",
enable_llm_translation=False,
llm_model="",
llm_prompt="",
output_formats=set(),
),
),
)
mock_config_changed = Mock()
widget.config_changed.connect(mock_config_changed)
qtbot.add_widget(widget)
delete_checkbox = widget.findChild(QCheckBox, "DeleteProcessedFilesCheckbox")
assert delete_checkbox is not None
assert not delete_checkbox.isChecked()
delete_checkbox.setChecked(True)
last_config = mock_config_changed.call_args_list[-1][0][0]
assert last_config.delete_processed_files is True
delete_checkbox.setChecked(False)
last_config = mock_config_changed.call_args_list[-1][0][0]
assert last_config.delete_processed_files is False

70
tests/widgets/transcription_task_folder_watcher_test.py
View file

@ -322,6 +322,76 @@ class TestTranscriptionTaskFolderWatcher:
task: FileTranscriptionTask = blocker.args[0]
assert task.file_path == os.path.join(input_directory, "whisper-french.mp3")
def test_should_set_delete_source_file_when_preference_enabled(self, qtbot: QtBot):
input_directory = mkdtemp()
output_directory = mkdtemp()
watcher = TranscriptionTaskFolderWatcher(
tasks={},
preferences=FolderWatchPreferences(
enabled=True,
input_directory=input_directory,
output_directory=output_directory,
delete_processed_files=True,
file_transcription_options=FileTranscriptionPreferences(
language=None,
task=Task.TRANSCRIBE,
model=self.default_model(),
word_level_timings=False,
extract_speech=False,
temperature=DEFAULT_WHISPER_TEMPERATURE,
initial_prompt="",
enable_llm_translation=False,
llm_model="",
llm_prompt="",
output_formats=set(),
),
),
)
shutil.copy(test_audio_path, input_directory)
with qtbot.wait_signal(watcher.task_found, timeout=10_000) as blocker:
pass
task: FileTranscriptionTask = blocker.args[0]
assert task.delete_source_file is True
def test_should_not_set_delete_source_file_when_preference_disabled(self, qtbot: QtBot):
input_directory = mkdtemp()
output_directory = mkdtemp()
watcher = TranscriptionTaskFolderWatcher(
tasks={},
preferences=FolderWatchPreferences(
enabled=True,
input_directory=input_directory,
output_directory=output_directory,
delete_processed_files=False,
file_transcription_options=FileTranscriptionPreferences(
language=None,
task=Task.TRANSCRIBE,
model=self.default_model(),
word_level_timings=False,
extract_speech=False,
temperature=DEFAULT_WHISPER_TEMPERATURE,
initial_prompt="",
enable_llm_translation=False,
llm_model="",
llm_prompt="",
output_formats=set(),
),
),
)
shutil.copy(test_audio_path, input_directory)
with qtbot.wait_signal(watcher.task_found, timeout=10_000) as blocker:
pass
task: FileTranscriptionTask = blocker.args[0]
assert task.delete_source_file is False
def test_should_set_original_file_path(self, qtbot: QtBot):
input_directory = mkdtemp()
output_directory = mkdtemp()