PoturaevPetr
525 lines
20 KiB
Python
525 lines
20 KiB
Python
"""
|
|
API endpoints для управления аудиофайлами (регистрация и пакетная обработка)
|
|
Используется Calls_WEB_Client_main для оркестрации процесса распознавания
|
|
"""
|
|
from fastapi import APIRouter, HTTPException, Depends, BackgroundTasks
|
|
from sqlalchemy.orm import Session
|
|
from pydantic import BaseModel
|
|
from typing import Optional, List
|
|
import os
|
|
import logging
|
|
from datetime import datetime
|
|
|
|
from apiApp.database import get_db
|
|
from apiApp.database.Audio import Audio
|
|
from apiApp.database.AiConclusion import AiConclusion
|
|
from apiApp.config import AUDIOFILES_PATH, MAX_RECOGNITION_ATTEMPTS
|
|
|
|
logger = logging.getLogger(__name__)
|
|
audio_management_router = APIRouter()
|
|
|
|
|
|
def query_audio_without_conclusion(db, limit=None):
|
|
"""
|
|
Возвращает запрос для поиска Audio без AiConclusion
|
|
|
|
Использует exists() подзапрос, так как AiConclusion - это relationship
|
|
"""
|
|
from sqlalchemy import exists, not_
|
|
|
|
# Используем более простой подзапрос
|
|
subquery = exists().where(
|
|
AiConclusion.audio_id == Audio.id
|
|
)
|
|
|
|
# Берём только те, которые еще можно/нужно распознавать
|
|
query = db.query(Audio).filter(
|
|
~subquery
|
|
).filter(
|
|
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())
|
|
|
|
if limit:
|
|
query = query.limit(limit)
|
|
|
|
return query
|
|
|
|
class AudioRegisterRequest(BaseModel):
|
|
"""Запрос на регистрацию аудиофайла"""
|
|
filename: str
|
|
file_path: str # Полный путь к файлу в общей папке audiofiles
|
|
|
|
|
|
class AudioProcessAllRequest(BaseModel):
|
|
"""Запрос на пакетное распознавание"""
|
|
limit: int = 100
|
|
|
|
|
|
class AudioRegisterResponse(BaseModel):
|
|
"""Ответ на регистрацию аудиофайла"""
|
|
id: str
|
|
filename: str
|
|
file_size: int
|
|
created_at: datetime
|
|
|
|
|
|
@audio_management_router.post("/audio/register", response_model=AudioRegisterResponse, status_code=201)
|
|
async def register_audio_file(
|
|
request: AudioRegisterRequest,
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""
|
|
Регистрация аудиофайла в БД (без копирования файла)
|
|
|
|
Создаёт запись в таблице Audio для файла, который уже находится
|
|
в общей папке audiofiles. НЕ копирует файл, только создаёт запись в БД.
|
|
|
|
Args:
|
|
request: {filename: "in-xxx.wav", file_path: "/app/audiofiles/in-xxx.wav"}
|
|
|
|
Returns:
|
|
201 Created + информация о созданной записи
|
|
400 Bad Request если файл уже зарегистрирован
|
|
404 Not Found если файл не существует на диске
|
|
"""
|
|
try:
|
|
filename = request.filename
|
|
file_path = request.file_path
|
|
|
|
logger.info(f"📝 Регистрация файла: {filename}")
|
|
|
|
# Проверяем, что файл уже существует в общей папке
|
|
if not os.path.exists(file_path):
|
|
logger.error(f"❌ Файл не найден: {file_path}")
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f'Файл не найден на диске: {file_path}'
|
|
)
|
|
|
|
# Проверяем, что файл не был уже зарегистрирован
|
|
existing_audio = db.query(Audio).filter(Audio.filename == filename).first()
|
|
if existing_audio:
|
|
logger.warning(f"⚠️ Файл уже зарегистрирован: {filename}")
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f'Файл {filename} уже зарегистрирован в БД'
|
|
)
|
|
|
|
# Получаем размер файла
|
|
file_size = os.path.getsize(file_path)
|
|
|
|
# Создаём запись в БД
|
|
audio = Audio()
|
|
audio.filename = filename
|
|
audio.file_size = file_size
|
|
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.commit()
|
|
db.refresh(audio)
|
|
|
|
logger.info(f"✅ Файл зарегистрирован: {filename} (audio_id={audio.id})")
|
|
|
|
return AudioRegisterResponse(
|
|
id=str(audio.id),
|
|
filename=audio.filename,
|
|
file_size=audio.file_size,
|
|
created_at=audio.index_date
|
|
)
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
db.rollback()
|
|
logger.error(f"❌ Ошибка при регистрации файла: {e}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f'Ошибка при регистрации: {str(e)}'
|
|
)
|
|
|
|
|
|
def process_audio_file(audio_id: str, db: Session):
|
|
"""
|
|
Фоновая обработка одного аудиофайла
|
|
|
|
Отправляет запрос в GigaAM API для распознавания
|
|
"""
|
|
try:
|
|
audio = db.query(Audio).filter(Audio.id == audio_id).first()
|
|
if not audio:
|
|
logger.error(f"❌ Audio {audio_id} не найден")
|
|
return
|
|
|
|
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
|
|
import os
|
|
file_path = os.path.join(AUDIOFILES_PATH, audio.filename)
|
|
|
|
if not os.path.exists(file_path):
|
|
logger.error(f"❌ Файл не найден на диске в FileAudioAPI: {file_path}")
|
|
audio.recognition_status = "failed"
|
|
audio.recognition_last_error = f"File not found on disk: {file_path}"
|
|
db.commit()
|
|
return
|
|
|
|
file_size = os.path.getsize(file_path)
|
|
logger.info(f"📁 Файл найден: {file_path} (размер: {file_size} bytes)")
|
|
|
|
# Формируем запрос в GigaAM API - ТОЛЬКО асинхронный endpoint
|
|
from apiApp.config import GIGAAM_API_URL
|
|
|
|
# Используем только асинхронный endpoint
|
|
api_url = f"{GIGAAM_API_URL}/api/call/process/async"
|
|
|
|
payload = {
|
|
"filename": audio.filename
|
|
}
|
|
|
|
import requests
|
|
try:
|
|
# Отправляем запрос в асинхронный endpoint
|
|
response = requests.post(api_url, json=payload, timeout=10)
|
|
|
|
if response.status_code == 200 or response.status_code == 202:
|
|
result = response.json()
|
|
task_id = result.get('task_id')
|
|
logger.info(f"✅ Задача добавлена в очередь для {audio.filename} (task_id: {task_id})")
|
|
else:
|
|
error_detail = response.text
|
|
logger.error(f"❌ Ошибка запуска распознавания для {audio.filename}: {response.status_code}")
|
|
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:
|
|
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:
|
|
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:
|
|
logger.error(f"❌ Ошибка при обработке {audio_id}: {e}")
|
|
|
|
|
|
@audio_management_router.post("/audio/process-all")
|
|
async def process_all_pending_audio(
|
|
request: AudioProcessAllRequest,
|
|
background_tasks: BackgroundTasks,
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""
|
|
Запуск распознавания для всех Audio без AiConclusion
|
|
|
|
Находит все записи Audio, у которых нет AiConclusion, и запускает
|
|
распознавание для них (до указанного лимита).
|
|
|
|
Args:
|
|
request: {limit: 100} - максимум файлов для обработки
|
|
|
|
Returns:
|
|
200 OK + {
|
|
"started_count": 15,
|
|
"pending_files": ["file1.wav", "file2.wav", ...],
|
|
"total_pending": 50,
|
|
"skipped_duplicates": 2
|
|
}
|
|
"""
|
|
try:
|
|
limit = request.limit
|
|
|
|
logger.info(f"🚀 Поиск Audio без AiConclusion (limit={limit})")
|
|
|
|
# Находим все Audio без AiConclusion используя вспомогательную функцию
|
|
pending_audio: list[Audio] = query_audio_without_conclusion(db, limit).all()
|
|
total_pending = query_audio_without_conclusion(db).count()
|
|
|
|
if not pending_audio:
|
|
logger.info("ℹ️ Нет файлов для распознавания")
|
|
return {
|
|
"started_count": 0,
|
|
"pending_files": [],
|
|
"total_pending": 0,
|
|
"skipped_duplicates": 0,
|
|
"message": "Нет файлов без AiConclusion"
|
|
}
|
|
|
|
logger.info(f"📋 Найдено файлов для обработки: {len(pending_audio)} из {total_pending}")
|
|
|
|
# Добавляем задачи в фон
|
|
started_count = 0
|
|
pending_files = []
|
|
skipped_duplicates = 0 # Счётчик дубликатов
|
|
|
|
for audio in pending_audio:
|
|
# Проверяем, что файл существует
|
|
file_path = os.path.join(AUDIOFILES_PATH, audio.filename)
|
|
if not os.path.exists(file_path):
|
|
logger.warning(f"⚠️ Файл не найден на диске: {audio.filename}")
|
|
continue
|
|
|
|
# ПРОВЕРКА НА ДУБЛИКАТЫ: Проверяем, не обрабатывается ли файл уже
|
|
import requests
|
|
try:
|
|
# Проверяем статус через GigaAM API
|
|
check_url = f"{os.getenv('GIGAAM_API_URL', 'http://gigaam_api:5001')}/api/call/check-status"
|
|
check_response = requests.post(
|
|
check_url,
|
|
json={"filename": audio.filename},
|
|
timeout=5
|
|
)
|
|
|
|
if check_response.status_code == 200:
|
|
check_data = check_response.json()
|
|
if check_data.get('in_queue', False):
|
|
logger.info(f"⏭️ Файл уже в очереди GigaAM: {audio.filename}")
|
|
skipped_duplicates += 1
|
|
continue
|
|
except Exception as e:
|
|
# Если проверка упала - всё равно продолжаем
|
|
logger.debug(f"Не удалось проверить статус {audio.filename}: {e}")
|
|
|
|
# Добавляем в фон (асинхронно)
|
|
pending_files.append(audio.filename)
|
|
started_count += 1
|
|
|
|
# Запускаем обработку в фоне
|
|
background_tasks.add_task(
|
|
process_single_audio,
|
|
str(audio.id)
|
|
)
|
|
|
|
logger.info(f"✅ Запущено распознавание для {started_count} файлов")
|
|
if skipped_duplicates > 0:
|
|
logger.info(f"⏭️ Пропущено дубликатов: {skipped_duplicates}")
|
|
|
|
return {
|
|
"started_count": started_count,
|
|
"pending_files": pending_files,
|
|
"total_pending": total_pending,
|
|
"skipped_duplicates": skipped_duplicates
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"❌ Ошибка при запуске пакетного распознавания: {e}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f'Ошибка: {str(e)}'
|
|
)
|
|
|
|
|
|
def process_single_audio(audio_id: str):
|
|
"""
|
|
Обработка одного аудиофайла в фоне
|
|
|
|
Создаёт новую DB сессию для обработки
|
|
"""
|
|
from apiApp.database import SessionLocal
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
process_audio_file(audio_id, db)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@audio_management_router.get("/audio/pending")
|
|
async def get_pending_audio(
|
|
db: Session = Depends(get_db),
|
|
limit: int = 100
|
|
):
|
|
"""
|
|
Получить список Audio без AiConclusion
|
|
|
|
Query Parameters:
|
|
limit: Максимум файлов (default: 100)
|
|
|
|
Returns:
|
|
Список файлов, ожидающих распознавания
|
|
"""
|
|
try:
|
|
pending_audio = query_audio_without_conclusion(db, limit).all()
|
|
|
|
files_info = []
|
|
for audio in pending_audio:
|
|
file_path = os.path.join(AUDIOFILES_PATH, audio.filename)
|
|
exists = os.path.exists(file_path)
|
|
|
|
files_info.append({
|
|
"audio_id": str(audio.id),
|
|
"filename": audio.filename,
|
|
"file_size": audio.file_size,
|
|
"created_at": audio.index_date.isoformat() if audio.index_date else None,
|
|
"exists_on_disk": exists
|
|
})
|
|
|
|
total_pending = query_audio_without_conclusion(db).count()
|
|
|
|
return {
|
|
"total_pending": total_pending,
|
|
"count": len(files_info),
|
|
"files": files_info
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"❌ Ошибка при получении списка: {e}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=str(e)
|
|
)
|
|
|
|
|
|
@audio_management_router.get("/audio/stats")
|
|
async def get_audio_stats(db: Session = Depends(get_db)):
|
|
"""
|
|
Получить статистику по аудиофайлам
|
|
|
|
Returns:
|
|
Статистика по Audio записям
|
|
"""
|
|
try:
|
|
total_audio = db.query(Audio).count()
|
|
with_conclusion = total_audio - query_audio_without_conclusion(db).count()
|
|
without_conclusion = query_audio_without_conclusion(db).count()
|
|
|
|
# Проверяем существование файлов на диске
|
|
all_audio = db.query(Audio).all()
|
|
existing_count = 0
|
|
missing_files = []
|
|
small_files = [] # Файлы меньше 1KB
|
|
|
|
for audio in all_audio:
|
|
file_path = os.path.join(AUDIOFILES_PATH, audio.filename)
|
|
if os.path.exists(file_path):
|
|
existing_count += 1
|
|
file_size = os.path.getsize(file_path)
|
|
if file_size < 1000:
|
|
small_files.append({
|
|
"audio_id": str(audio.id),
|
|
"filename": audio.filename,
|
|
"file_size": file_size,
|
|
"index_date": audio.index_date.isoformat() if audio.index_date else None
|
|
})
|
|
else:
|
|
missing_files.append({
|
|
"audio_id": str(audio.id),
|
|
"filename": audio.filename,
|
|
"index_date": audio.index_date.isoformat() if audio.index_date else None
|
|
})
|
|
|
|
stats = {
|
|
"total_audio": total_audio,
|
|
"with_conclusion": with_conclusion,
|
|
"without_conclusion": without_conclusion,
|
|
"existing_on_disk": existing_count,
|
|
"missing_on_disk": total_audio - existing_count,
|
|
"small_files_count": len(small_files)
|
|
}
|
|
|
|
# Добавляем списки проблемных файлов (первые 50 каждого типа)
|
|
if missing_files:
|
|
stats["missing_files_sample"] = missing_files[:50]
|
|
logger.warning(f"⚠️ Найдено {len(missing_files)} отсутствующих файлов")
|
|
|
|
if small_files:
|
|
stats["small_files_sample"] = small_files[:50]
|
|
logger.warning(f"⚠️ Найдено {len(small_files)} файлов меньше 1KB")
|
|
|
|
return stats
|
|
|
|
except Exception as e:
|
|
logger.error(f"❌ Ошибка при получении статистики: {e}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=str(e)
|
|
)
|
|
|
|
|
|
@audio_management_router.delete("/audio/cleanup")
|
|
async def cleanup_invalid_audio_files(
|
|
delete_missing: bool = False,
|
|
delete_small: bool = True,
|
|
min_size_bytes: int = 1000,
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""
|
|
Удаляет записи Audio для проблемных файлов
|
|
|
|
Query Parameters:
|
|
delete_missing: Удалять записи с отсутствующими файлами (default: False)
|
|
delete_small: Удалять записи с маленькими файлами (default: True)
|
|
min_size_bytes: Минимальный размер файла в bytes (default: 1000)
|
|
|
|
Returns:
|
|
Статистику удаления
|
|
"""
|
|
try:
|
|
all_audio = db.session.query(Audio).all()
|
|
|
|
deleted_missing = 0
|
|
deleted_small = 0
|
|
|
|
for audio in all_audio:
|
|
file_path = os.path.join(AUDIOFILES_PATH, audio.filename)
|
|
|
|
# Проверяем отсутствие файла
|
|
if delete_missing and not os.path.exists(file_path):
|
|
logger.info(f"🗑️ Удаление записи с отсутствующим файлом: {audio.filename}")
|
|
db.session.delete(audio)
|
|
deleted_missing += 1
|
|
continue
|
|
|
|
# Проверяем размер файла
|
|
if delete_small and os.path.exists(file_path):
|
|
file_size = os.path.getsize(file_path)
|
|
if file_size < min_size_bytes:
|
|
logger.info(f"🗑️ Удаление записи с маленьким файлом ({file_size} bytes): {audio.filename}")
|
|
# Удаляем и файл тоже
|
|
try:
|
|
os.remove(file_path)
|
|
logger.info(f" Файл удалён: {file_path}")
|
|
except Exception as e:
|
|
logger.warning(f" Не удалось удалить файл: {e}")
|
|
|
|
db.session.delete(audio)
|
|
deleted_small += 1
|
|
|
|
db.session.commit()
|
|
|
|
return {
|
|
"success": True,
|
|
"deleted_missing": deleted_missing,
|
|
"deleted_small": deleted_small,
|
|
"total_deleted": deleted_missing + deleted_small,
|
|
"message": f"Удалено {deleted_missing + deleted_small} записей"
|
|
}
|
|
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
logger.error(f"❌ Ошибка при очистке: {e}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=str(e)
|
|
)
|
|
|
|
|