feat: transcribe Telegram voice messages via APInex

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-провайдера и подключить генерацию текстов
 - [ ] Утвердить список соцсетей и формат запроса

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`.

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("/"))

requirements.txt

@@ -1 +1,2 @@
 aiogram==3.26.0
+aiohttp>=3.9,<4

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()