Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion buzz/db/entity/transcription.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@

from buzz.db.entity.entity import Entity
from buzz.model_loader import ModelType
from buzz.paths import safe_filename_component
from buzz.settings.settings import Settings
from buzz.transcriber.transcriber import OutputFormat, Task, FileTranscriptionTask

Expand Down Expand Up @@ -46,7 +47,9 @@ def get_output_file_path(
output_format: OutputFormat,
output_directory: str | None = None,
):
input_file_name = os.path.splitext(os.path.basename(self.file))[0]
input_file_name = safe_filename_component(
self.name or os.path.splitext(os.path.basename(self.file))[0]
)

date_time_now = datetime.datetime.now().strftime("%d-%b-%Y %H-%M-%S")

Expand Down
27 changes: 27 additions & 0 deletions buzz/paths.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,32 @@
import os
import re


WINDOWS_INVALID_FILENAME_CHARS = re.compile(r'[<>:"/\\|?*\x00-\x1f]')
WINDOWS_RESERVED_FILENAMES = {
"CON",
"PRN",
"AUX",
"NUL",
*(f"COM{number}" for number in range(1, 10)),
*(f"LPT{number}" for number in range(1, 10)),
}


def file_path_as_title(file_path: str):
return os.path.basename(file_path)


def safe_filename_component(name: str, fallback: str = "transcript") -> str:
"""Make metadata safe to use as one portable filename component."""
sanitized = WINDOWS_INVALID_FILENAME_CHARS.sub("_", name)
sanitized = re.sub(r"[ .]$", "#", sanitized)

if not sanitized:
return fallback

windows_stem = sanitized.split(".", 1)[0].rstrip(" .").upper()
if windows_stem in WINDOWS_RESERVED_FILENAMES:
sanitized = f"_{sanitized}"

return sanitized
92 changes: 57 additions & 35 deletions buzz/transcriber/file_transcriber.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
import shutil
import tempfile
from abc import abstractmethod
from typing import Optional, List
from typing import Any, Optional, List
from pathlib import Path

from PyQt6.QtCore import QObject, pyqtSignal, pyqtSlot
Expand All @@ -24,6 +24,29 @@
app_env['PATH'] = os.pathsep.join([os.path.join(APP_BASE_DIR, "_internal")] + [app_env['PATH']])


def _downloaded_file_path(info: Any) -> Path:
"""Return the actual file path reported by yt-dlp after downloading."""
candidates = []

if info is not None:
filepath = info.get("filepath")
if filepath:
candidates.append(filepath)

for download in info.get("requested_downloads") or []:
filepath = download.get("filepath")
if filepath:
candidates.append(filepath)

for candidate in candidates:
path = Path(candidate).resolve()
if path.is_file():
return path

reported_path = candidates[0] if candidates else "<no filepath reported>"
raise FileNotFoundError(f"yt-dlp output does not exist: {reported_path}")


class FileTranscriber(QObject):
transcription_task: FileTranscriptionTask
progress = pyqtSignal(tuple) # (current, total)
Expand Down Expand Up @@ -63,6 +86,7 @@ def run(self):
output_directory=self.transcription_task.output_directory,
model=self.transcription_task.transcription_options.model,
task=self.transcription_task.transcription_options.task,
display_name=self.transcription_task.display_name,
)

write_output(
Expand All @@ -75,59 +99,46 @@ def run(self):
def _download_from_url(self) -> bool:
cookiefile = os.getenv("BUZZ_DOWNLOAD_COOKIEFILE")

extract_options = {
"logger": logging.getLogger(),
}
if cookiefile:
extract_options["cookiefile"] = cookiefile

try:
with YoutubeDL(extract_options) as ydl_info:
info = ydl_info.extract_info(self.transcription_task.url, download=False)
video_title = info.get("title", "audio")
except Exception as exc:
logging.debug(f"Error extracting video info: {exc}")
video_title = "audio"

video_title = YoutubeDL.sanitize_info({"title": video_title})["title"]
for char in ['/', '\\', ':', '*', '?', '"', '<', '>', '|']:
video_title = video_title.replace(char, '_')

temp_dir = tempfile.mkdtemp()
temp_output_path = os.path.join(temp_dir, video_title)
wav_file = temp_output_path + ".wav"
wav_file = str(Path(wav_file).resolve())
temp_dir = Path(tempfile.mkdtemp()).resolve()
# Never derive a working path from the media title. In particular,
# yt-dlp normalizes trailing dots and spaces differently on Windows.
download_template = str(temp_dir / "source.%(ext)s")
wav_file = temp_dir / "audio.wav"

options = {
"format": "bestaudio/best",
"progress_hooks": [self.on_download_progress],
"outtmpl": temp_output_path,
"outtmpl": download_template,
"logger": logging.getLogger(),
}

if cookiefile:
options["cookiefile"] = cookiefile

ydl = YoutubeDL(options)

try:
logging.debug(f"Downloading audio file from URL: {self.transcription_task.url}")
ydl.download([self.transcription_task.url])
with YoutubeDL(options) as ydl:
info = ydl.extract_info(self.transcription_task.url, download=True)
downloaded_file = _downloaded_file_path(info)
title = info.get("title")
if title:
self.transcription_task.display_name = title
except Exception as exc:
logging.debug(f"Error downloading audio: {exc.msg}")
self.error.emit(exc.msg)
message = getattr(exc, "msg", str(exc))
logging.debug(f"Error downloading audio: {message}")
self.error.emit(message)
return False

cmd = [
"ffmpeg",
"-nostdin",
"-threads", "0",
"-i", temp_output_path,
"-i", str(downloaded_file),
"-ac", "1",
"-ar", str(whisper_audio.SAMPLE_RATE),
"-acodec", "pcm_s16le",
"-loglevel", "panic",
wav_file
str(wav_file)
]

if sys.platform == "win32":
Expand All @@ -144,11 +155,22 @@ def _download_from_url(self) -> bool:
else:
result = subprocess.run(cmd, capture_output=True)

if len(result.stderr):
logging.warning(f"Error processing downloaded audio. Error: {result.stderr.decode()}")
raise Exception(f"Error processing downloaded audio: {result.stderr.decode()}")
if result.returncode != 0:
error = result.stderr.decode("utf-8", errors="replace")
if not error:
error = f"ffmpeg exited with code {result.returncode}"
message = f"Error processing downloaded audio: {error}"
logging.warning(message)
self.error.emit(message)
return False

if not wav_file.is_file():
message = f"FFmpeg output does not exist: {wav_file}"
logging.warning(message)
self.error.emit(message)
return False

self.transcription_task.file_path = wav_file
self.transcription_task.file_path = str(wav_file)
logging.debug(f"Downloaded audio to file: {self.transcription_task.file_path}")
return True

Expand Down
6 changes: 5 additions & 1 deletion buzz/transcriber/transcriber.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@

from buzz.locale import _
from buzz.model_loader import TranscriptionModel
from buzz.paths import safe_filename_component
from buzz.settings.settings import Settings

DEFAULT_WHISPER_TEMPERATURE = (0.0, 0.2, 0.4, 0.6, 0.8, 1.0)
Expand Down Expand Up @@ -208,6 +209,7 @@ class Source(enum.Enum):
original_file_path: Optional[str] = None # Original path before speech extraction
delete_source_file: bool = False
url: Optional[str] = None
display_name: Optional[str] = None
fraction_downloaded: float = 0.0

def __post_init__(self):
Expand Down Expand Up @@ -258,11 +260,13 @@ def get_output_file_path(
output_format: OutputFormat,
output_directory: str | None = None,
export_file_name_template: str | None = None,
display_name: str | None = None,
):
input_file_name = os.path.splitext(os.path.basename(file_path))[0]
input_file_name = display_name or 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]
input_file_name = safe_filename_component(input_file_name)
date_time_now = datetime.datetime.now().strftime("%d-%b-%Y %H-%M-%S")

export_file_name_template = (
Expand Down
4 changes: 2 additions & 2 deletions buzz/widgets/main_window.py
Original file line number Diff line number Diff line change
Expand Up @@ -492,9 +492,9 @@ def on_task_completed(self, task: FileTranscriptionTask, segments: List[Segment]
# Update file path in database only for URL imports where file is downloaded
if task.source == FileTranscriptionTask.Source.URL_IMPORT and task.file_path:
logging.debug(f"Updating transcription file path: {task.file_path}")
# Use the file basename (video title) as the display name
# URL titles are UI metadata and must not be used as working paths.
basename = os.path.basename(task.file_path)
name = os.path.splitext(basename)[0] # Remove .wav extension
name = task.display_name or os.path.splitext(basename)[0]
self.transcription_service.update_transcription_file_and_name(task.uid, task.file_path, name)

# When plugins are enabled, run the after_transcription / save / on_complete
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
from buzz.db.entity.transcription import Transcription
from buzz.db.service.transcription_service import TranscriptionService
from buzz.locale import _
from buzz.paths import safe_filename_component
from buzz.transcriber.file_transcriber import write_output
from buzz.transcriber.docx_writer import write_speaker_docx
from buzz.transcriber.transcriber import (
Expand Down Expand Up @@ -139,11 +140,12 @@ def export_speakers_docx(self):

source_path = self.transcription.file or "transcript"
source_directory = os.path.dirname(source_path)
source_stem = (
source_title = (
self.transcription.name
or os.path.splitext(os.path.basename(source_path))[0]
or "transcript"
)
source_stem = safe_filename_component(source_title)
default_path = os.path.join(
source_directory,
f"{source_stem}_speakers.docx",
Expand All @@ -162,7 +164,7 @@ def export_speakers_docx(self):
try:
write_speaker_docx(
output_file_path,
source_stem,
source_title,
segments,
include_timestamps=(
timestamp_choice == QMessageBox.StandardButton.Yes
Expand Down
22 changes: 21 additions & 1 deletion tests/db/entity/transcription_test.py
Original file line number Diff line number Diff line change
@@ -1,10 +1,30 @@
import pytest
from uuid import uuid4

from buzz.db.entity.transcription import Transcription
from buzz.transcriber.transcriber import FileTranscriptionTask, OutputFormat


class TestTranscription:
def test_url_export_uses_safe_name_instead_of_internal_audio_name(
self, monkeypatch, tmp_path
):
title = "中文 | test?."
transcription = Transcription(
file=str(tmp_path / "audio.wav"),
name=title,
source=FileTranscriptionTask.Source.URL_IMPORT.value,
)
monkeypatch.setattr(
"buzz.db.entity.transcription.Settings.get_default_export_file_template",
lambda _settings: "{{ input_file_name }}",
)

output_path = transcription.get_output_file_path(OutputFormat.SRT)

assert transcription.name == title
assert output_path != str(tmp_path / "audio.srt")
assert output_path == str(tmp_path / "中文 _ test_#.srt")

def test_transcription_creation_with_name_and_notes(self):
"""Test creating a transcription with name and notes fields"""
transcription = Transcription(
Expand Down
18 changes: 18 additions & 0 deletions tests/paths_test.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,18 @@
import pytest

from buzz.paths import safe_filename_component


@pytest.mark.parametrize(
"name,expected",
[
("中文 | QVD-123.", "中文 _ QVD-123#"),
("foo:bar?.wav", "foo_bar_.wav"),
("trailing-dot.", "trailing-dot#"),
("trailing-space ", "trailing-space#"),
("emoji-🚀", "emoji-🚀"),
("CON", "_CON"),
],
)
def test_safe_filename_component(name, expected):
assert safe_filename_component(name) == expected
Loading
Loading