From ea68cfba29fd1d6a05f0411e7981841b02cfad32 Mon Sep 17 00:00:00 2001 From: admin Date: Sun, 4 Oct 2026 20:02:00 +0200 Subject: [PATCH] feat: transcribe Telegram voice messages via APInex --- IDEA.MD | 10 ++++- README.md | 17 +++++--- bot.py | 97 ++++++++++++++++++++++++++++++++++++++++- requirements.txt | 1 + tests/test_voice_stt.py | 73 +++++++++++++++++++++++++++++++ 5 files changed, 191 insertions(+), 7 deletions(-) create mode 100644 tests/test_voice_stt.py diff --git a/IDEA.MD b/IDEA.MD index 104ca17..005587a 100644 --- a/IDEA.MD +++ b/IDEA.MD @@ -21,11 +21,19 @@ Telegram-бот, который по запросу пользователя г - Формат: Telegram-бот. - Предварительный стек: Python, aiogram 3.x. - LLM-провайдер и модель: выбрать отдельно; ключи хранить только в переменных окружения, не в репозитории. +- Поддержка голосовых: APInex Voice STT, ключ — Coolify env `APINEX_API_KEY` в development. - Целевые соцсети, хранение истории и дополнительные функции уточнить при реализации. +## Деплой и конфигурация +- Gitea: `admin/social-post-writer-bot`, ветка `dev`. +- Coolify проект: `Social Post Writer Bot`, окружение `development`. +- Coolify приложение: `social-post-writer-bot-development` на `spb-easy-dale`. +- Секреты хранятся в переменных окружения приложения Coolify; не сохраняются в репозитории. + ## Статус - [x] Идея проекта зафиксирована - [x] Echo MVP реализован на aiogram 3.x: `/start` и повтор текста -- [x] Тестовый деплой в Coolify на `spb-easy-dale` +- [x] Добавлена расшифровка голосовых через APInex Voice STT +- [x] Development-развёртывание в Coolify - [ ] Выбрать LLM-провайдера и подключить генерацию текстов - [ ] Утвердить список соцсетей и формат запроса diff --git a/README.md b/README.md index c4c7c47..c697022 100644 --- a/README.md +++ b/README.md @@ -1,8 +1,15 @@ -# Social Post Writer Bot — test echo deployment +# Social Post Writer Bot — development -Temporary MVP: `/start` sends a short prompt, and ordinary text is echoed unchanged. +Telegram bot that currently echoes text and transcribes voice/audio messages into Russian using APInex Voice STT. - Runtime: Python 3.12, aiogram 3.x -- `BOT_TOKEN`: supplied only through Coolify environment variables -- `BOT_ENV=test`: routes Bot API calls to Telegram's test environment -- No LLM integration or automatic social-media publishing yet +- Coolify environment: `development` on `spb-easy-dale` +- Gitea branch: `dev` +- Runtime secrets are configured only in Coolify environment variables: `BOT_TOKEN`, `BOT_ENV`, `APINEX_API_KEY`. +- For test bot tokens, set `BOT_ENV=test` to use Telegram's test Bot API. +- Voice transcription uses raw audio upload to APInex `POST /v1/tools/voice/stt`; voice notes are sent as `audio/ogg`, with Russian language hint. +- No LLM post writing or automatic social-media publishing yet. + +## Local tests + +Run `python -m unittest discover -s tests -v`. diff --git a/bot.py b/bot.py index 3aaf74b..d166532 100644 --- a/bot.py +++ b/bot.py @@ -1,7 +1,10 @@ import asyncio +import io import logging import os +from typing import Any +import aiohttp from aiogram import Bot, Dispatcher, F, Router from aiogram.client.session.aiohttp import AiohttpSession from aiogram.client.telegram import TEST @@ -9,13 +12,105 @@ from aiogram.filters import CommandStart from aiogram.types import Message logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) router = Router() +APINEX_STT_URL = "https://api.apinex.bond/v1/tools/voice/stt?language=ru" +MAX_AUDIO_BYTES = 20 * 1024 * 1024 + + +def extract_transcript(payload: dict[str, Any]) -> str: + """Return the first APInex voice/STT transcript, or an empty string.""" + channels = payload.get("results", {}).get("channels", []) + for channel in channels: + alternatives = channel.get("alternatives", []) + if alternatives: + transcript = alternatives[0].get("transcript", "") + if transcript and transcript.strip(): + return transcript.strip() + return "" + + +def split_text(text: str, limit: int = 4000) -> list[str]: + """Split a long transcript into Telegram-safe chunks, preferring word boundaries.""" + chunks: list[str] = [] + remaining = text.strip() + while len(remaining) > limit: + split_at = remaining.rfind(" ", 0, limit + 1) + if split_at < limit // 2: + split_at = limit + chunks.append(remaining[:split_at].strip()) + remaining = remaining[split_at:].strip() + if remaining: + chunks.append(remaining) + return chunks + + +async def transcribe_audio(audio: bytes, content_type: str, api_key: str) -> str: + timeout = aiohttp.ClientTimeout(total=120) + headers = { + "Authorization": f"Bearer {api_key}", + "Content-Type": content_type, + "Accept": "application/json", + } + async with aiohttp.ClientSession(timeout=timeout) as session: + async with session.post(APINEX_STT_URL, data=audio, headers=headers) as response: + if response.status != 200: + logger.warning("APInex STT returned HTTP %s", response.status) + raise RuntimeError("APInex STT request failed") + try: + payload = await response.json(content_type=None) + except (aiohttp.ContentTypeError, ValueError) as exc: + raise RuntimeError("APInex STT returned invalid JSON") from exc + return extract_transcript(payload) @router.message(CommandStart()) async def start(message: Message) -> None: - await message.answer("Тестовый эхо-бот запущен. Отправь мне текст — я повторю его.") + await message.answer("Тестовый бот запущен. Отправь текст — я повторю его, или голосовое — расшифрую.") + + +@router.message(F.voice | F.audio) +async def transcribe_voice(message: Message, bot: Bot) -> None: + api_key = os.environ.get("APINEX_API_KEY", "").strip() + if not api_key: + await message.answer("Расшифровка пока не настроена.") + return + + audio_file = message.voice or message.audio + if audio_file is None: + return + if audio_file.file_size and audio_file.file_size > MAX_AUDIO_BYTES: + await message.answer("Аудиофайл слишком большой. Максимальный размер — 20 МБ.") + return + + content_type = "audio/ogg" if message.voice else (message.audio.mime_type or "audio/mpeg") + await bot.send_chat_action(message.chat.id, "typing") + try: + audio_buffer = io.BytesIO() + await bot.download(audio_file, destination=audio_buffer) + audio = audio_buffer.getvalue() + if not audio: + await message.answer("Не удалось получить аудиофайл.") + return + if len(audio) > MAX_AUDIO_BYTES: + await message.answer("Аудиофайл слишком большой. Максимальный размер — 20 МБ.") + return + transcript = await transcribe_audio(audio, content_type, api_key) + except (aiohttp.ClientError, asyncio.TimeoutError, RuntimeError) as exc: + logger.warning("Voice transcription failed: %s", type(exc).__name__) + await message.answer("Не удалось расшифровать аудио. Попробуй ещё раз позже.") + return + except Exception: + logger.exception("Unexpected voice transcription failure") + await message.answer("Произошла ошибка при обработке аудио.") + return + + if not transcript: + await message.answer("Не удалось распознать речь в аудио.") + return + for chunk in split_text(transcript): + await message.answer(chunk) @router.message(F.text & ~F.text.startswith("/")) diff --git a/requirements.txt b/requirements.txt index d5a4d9f..beac509 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1 +1,2 @@ aiogram==3.26.0 +aiohttp>=3.9,<4 diff --git a/tests/test_voice_stt.py b/tests/test_voice_stt.py new file mode 100644 index 0000000..a43019e --- /dev/null +++ b/tests/test_voice_stt.py @@ -0,0 +1,73 @@ +import unittest +from unittest.mock import patch + +import bot + + +class FakeResponse: + def __init__(self, status, payload): + self.status = status + self.payload = payload + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + return False + + async def json(self, content_type=None): + return self.payload + + +class FakeClientSession: + def __init__(self, response, **kwargs): + self.response = response + self.kwargs = kwargs + self.request: tuple[str, dict] = ("", {}) + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + return False + + def post(self, url, **kwargs): + self.request = (url, kwargs) + return self.response + + +class VoiceSttTests(unittest.IsolatedAsyncioTestCase): + def test_extract_transcript(self): + payload = {"results": {"channels": [{"alternatives": [{"transcript": " Привет, мир! "}]}]}} + self.assertEqual(bot.extract_transcript(payload), "Привет, мир!") + + def test_empty_response(self): + self.assertEqual(bot.extract_transcript({"results": {"channels": []}}), "") + self.assertEqual(bot.extract_transcript({}), "") + + def test_split_long_transcript(self): + chunks = bot.split_text("слово " * 2000, 4000) + self.assertTrue(all(len(chunk) <= 4000 for chunk in chunks)) + self.assertEqual(" ".join(chunks).split(), ("слово " * 2000).split()) + + async def test_transcribe_sends_raw_ogg_and_parses_response(self): + response = FakeResponse(200, {"results": {"channels": [{"alternatives": [{"transcript": "Распознано"}]}]}}) + fake_session = FakeClientSession(response) + with patch.object(bot.aiohttp, "ClientSession", return_value=fake_session): + result = await bot.transcribe_audio(b"ogg-data", "audio/ogg", "not-logged") + self.assertEqual(result, "Распознано") + url, kwargs = fake_session.request + self.assertEqual(url, bot.APINEX_STT_URL) + self.assertEqual(kwargs["data"], b"ogg-data") + self.assertEqual(kwargs["headers"]["Content-Type"], "audio/ogg") + self.assertEqual(kwargs["headers"]["Authorization"], "Bearer not-logged") + + async def test_transcribe_raises_on_api_error(self): + fake_session = FakeClientSession(FakeResponse(401, {})) + with patch.object(bot.aiohttp, "ClientSession", return_value=fake_session): + with self.assertRaises(RuntimeError): + await bot.transcribe_audio(b"ogg-data", "audio/ogg", "not-logged") + + +if __name__ == "__main__": + unittest.main()