feat(transcribe): Whisper transcription module with Groq→OpenAI fallback (#277)

Adds agent_reach/transcribe.py (download → compress → chunk → transcribe with provider fallback, fully mocked tests). Maintainer follow-up on the branch: wired an agent-reach transcribe CLI subcommand + skill docs so agents can actually invoke it, added the missing configure openai-key branch the error hint referenced, removed two dead static methods. 103 tests pass; wheel-gate clean.

.env.example

@@ -13,3 +13,6 @@
 
 # Groq Whisper (optional, for video transcription) — https://console.groq.com
 # GROQ_API_KEY=gsk_your_key_here
+
+# OpenAI Whisper (optional fallback when Groq is rate-limited) — https://platform.openai.com
+# OPENAI_API_KEY=sk-your_key_here

agent_reach/channels/youtube.py

@@ -17,6 +17,7 @@ class YouTubeChannel(Channel):
 
     def can_handle(self, url: str) -> bool:
         from urllib.parse import urlparse
+
         d = urlparse(url).netloc.lower()
         return "youtube.com" in d or "youtu.be" in d
 
@@ -40,7 +41,31 @@ class YouTubeChannel(Channel):
                 has_js_config = "--js-runtimes" in read_utf8_text(ytdlp_config)
             if not has_js_config:
                 return "warn", (
-                    "yt-dlp 已安装但未配置 JS runtime。运行：\n"
-                    f"  {render_ytdlp_fix_command()}"
+                    f"yt-dlp 已安装但未配置 JS runtime。运行：\n  {render_ytdlp_fix_command()}"
                 )
-        return "ok", "可提取视频信息和字幕"
+        # Surface transcription readiness so `doctor` reports it.
+        msg = "可提取视频信息和字幕"
+        if config is not None:
+            providers = []
+            if config.is_configured("groq_whisper"):
+                providers.append("groq")
+            if config.is_configured("openai_whisper"):
+                providers.append("openai")
+            if providers:
+                if not shutil.which("ffmpeg"):
+                    msg += "（音频转写需安装 ffmpeg）"
+                else:
+                    msg += f"，可转写音频（{'→'.join(providers)}）"
+        return "ok", msg
+
+    def transcribe(self, url: str, *, provider: str = "auto", config=None) -> str:
+        """Download a YouTube video's audio and return its transcript.
+
+        Delegates to :func:`agent_reach.transcribe.transcribe`. Imported lazily
+        so the channel module stays cheap to import for users who never
+        transcribe.
+        """
+        from agent_reach.transcribe import transcribe as _transcribe
+
+        return _transcribe(url, provider=provider, config=config)
+

agent_reach/cli.py

@@ -79,7 +79,7 @@ def main():
     # ── configure ──
     p_conf = sub.add_parser("configure", help="Set a config value or auto-extract from browser")
     p_conf.add_argument("key", nargs="?", default=None,
-                        choices=["proxy", "github-token", "groq-key",
+                        choices=["proxy", "github-token", "groq-key", "openai-key",
                                  "twitter-cookies", "youtube-cookies",
                                  "xhs-cookies"],
                         help="What to configure (omit if using --from-browser)")
@@ -111,6 +111,14 @@ def main():
     p_format.add_argument("platform", choices=["xhs"], help="Platform to format (xhs)")
 
     # ── check-update ──
+    # ── transcribe ──
+    p_tr = sub.add_parser("transcribe", help="Transcribe a URL or local audio file (Whisper via Groq/OpenAI)")
+    p_tr.add_argument("source", help="Audio/video URL or local file path")
+    p_tr.add_argument("--provider", choices=["auto", "groq", "openai"], default="auto",
+                      help="Transcription provider (default: auto = groq → openai fallback)")
+    p_tr.add_argument("-o", "--output", default=None,
+                      help="Write transcript to a file instead of stdout")
+
     sub.add_parser("check-update", help="Check for new versions and changes")
 
     # ── watch ──
@@ -150,6 +158,8 @@ def main():
         _cmd_skill(args)
     elif args.command == "format":
         _cmd_format(args)
+    elif args.command == "transcribe":
+        _cmd_transcribe(args)
 
 
 # ── Command handlers ────────────────────────────────
@@ -1125,6 +1135,29 @@ def _cmd_configure(args):
         config.set("groq_api_key", value)
         print(f"✅ Groq key configured!")
 
+    elif args.key == "openai-key":
+        config.set("openai_api_key", value)
+        print(f"✅ OpenAI key configured!")
+
+
+def _cmd_transcribe(args):
+    """Transcribe a URL or local audio file via Whisper (Groq → OpenAI fallback)."""
+    from pathlib import Path
+
+    from agent_reach.transcribe import TranscribeError, transcribe
+
+    try:
+        text = transcribe(args.source, provider=args.provider)
+    except TranscribeError as e:
+        print(f"❌ {e}")
+        sys.exit(1)
+
+    if args.output:
+        Path(args.output).write_text(text + "\n", encoding="utf-8")
+        print(f"✅ Transcript written to {args.output}")
+    else:
+        print(text)
+
 
 def _parse_twitter_cookie_input(value: str):
     """Parse Twitter cookie input from either separate values or a cookie header."""

agent_reach/config.py

@@ -23,6 +23,7 @@ class Config:
         "exa_search": ["exa_api_key"],
         "twitter_xreach": ["twitter_auth_token", "twitter_ct0"],  # legacy key name; used by twitter-cli
         "groq_whisper": ["groq_api_key"],
+        "openai_whisper": ["openai_api_key"],
         "github_token": ["github_token"],
     }
 

agent_reach/skill/references/video.md

@@ -39,6 +39,17 @@ yt-dlp --dump-json "ytsearch5:query"
 > **字幕注意**: 手动上传的字幕提取可靠；自动生成字幕可能存在行间重复，需后处理。
 > **评论注意**: `--write-comments` 基于网页抓取（非 YouTube Data API），部分评论可能丢失。
 
+### 无字幕兜底：Whisper 音频转写
+
+```bash
+# 视频没有字幕时的兜底：下载音频并用 Whisper 转写（Groq 免费 key 即可）
+agent-reach transcribe "https://www.youtube.com/watch?v=VIDEO_ID"
+agent-reach transcribe ./local_audio.mp3 -o /tmp/transcript.txt
+```
+
+> 需要先配置 key：`agent-reach configure groq-key gsk_xxx`（免费，console.groq.com）
+> 或 `agent-reach configure openai-key sk-xxx`。默认 auto 模式：groq 失败自动降级 openai。
+
 ## B站 / Bilibili (yt-dlp + bili-cli)
 
 ### 视频元数据 (yt-dlp)

agent_reach/transcribe.py

@@ -0,0 +1,253 @@
+# -*- coding: utf-8 -*-
+"""Whisper audio transcription with Groq → OpenAI fallback.
+
+Downloads audio (yt-dlp), compresses + chunks (ffmpeg), and posts to a
+Whisper-compatible API. Defaults to Groq's free `whisper-large-v3` and falls
+back to OpenAI's `whisper-1` on HTTP error.
+
+Public entry point:
+    transcribe(source, *, provider="auto", out_dir=None, config=None) -> str
+
+Designed to be importable from channels (e.g. YouTubeChannel.transcribe).
+"""
+
+from __future__ import annotations
+
+import shutil
+import subprocess
+import tempfile
+from pathlib import Path
+from typing import List, Optional
+
+import requests
+
+from agent_reach.config import Config
+
+# Whisper API limit is 25MB; leave headroom for multipart overhead.
+SIZE_LIMIT_BYTES = 24 * 1024 * 1024
+CHUNK_SECONDS = 600  # 10 min — small enough that boundary cuts rarely lose meaning
+
+PROVIDERS = {
+    "groq": {
+        "endpoint": "https://api.groq.com/openai/v1/audio/transcriptions",
+        "model": "whisper-large-v3",
+        "key_field": "groq_api_key",
+    },
+    "openai": {
+        "endpoint": "https://api.openai.com/v1/audio/transcriptions",
+        "model": "whisper-1",
+        "key_field": "openai_api_key",
+    },
+}
+
+
+class TranscribeError(RuntimeError):
+    """Raised when transcription cannot complete."""
+
+
+class MissingDependency(TranscribeError):
+    """Raised when a required external binary is missing."""
+
+
+class NoProviderConfigured(TranscribeError):
+    """Raised when no provider has an API key configured."""
+
+
+def _require(binary: str) -> None:
+    if not shutil.which(binary):
+        raise MissingDependency(f"{binary} not found in PATH")
+
+
+def _run(cmd: List[str]) -> None:
+    """Run a subprocess, raising TranscribeError on nonzero exit."""
+    proc = subprocess.run(cmd, capture_output=True, text=True)
+    if proc.returncode != 0:
+        raise TranscribeError(
+            f"{cmd[0]} failed (exit {proc.returncode}): {proc.stderr.strip()[:300]}"
+        )
+
+
+def download_audio(url: str, out_dir: Path) -> Path:
+    """Download audio with yt-dlp into out_dir; return the resulting file path."""
+    _require("yt-dlp")
+    template = out_dir / "source.%(ext)s"
+    _run(
+        [
+            "yt-dlp",
+            "-x",
+            "--audio-format",
+            "m4a",
+            "--audio-quality",
+            "0",
+            "-o",
+            str(template),
+            url,
+        ]
+    )
+    files = sorted(out_dir.glob("source.*"))
+    if not files:
+        raise TranscribeError("yt-dlp produced no output file")
+    return files[0]
+
+
+def compress_audio(src: Path, out_dir: Path) -> Path:
+    """Re-encode to mono / 16kHz / 32kbps m4a — keeps most content under 25MB."""
+    _require("ffmpeg")
+    dst = out_dir / "compressed.m4a"
+    _run(
+        [
+            "ffmpeg",
+            "-loglevel",
+            "error",
+            "-y",
+            "-i",
+            str(src),
+            "-vn",
+            "-ac",
+            "1",
+            "-ar",
+            "16000",
+            "-b:a",
+            "32k",
+            str(dst),
+        ]
+    )
+    return dst
+
+
+def chunk_audio(src: Path, out_dir: Path, segment_seconds: int = CHUNK_SECONDS) -> List[Path]:
+    """Split src into segments. Re-encodes each segment so cuts align to keyframes."""
+    _require("ffmpeg")
+    pattern = out_dir / "chunk_%03d.m4a"
+    _run(
+        [
+            "ffmpeg",
+            "-loglevel",
+            "error",
+            "-y",
+            "-i",
+            str(src),
+            "-f",
+            "segment",
+            "-segment_time",
+            str(segment_seconds),
+            "-ac",
+            "1",
+            "-ar",
+            "16000",
+            "-b:a",
+            "32k",
+            str(pattern),
+        ]
+    )
+    chunks = sorted(out_dir.glob("chunk_*.m4a"))
+    if not chunks:
+        raise TranscribeError("ffmpeg produced no chunks")
+    return chunks
+
+
+def _provider_key(provider: str, config: Config) -> Optional[str]:
+    field = PROVIDERS[provider]["key_field"]
+    val = config.get(field)
+    return val or None
+
+
+def transcribe_chunk(
+    chunk: Path,
+    provider: str,
+    *,
+    config: Optional[Config] = None,
+    timeout: int = 120,
+) -> str:
+    """Transcribe one chunk via the named provider. Raises TranscribeError on failure."""
+    if provider not in PROVIDERS:
+        raise TranscribeError(f"unknown provider: {provider}")
+    cfg = config or Config()
+    key = _provider_key(provider, cfg)
+    if not key:
+        raise NoProviderConfigured(
+            f"{provider}: missing {PROVIDERS[provider]['key_field']} "
+            f"(configure with `agent-reach configure {provider}-key ...`)"
+        )
+
+    info = PROVIDERS[provider]
+    with chunk.open("rb") as fh:
+        try:
+            resp = requests.post(
+                info["endpoint"],
+                headers={"Authorization": f"Bearer {key}"},
+                files={"file": (chunk.name, fh, "audio/m4a")},
+                data={"model": info["model"], "response_format": "text"},
+                timeout=timeout,
+            )
+        except requests.RequestException as e:
+            raise TranscribeError(f"{provider}: network error: {e}") from e
+
+    if not resp.ok:
+        raise TranscribeError(f"{provider}: HTTP {resp.status_code}: {resp.text[:300]}")
+    return resp.text
+
+
+def _provider_order(provider: str) -> List[str]:
+    if provider == "auto":
+        return ["groq", "openai"]
+    if provider in PROVIDERS:
+        return [provider]
+    raise TranscribeError(f"unknown provider: {provider} (use groq|openai|auto)")
+
+
+def transcribe(
+    source: str,
+    *,
+    provider: str = "auto",
+    out_dir: Optional[Path] = None,
+    config: Optional[Config] = None,
+) -> str:
+    """Transcribe a URL or local file path. Returns the joined transcript text.
+
+    `provider` is one of `auto` (groq → openai), `groq`, or `openai`.
+    `out_dir` defaults to a fresh temp directory; intermediate files stay there.
+    """
+    cfg = config or Config()
+    order = _provider_order(provider)
+
+    # Validate at least one provider is configured before doing expensive work.
+    if not any(_provider_key(p, cfg) for p in order):
+        names = ", ".join(PROVIDERS[p]["key_field"] for p in order)
+        raise NoProviderConfigured(f"no provider key configured (need one of: {names})")
+
+    work_dir = Path(out_dir) if out_dir else Path(tempfile.mkdtemp(prefix="transcribe-"))
+    work_dir.mkdir(parents=True, exist_ok=True)
+
+    src_path = Path(source)
+    if src_path.is_file():
+        audio = src_path
+    else:
+        audio = download_audio(source, work_dir)
+
+    compressed = compress_audio(audio, work_dir)
+    if compressed.stat().st_size <= SIZE_LIMIT_BYTES:
+        chunks = [compressed]
+    else:
+        chunks = chunk_audio(compressed, work_dir)
+
+    pieces: List[str] = []
+    for chunk in chunks:
+        text = _transcribe_with_fallback(chunk, order, cfg)
+        pieces.append(text.strip())
+    return "\n".join(p for p in pieces if p)
+
+
+def _transcribe_with_fallback(chunk: Path, order: List[str], config: Config) -> str:
+    """Try each provider in order; return first success or raise the last error."""
+    last_err: Optional[Exception] = None
+    for p in order:
+        if not _provider_key(p, config):
+            # Skip silently — caller already validated at least one is configured.
+            continue
+        try:
+            return transcribe_chunk(chunk, p, config=config)
+        except TranscribeError as e:
+            last_err = e
+            continue
+    raise TranscribeError(f"all providers failed for {chunk.name}: {last_err}")

tests/test_cli.py

@@ -33,6 +33,21 @@ class TestCLI:
         assert "Agent Reach" in captured.out
         assert "✅" in captured.out
 
+    def test_transcribe_command_prints_text(self, capsys):
+        with patch("agent_reach.transcribe.transcribe", return_value="hello transcript"):
+            with patch("sys.argv", ["agent-reach", "transcribe", "audio.mp3"]):
+                main()
+        captured = capsys.readouterr()
+        assert "hello transcript" in captured.out
+
+    def test_transcribe_command_writes_output_file(self, capsys, tmp_path):
+        out_file = tmp_path / "t.txt"
+        with patch("agent_reach.transcribe.transcribe", return_value="saved text"):
+            with patch("sys.argv", ["agent-reach", "transcribe", "audio.mp3", "-o", str(out_file)]):
+                main()
+        assert out_file.read_text(encoding="utf-8").strip() == "saved text"
+        assert "Transcript written" in capsys.readouterr().out
+
     def test_parse_twitter_cookie_input_separate_values(self):
         auth_token, ct0 = cli._parse_twitter_cookie_input("token123 ct0abc")
         assert auth_token == "token123"

tests/test_transcribe.py

@@ -0,0 +1,262 @@
+# -*- coding: utf-8 -*-
+"""Tests for agent_reach.transcribe — provider routing, fallback, and errors."""
+
+from typing import List
+
+import pytest
+
+from agent_reach import transcribe as tr
+from agent_reach.config import Config
+
+# --- Fixtures ----------------------------------------------------------- #
+
+
+@pytest.fixture
+def fake_config(tmp_path, monkeypatch):
+    """A Config that writes to a temp dir and never touches the user's HOME."""
+    cfg_path = tmp_path / "config.yaml"
+    monkeypatch.setattr(Config, "CONFIG_DIR", tmp_path)
+    monkeypatch.setattr(Config, "CONFIG_FILE", cfg_path)
+    cfg = Config(config_path=cfg_path)
+    return cfg
+
+
+@pytest.fixture
+def chunk_file(tmp_path):
+    p = tmp_path / "chunk.m4a"
+    p.write_bytes(b"\x00fake-m4a-bytes")
+    return p
+
+
+class FakeResponse:
+    def __init__(self, status_code: int, text: str = ""):
+        self.status_code = status_code
+        self.text = text
+
+    @property
+    def ok(self) -> bool:
+        return 200 <= self.status_code < 300
+
+
+# --- transcribe_chunk: provider routing -------------------------------- #
+
+
+class TestTranscribeChunk:
+    def test_routes_to_groq_endpoint(self, monkeypatch, fake_config, chunk_file):
+        fake_config.set("groq_api_key", "gsk_test")
+        captured = {}
+
+        def fake_post(url, headers=None, files=None, data=None, timeout=None):
+            captured["url"] = url
+            captured["headers"] = headers
+            captured["model"] = data["model"]
+            return FakeResponse(200, "hello world")
+
+        monkeypatch.setattr(tr.requests, "post", fake_post)
+        text = tr.transcribe_chunk(chunk_file, "groq", config=fake_config)
+        assert text == "hello world"
+        assert captured["url"] == tr.PROVIDERS["groq"]["endpoint"]
+        assert captured["model"] == "whisper-large-v3"
+        assert captured["headers"]["Authorization"] == "Bearer gsk_test"
+
+    def test_routes_to_openai_endpoint(self, monkeypatch, fake_config, chunk_file):
+        fake_config.set("openai_api_key", "sk-test")
+        captured = {}
+
+        def fake_post(url, headers=None, files=None, data=None, timeout=None):
+            captured["url"] = url
+            captured["model"] = data["model"]
+            return FakeResponse(200, "openai output")
+
+        monkeypatch.setattr(tr.requests, "post", fake_post)
+        text = tr.transcribe_chunk(chunk_file, "openai", config=fake_config)
+        assert text == "openai output"
+        assert captured["url"] == tr.PROVIDERS["openai"]["endpoint"]
+        assert captured["model"] == "whisper-1"
+
+    def test_raises_when_key_missing(self, fake_config, chunk_file):
+        with pytest.raises(tr.NoProviderConfigured):
+            tr.transcribe_chunk(chunk_file, "groq", config=fake_config)
+
+    def test_raises_on_http_error(self, monkeypatch, fake_config, chunk_file):
+        fake_config.set("groq_api_key", "gsk_test")
+        monkeypatch.setattr(
+            tr.requests,
+            "post",
+            lambda *a, **k: FakeResponse(429, "rate limited"),
+        )
+        with pytest.raises(tr.TranscribeError, match="HTTP 429"):
+            tr.transcribe_chunk(chunk_file, "groq", config=fake_config)
+
+    def test_unknown_provider(self, fake_config, chunk_file):
+        with pytest.raises(tr.TranscribeError, match="unknown provider"):
+            tr.transcribe_chunk(chunk_file, "azure", config=fake_config)
+
+
+# --- _transcribe_with_fallback ----------------------------------------- #
+
+
+class TestFallback:
+    def test_groq_succeeds_no_openai_call(self, monkeypatch, fake_config, chunk_file):
+        fake_config.set("groq_api_key", "gsk_test")
+        fake_config.set("openai_api_key", "sk-test")
+        calls: List[str] = []
+
+        def fake_post(url, headers=None, files=None, data=None, timeout=None):
+            calls.append(url)
+            return FakeResponse(200, "from-groq")
+
+        monkeypatch.setattr(tr.requests, "post", fake_post)
+        text = tr._transcribe_with_fallback(chunk_file, ["groq", "openai"], fake_config)
+        assert text == "from-groq"
+        assert calls == [tr.PROVIDERS["groq"]["endpoint"]]
+
+    def test_groq_429_falls_back_to_openai(self, monkeypatch, fake_config, chunk_file):
+        fake_config.set("groq_api_key", "gsk_test")
+        fake_config.set("openai_api_key", "sk-test")
+        calls: List[str] = []
+
+        def fake_post(url, headers=None, files=None, data=None, timeout=None):
+            calls.append(url)
+            if url == tr.PROVIDERS["groq"]["endpoint"]:
+                return FakeResponse(429, "rate limited")
+            return FakeResponse(200, "from-openai")
+
+        monkeypatch.setattr(tr.requests, "post", fake_post)
+        text = tr._transcribe_with_fallback(chunk_file, ["groq", "openai"], fake_config)
+        assert text == "from-openai"
+        assert calls == [
+            tr.PROVIDERS["groq"]["endpoint"],
+            tr.PROVIDERS["openai"]["endpoint"],
+        ]
+
+    def test_skip_unconfigured_provider(self, monkeypatch, fake_config, chunk_file):
+        # Only openai key configured — fallback should skip groq silently.
+        fake_config.set("openai_api_key", "sk-test")
+        calls: List[str] = []
+
+        def fake_post(url, headers=None, files=None, data=None, timeout=None):
+            calls.append(url)
+            return FakeResponse(200, "via-openai")
+
+        monkeypatch.setattr(tr.requests, "post", fake_post)
+        text = tr._transcribe_with_fallback(chunk_file, ["groq", "openai"], fake_config)
+        assert text == "via-openai"
+        assert calls == [tr.PROVIDERS["openai"]["endpoint"]]
+
+    def test_all_fail_raises_with_last_error(self, monkeypatch, fake_config, chunk_file):
+        fake_config.set("groq_api_key", "gsk_test")
+        fake_config.set("openai_api_key", "sk-test")
+        monkeypatch.setattr(
+            tr.requests,
+            "post",
+            lambda *a, **k: FakeResponse(500, "boom"),
+        )
+        with pytest.raises(tr.TranscribeError, match="all providers failed"):
+            tr._transcribe_with_fallback(chunk_file, ["groq", "openai"], fake_config)
+
+
+# --- transcribe (orchestrator) ---------------------------------------- #
+
+
+class TestOrchestrator:
+    def test_local_file_skips_yt_dlp(self, monkeypatch, fake_config, tmp_path, chunk_file):
+        fake_config.set("groq_api_key", "gsk_test")
+
+        def boom_download(*a, **k):
+            raise AssertionError("yt-dlp must not be called for local files")
+
+        # Stub heavy external steps to no-ops that keep file paths valid.
+        compressed = tmp_path / "compressed.m4a"
+        compressed.write_bytes(b"x" * 1024)
+
+        def fake_compress(src, out_dir):
+            return compressed
+
+        monkeypatch.setattr(tr, "download_audio", boom_download)
+        monkeypatch.setattr(tr, "compress_audio", fake_compress)
+        monkeypatch.setattr(
+            tr.requests,
+            "post",
+            lambda *a, **k: FakeResponse(200, "transcript text"),
+        )
+
+        text = tr.transcribe(
+            str(chunk_file),
+            out_dir=tmp_path / "work",
+            config=fake_config,
+        )
+        assert text == "transcript text"
+
+    def test_chunks_concatenated_with_newlines(
+        self, monkeypatch, fake_config, tmp_path, chunk_file
+    ):
+        fake_config.set("groq_api_key", "gsk_test")
+        # Force the "needs chunking" path by writing a file above the size limit.
+        big = tmp_path / "compressed.m4a"
+        big.write_bytes(b"x" * (tr.SIZE_LIMIT_BYTES + 1))
+        monkeypatch.setattr(tr, "compress_audio", lambda src, out_dir: big)
+        c1 = tmp_path / "chunk_001.m4a"
+        c2 = tmp_path / "chunk_002.m4a"
+        c1.write_bytes(b"a")
+        c2.write_bytes(b"b")
+        monkeypatch.setattr(tr, "chunk_audio", lambda src, out_dir: [c1, c2])
+
+        responses = iter(["part one ", "part two "])
+        monkeypatch.setattr(
+            tr.requests,
+            "post",
+            lambda *a, **k: FakeResponse(200, next(responses)),
+        )
+
+        text = tr.transcribe(
+            str(chunk_file),
+            out_dir=tmp_path / "work",
+            config=fake_config,
+        )
+        assert text == "part one\npart two"
+
+    def test_no_provider_configured_fails_fast(self, fake_config, chunk_file):
+        with pytest.raises(tr.NoProviderConfigured):
+            tr.transcribe(str(chunk_file), config=fake_config)
+
+    def test_invalid_provider_string(self, fake_config, chunk_file):
+        with pytest.raises(tr.TranscribeError, match="unknown provider"):
+            tr.transcribe(str(chunk_file), provider="azure", config=fake_config)
+
+
+# --- YouTubeChannel integration --------------------------------------- #
+
+
+class TestYouTubeChannelTranscribe:
+    def test_delegates_to_transcribe(self, monkeypatch, fake_config):
+        from agent_reach.channels.youtube import YouTubeChannel
+
+        captured = {}
+
+        def fake_transcribe(source, *, provider="auto", out_dir=None, config=None):
+            captured["source"] = source
+            captured["provider"] = provider
+            captured["config"] = config
+            return "delegated text"
+
+        monkeypatch.setattr(tr, "transcribe", fake_transcribe)
+        out = YouTubeChannel().transcribe(
+            "https://youtu.be/abc", provider="groq", config=fake_config
+        )
+        assert out == "delegated text"
+        assert captured["source"] == "https://youtu.be/abc"
+        assert captured["provider"] == "groq"
+        assert captured["config"] is fake_config
+
+
+# --- Config feature requirement --------------------------------------- #
+
+
+class TestConfigOpenAIWhisper:
+    def test_openai_whisper_feature_registered(self, fake_config):
+        assert "openai_whisper" in Config.FEATURE_REQUIREMENTS
+        assert Config.FEATURE_REQUIREMENTS["openai_whisper"] == ["openai_api_key"]
+        assert not fake_config.is_configured("openai_whisper")
+        fake_config.set("openai_api_key", "sk-test")
+        assert fake_config.is_configured("openai_whisper")