@ -1,18 +1,42 @@
import os
import re
import logging
import aiohttp
import json
import asyncio
import sqlite3
import subprocess
from datetime import datetime
from aiogram import Router , F , Bot
from aiogram . types import Message
from aiogram . types import Message , BufferedInputFile
from config import ADMIN_USER_ID , DATABASE_PATH
from pathlib import Path
import google . generativeai as genai
import edge_tts
logger = logging . getLogger ( __name__ )
router = Router ( )
WHISPER_API_URL = " http://192.168.10.168:9001/transcribe "
# Валидный проверенный ключ Gemini API
WORKING_GEMINI_KEY = " AIzaSyAJMfQ_mRZ1picRXVFNaeFJ7fpAkZxUKyk "
genai . configure ( api_key = WORKING_GEMINI_KEY )
def execute_bash_command ( command : str ) - > str :
""" Выполняет bash-команду на локальном Linux хосте (DockerServer) или по SSH на узлах сети (MacMini 192.168.10.167, Synology 192.168.10.105, MikroTik 192.168.10.215, NanoPi 192.168.10.1) и возвращает результат вывода. """
logger . info ( f " ⚡️ [Voice Agent Command] Executing bash: { command } " )
try :
res = subprocess . run (
command ,
shell = True ,
capture_output = True ,
text = True ,
timeout = 45
)
out = ( res . stdout or " " ) + ( " \n " + res . stderr if res . stderr else " " )
output_str = out . strip ( )
if len ( output_str ) > 3000 :
output_str = output_str [ : 3000 ] + " \n ... [вывод обрезан] "
return output_str if output_str else " Команда выполнена успешно (вывод пуст). "
except Exception as e :
logger . error ( f " ❌ Bash Execution Error: { e } " )
return f " Ошибка выполнения bash-команды: { e } "
def log_voice_to_db ( user_id , username , msg_type , duration , text ) :
""" Фиксация статистики в основную базу Домового """
@ -39,68 +63,192 @@ def log_voice_to_db(user_id, username, msg_type, duration, text):
except Exception as e :
logger . error ( f " ❌ DB Log error: { e } " )
@router.message ( F . voice | F . video_note )
async def text_to_speech_bytes ( text : str ) - > bytes :
""" Синтез качественного женского голосового ответа на русском через Edge TTS с """
# Если в тексте есть тег <voice>...</voice>, озвучиваем именно е г о
match = re . search ( r " <voice>(.*?)</voice> " , text , re . DOTALL | re . IGNORECASE )
if match :
speech_text = match . group ( 1 ) . strip ( )
else :
# Иначе берем выжимку до 350 символов
speech_text = text [ : 350 ] if len ( text ) > 350 else text
# Очищаем от Markdown и спецсимволов для чистоты звучания
for ch in [ " * " , " # " , " _ " , " ` " , " ~ " , " > " , " • " , " 🎯 " , " ⚙️ " , " 📋 " , " 💡 " , " 🗄️ " , " 🚨 " , " ⚠️ " , " ✅ " , " ❌ " ] :
speech_text = speech_text . replace ( ch , " " )
speech_text = re . sub ( r " \ [.*? \ ] \ (.*? \ ) " , " " , speech_text ) . strip ( )
if not speech_text :
speech_text = " Приняла, Александр. Всё структурировано и зафиксировано. "
communicate = edge_tts . Communicate ( speech_text , " ru-RU-SvetlanaNeural " , pitch = " -8Hz " , rate = " -3 % " )
out_path = f " /tmp/tts_ { os . getpid ( ) } _ { int ( datetime . now ( ) . timestamp ( ) ) } .mp3 "
await communicate . save ( out_path )
audio_bytes = b " "
if os . path . exists ( out_path ) :
with open ( out_path , " rb " ) as f :
audio_bytes = f . read ( )
os . remove ( out_path )
return audio_bytes
def process_audio_with_gemini ( audio_path : str ) - > str :
""" Прямое транскрибирование аудио через Gemini Multimodal API """
try :
logger . info ( f " Выгрузка { audio_path } в Gemini Multimodal Audio API... " )
audio_file = genai . upload_file ( path = audio_path )
prompt = (
" Точно транскрибируй эту аудиозапись на русском языке в текст. "
" Н е "
)
model = genai . GenerativeModel ( ' gemini-2.5-flash ' )
response = model . generate_content ( [ audio_file , prompt ] )
try :
genai . delete_file ( name = audio_file . name )
except Exception :
pass
return response . text . strip ( ) if response . text else " "
except Exception as e :
logger . error ( f " Ошибка Gemini Audio API: { e } " )
return " "
async def execute_antigravity_agent ( user_text : str ) - > str :
""" Обработка команды и мыслей агентом Майа с """
system_instruction = (
" Ты — Майа (Maya / Antigravity Lead Agent), суперинтеллектуальный помощник Александра. "
" Александр обращается к тебе голосом в Телеграмме или надиктованной аудиозаписью с "
" Твоя задача — быть идеальным личным инженером и аналитиком: \n \n "
" 1. СТРУКТУРИРОВАНИЕ МЫСЛЕЙ (если Александр надиктовал идеи, мысли, список дел или сумбурный поток): \n "
" Разложи всё по четким блокам: \n "
" 🎯 Суть: краткая главная мысль в 1-2 предложениях. \n "
" 📋 Задачи к выполнению (Action Items): конкретные пункты с \n "
" 💡 Идеи в копилку / Бэклог: новые задумки, архитектурные гипотезы. \n "
" 🗄️ Куда зафиксировать: в какую заметку Obsidian (HomeLab или Work) это стоит сохранить. \n \n "
" 2. СИСТЕМНЫЕ КОМАНДЫ (если Александр просит проверить ресурсы, диски, сеть, контейнеры, запустить что-то): \n "
" ОБЯЗАТЕЛЬНО используй execute_bash_command! Выполни команду, проанализируй вывод и дай четкий отчет. \n "
" Узлы сети: NanoPi R3S (root@192.168.10.1), Synology (matrixhasyou@192.168.10.105), MacMini (matrixhasyou@192.168.10.167), MikroTik (matrixhasyou@192.168.10.215). \n \n "
" 3. ГОЛОСОВОЙ ОТКЛИК ДЛЯ ОЗВУЧКИ: \n "
" В С А М О М \n "
" Внутри него напиши короткое, живое, теплое и емкое резюме от Майи (1-3 предложения), которое будет озвучено вслух! "
" Внутри <voice> пиши только чистый русский текст без звездочек, эмодзи, списков и Markdown. "
)
models_to_try = [ ' gemini-2.5-flash ' , ' gemini-3.5-flash ' , ' gemini-3.5-flash-lite ' ]
last_error = None
for model_name in models_to_try :
try :
logger . info ( f " 🚀 [Maya Agent] Запуск с { model_name } " )
model = genai . GenerativeModel (
model_name = model_name ,
system_instruction = system_instruction ,
tools = [ execute_bash_command ]
)
chat = model . start_chat ( enable_automatic_function_calling = True )
response = chat . send_message ( user_text )
if response and response . text :
return response . text . strip ( )
except Exception as e :
logger . warning ( f " ⚠️ [Maya Agent] Сбой на модели { model_name } : { e } " )
last_error = e
await asyncio . sleep ( 1 )
return f " ❌ Ошибка исполнения Maya Agent: { last_error } "
@router.message ( F . voice | F . video_note | F . audio | F . document )
async def handle_voice_message ( message : Message , bot : Bot ) :
# ПЕРЕВОДИМ ТОЛЬКО ДЛЯ ТЕБЯ (СУПЕРАДМИНА)
# ПЕРЕВОДИМ И ИСПОЛНЯЕМ СТРОГО ДЛЯ АЛЕКСАНДРА (ЧУЖИХ Н Е )
if int ( message . from_user . id ) != int ( ADMIN_USER_ID ) :
return
status_msg = await message . reply ( " ⏳ <i>Распознаю голос (Whisper AI)...</i> " , parse_mode = " HTML " )
# Определяем тип медиафайла
msg_type = " VOICE "
duration = 0
filename = " audio.ogg "
file_id = None
if message . voice :
file_id = message . voice . file_id
filename = f " voice_ { file_id } .ogg "
duration = message . voice . duration
msg_type = " VOICE "
elif message . video_note :
file_id = message . video_note . file_id
filename = f " note_ { file_id } .mp4 "
duration = message . video_note . duration
msg_type = " VIDEO_NOTE "
elif message . audio :
file_id = message . audio . file_id
ext = os . path . splitext ( message . audio . file_name or " audio.mp3 " ) [ 1 ] or " .mp3 "
filename = f " audio_ { file_id } { ext } "
duration = message . audio . duration or 0
msg_type = " AUDIO_RECORDING "
elif message . document :
doc = message . document
fn = ( doc . file_name or " " ) . lower ( )
mime = ( doc . mime_type or " " ) . lower ( )
allowed_exts = ( ' .mp3 ' , ' .m4a ' , ' .ogg ' , ' .wav ' , ' .aac ' , ' .flac ' , ' .3gp ' , ' .oga ' , ' .wma ' )
if not ( mime . startswith ( " audio/ " ) or fn . endswith ( allowed_exts ) ) :
return # Н е
file_id = doc . file_id
ext = os . path . splitext ( doc . file_name or " rec.m4a " ) [ 1 ] or " .m4a "
filename = f " doc_ { file_id } { ext } "
duration = 0
msg_type = " RECORDER_FILE "
else :
return
status_msg = await message . reply ( " 🎙 <i>Слушаю и анализирую аудиозапись...</i> " , parse_mode = " HTML " )
await bot . send_chat_action ( message . chat . id , " typing " )
try :
if message . voice :
file_id = message . voice . file_id
filename = f " { file_id } .oga "
msg_type = " VOICE "
else :
file_id = message . video_note . file_id
filename = f " { file_id } .mp4 "
msg_type = " VIDEO_NOTE "
local_audio_path = f " /tmp/ { filename } "
file_info = await bot . get_file ( file_id )
await bot . download_file ( file_info . file_path , local_audio_path )
file = await bot . get_file ( file_id )
# 1. Прямая расшифровка через Gemini Audio
transcribed_text = process_audio_with_gemini ( local_audio_path )
# Используем aiohttp для отправки файла в микросервис
async with aiohttp . ClientSession ( ) as session :
# Скачиваем файл во временный буфер
file_url = f " https://api.telegram.org/file/bot { bot . token } / { file . file_path } "
# Прокси если нужно (подтянется из конфига если настроено, но тут идем напрямую)
async with session . get ( file_url ) as file_resp :
if file_resp . status != 200 :
raise Exception ( f " Н е { file_resp . status } " )
file_content = await file_resp . read ( )
# Отправляем в наш Whisper-Service
data = aiohttp . FormData ( )
data . add_field ( ' file ' , file_content , filename = filename )
async with session . post ( WHISPER_API_URL , data = data ) as whisper_resp :
if whisper_resp . status != 200 :
err_text = await whisper_resp . text ( )
raise Exception ( f " Ошибка Whisper API: { err_text } " )
result = await whisper_resp . json ( )
text = result . get ( " text " , " " )
duration = result . get ( " duration " , 0 )
if os . path . exists ( local_audio_path ) :
os . remove ( local_audio_path )
if text :
# Логируем в базу
log_voice_to_db ( message . from_user . id , message . from_user . username , msg_type , int ( duration ) , text )
if transcribed_text :
log_voice_to_db ( message . from_user . id , message . from_user . username , msg_type , int ( duration ) , transcribed_text )
# Формируем красивый ответ
response = f " 📝 <b>РАСШИФРОВКА</b> ( { msg_type } ) \n "
response + = f " ⏱ Длительность: { int ( duration ) } с \n "
response + = f " ━━━━━━━━━━━━━━ \n "
response + = f " <i> { text } </i> "
await status_msg . edit_text ( f " 🗣 <b>Услышано:</b> «<i> { transcribed_text } </i>» \n \n ⚙️ <b>Майа систематизирует и готовит ответ...</b> " , parse_mode = " HTML " )
await bot . send_chat_action ( message . chat . id , " record_voice " )
# 2. Исполнение Maya Lead Agent (систематизация + команды хоста)
agent_response = await execute_antigravity_agent ( transcribed_text )
# 3. Подготовка текста для отправки (вырезаем тег <voice> из визуального сообщения)
clean_text = re . sub ( r " <voice>.*?</voice> " , " " , agent_response , flags = re . DOTALL ) . strip ( )
formatted_text = f " 🎯 <b>Майа // Структурированный отклик:</b> \n \n { clean_text } "
await status_msg . edit_text ( response , parse_mode = " HTML " )
# Разделяем длинный текст Telegram если нужно (TG limit 4096)
if len ( formatted_text ) > 4000 :
await status_msg . edit_text ( formatted_text [ : 4000 ] , parse_mode = " HTML " )
await message . reply ( formatted_text [ 4000 : ] , parse_mode = " HTML " )
else :
await status_msg . edit_text ( formatted_text , parse_mode = " HTML " )
# 4. Синтез и отправка голосовой заметки (Voice Note)
try :
voice_bytes = await text_to_speech_bytes ( agent_response )
if voice_bytes :
voice_file = BufferedInputFile ( voice_bytes , filename = " reply_voice.mp3 " )
await message . reply_voice ( voice = voice_file , caption = " 🎙 Голосовой отклик Майи " )
except Exception as tts_err :
logger . error ( f " TTS Send error: { tts_err } " )
else :
await status_msg . edit_text ( " 🔇 Голос обнаружен, но слов не разобрал. " )
await status_msg . edit_text ( " 🔇 Аудиозапись принята, но слова не распознаны. Попробуй наговорить еще раз ." )
except Exception as e :
logger . error ( f " ❌ Voice Integration Error: { e } " )
await status_msg . edit_text ( f " ❌ Ошибка модуля Whisper: { e } " )
logger . error ( f " ❌ Maya Voice Integration Error: { e } " )
await status_msg . edit_text ( f " ❌ Ошибка обработки аудио: { e } " , parse_mode = " HTML " )
voice_router = router
