80 lines
3.1 KiB
Python
80 lines
3.1 KiB
Python
import os
|
||
import logging
|
||
import requests
|
||
import sqlite3
|
||
from datetime import datetime
|
||
from pydub import AudioSegment
|
||
from pathlib import Path
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Настройки внешнего Whisper-сервиса на MacMini
|
||
WHISPER_API_URL = "http://192.168.10.167:9001/transcribe"
|
||
|
||
class VoiceService:
|
||
def __init__(self, db_path: str, model_size: str = "large-v3"):
|
||
self.db_path = db_path
|
||
self.model_size = model_size
|
||
logger.info(f"🎙️ Использование внешнего Whisper-сервиса ({model_size}) на MacMini")
|
||
self._init_db()
|
||
|
||
def _init_db(self):
|
||
conn = sqlite3.connect(self.db_path)
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
CREATE TABLE IF NOT EXISTS voice_recognition_logs (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
user_id INTEGER,
|
||
username TEXT,
|
||
message_type TEXT,
|
||
duration INTEGER,
|
||
text TEXT,
|
||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||
)
|
||
''')
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
def transcribe(self, file_path: str, user_id: int, username: str, msg_type: str) -> str:
|
||
f_path = Path(file_path)
|
||
if not f_path.exists():
|
||
raise FileNotFoundError(f"Файл не найден: {file_path}")
|
||
|
||
try:
|
||
logger.info(f"🧠 Отправка {f_path.name} на MacMini M4...")
|
||
|
||
# Определяем длительность (локально через pydub)
|
||
audio = AudioSegment.from_file(str(f_path))
|
||
duration = int(len(audio) / 1000)
|
||
|
||
# Отправляем файл в API
|
||
with open(str(f_path), 'rb') as f:
|
||
files = {'file': (f_path.name, f, 'audio/mpeg')}
|
||
response = requests.post(WHISPER_API_URL, files=files, timeout=300)
|
||
|
||
if response.status_code != 200:
|
||
raise Exception(f"API Error: {response.text}")
|
||
|
||
result = response.json()
|
||
text = result.get("text", "").strip()
|
||
|
||
logger.info(f"✅ Распознано успешно ({duration}с)")
|
||
|
||
self._log_to_db(user_id, username, msg_type, duration, text)
|
||
return text
|
||
except Exception as e:
|
||
logger.error(f"❌ Ошибка внешней транскрибации: {e}")
|
||
return "Не удалось распознать голос (ошибка связи с M4-сервером)."
|
||
|
||
def _log_to_db(self, user_id, username, msg_type, duration, text):
|
||
try:
|
||
conn = sqlite3.connect(self.db_path)
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
INSERT INTO voice_recognition_logs (user_id, username, message_type, duration, text)
|
||
VALUES (?, ?, ?, ?, ?)
|
||
''', (user_id, username, msg_type, duration, text))
|
||
conn.commit()
|
||
conn.close()
|
||
except Exception as e:
|
||
logger.error(f"DB Log error: {e}")
|