diff --git a/.env.example b/.env.example index 6e41ac3..60e70b1 100644 --- a/.env.example +++ b/.env.example @@ -84,6 +84,11 @@ TELEGRAM_ENABLED=false # TELEGRAM_ACCESS_CODE=replace-with-a-dedicated-high-entropy-code # Telegram turns automatically show one live-updating tool status message. +# Optional Telegram voice-note transcription through Cloudflare Workers AI. +# Use a Cloudflare API token with Workers AI Read and Write permissions. +# CLOUDFLARE_ACCOUNT_ID=replace-me +# CLOUDFLARE_API_TOKEN=replace-me + # Optional private text-to-speech delivery for Telegram replies. The Kokoro # server must be reachable from the Blacki container; localhost points back to # the container itself. A Tailscale address is appropriate for a private API. diff --git a/.github/workflows/docker-publish.yml b/.github/workflows/docker-publish.yml index 0a0e9ff..dfee28f 100644 --- a/.github/workflows/docker-publish.yml +++ b/.github/workflows/docker-publish.yml @@ -110,6 +110,8 @@ jobs: TELEGRAM_ACCESS_CODE: ${{ secrets.TELEGRAM_ACCESS_CODE }} TELEGRAM_BOT_TOKEN: ${{ secrets.TELEGRAM_BOT_TOKEN }} TELEGRAM_TOOL_NOTIFICATIONS: ${{ secrets.TELEGRAM_TOOL_NOTIFICATIONS }} + CLOUDFLARE_ACCOUNT_ID: ${{ secrets.CLOUDFLARE_ACCOUNT_ID }} + CLOUDFLARE_API_TOKEN: ${{ secrets.CLOUDFLARE_API_TOKEN }} KOKORO_TTS_BASE_URL: ${{ secrets.KOKORO_TTS_BASE_URL }} KOKORO_TTS_VOICE: ${{ secrets.KOKORO_TTS_VOICE }} ZEPTO_MCP_ENABLED: ${{ secrets.ZEPTO_MCP_ENABLED }} @@ -189,6 +191,8 @@ jobs: TELEGRAM_ACCESS_CODE \ TELEGRAM_BOT_TOKEN \ TELEGRAM_TOOL_NOTIFICATIONS \ + CLOUDFLARE_ACCOUNT_ID \ + CLOUDFLARE_API_TOKEN \ KOKORO_TTS_BASE_URL \ KOKORO_TTS_VOICE \ ZEPTO_MCP_ENABLED \ diff --git a/docs/base-infra/environment-variables.md b/docs/base-infra/environment-variables.md index 9c8dead..ac0bec0 100644 --- a/docs/base-infra/environment-variables.md +++ b/docs/base-infra/environment-variables.md @@ -95,6 +95,8 @@ its presence changes model routing. | `TELEGRAM_ENABLED` | `false` | Start Telegram long polling | | `TELEGRAM_BOT_TOKEN` | unset | Token from BotFather | | `TELEGRAM_ACCESS_CODE` | unset | Shared code required for new private Telegram chats | +| `CLOUDFLARE_ACCOUNT_ID` | unset | Cloudflare account ID for Telegram voice transcription | +| `CLOUDFLARE_API_TOKEN` | unset | Cloudflare Workers AI API token for Telegram voice transcription | | `KOKORO_TTS_BASE_URL` | unset | Register private Kokoro speech delivery for Telegram | | `KOKORO_TTS_VOICE` | `af_heart` | Kokoro voice ID used for generated MP3 audio | @@ -111,6 +113,14 @@ Kokoro process on another Tailscale host; configure that host's Tailscale IP or MagicDNS name instead. Plain HTTP is suitable only across a trusted private network such as the encrypted Tailscale connection. +`CLOUDFLARE_ACCOUNT_ID` and `CLOUDFLARE_API_TOKEN` are an optional pair. When +both are configured, Telegram voice notes are transcribed with Cloudflare +Workers AI's `@cf/openai/whisper-large-v3-turbo` model. The token should have +Workers AI Read and Write permissions. Blacki keeps the downloaded voice note +and provider request payload transient; it does not send the raw voice note to +the normal sandbox/file-ingestion path. The resulting transcript follows the +same conversation-history and logging/privacy policy as ordinary Telegram text. + ### Google Health connector The optional connector is enabled when all three required values are present. diff --git a/docs/telegram-setup.md b/docs/telegram-setup.md index 0ec0d40..3ed888d 100644 --- a/docs/telegram-setup.md +++ b/docs/telegram-setup.md @@ -41,6 +41,27 @@ delete any chat history, preferences, reminders, files, or health data. At least one model provider must also be configured. See [Configuration](base-infra/environment-variables.md). +### Optional voice-note transcription + +To let Blacki turn Telegram's native voice-note button into a normal text +conversation, add a Cloudflare Workers AI account ID and API token: + +```dotenv +CLOUDFLARE_ACCOUNT_ID=replace-me +CLOUDFLARE_API_TOKEN=replace-me +``` + +The token should have Workers AI Read and Write permissions. Blacki sends +voice notes to Cloudflare's hosted +`@cf/openai/whisper-large-v3-turbo` model, then passes the transcript through +the existing Telegram text-turn path so conversation history and tool +confirmations continue to work. Voice notes are kept transiently in memory; +regular audio files continue to use the existing file-upload behavior. If the +credentials are absent, Blacki reports that voice transcription is not +configured instead of failing startup. The raw voice bytes are transient; the +resulting transcript is handled like ordinary Telegram text and follows the +existing conversation-history and logging/privacy settings. + ### Optional Kokoro speech replies To let the Telegram-only root agent turn text into playable MP3 audio, add: diff --git a/src/blacki/telegram/bot.py b/src/blacki/telegram/bot.py index bdf4586..65282cc 100644 --- a/src/blacki/telegram/bot.py +++ b/src/blacki/telegram/bot.py @@ -9,7 +9,7 @@ from collections.abc import Coroutine, Sequence from contextvars import ContextVar from dataclasses import dataclass -from typing import TYPE_CHECKING, Any, cast +from typing import TYPE_CHECKING, Any, Protocol, cast from google.genai import types @@ -36,6 +36,12 @@ from .formatting import escape_markdown_plain, format_for_telegram from .settings_menu import SettingsMenu from .streaming import split_long_message +from .transcription import ( + MAX_CLOUDFLARE_WHISPER_AUDIO_BYTES, + MAX_CONCURRENT_CLOUDFLARE_TRANSCRIPTIONS, + CloudflareWhisperError, + CloudflareWhisperTranscriber, +) from .types import ( BotCommand, CallbackQuery, @@ -93,6 +99,16 @@ class TelegramSessionIdentity: session_id_prefix: str +class VoiceTranscriber(Protocol): + """Protocol implemented by the Telegram voice transcription service.""" + + async def transcribe(self, audio_bytes: bytes) -> str: + """Return the transcript for one voice note.""" + + async def close(self) -> None: + """Release any provider resources.""" + + class TelegramBot: """Telegram bot client that sends typing indicators and final replies.""" @@ -102,12 +118,19 @@ def __init__( runtime: AdkRuntime, google_health_service: GoogleHealthService | None = None, access_storage: TelegramAccessStorage | None = None, + voice_transcriber: VoiceTranscriber | None = None, ) -> None: """Initialize the Telegram bot.""" self.config = config self.runtime = runtime self.google_health_service = google_health_service self.access_storage = access_storage + self._voice_transcriber = ( + voice_transcriber or CloudflareWhisperTranscriber.from_environment() + ) + self._voice_transcription_semaphore = asyncio.Semaphore( + MAX_CONCURRENT_CLOUDFLARE_TRANSCRIPTIONS + ) self._api: TelegramApiClient | None = None self._running = False self._polling_task: asyncio.Task[None] | None = None @@ -166,7 +189,11 @@ async def stop(self) -> None: with contextlib.suppress(asyncio.CancelledError): await asyncio.gather(*self._background_tasks, return_exceptions=True) - await self.runtime.close() + try: + await self.runtime.close() + finally: + if self._voice_transcriber is not None: + await self._voice_transcriber.close() await self._settings_menu.aclose() @@ -737,12 +764,16 @@ async def _route_non_text_message(self, message: Message) -> None: mime_type = message.video.mime_type media_kind = "video" elif message.voice: - file_id = message.voice.file_id - file_unique_id = message.voice.file_unique_id - file_name = "voice.ogg" - file_size = message.voice.file_size - mime_type = message.voice.mime_type - media_kind = "voice" + await self._handle_voice_upload( + chat_id=chat_id, + message_thread_id=message_thread_id, + file_id=message.voice.file_id, + file_size=message.voice.file_size, + mime_type=message.voice.mime_type, + caption=message.caption, + sender_user_id=sender_user_id, + ) + return else: logger.debug("Unsupported non-text message from chat %s", chat_id) return @@ -760,6 +791,92 @@ async def _route_non_text_message(self, message: Message) -> None: sender_user_id=sender_user_id, ) + async def _handle_voice_upload( + self, + *, + chat_id: int, + message_thread_id: int | None, + file_id: str, + file_size: int | None, + mime_type: str | None, + caption: str | None, + sender_user_id: int | None = None, + ) -> None: + """Transcribe a Telegram voice note and process it as a text turn.""" + if self._voice_transcriber is None: + await self.api.send_message( + chat_id=chat_id, + text=( + "Voice transcription is not configured. Add " + "CLOUDFLARE_ACCOUNT_ID and CLOUDFLARE_API_TOKEN to Blacki's " + "environment." + ), + message_thread_id=message_thread_id, + ) + return + + if file_size is not None and file_size > MAX_CLOUDFLARE_WHISPER_AUDIO_BYTES: + await self.api.send_message( + chat_id=chat_id, + text="❌ Voice notes must be 8 MB or smaller to transcribe.", + message_thread_id=message_thread_id, + ) + return + + try: + async with self._voice_transcription_semaphore: + await self.api.send_chat_action( + chat_id=chat_id, + action="typing", + message_thread_id=message_thread_id, + ) + file_info = await self.api.get_file(file_id) + file_path_api = file_info.get("file_path") + if not file_path_api: + raise ValueError("Failed to get voice file path from Telegram API") + + audio_bytes = await self.api.download_file(file_path_api) + if not audio_bytes: + raise ValueError("Telegram returned an empty voice note") + if len(audio_bytes) > MAX_CLOUDFLARE_WHISPER_AUDIO_BYTES: + raise ValueError("Telegram voice note exceeds the 8 MB limit") + + transcript = await self._voice_transcriber.transcribe(audio_bytes) + del audio_bytes + + user_message = transcript + if caption and caption.strip(): + user_message = f"{caption.strip()}\n\n{transcript}" + + await self._handle_message( + chat_id=chat_id, + message_thread_id=message_thread_id, + user_message=user_message, + sender_user_id=sender_user_id, + ) + except CloudflareWhisperError as exc: + logger.warning( + "Failed to transcribe Telegram voice note (%s)", + type(exc).__name__, + ) + await self.api.send_message( + chat_id=chat_id, + text=( + "❌ Sorry, I couldn't transcribe that voice note. Please try again." + ), + message_thread_id=message_thread_id, + ) + except Exception as exc: + logger.warning( + "Failed to handle Telegram voice note (%s)", + type(exc).__name__, + ) + await self.api.send_message( + chat_id=chat_id, + text="❌ Sorry, I failed to process the voice note.", + message_thread_id=message_thread_id, + ) + async def _handle_photo_upload( self, *, diff --git a/src/blacki/telegram/transcription.py b/src/blacki/telegram/transcription.py new file mode 100644 index 0000000..d1b6d59 --- /dev/null +++ b/src/blacki/telegram/transcription.py @@ -0,0 +1,130 @@ +"""Cloudflare Workers AI transcription for Telegram voice notes.""" + +from __future__ import annotations + +import asyncio +import base64 +import os +from dataclasses import dataclass, field +from urllib.parse import quote + +import httpx + +CLOUDFLARE_WHISPER_MODEL = "@cf/openai/whisper-large-v3-turbo" +CLOUDFLARE_WHISPER_RUN_PATH = "/ai/run/" +MAX_CLOUDFLARE_WHISPER_AUDIO_BYTES = 8 * 1024 * 1024 +CLOUDFLARE_WHISPER_CONNECT_TIMEOUT_SECONDS = 5.0 +CLOUDFLARE_WHISPER_READ_TIMEOUT_SECONDS = 90.0 +MAX_CONCURRENT_CLOUDFLARE_TRANSCRIPTIONS = 1 + + +class CloudflareWhisperError(RuntimeError): + """Raised when Cloudflare returns an unusable transcription response.""" + + +@dataclass(frozen=True, slots=True) +class CloudflareWhisperConfig: + """Credentials for the Cloudflare Workers AI REST API.""" + + account_id: str + api_token: str = field(repr=False) + + +@dataclass(slots=True) +class CloudflareWhisperTranscriber: + """Call Cloudflare's hosted Whisper model with transient audio bytes.""" + + config: CloudflareWhisperConfig + http_transport: httpx.AsyncBaseTransport | None = field( + default=None, + repr=False, + ) + _client: httpx.AsyncClient | None = field(default=None, init=False, repr=False) + _semaphore: asyncio.Semaphore = field( + default_factory=lambda: asyncio.Semaphore( + MAX_CONCURRENT_CLOUDFLARE_TRANSCRIPTIONS + ), + init=False, + repr=False, + ) + + @classmethod + def from_environment(cls) -> CloudflareWhisperTranscriber | None: + """Create a transcriber when both Cloudflare credentials are present.""" + account_id = os.getenv("CLOUDFLARE_ACCOUNT_ID", "").strip() + api_token = os.getenv("CLOUDFLARE_API_TOKEN", "").strip() + if not account_id or not api_token: + return None + return cls(CloudflareWhisperConfig(account_id, api_token)) + + @property + def endpoint(self) -> str: + """Return the Workers AI model-run endpoint.""" + account_id = quote(self.config.account_id, safe="") + model = quote(CLOUDFLARE_WHISPER_MODEL, safe="@/") + return ( + "https://api.cloudflare.com/client/v4/accounts/" + f"{account_id}{CLOUDFLARE_WHISPER_RUN_PATH}{model}" + ) + + def _get_client(self) -> httpx.AsyncClient: + """Return the shared HTTP client, creating it lazily.""" + if self._client is None: + timeout = httpx.Timeout( + CLOUDFLARE_WHISPER_READ_TIMEOUT_SECONDS, + connect=CLOUDFLARE_WHISPER_CONNECT_TIMEOUT_SECONDS, + ) + self._client = httpx.AsyncClient( + timeout=timeout, + follow_redirects=False, + transport=self.http_transport, + ) + return self._client + + async def transcribe(self, audio_bytes: bytes) -> str: + """Transcribe one OGG voice note without persisting its audio.""" + if not audio_bytes: + raise CloudflareWhisperError("Cloudflare transcription audio is empty") + if len(audio_bytes) > MAX_CLOUDFLARE_WHISPER_AUDIO_BYTES: + raise CloudflareWhisperError("Cloudflare transcription audio is too large") + + async with self._semaphore: + payload = { + "audio": base64.b64encode(audio_bytes).decode("ascii"), + "task": "transcribe", + } + try: + response = await self._get_client().post( + self.endpoint, + headers={"Authorization": f"Bearer {self.config.api_token}"}, + json=payload, + ) + response.raise_for_status() + body = response.json() + except (httpx.HTTPError, ValueError) as exc: + raise CloudflareWhisperError( + "Cloudflare transcription request failed" + ) from exc + finally: + payload["audio"] = "" + + if not isinstance(body, dict) or body.get("success") is not True: + raise CloudflareWhisperError( + "Cloudflare returned an unsuccessful transcription response" + ) + + result = body.get("result") + if not isinstance(result, dict): + raise CloudflareWhisperError("Cloudflare returned no transcription result") + + text = result.get("text") + if not isinstance(text, str) or not text.strip(): + raise CloudflareWhisperError("Cloudflare returned an empty transcription") + return text.strip() + + async def close(self) -> None: + """Close the shared HTTP client if a transcription created it.""" + client = self._client + self._client = None + if client is not None: + await client.aclose() diff --git a/tests/test_deployment_contract.py b/tests/test_deployment_contract.py index 8f7c892..119bcf7 100644 --- a/tests/test_deployment_contract.py +++ b/tests/test_deployment_contract.py @@ -597,6 +597,24 @@ def test_production_deployment_serializes_kokoro_tts_settings() -> None: assert setting in writer_names +def test_production_deployment_serializes_cloudflare_whisper_settings() -> None: + """Cloudflare Workers AI credentials must reach production Compose.""" + workflow = _load_yaml(".github/workflows/docker-publish.yml") + deploy_step = next( + step + for step in workflow["jobs"]["deploy"]["steps"] + if step["name"] == "Deploy to Server via Tailscale" + ) + deploy_script = deploy_step["run"] + writer_start = deploy_script.index("python3 scripts/write_compose_env.py") + writer_end = deploy_script.index("printf '%s' \"$GH_TOKEN\"") + writer_names = deploy_script[writer_start:writer_end].split() + + for setting in ("CLOUDFLARE_ACCOUNT_ID", "CLOUDFLARE_API_TOKEN"): + assert deploy_step["env"][setting] == f"${{{{ secrets.{setting} }}}}" + assert setting in writer_names + + def test_production_deployment_serializes_google_health_settings() -> None: """Google Health OAuth secrets must reach the remote Compose environment.""" workflow = _load_yaml(".github/workflows/docker-publish.yml") diff --git a/tests/test_google_health.py b/tests/test_google_health.py index ff762c4..71c2617 100644 --- a/tests/test_google_health.py +++ b/tests/test_google_health.py @@ -1366,6 +1366,10 @@ async def failed_points(*args: object, **kwargs: object) -> list[dict[str, objec client.refresh_access_token = AsyncMock( return_value=GoogleTokenResponse("access", 3600, None, GOOGLE_HEALTH_SCOPES) ) + stale_date, _ = _date_window(7) + provider_date = ( + (datetime.fromisoformat(stale_date) + timedelta(days=1)).date().isoformat() + ) async def list_points(*args: object, **kwargs: object) -> list[dict[str, object]]: data_type = str(args[1]) @@ -1379,8 +1383,8 @@ async def list_points(*args: object, **kwargs: object) -> list[dict[str, object] "steps": { "count": 8420, "interval": { - "startTime": "2026-08-16T00:00:00Z", - "endTime": "2026-08-17T00:00:00Z", + "startTime": f"{provider_date}T00:00:00Z", + "endTime": f"{provider_date}T23:59:59Z", }, } } @@ -1388,7 +1392,6 @@ async def list_points(*args: object, **kwargs: object) -> list[dict[str, object] return [] client.list_data_points = AsyncMock(side_effect=list_points) - stale_date, _ = _date_window(7) await health_storage.upsert_daily_summaries( "telegram-chat-42", [{"date": stale_date, "steps": 1}] ) diff --git a/tests/test_telegram_bot.py b/tests/test_telegram_bot.py index b4661f8..2f3f1ad 100644 --- a/tests/test_telegram_bot.py +++ b/tests/test_telegram_bot.py @@ -41,6 +41,11 @@ _merge_stream_text, split_long_message, ) +from blacki.telegram.transcription import ( + MAX_CLOUDFLARE_WHISPER_AUDIO_BYTES, + CloudflareWhisperError, + CloudflareWhisperTranscriber, +) from blacki.telegram.types import BotCommand, ChatType, Message, ParseMode, Update from blacki.user_files.service import IngestResult, StoredUserFile @@ -1537,6 +1542,28 @@ async def test_stop_closes_api( mock_api.close.assert_called_once() assert runtime_recorder.closed is True + @pytest.mark.asyncio + async def test_stop_closes_voice_transcriber( + self, + telegram_config: TelegramConfig, + runtime_recorder: RecordingRuntime, + ) -> None: + transcriber = create_autospec( + CloudflareWhisperTranscriber, + spec_set=True, + instance=True, + ) + transcriber.close = AsyncMock() + bot = TelegramBot( + telegram_config, + cast(AdkRuntime, runtime_recorder), + voice_transcriber=cast(Any, transcriber), + ) + + await bot.stop() + + transcriber.close.assert_awaited_once() + @pytest.mark.asyncio async def test_register_commands_success( self, @@ -3277,7 +3304,7 @@ async def test_handles_voice( ) -> None: """Test routing a voice message.""" bot = TelegramBot(telegram_config, cast(AdkRuntime, runtime_recorder)) - bot._handle_file_upload = AsyncMock() # type: ignore[method-assign] + bot._handle_voice_upload = AsyncMock() # type: ignore[method-assign] message = Message.model_validate( { @@ -3294,15 +3321,12 @@ async def test_handles_voice( await bot._route_non_text_message(message) - bot._handle_file_upload.assert_called_once_with( + bot._handle_voice_upload.assert_called_once_with( chat_id=123, message_thread_id=None, file_id="voi123", - file_unique_id="uniq123", - file_name="voice.ogg", file_size=None, mime_type=None, - media_kind="voice", caption=None, sender_user_id=None, ) @@ -3330,6 +3354,247 @@ async def test_handles_unsupported_message( bot._handle_file_upload.assert_not_called() +class TestHandleVoiceUpload: + """Tests for transient voice-note transcription.""" + + def _bot_with_transcriber( + self, + telegram_config: TelegramConfig, + runtime_recorder: RecordingRuntime, + ) -> tuple[TelegramBot, Any, Any]: + transcriber = create_autospec( + CloudflareWhisperTranscriber, + spec_set=True, + instance=True, + ) + transcriber.transcribe = AsyncMock(return_value="Transcript text") + transcriber.close = AsyncMock() + bot = TelegramBot( + telegram_config, + cast(AdkRuntime, runtime_recorder), + voice_transcriber=cast(Any, transcriber), + ) + mock_api = create_autospec(TelegramApiClient, spec_set=True, instance=True) + mock_api.send_chat_action = AsyncMock() + mock_api.send_message = AsyncMock() + mock_api.get_file = AsyncMock(return_value={"file_path": "voice.ogg"}) + mock_api.download_file = AsyncMock(return_value=b"ogg-audio") + bot._api = mock_api + return bot, transcriber, mock_api + + @pytest.mark.asyncio + async def test_transcribes_voice_and_preserves_caption( + self, + telegram_config: TelegramConfig, + runtime_recorder: RecordingRuntime, + ) -> None: + bot, transcriber, mock_api = self._bot_with_transcriber( + telegram_config, + runtime_recorder, + ) + bot._handle_message = AsyncMock() # type: ignore[method-assign] + + await bot._handle_voice_upload( + chat_id=123, + message_thread_id=77, + file_id="voice123", + file_size=100, + mime_type="audio/ogg", + caption="Summarize this", + sender_user_id=456, + ) + + mock_api.get_file.assert_awaited_once_with("voice123") + mock_api.download_file.assert_awaited_once_with("voice.ogg") + transcriber.transcribe.assert_awaited_once_with(b"ogg-audio") + bot._handle_message.assert_awaited_once_with( + chat_id=123, + message_thread_id=77, + user_message="Summarize this\n\nTranscript text", + sender_user_id=456, + ) + + @pytest.mark.asyncio + async def test_transcribes_voice_without_caption( + self, + telegram_config: TelegramConfig, + runtime_recorder: RecordingRuntime, + ) -> None: + bot, _, _ = self._bot_with_transcriber(telegram_config, runtime_recorder) + bot._handle_message = AsyncMock() # type: ignore[method-assign] + + await bot._handle_voice_upload( + chat_id=123, + message_thread_id=None, + file_id="voice123", + file_size=None, + mime_type=None, + caption=" ", + ) + + bot._handle_message.assert_awaited_once_with( + chat_id=123, + message_thread_id=None, + user_message="Transcript text", + sender_user_id=None, + ) + + @pytest.mark.asyncio + async def test_reports_missing_configuration_without_downloading( + self, + monkeypatch: pytest.MonkeyPatch, + telegram_config: TelegramConfig, + runtime_recorder: RecordingRuntime, + ) -> None: + monkeypatch.delenv("CLOUDFLARE_ACCOUNT_ID", raising=False) + monkeypatch.delenv("CLOUDFLARE_API_TOKEN", raising=False) + bot = TelegramBot(telegram_config, cast(AdkRuntime, runtime_recorder)) + mock_api = create_autospec(TelegramApiClient, spec_set=True, instance=True) + mock_api.send_message = AsyncMock() + mock_api.get_file = AsyncMock() + bot._api = mock_api + + await bot._handle_voice_upload( + chat_id=123, + message_thread_id=None, + file_id="voice123", + file_size=None, + mime_type=None, + caption=None, + ) + + assert "CLOUDFLARE_ACCOUNT_ID" in mock_api.send_message.call_args.kwargs["text"] + mock_api.get_file.assert_not_awaited() + + @pytest.mark.asyncio + async def test_rejects_reported_oversize_before_download( + self, + telegram_config: TelegramConfig, + runtime_recorder: RecordingRuntime, + ) -> None: + bot, transcriber, mock_api = self._bot_with_transcriber( + telegram_config, + runtime_recorder, + ) + + await bot._handle_voice_upload( + chat_id=123, + message_thread_id=None, + file_id="voice123", + file_size=MAX_CLOUDFLARE_WHISPER_AUDIO_BYTES + 1, + mime_type=None, + caption=None, + ) + + assert "8 MB" in mock_api.send_message.call_args.kwargs["text"] + mock_api.get_file.assert_not_awaited() + transcriber.transcribe.assert_not_awaited() + + @pytest.mark.asyncio + @pytest.mark.parametrize("audio_bytes", [b"", b"12345"]) + async def test_reports_invalid_downloaded_audio( + self, + monkeypatch: pytest.MonkeyPatch, + audio_bytes: bytes, + telegram_config: TelegramConfig, + runtime_recorder: RecordingRuntime, + ) -> None: + import blacki.telegram.bot as bot_module + + monkeypatch.setattr(bot_module, "MAX_CLOUDFLARE_WHISPER_AUDIO_BYTES", 4) + bot, transcriber, mock_api = self._bot_with_transcriber( + telegram_config, + runtime_recorder, + ) + mock_api.download_file = AsyncMock(return_value=audio_bytes) + + await bot._handle_voice_upload( + chat_id=123, + message_thread_id=None, + file_id="voice123", + file_size=None, + mime_type=None, + caption=None, + ) + + assert "failed to process" in mock_api.send_message.call_args.kwargs["text"] + transcriber.transcribe.assert_not_awaited() + + @pytest.mark.asyncio + async def test_reports_provider_failure_without_exposing_details( + self, + caplog: pytest.LogCaptureFixture, + telegram_config: TelegramConfig, + runtime_recorder: RecordingRuntime, + ) -> None: + bot, transcriber, mock_api = self._bot_with_transcriber( + telegram_config, + runtime_recorder, + ) + transcriber.transcribe = AsyncMock( + side_effect=CloudflareWhisperError("private provider response") + ) + + await bot._handle_voice_upload( + chat_id=123, + message_thread_id=None, + file_id="voice123", + file_size=None, + mime_type=None, + caption=None, + ) + + assert "couldn't transcribe" in mock_api.send_message.call_args.kwargs["text"] + assert "private provider response" not in caplog.text + + @pytest.mark.asyncio + async def test_reports_unexpected_voice_failure( + self, + telegram_config: TelegramConfig, + runtime_recorder: RecordingRuntime, + ) -> None: + bot, _, mock_api = self._bot_with_transcriber( + telegram_config, + runtime_recorder, + ) + mock_api.get_file = AsyncMock(side_effect=RuntimeError("telegram detail")) + + await bot._handle_voice_upload( + chat_id=123, + message_thread_id=None, + file_id="voice123", + file_size=None, + mime_type=None, + caption=None, + ) + + assert "failed to process" in mock_api.send_message.call_args.kwargs["text"] + + @pytest.mark.asyncio + async def test_reports_missing_telegram_file_path( + self, + telegram_config: TelegramConfig, + runtime_recorder: RecordingRuntime, + ) -> None: + bot, transcriber, mock_api = self._bot_with_transcriber( + telegram_config, + runtime_recorder, + ) + mock_api.get_file = AsyncMock(return_value={}) + + await bot._handle_voice_upload( + chat_id=123, + message_thread_id=None, + file_id="voice123", + file_size=None, + mime_type=None, + caption=None, + ) + + assert "failed to process" in mock_api.send_message.call_args.kwargs["text"] + transcriber.transcribe.assert_not_awaited() + + class TestHandlePhotoUpload: """Tests for native multimodal Telegram photo handling.""" diff --git a/tests/test_telegram_transcription.py b/tests/test_telegram_transcription.py new file mode 100644 index 0000000..63876d3 --- /dev/null +++ b/tests/test_telegram_transcription.py @@ -0,0 +1,148 @@ +"""Tests for the Cloudflare Workers AI Telegram transcription client.""" + +from __future__ import annotations + +import base64 +import json + +import httpx +import pytest + +from blacki.telegram.transcription import ( + CLOUDFLARE_WHISPER_MODEL, + MAX_CLOUDFLARE_WHISPER_AUDIO_BYTES, + CloudflareWhisperConfig, + CloudflareWhisperError, + CloudflareWhisperTranscriber, +) + + +def _config() -> CloudflareWhisperConfig: + return CloudflareWhisperConfig( + account_id="account-id", + api_token="cloudflare-token", + ) + + +@pytest.mark.asyncio +async def test_transcriber_uses_cloudflare_rest_contract() -> None: + requests: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + requests.append(request) + return httpx.Response( + 200, + json={"success": True, "result": {"text": " Hello from audio "}}, + ) + + transcriber = CloudflareWhisperTranscriber( + _config(), + http_transport=httpx.MockTransport(handler), + ) + + assert await transcriber.transcribe(b"\x00\x01\xff") == "Hello from audio" + assert await transcriber.transcribe(b"\x02") == "Hello from audio" + await transcriber.close() + + assert len(requests) == 2 + assert requests[0].method == "POST" + assert str(requests[0].url) == ( + "https://api.cloudflare.com/client/v4/accounts/account-id/ai/run/" + f"{CLOUDFLARE_WHISPER_MODEL}" + ) + assert requests[0].headers["authorization"] == "Bearer cloudflare-token" + assert json.loads(requests[0].content) == { + "audio": base64.b64encode(b"\x00\x01\xff").decode("ascii"), + "task": "transcribe", + } + + +def test_from_environment_requires_both_credentials( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("CLOUDFLARE_ACCOUNT_ID", raising=False) + monkeypatch.delenv("CLOUDFLARE_API_TOKEN", raising=False) + assert CloudflareWhisperTranscriber.from_environment() is None + + monkeypatch.setenv("CLOUDFLARE_ACCOUNT_ID", "account-id") + assert CloudflareWhisperTranscriber.from_environment() is None + + monkeypatch.setenv("CLOUDFLARE_API_TOKEN", " cloudflare-token ") + transcriber = CloudflareWhisperTranscriber.from_environment() + assert transcriber is not None + assert transcriber.config.account_id == "account-id" + assert transcriber.config.api_token == "cloudflare-token" # noqa: S105 + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("body", "error"), + [ + ( + {"success": False, "errors": [{"message": "provider detail"}]}, + "unsuccessful", + ), + ({"success": True}, "no transcription result"), + ({"success": True, "result": {"text": " "}}, "empty transcription"), + ], +) +async def test_transcriber_rejects_invalid_cloudflare_results( + body: dict[str, object], + error: str, +) -> None: + transcriber = CloudflareWhisperTranscriber( + _config(), + http_transport=httpx.MockTransport(lambda _: httpx.Response(200, json=body)), + ) + + with pytest.raises(CloudflareWhisperError, match=error): + await transcriber.transcribe(b"audio") + await transcriber.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "response", + [ + httpx.Response(503), + httpx.Response(200, content=b"not-json"), + ], +) +async def test_transcriber_sanitizes_http_and_json_failures( + response: httpx.Response, +) -> None: + transcriber = CloudflareWhisperTranscriber( + _config(), + http_transport=httpx.MockTransport(lambda _: response), + ) + + with pytest.raises(CloudflareWhisperError, match="request failed"): + await transcriber.transcribe(b"private audio") + await transcriber.close() + + +@pytest.mark.asyncio +async def test_transcriber_sanitizes_network_failures() -> None: + def handler(request: httpx.Request) -> httpx.Response: + raise httpx.ConnectError("private provider detail", request=request) + + transcriber = CloudflareWhisperTranscriber( + _config(), + http_transport=httpx.MockTransport(handler), + ) + + with pytest.raises(CloudflareWhisperError, match="request failed"): + await transcriber.transcribe(b"audio") + await transcriber.close() + + +@pytest.mark.asyncio +async def test_transcriber_rejects_empty_and_oversized_audio_without_network() -> None: + transcriber = CloudflareWhisperTranscriber(_config()) + + with pytest.raises(CloudflareWhisperError, match="audio is empty"): + await transcriber.transcribe(b"") + with pytest.raises(CloudflareWhisperError, match="audio is too large"): + await transcriber.transcribe(b"x" * (MAX_CLOUDFLARE_WHISPER_AUDIO_BYTES + 1)) + + await transcriber.close()