refactor(file_transcriber): extract methods from FileTranscriber.run to fix R0915

buzz/transcriber/file_transcriber.py

@@ -38,88 +38,9 @@ def __init__(self, task: FileTranscriptionTask, parent: Optional["QObject"] = No
     @pyqtSlot()
     def run(self):
         if self.transcription_task.source == FileTranscriptionTask.Source.URL_IMPORT:
-            cookiefile = os.getenv("BUZZ_DOWNLOAD_COOKIEFILE")
-
-            # First extract info to get the video title
-            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"
-
-            # Sanitize title for use as filename
-            video_title = YoutubeDL.sanitize_info({"title": video_title})["title"]
-            # Remove characters that are problematic in filenames
-            for char in ['/', '\\', ':', '*', '?', '"', '<', '>', '|']:
-                video_title = video_title.replace(char, '_')
-
-            # Create temp directory and use video title as filename
-            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())
-
-            options = {
-                "format": "bestaudio/best",
-                "progress_hooks": [self.on_download_progress],
-                "outtmpl": temp_output_path,
-                "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])
-            except Exception as exc:
-                logging.debug(f"Error downloading audio: {exc.msg}")
-                self.error.emit(exc.msg)
+            if not self._download_from_url():
                 return
 
-            cmd = [
-                "ffmpeg",
-                "-nostdin",
-                "-threads", "0",
-                "-i", temp_output_path,
-                "-ac", "1",
-                "-ar", str(whisper_audio.SAMPLE_RATE),
-                "-acodec", "pcm_s16le",
-                "-loglevel", "panic",
-                wav_file
-            ]
-
-            if sys.platform == "win32":
-                si = subprocess.STARTUPINFO()
-                si.dwFlags |= subprocess.STARTF_USESHOWWINDOW
-                si.wShowWindow = subprocess.SW_HIDE
-                result = subprocess.run(
-                    cmd,
-                    capture_output=True,
-                    startupinfo=si,
-                    env=app_env,
-                    creationflags=subprocess.CREATE_NO_WINDOW
-                )
-            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()}")
-
-            self.transcription_task.file_path = wav_file
-            logging.debug(f"Downloaded audio to file: {self.transcription_task.file_path}")
-
         try:
             segments = self.transcribe()
         except Exception as exc:
@@ -149,22 +70,104 @@ def run(self):
             )
 
         if self.transcription_task.source == FileTranscriptionTask.Source.FOLDER_WATCH:
-            # 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
+            self._handle_folder_watch()
+
+    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())
+
+        options = {
+            "format": "bestaudio/best",
+            "progress_hooks": [self.on_download_progress],
+            "outtmpl": temp_output_path,
+            "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])
+        except Exception as exc:
+            logging.debug(f"Error downloading audio: {exc.msg}")
+            self.error.emit(exc.msg)
+            return False
+
+        cmd = [
+            "ffmpeg",
+            "-nostdin",
+            "-threads", "0",
+            "-i", temp_output_path,
+            "-ac", "1",
+            "-ar", str(whisper_audio.SAMPLE_RATE),
+            "-acodec", "pcm_s16le",
+            "-loglevel", "panic",
+            wav_file
+        ]
+
+        if sys.platform == "win32":
+            si = subprocess.STARTUPINFO()
+            si.dwFlags |= subprocess.STARTF_USESHOWWINDOW
+            si.wShowWindow = subprocess.SW_HIDE
+            result = subprocess.run(
+                cmd,
+                capture_output=True,
+                startupinfo=si,
+                env=app_env,
+                creationflags=subprocess.CREATE_NO_WINDOW
             )
-            if source_path and os.path.exists(source_path):
-                if self.transcription_task.delete_source_file:
-                    os.remove(source_path)
-                else:
-                    shutil.move(
-                        source_path,
-                        os.path.join(
-                            self.transcription_task.output_directory,
-                            os.path.basename(source_path),
-                        ),
-                    )
+        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()}")
+
+        self.transcription_task.file_path = wav_file
+        logging.debug(f"Downloaded audio to file: {self.transcription_task.file_path}")
+        return True
+
+    def _handle_folder_watch(self):
+        source_path = (
+            self.transcription_task.original_file_path
+            or self.transcription_task.file_path
+        )
+        if source_path and os.path.exists(source_path):
+            if self.transcription_task.delete_source_file:
+                os.remove(source_path)
+            else:
+                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":