fix: Suppress all WhisperX/pyannote and Gemini console output

- Add suppress_whisperx_output() context manager to silence WhisperX output
- Add suppress_gemini_output() context manager to silence Gemini API output
- Suppress warnings, logging, and stdout/stderr from both libraries
- Output is still shown when --debug flag is enabled
- Fixes #71

Author: dohyeondk

src/sub_tools/intelligence/gemini.py

@@ -1,4 +1,9 @@
 import asyncio
+import logging
+import os
+import sys
+import warnings
+from contextlib import contextmanager
 from typing import Callable, Optional
 
 from google import genai
@@ -14,7 +19,7 @@
 
 
 def proofread() -> None:
-    """Proofread the source SRT file with Gemini."""
+    """Proofread the source SRT file using Gemini."""
 
     if should_skip(f"{config.source_language}.srt"):
         return
@@ -23,7 +28,7 @@ def proofread() -> None:
 
 
 async def _proofread() -> None:
-    info("Proofreading with Gemini...")
+    info("Proofreading using Gemini...")
 
     language_code = config.source_language
     language = get_language_name(language_code)
@@ -65,7 +70,7 @@ async def _proofread() -> None:
 
 
 def translate() -> None:
-    """Translate the source SRT file with Gemini."""
+    """Translate the source SRT file using Gemini."""
 
     asyncio.run(_translate())
 
@@ -87,7 +92,7 @@ async def _translate() -> None:
     if not target_language_codes:
         return
 
-    info("Translating with Gemini...")
+    info("Translating using Gemini...")
 
     tasks = []
 
@@ -116,6 +121,57 @@ async def _translate() -> None:
         await asyncio.gather(*tasks)
 
 
+@contextmanager
+def _suppress_gemini_output():
+    """
+    Context manager to suppress all Gemini API output unless in debug mode.
+
+    Suppresses:
+    - All warnings from google.genai and related libraries
+    - All logging output from these libraries
+    - All stdout/stderr output
+    """
+    if config.debug:
+        yield
+        return
+
+    # Suppress warnings
+    with warnings.catch_warnings():
+        warnings.filterwarnings("ignore")
+
+        # Suppress logging
+        logging_modules = [
+            "google",
+            "google.genai",
+            "google.api_core",
+            "grpc",
+        ]
+        original_levels = {}
+        for module_name in logging_modules:
+            logger = logging.getLogger(module_name)
+            original_levels[module_name] = logger.level
+            logger.setLevel(logging.ERROR)
+
+        # Suppress stdout/stderr
+        original_stdout = sys.stdout
+        original_stderr = sys.stderr
+
+        with open(os.devnull, "w") as devnull:
+            sys.stdout = devnull
+            sys.stderr = devnull
+
+            try:
+                yield
+            finally:
+                # Restore stdout/stderr
+                sys.stdout = original_stdout
+                sys.stderr = original_stderr
+
+                # Restore logging levels
+                for module_name, level in original_levels.items():
+                    logging.getLogger(module_name).setLevel(level)
+
+
 async def _translate_language(
     file: types.File,
     srt_content: str,
@@ -173,50 +229,53 @@ async def _call_gemini_api(
     """
     Helper method to call Gemini API with retries for rate limits.
     """
-    client = genai.Client(api_key=config.gemini_api_key)
-
-    # Build parts for the content
-    parts = []
-    if file:
-        parts.append(file)
-    if text:
-        parts.append(types.Part.from_text(text=text))
-
-    tools = [
-        types.Tool(google_search=types.GoogleSearch()),
-    ]
-
-    for attempt in range(config.retry):
-        try:
-            response = await client.aio.models.generate_content(
-                model=config.gemini_model,
-                contents=parts,
-                config=types.GenerateContentConfig(
-                    system_instruction=system_instruction,
-                    thinking_config=types.ThinkingConfig(
-                        include_thoughts=True, thinking_level=types.ThinkingLevel.HIGH
+    with _suppress_gemini_output():
+        client = genai.Client(api_key=config.gemini_api_key)
+
+        # Build parts for the content
+        parts = []
+        if file:
+            parts.append(file)
+        if text:
+            parts.append(types.Part.from_text(text=text))
+
+        tools = [
+            types.Tool(google_search=types.GoogleSearch()),
+        ]
+
+        for attempt in range(config.retry):
+            try:
+                response = await client.aio.models.generate_content(
+                    model=config.gemini_model,
+                    contents=parts,
+                    config=types.GenerateContentConfig(
+                        system_instruction=system_instruction,
+                        thinking_config=types.ThinkingConfig(
+                            include_thoughts=True,
+                            thinking_level=types.ThinkingLevel.HIGH,
+                        ),
+                        tools=tools,
                     ),
-                    tools=tools,
-                ),
-            )
-            text = response.text
-            if text:
-                with open(output_file, "w") as f:
-                    f.write(text)
-                return
-
-        except google_exceptions.ResourceExhausted as e:
-            if attempt < config.retry - 1:
-                wait_time = 2**attempt  # Exponential backoff: 1, 2, 4 seconds
-                await asyncio.sleep(wait_time)
-                continue
-            else:
+                )
+                text = response.text
+                if text:
+                    with open(output_file, "w") as f:
+                        f.write(text)
+                    return
+
+            except google_exceptions.ResourceExhausted as e:
+                if attempt < config.retry - 1:
+                    wait_time = 2**attempt  # Exponential backoff: 1, 2, 4 seconds
+                    await asyncio.sleep(wait_time)
+                    continue
+                else:
+                    raise e
+            except Exception as e:
                 raise e
-        except Exception as e:
-            raise e
 
 
 def _upload_file(file_path: str) -> types.File:
-    client = genai.Client(api_key=config.gemini_api_key)
-    file = client.files.upload(file=file_path)
-    return file
+    with _suppress_gemini_output():
+        client = genai.Client(api_key=config.gemini_api_key)
+        file = client.files.upload(file=file_path)
+        return file

src/sub_tools/intelligence/whisperx.py

@@ -1,5 +1,12 @@
+import logging
+import os
+import sys
+import warnings
+from contextlib import contextmanager
+
 import whisperx
 
+from sub_tools.system.console import info
 from sub_tools.system.file import should_skip
 
 from ..config import config
@@ -10,35 +17,90 @@ def transcribe() -> None:
     if should_skip(config.srt_file):
         return
 
+    info("Transcribing audio using WhisperX...")
+
     try:
-        # Load model
-        model = whisperx.load_model(
-            config.whisperx_model,
-            device=config.whisperx_device,
-            compute_type=config.whisperx_compute_type,
-            language=config.source_language,
-        )
-
-        # Transcribe audio
-        audio = whisperx.load_audio(config.audio_file)
-        result = model.transcribe(audio, batch_size=config.whisperx_batch_size)
-
-        # Align whisper output
-        model_a, metadata = whisperx.load_align_model(
-            language_code=config.source_language, device=config.whisperx_device
-        )
-        result = whisperx.align(
-            result["segments"],
-            model_a,
-            metadata,
-            audio,
-            config.whisperx_device,
-            return_char_alignments=False,
-        )
-
-        # Save as SRT
-        serialize_subtitles(result["segments"], config.srt_file)
+        with _suppress_whisperx_output():
+            # Load model
+            model = whisperx.load_model(
+                config.whisperx_model,
+                device=config.whisperx_device,
+                compute_type=config.whisperx_compute_type,
+                language=config.source_language,
+            )
+
+            # Transcribe audio
+            audio = whisperx.load_audio(config.audio_file)
+            result = model.transcribe(audio, batch_size=config.whisperx_batch_size)
+
+            # Align whisper output
+            model_a, metadata = whisperx.load_align_model(
+                language_code=config.source_language, device=config.whisperx_device
+            )
+            result = whisperx.align(
+                result["segments"],
+                model_a,
+                metadata,
+                audio,
+                config.whisperx_device,
+                return_char_alignments=False,
+            )
+
+            # Save as SRT
+            serialize_subtitles(result["segments"], config.srt_file)
 
     except Exception as e:
         print(f"WhisperX transcription failed: {str(e)}")
         raise
+
+
+@contextmanager
+def _suppress_whisperx_output():
+    """
+    Context manager to suppress all WhisperX/pyannote output unless in debug mode.
+
+    Suppresses:
+    - All warnings from whisperx, pyannote, pytorch_lightning, and torch
+    - All logging output from these libraries
+    - All stdout/stderr output
+    """
+    if config.debug:
+        yield
+        return
+
+    # Suppress warnings
+    with warnings.catch_warnings():
+        warnings.filterwarnings("ignore")
+
+        # Suppress logging
+        logging_modules = [
+            "whisperx",
+            "pyannote",
+            "pytorch_lightning",
+            "lightning",
+            "torch",
+        ]
+        original_levels = {}
+        for module_name in logging_modules:
+            logger = logging.getLogger(module_name)
+            original_levels[module_name] = logger.level
+            logger.setLevel(logging.ERROR)
+
+        # Suppress stdout/stderr
+        original_stdout = sys.stdout
+        original_stderr = sys.stderr
+
+        with open(os.devnull, "w") as devnull:
+            sys.stdout = devnull
+            sys.stderr = devnull
+
+            try:
+                yield
+            finally:
+                # Restore stdout/stderr
+                sys.stdout = original_stdout
+                sys.stderr = original_stderr
+
+                # Restore logging levels
+                for module_name, level in original_levels.items():
+                    logging.getLogger(module_name).setLevel(level)