Compare commits

...

10 Commits

7 changed files with 167 additions and 42 deletions
+4 -2
View File
@@ -13,7 +13,7 @@ APP_VERSION=1.0.0
# Server # Server
HOST=0.0.0.0 HOST=0.0.0.0
PORT=5056 PORT=5008
RELOAD=True RELOAD=True
#SFTP #SFTP
@@ -23,4 +23,6 @@ SFPT_PASSWORD = Audio4analy6!6
FILESAPTH = audiofiles FILESAPTH = audiofiles
GIGAAM_API_URL = http://localhost:5001 GIGAAM_API_URL = http://host.docker.internal:5002
CALLS_WEB_CLIENT_URL=http://web_client_test:8642
AUDIO_FILES_PATH_TEST=D:\SpeechAnalitics\audiofiles
+16 -2
View File
@@ -1,12 +1,26 @@
# FileAudioAPI # FileAudioAPI
## Назначение и роль в экосистеме
**FileAudioAPI** — сервис **учёта аудио и результатов распознавания** между диском, GPU-распознаванием и веб-клиентом. На **FastAPI**: хранит метаданные и статусы в БД, складывает файлы в общий каталог **`AUDIOFILES_PATH`** (тот же volume `audiofiles`, что у **GigaAM_API** и **Calls_WEB_Client_main**), инициирует ASR в **GigaAM_API**, принимает обратно текст/заключение, при необходимости дергает webhook в **Calls_WEB_Client_main**.
| Направление | Роль |
|-------------|------|
| **Внутренние файлы** | Регистрация, очереди `process-all`, статусы `recognition_*`, связь с `AiConclusion`. |
| **Внешние файлы** | Загрузка с опциональным `callback_url`, сохранение под UUID-именем в общей папке, **всегда** постановка на распознавание в GigaAM. |
| **Интеграция** | `GIGAAM_API_URL`, `CALLS_WEB_CLIENT_URL` / `WEBHOOK_API_KEY` для доставки результатов в основное приложение. |
Swagger: **`/api/v1/docs`** (см. ниже).
---
Сервис для хранения аудиофайлов, индексации файлов, записи и выдачи результатов распознавания, реализованный на **FastAPI**. Сервис для хранения аудиофайлов, индексации файлов, записи и выдачи результатов распознавания, реализованный на **FastAPI**.
## 🚀 Быстрый старт ## 🚀 Быстрый старт
```bash ```bash
# Клонировать и перейти в директорию # Перейти в директорию сервиса (от корня репозитория SpeechAnalytics)
cd /Users/petr/SpeechAnalytics/FileAudioAPI cd FileAudioAPI
# Запуск с Docker (рекомендуется) # Запуск с Docker (рекомендуется)
docker-compose up -d docker-compose up -d
+6 -3
View File
@@ -25,14 +25,14 @@ PORT = int(os.getenv("PORT", "5008"))
HOST = os.getenv("HOST", "localhost") HOST = os.getenv("HOST", "localhost")
# GigaAM API Configuration # GigaAM API Configuration
GIGAAM_API_URL = os.getenv("GIGAAM_API_URL", "http://192.168.1.73:5002") GIGAAM_API_URL = os.getenv("GIGAAM_API_URL", "http://host.docker.internal:5002")
AUDIOFILES_PATH = os.path.join(os.getcwd(), os.getenv("FILESAPTH", "audiofiles")) AUDIOFILES_PATH = os.path.join(os.getcwd(), os.getenv("FILESAPTH", "audiofiles"))
# Calls_WEB_Client_main Webhook Configuration # Calls_WEB_Client_main Webhook Configuration
CALLS_WEB_CLIENT_URL = os.getenv( CALLS_WEB_CLIENT_URL = os.getenv(
"CALLS_WEB_CLIENT_URL", "CALLS_WEB_CLIENT_URL",
"http://192.168.1.73:8642" "http://host.docker.internal:8642"
) )
WEBHOOK_ENDPOINT = f"{CALLS_WEB_CLIENT_URL}/api/transcription/webhook" WEBHOOK_ENDPOINT = f"{CALLS_WEB_CLIENT_URL}/api/transcription/webhook"
WEBHOOK_API_KEY = os.getenv("WEBHOOK_API_KEY", "webhook_secret_key") WEBHOOK_API_KEY = os.getenv("WEBHOOK_API_KEY", "webhook_secret_key")
@@ -40,4 +40,7 @@ WEBHOOK_API_KEY = os.getenv("WEBHOOK_API_KEY", "webhook_secret_key")
# Auto-restore recognition on startup # Auto-restore recognition on startup
ENABLE_AUTO_RESTORE = os.getenv("ENABLE_AUTO_RESTORE", "true").lower() == "true" ENABLE_AUTO_RESTORE = os.getenv("ENABLE_AUTO_RESTORE", "true").lower() == "true"
AUTO_RESTORE_LIMIT = int(os.getenv("AUTO_RESTORE_LIMIT", "100")) # Максимум файлов для восстановления AUTO_RESTORE_LIMIT = int(os.getenv("AUTO_RESTORE_LIMIT", "100")) # Максимум файлов для восстановления
AUTO_RESTORE_DELAY = int(os.getenv("AUTO_RESTORE_DELAY", "5")) # Задержка перед запуском (секунды) AUTO_RESTORE_DELAY = int(os.getenv("AUTO_RESTORE_DELAY", "5")) # Задержка перед запуском (секунды)
# Recognition retry policy (FileAudioAPI side)
MAX_RECOGNITION_ATTEMPTS = int(os.getenv("MAX_RECOGNITION_ATTEMPTS", "3"))
+9 -1
View File
@@ -15,6 +15,10 @@ class Audio(Base):
duration = Column(Float) duration = Column(Float)
file_size = Column(Integer) file_size = Column(Integer)
sourse = Column(Text, default="internal") sourse = Column(Text, default="internal")
recognition_status = Column(Text, default="pending", index=True) # pending, processing, completed, failed
recognition_attempts = Column(Integer, default=0)
recognition_last_error = Column(Text, nullable=True)
recognition_last_attempt_at = Column(DateTime, nullable=True)
ai_conclusion = relationship("AiConclusion", back_populates="audio", cascade="all, delete-orphan") ai_conclusion = relationship("AiConclusion", back_populates="audio", cascade="all, delete-orphan")
@@ -26,5 +30,9 @@ class Audio(Base):
"file_path": self.file_path, "file_path": self.file_path,
"duration": self.duration, "duration": self.duration,
"file_size": self.file_size, "file_size": self.file_size,
"sourse": self.sourse "sourse": self.sourse,
"recognition_status": self.recognition_status,
"recognition_attempts": self.recognition_attempts,
"recognition_last_error": self.recognition_last_error,
"recognition_last_attempt_at": self.recognition_last_attempt_at.isoformat() if self.recognition_last_attempt_at else None,
} }
+96 -30
View File
@@ -9,9 +9,10 @@ from apiApp.database import get_db
from apiApp.database.Audio import Audio from apiApp.database.Audio import Audio
from apiApp.database.AiConclusion import AiConclusion from apiApp.database.AiConclusion import AiConclusion
from datetime import datetime from datetime import datetime
import time
import logging import logging
import requests import requests
from apiApp.config import WEBHOOK_ENDPOINT, WEBHOOK_API_KEY from apiApp.config import WEBHOOK_ENDPOINT, WEBHOOK_API_KEY, CALLS_WEB_CLIENT_URL
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
ai_conclusion_router = APIRouter() ai_conclusion_router = APIRouter()
@@ -62,6 +63,11 @@ class AiConclusionResponse(BaseModel):
error: Optional[str] = None error: Optional[str] = None
class RecognitionFailedRequest(BaseModel):
filename: str
error: str
class ConclusionByFilenameResponse(BaseModel): class ConclusionByFilenameResponse(BaseModel):
"""Заключение по имени файла""" """Заключение по имени файла"""
filename: str filename: str
@@ -154,43 +160,57 @@ async def save_ai_conclusion(request: AiConclusionRequest, db: Session = Depends
db.commit() db.commit()
logger.info(f"✅ Заключение сохранено для {request.filename}") logger.info(f"✅ Заключение сохранено для {request.filename}")
# Обновляем статус распознавания у Audio
audio.recognition_status = "completed"
audio.recognition_last_error = None
db.commit()
# Для внешних файлов — отправляем результат клиенту из FileAudioAPI # Для внешних файлов — отправляем результат клиенту из FileAudioAPI
if (audio.sourse or "").lower() == "external" and request.callback_url: if (audio.sourse or "").lower() == "external" and request.callback_url:
_send_callback(request.callback_url, audio, conclusion_data) _send_callback(request.callback_url, audio, conclusion_data)
# Отправляем webhook в Calls_WEB_Client_main для анализа # Отправляем webhook в Calls_WEB_Client_main для анализа (с retry при перезагрузке контейнера)
try: if (audio.sourse or "").lower() != "external":
if (audio.sourse or "").lower() != "external": logger.info(f"📤 Отправка webhook в Calls_WEB_Client_main для {request.filename}")
logger.info(f"📤 Отправка webhook в Calls_WEB_Client_main для {request.filename}")
webhook_payload = { webhook_payload = {
"audio_id": str(audio.id), "audio_id": str(audio.id),
"filename": request.filename, "filename": request.filename,
"transcription": request.transcription, "transcription": request.transcription,
"corrected_transcription": request.corrected_transcription, "corrected_transcription": request.corrected_transcription,
"segments": request.segments, "segments": request.segments,
"processing_time_seconds": request.processing_time_seconds "processing_time_seconds": request.processing_time_seconds
} }
webhook_response = requests.post( max_retries = 5
WEBHOOK_ENDPOINT, delays = [2, 4, 8, 15, 30]
json=webhook_payload, delivered = False
headers={"X-Webhook-Key": WEBHOOK_API_KEY},
timeout=30
)
if webhook_response.status_code == 200: for attempt in range(1, max_retries + 1):
logger.info(f"✅ Webhook успешно отправлен для {request.filename}") try:
else: webhook_response = requests.post(
logger.warning(f"⚠️ Webhook вернул статус {webhook_response.status_code}") WEBHOOK_ENDPOINT,
logger.warning(f"Response: {webhook_response.text}") json=webhook_payload,
headers={"X-Webhook-Key": WEBHOOK_API_KEY},
timeout=30
)
if webhook_response.status_code in (200, 201):
logger.info(f"✅ Webhook успешно отправлен для {request.filename} (попытка {attempt}/{max_retries})")
delivered = True
break
else:
logger.warning(f"⚠️ Webhook вернул статус {webhook_response.status_code} (попытка {attempt}/{max_retries})")
logger.warning(f"Response: {webhook_response.text[:200]}")
except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as e:
logger.warning(f"⏳ Calls_WEB_Client_main временно недоступен ({type(e).__name__}), попытка {attempt}/{max_retries}...")
except Exception as e:
logger.error(f"❌ Непредвиденная ошибка при отправке webhook (попытка {attempt}): {e}")
except requests.exceptions.ConnectionError: if attempt < max_retries:
logger.error(f"❌ Не удалось подключиться к Calls_WEB_Client_main webhook: {WEBHOOK_ENDPOINT}") time.sleep(delays[attempt - 1])
except requests.exceptions.Timeout:
logger.warning(f"⚠️ Таймаут при отправке webhook для {request.filename}") if not delivered:
except Exception as e: logger.error(f"❌ Не удалось доставить webhook для {request.filename} после {max_retries} попыток")
logger.error(f"❌ Ошибка при отправке webhook: {e}")
return AiConclusionResponse( return AiConclusionResponse(
success=True, success=True,
@@ -208,3 +228,49 @@ async def save_ai_conclusion(request: AiConclusionRequest, db: Session = Depends
status_code=500, status_code=500,
detail=str(e) detail=str(e)
) )
@ai_conclusion_router.post("/conclusion/failed", response_model=AiConclusionResponse)
async def mark_recognition_failed(request: RecognitionFailedRequest, db: Session = Depends(get_db)):
"""
Помечает распознавание как failed для файла (чтобы auto-restore не пытался бесконечно).
Используется GigaAM_API при невозможности получить результат.
"""
audio = db.query(Audio).filter(Audio.filename == request.filename).first()
if not audio:
logger.warning(f"⚠️ Попытка пометить failed для несуществующего аудио: {request.filename}, error={request.error}")
raise HTTPException(status_code=404, detail=f"Файл не найден: {request.filename}")
audio.recognition_status = "failed"
audio.recognition_last_error = request.error
db.commit()
logger.warning(f"❌ Распознавание помечено failed для {request.filename}. Причина: {request.error}")
# Отправляем вебхук об ошибке в Calls_WEB_Client_main, чтобы синхронизировать pipeline_status
try:
if (audio.sourse or "").lower() != "external":
failed_webhook_payload = {
"audio_id": str(audio.id),
"filename": request.filename,
"status": "failed",
"error": request.error
}
webhook_url = f"{CALLS_WEB_CLIENT_URL}/api/transcription/failed"
logger.info(f"📤 Отправка failed-webhook в Calls_WEB_Client_main: {webhook_url}")
requests.post(
webhook_url,
json=failed_webhook_payload,
headers={"X-Webhook-Key": WEBHOOK_API_KEY},
timeout=10
)
except Exception as e:
logger.error(f"❌ Ошибка отправки failed-webhook в Calls_WEB: {e}")
return AiConclusionResponse(
success=True,
message="Recognition marked as failed",
audio_id=str(audio.id),
filename=audio.filename
)
+27 -3
View File
@@ -13,7 +13,7 @@ from datetime import datetime
from apiApp.database import get_db from apiApp.database import get_db
from apiApp.database.Audio import Audio from apiApp.database.Audio import Audio
from apiApp.database.AiConclusion import AiConclusion from apiApp.database.AiConclusion import AiConclusion
from apiApp.config import AUDIOFILES_PATH from apiApp.config import AUDIOFILES_PATH, MAX_RECOGNITION_ATTEMPTS
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
audio_management_router = APIRouter() audio_management_router = APIRouter()
@@ -32,10 +32,14 @@ def query_audio_without_conclusion(db, limit=None):
AiConclusion.audio_id == Audio.id AiConclusion.audio_id == Audio.id
) )
# Берём только те, которые еще можно/нужно распознавать
query = db.query(Audio).filter( query = db.query(Audio).filter(
~subquery ~subquery
).filter( ).filter(
Audio.sourse == "internal" Audio.sourse == "internal"
).filter(
(Audio.recognition_status.in_(["pending", "processing"])) |
((Audio.recognition_status == "failed") & (Audio.recognition_attempts < MAX_RECOGNITION_ATTEMPTS))
).order_by(Audio.index_date.asc()) ).order_by(Audio.index_date.asc())
if limit: if limit:
@@ -112,6 +116,10 @@ async def register_audio_file(
audio.filename = filename audio.filename = filename
audio.file_size = file_size audio.file_size = file_size
audio.index_date = datetime.utcnow() audio.index_date = datetime.utcnow()
audio.recognition_status = "pending"
audio.recognition_attempts = 0
audio.recognition_last_error = None
audio.recognition_last_attempt_at = None
db.add(audio) db.add(audio)
db.commit() db.commit()
@@ -150,6 +158,11 @@ def process_audio_file(audio_id: str, db: Session):
return return
logger.info(f"🎵 Запуск распознавания для {audio.filename}") logger.info(f"🎵 Запуск распознавания для {audio.filename}")
audio.recognition_status = "processing"
audio.recognition_attempts = (audio.recognition_attempts or 0) + 1
audio.recognition_last_attempt_at = datetime.utcnow()
audio.recognition_last_error = None
db.commit()
# Проверяем что файл существует на диске # Проверяем что файл существует на диске
from apiApp.config import AUDIOFILES_PATH from apiApp.config import AUDIOFILES_PATH
@@ -158,7 +171,9 @@ def process_audio_file(audio_id: str, db: Session):
if not os.path.exists(file_path): if not os.path.exists(file_path):
logger.error(f"❌ Файл не найден на диске в FileAudioAPI: {file_path}") logger.error(f"❌ Файл не найден на диске в FileAudioAPI: {file_path}")
# Помечаем audio как проблемный audio.recognition_status = "failed"
audio.recognition_last_error = f"File not found on disk: {file_path}"
db.commit()
return return
file_size = os.path.getsize(file_path) file_size = os.path.getsize(file_path)
@@ -187,11 +202,20 @@ def process_audio_file(audio_id: str, db: Session):
error_detail = response.text error_detail = response.text
logger.error(f"❌ Ошибка запуска распознавания для {audio.filename}: {response.status_code}") logger.error(f"❌ Ошибка запуска распознавания для {audio.filename}: {response.status_code}")
logger.error(f" Detail: {error_detail}") logger.error(f" Detail: {error_detail}")
audio.recognition_status = "failed"
audio.recognition_last_error = f"GigaAM start failed: {response.status_code} {error_detail}"
db.commit()
except requests.exceptions.Timeout: except requests.exceptions.Timeout:
logger.error(f"❌ Таймаут при отправке задачи для {audio.filename}") logger.error(f"❌ Таймаут при отправке задачи для {audio.filename}")
audio.recognition_status = "failed"
audio.recognition_last_error = "Timeout when starting recognition in GigaAM"
db.commit()
except requests.exceptions.ConnectionError as e: except requests.exceptions.ConnectionError as e:
logger.error(f"❌ Ошибка подключения к GigaAM API для {audio.filename}: {e}") logger.error(f"❌ Ошибка подключения к GigaAM API для {audio.filename}: {e}")
audio.recognition_status = "failed"
audio.recognition_last_error = f"Connection error when starting recognition in GigaAM: {e}"
db.commit()
except Exception as e: except Exception as e:
logger.error(f"❌ Ошибка при обработке {audio_id}: {e}") logger.error(f"❌ Ошибка при обработке {audio_id}: {e}")
@@ -226,7 +250,7 @@ async def process_all_pending_audio(
logger.info(f"🚀 Поиск Audio без AiConclusion (limit={limit})") logger.info(f"🚀 Поиск Audio без AiConclusion (limit={limit})")
# Находим все Audio без AiConclusion используя вспомогательную функцию # Находим все Audio без AiConclusion используя вспомогательную функцию
pending_audio = query_audio_without_conclusion(db, limit).all() pending_audio: list[Audio] = query_audio_without_conclusion(db, limit).all()
total_pending = query_audio_without_conclusion(db).count() total_pending = query_audio_without_conclusion(db).count()
if not pending_audio: if not pending_audio:
+9 -1
View File
@@ -11,7 +11,15 @@ class Loader():
self.call_types = ['in'] self.call_types = ['in']
self.loaded_files = [] # Список загруженных файлов self.loaded_files = [] # Список загруженных файлов
def filter_call(self, filename: str): def filter_call(self, filename: str) -> bool:
if not filename or not filename.endswith('.wav'):
return False
# Пропускаем раздельные каналы звонков (_r = receive/клиент, _t = transmit/оператор)
name_without_ext = filename[:-4]
if name_without_ext.endswith(('_r', '_t', '_in', '_out', '-r', '-t')):
return False
if filename.split("-")[0] in self.call_types: if filename.split("-")[0] in self.call_types:
return True return True
return False return False