Initial commit: MIB bot

yandex_cloud_sk.py

@@ -0,0 +1,63 @@
+import asyncio
+import tempfile
+from pathlib import Path
+
+import aiohttp
+from speechkit import configure_credentials, creds, model_repository
+from speechkit.stt import AudioProcessingType
+
+from config import YSK_API_KEY, create_aiohttp_session, get_aiohttp_request_kwargs
+
+if YSK_API_KEY:
+    configure_credentials(
+        yandex_credentials=creds.YandexCredentials(api_key=YSK_API_KEY)
+    )
+
+
+def _transcribe_file_sync(file_path: str) -> str | None:
+    model = model_repository.recognition_model()
+    model.model = "general"
+    model.language = "ru-RU"
+    model.audio_processing_type = AudioProcessingType.Full
+
+    result = model.transcribe_file(file_path)
+    for res in result:
+        normalized_text = getattr(res, "normalized_text", None)
+        if normalized_text:
+            return normalized_text
+    return None
+
+
+async def get_text_from_speech(file_url: str) -> str | None:
+    temp_path = None
+    timeout = aiohttp.ClientTimeout(total=300, sock_connect=30, sock_read=300)
+
+    try:
+        async with create_aiohttp_session(timeout=timeout) as session:
+            async with session.get(file_url, **get_aiohttp_request_kwargs()) as response:
+                response.raise_for_status()
+                content = await response.read()
+
+        with tempfile.NamedTemporaryFile(delete=False, suffix=".ogg") as temp_file:
+            temp_file.write(content)
+            temp_path = temp_file.name
+
+        return await asyncio.to_thread(_transcribe_file_sync, temp_path)
+    finally:
+        if temp_path:
+            try:
+                Path(temp_path).unlink(missing_ok=True)
+            except Exception:
+                pass
+
+
+def synthesize(text: str) -> str:
+    model = model_repository.synthesis_model()
+    model.voice = "lera"
+    result = model.synthesize(text, raw_format=False)
+
+    with tempfile.NamedTemporaryFile(delete=False, suffix=".ogg") as temp_file:
+        output_path = temp_file.name
+
+    result.export(output_path, "ogg")
+    return output_path