feat: 默认只下载音频（--keep-video 可选）+ Apple Silicon MPS 支持

pyproject.toml

@@ -15,7 +15,7 @@ dependencies = [
   "rich>=13.9.4",
   "tqdm>=4.67.3",
   "typer>=0.12.5",
-  "yt-dlp>=2025.3.31",
+  "yt-dlp>=2026.7.4",
 ]
 
 [project.optional-dependencies]

src/b2t/cli.py

@@ -49,8 +49,11 @@ def transcribe(
         prompt: str = typer.Option("", "--prompt", help=tr(language, "opt_prompt_help")),
         output: Path | None = typer.Option(None, "--output", help=tr(language, "opt_output_help")),
         workspace: Path | None = typer.Option(None, "--workspace", help=tr(language, "opt_workspace_help")),
+        keep_video: bool = typer.Option(False, "--keep-video", help=tr(language, "opt_keep_video_help")),
     ) -> None:
         """Download or open media, then transcribe it with the selected provider."""
+        if keep_video:
+            os.environ["B2T_KEEP_VIDEO"] = "1"
         try:
             settings, config = _load_runtime(workspace=workspace, provider=provider, model=model)
             renderer = TqdmTaskRenderer(config.language)
@@ -91,8 +94,11 @@ def batch_transcribe(
         model: str | None = typer.Option(None, "--model", help=tr(language, "opt_model_help")),
         prompt: str = typer.Option("", "--prompt", help=tr(language, "opt_prompt_help")),
         workspace: Path | None = typer.Option(None, "--workspace", help=tr(language, "opt_workspace_help")),
+        keep_video: bool = typer.Option(False, "--keep-video", help=tr(language, "opt_keep_video_help")),
     ) -> None:
         """Submit multiple transcription tasks from arguments or a newline-separated file."""
+        if keep_video:
+            os.environ["B2T_KEEP_VIDEO"] = "1"
         selected_language = _detect_preferred_language(workspace)
         try:
             settings, config = _load_runtime(workspace=workspace, provider=provider, model=model)

src/b2t/downloaders/ytdlp.py

@@ -76,15 +76,24 @@ def progress_hook(data: dict[str, Any]) -> None:
 
     def _build_ydl_opts(self, source: SourceRef, settings: Settings) -> dict[str, Any]:
         ydl_opts: dict[str, Any] = {
-            "format": "bv*+ba/b",
-            "merge_output_format": "mp4",
             "noplaylist": True,
             "outtmpl": str(settings.downloads_dir / "%(id)s.%(ext)s"),
             "noprogress": True,
             "quiet": True,
             "no_warnings": True,
         }
 
+        # Transcription only needs the audio track, so download audio-only by
+        # default instead of the best video (which can be a multi-GB 4K file
+        # that `ffmpeg -vn` immediately discards). Set B2T_KEEP_VIDEO=1 or pass
+        # --keep-video to download and keep the full video as well.
+        keep_video = os.getenv("B2T_KEEP_VIDEO", "").strip().lower() in {"1", "true", "yes", "on"}
+        if keep_video:
+            ydl_opts["format"] = "bv*+ba/b"
+            ydl_opts["merge_output_format"] = "mp4"
+        else:
+            ydl_opts["format"] = "ba/bestaudio/best"
+
         # Support cookies for authenticated access to Bilibili.
         # Priority: B2T_COOKIE_FILE env var > cookies.txt in workspace.
         cookie_file = os.getenv("B2T_COOKIE_FILE")

src/b2t/i18n.py

@@ -29,6 +29,7 @@
         "opt_model_help": "模型名称。",
         "opt_prompt_help": "转写提示词（可选）。",
         "opt_output_help": "输出文件或目录。",
+        "opt_keep_video_help": "下载并保留完整视频（默认只下载音频）。",
         "opt_workspace_help": "工作目录，默认 ./.b2t。",
         "opt_host_help": "监听地址。",
         "opt_port_help": "监听端口。",
@@ -219,6 +220,7 @@
         "opt_model_help": "Model name.",
         "opt_prompt_help": "Optional transcription prompt.",
         "opt_output_help": "Output file or directory.",
+        "opt_keep_video_help": "Download and keep the full video (audio-only by default).",
         "opt_workspace_help": "Workspace root, defaults to ./.b2t.",
         "opt_host_help": "Bind address.",
         "opt_port_help": "Bind port.",

src/b2t/transcribers/whisper_local.py

@@ -34,7 +34,9 @@ def transcribe(
             # `verbose=False` keeps Whisper text output quiet while still driving the internal tqdm loop.
             "verbose": False,
         }
-        if self.device == "cpu":
+        if self.device in ("cpu", "mps"):
+            # fp16 is NaN-prone on some MPS/torch builds and Whisper re-casts
+            # weights per forward pass anyway; fp32 is the safe choice off CUDA.
             transcribe_options["fp16"] = False
         with whisper_progress(progress):
             result = model.transcribe(str(audio_path), **transcribe_options)
@@ -52,13 +54,30 @@ def _ensure_model(self) -> Any:
             return self._model
 
         try:
+            import torch
             import whisper
         except ImportError as exc:
             raise RuntimeError(build_whisper_import_error_message()) from exc
 
         if self.device is None:
-            self.device = "cuda" if whisper.torch.cuda.is_available() else "cpu"
-        self._model = whisper.load_model(self.model_name, device=self.device)
+            if torch.cuda.is_available():
+                self.device = "cuda"
+            elif torch.backends.mps.is_available():
+                # Apple Silicon GPU (Metal Performance Shaders).
+                self.device = "mps"
+            else:
+                self.device = "cpu"
+
+        if self.device == "mps":
+            # Whisper registers alignment_heads as a sparse buffer, which some
+            # torch builds cannot move to MPS (NotImplementedError at load).
+            try:
+                self._model = whisper.load_model(self.model_name, device="mps")
+            except Exception:
+                self.device = "cpu"
+                self._model = whisper.load_model(self.model_name, device="cpu")
+        else:
+            self._model = whisper.load_model(self.model_name, device=self.device)
         return self._model
 
 

tests/test_whisper_local.py

@@ -1,10 +1,73 @@
+import sys
+import types
+from pathlib import Path
+
 from b2t.progress import ProgressReporter
 from b2t.transcribers.whisper_local import (
+    LocalWhisperTranscriber,
     WhisperProgressTqdm,
     build_whisper_import_error_message,
 )
 
 
+def _install_fake_whisper(monkeypatch, *, cuda: bool = False, mps: bool = False, mps_load_fails: bool = False) -> dict:
+    calls: dict = {"load_devices": []}
+
+    fake_torch = types.ModuleType("torch")
+    fake_torch.cuda = types.SimpleNamespace(is_available=lambda: cuda)
+    fake_torch.backends = types.SimpleNamespace(
+        mps=types.SimpleNamespace(is_available=lambda: mps),
+    )
+
+    class FakeModel:
+        def transcribe(self, audio_path, **options):
+            calls["transcribe_options"] = options
+            return {"text": "hello", "segments": [], "language": "en"}
+
+    def load_model(name, device=None):
+        calls["load_devices"].append(device)
+        if device == "mps" and mps_load_fails:
+            raise NotImplementedError("sparse tensors not supported on MPS")
+        return FakeModel()
+
+    fake_whisper = types.ModuleType("whisper")
+    fake_whisper.load_model = load_model
+
+    monkeypatch.setitem(sys.modules, "torch", fake_torch)
+    monkeypatch.setitem(sys.modules, "whisper", fake_whisper)
+    return calls
+
+
+def test_device_auto_selects_mps_when_cuda_unavailable(monkeypatch) -> None:
+    calls = _install_fake_whisper(monkeypatch, cuda=False, mps=True)
+    transcriber = LocalWhisperTranscriber(model="small")
+
+    transcriber._ensure_model()
+
+    assert transcriber.device == "mps"
+    assert calls["load_devices"] == ["mps"]
+
+
+def test_mps_load_failure_falls_back_to_cpu(monkeypatch) -> None:
+    calls = _install_fake_whisper(monkeypatch, cuda=False, mps=True, mps_load_fails=True)
+    transcriber = LocalWhisperTranscriber(model="small")
+
+    transcriber._ensure_model()
+
+    assert transcriber.device == "cpu"
+    assert calls["load_devices"] == ["mps", "cpu"]
+
+
+def test_transcribe_disables_fp16_on_mps(monkeypatch) -> None:
+    calls = _install_fake_whisper(monkeypatch, cuda=False, mps=True)
+    transcriber = LocalWhisperTranscriber(model="small")
+
+    result = transcriber.transcribe(Path("dummy.wav"))
+
+    assert calls["transcribe_options"]["fp16"] is False
+    assert result["device"] == "mps"
+
+
 def test_build_whisper_import_error_message_reports_missing_install() -> None:
     message = build_whisper_import_error_message(
         whisper_available=False,

tests/test_ytdlp_downloader.py

@@ -38,6 +38,56 @@ def test_ytdlp_options_select_playlist_item_when_page_is_set(tmp_path) -> None:
     assert opts["outtmpl"] == str(settings.downloads_dir / "%(id)s.%(playlist_index)02d.%(ext)s")
 
 
+def test_ytdlp_options_download_audio_only_by_default(tmp_path, monkeypatch) -> None:
+    monkeypatch.delenv("B2T_KEEP_VIDEO", raising=False)
+    settings = Settings.from_workspace(tmp_path / ".b2t")
+    source = SourceRef(
+        raw_input="BV1xx411c7XD",
+        kind="bilibili",
+        display_name="BV1xx411c7XD",
+        bv="BV1xx411c7XD",
+        url="https://www.bilibili.com/video/BV1xx411c7XD",
+    )
+
+    opts = YtDlpDownloader()._build_ydl_opts(source, settings)
+
+    assert opts["format"] == "ba/bestaudio/best"
+    assert "merge_output_format" not in opts
+
+
+def test_ytdlp_options_download_full_video_when_keep_video_enabled(tmp_path, monkeypatch) -> None:
+    monkeypatch.setenv("B2T_KEEP_VIDEO", "1")
+    settings = Settings.from_workspace(tmp_path / ".b2t")
+    source = SourceRef(
+        raw_input="BV1xx411c7XD",
+        kind="bilibili",
+        display_name="BV1xx411c7XD",
+        bv="BV1xx411c7XD",
+        url="https://www.bilibili.com/video/BV1xx411c7XD",
+    )
+
+    opts = YtDlpDownloader()._build_ydl_opts(source, settings)
+
+    assert opts["format"] == "bv*+ba/b"
+    assert opts["merge_output_format"] == "mp4"
+
+
+def test_ytdlp_options_falsey_keep_video_still_audio_only(tmp_path, monkeypatch) -> None:
+    monkeypatch.setenv("B2T_KEEP_VIDEO", "0")
+    settings = Settings.from_workspace(tmp_path / ".b2t")
+    source = SourceRef(
+        raw_input="BV1xx411c7XD",
+        kind="bilibili",
+        display_name="BV1xx411c7XD",
+        bv="BV1xx411c7XD",
+        url="https://www.bilibili.com/video/BV1xx411c7XD",
+    )
+
+    opts = YtDlpDownloader()._build_ydl_opts(source, settings)
+
+    assert opts["format"] == "ba/bestaudio/best"
+
+
 def test_ytdlp_options_bypass_proxy_by_default(tmp_path, monkeypatch) -> None:
     monkeypatch.delenv("B2T_USE_PROXY", raising=False)
     settings = Settings.from_workspace(tmp_path / ".b2t")