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 fix folder watch (#1335)

Browse source
This commit is contained in:
Raivis Dejus 2026-01-03 13:53:33 +02:00 • committed by GitHub
commit b0a53b4c2f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 96 additions and 6 deletions
Download patch file Download diff file Expand all files Collapse all files

18
buzz/transcriber/file_transcriber.py
View file

@ -149,13 +149,19 @@ class FileTranscriber(QObject):
)
if self.transcription_task.source == FileTranscriptionTask.Source.FOLDER_WATCH:
shutil.move(
self.transcription_task.file_path,
os.path.join(
self.transcription_task.output_directory,
os.path.basename(self.transcription_task.file_path),
),
# Use original_file_path if available (before speech extraction changed file_path)
source_path = (
self.transcription_task.original_file_path
or self.transcription_task.file_path
)
if source_path and os.path.exists(source_path):
shutil.move(
source_path,
os.path.join(
self.transcription_task.output_directory,
os.path.basename(source_path),
),
)
def on_download_progress(self, data: dict):
if data["status"] == "downloading":

4
buzz/transcriber/transcriber.py
View file

@ -199,6 +199,7 @@ class FileTranscriptionTask:
output_directory: Optional[str] = None
source: Source = Source.FILE_IMPORT
file_path: Optional[str] = None
original_file_path: Optional[str] = None # Original path before speech extraction
url: Optional[str] = None
fraction_downloaded: float = 0.0
@ -229,6 +230,9 @@ def get_output_file_path(
export_file_name_template: str | None = None,
):
input_file_name = os.path.splitext(os.path.basename(file_path))[0]
# Remove "_speech" suffix from extracted speech files
if input_file_name.endswith("_speech"):
input_file_name = input_file_name[:-7]
date_time_now = datetime.datetime.now().strftime("%d-%b-%Y %H-%M-%S")
export_file_name_template = (

1
buzz/widgets/main_window.py
View file

@ -392,6 +392,7 @@ class MainWindow(QMainWindow):
basename = os.path.basename(task.file_path)
name = os.path.splitext(basename)[0] # Remove .wav extension
self.transcription_service.update_transcription_file_and_name(task.uid, task.file_path, name)
self.transcription_service.update_transcription_as_completed(task.uid, segments)
self.table_widget.refresh_row(task.uid)

2
buzz/widgets/transcription_task_folder_watcher.py
View file

@ -71,6 +71,7 @@ class TranscriptionTaskFolderWatcher(QFileSystemWatcher):
filename.startswith(".") # hidden files
or file_ext not in SUPPORTED_EXTENSIONS # non-media files
or is_temp_conversion_file # temp conversion files like .ogg.wav
or "_speech.mp3" in filename # extracted speech output files
or file_path in tasks # file already in tasks
or file_path in self.paths_emitted # file already emitted
):
@ -104,6 +105,7 @@ class TranscriptionTaskFolderWatcher(QFileSystemWatcher):
task = FileTranscriptionTask(
file_path=file_path,
original_file_path=file_path,
transcription_options=transcription_options,
file_transcription_options=file_transcription_options,
model_path=model_path,

77
tests/widgets/transcription_task_folder_watcher_test.py
View file

@ -280,3 +280,80 @@ class TestTranscriptionTaskFolderWatcher:
task: FileTranscriptionTask = blocker.args[0]
assert task.file_path == os.path.join(input_directory, "whisper-french.mp3")
def test_should_ignore_extracted_speech_files(self, qtbot: QtBot):
input_directory = mkdtemp()
output_directory = mkdtemp()
watcher = TranscriptionTaskFolderWatcher(
tasks={},
preferences=FolderWatchPreferences(
enabled=True,
input_directory=input_directory,
output_directory=output_directory,
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(),
),
),
)
# Create extracted speech file (should be ignored)
shutil.copy(
test_audio_path,
os.path.join(input_directory, "video_speech.mp3"),
)
# Copy normal media file (should be found)
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.file_path == os.path.join(input_directory, "whisper-french.mp3")
def test_should_set_original_file_path(self, qtbot: QtBot):
input_directory = mkdtemp()
output_directory = mkdtemp()
watcher = TranscriptionTaskFolderWatcher(
tasks={},
preferences=FolderWatchPreferences(
enabled=True,
input_directory=input_directory,
output_directory=output_directory,
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]
expected_path = os.path.join(input_directory, "whisper-french.mp3")
assert task.file_path == expected_path
assert task.original_file_path == expected_path