Evgeny Naumov
Переделан app.py для правильной работы API endpoints в Gradio
0e13ad6
import gradio as gr
import requests
import json
import time
import threading
from typing import Dict, Any
# Глобальное хранилище задач транскрипции
transcription_tasks = {}
def generate_task_id():
"""Генерирует уникальный ID задачи"""
import uuid
return str(uuid.uuid4())
def create_task(task_id: str, filename: str, language: str, model: str) -> Dict[str, Any]:
"""Создает новую задачу транскрипции"""
task = {
"id": task_id,
"filename": filename,
"language": language,
"model": model,
"status": "processing",
"progress": 0,
"result": None,
"error": None,
"created_at": time.time(),
"updated_at": time.time()
}
transcription_tasks[task_id] = task
return task
def update_task(task_id: str, updates: Dict[str, Any]) -> Dict[str, Any]:
"""Обновляет задачу"""
if task_id in transcription_tasks:
transcription_tasks[task_id].update(updates)
transcription_tasks[task_id]["updated_at"] = time.time()
return transcription_tasks.get(task_id)
def simulate_transcription(task_id: str, language: str, model: str):
"""Имитирует процесс транскрипции"""
task = transcription_tasks.get(task_id)
if not task:
return
try:
# Имитируем прогресс
for i in range(0, 101, 10):
update_task(task_id, {"progress": i})
time.sleep(1) # 1 секунда на 10%
# Имитируем результат
result = {
"text": f"[Транскрипция завершена: {task['filename']}]",
"language": language,
"model": model,
"duration": 120.5,
"segments": [],
"timestamp": time.time()
}
update_task(task_id, {
"status": "completed",
"progress": 100,
"result": result
})
except Exception as e:
update_task(task_id, {
"status": "error",
"error": str(e)
})
def create_interface():
"""Создает Gradio интерфейс с API endpoints"""
with gr.Blocks(title="Whisper Web API", theme=gr.themes.Soft()) as demo:
gr.Markdown("""
# 🎤 Whisper Web API
Веб-приложение для транскрипции аудио с использованием моделей Whisper.
## 📋 API Endpoints
- **GET** `/api/health` - проверка здоровья сервера
- **GET** `/api/models` - список доступных моделей
- **GET** `/api/languages` - список поддерживаемых языков
- **POST** `/api/transcribe` - транскрипция аудио файла
- **POST** `/api/transcribe/url` - транскрипция аудио по URL
- **GET** `/api/status/{taskId}` - проверка статуса задачи
- **GET** `/api/result/{taskId}` - получение результата
## 🔗 Ссылки
- **API Base URL**: `https://naumxv-whisper-web-api.hf.space/api`
- **Веб-интерфейс**: `https://naumxv-whisper-web-api.hf.space/`
""")
with gr.Row():
with gr.Column():
gr.HTML("""
<div style="text-align: center; padding: 20px;">
<h3>🌐 Веб-интерфейс</h3>
<p>Откройте веб-интерфейс для интерактивной транскрипции:</p>
<a href="/" target="_blank" style="
display: inline-block;
padding: 10px 20px;
background: linear-gradient(45deg, #667eea 0%, #764ba2 100%);
color: white;
text-decoration: none;
border-radius: 5px;
font-weight: bold;
">Открыть Whisper Web</a>
</div>
""")
with gr.Column():
gr.HTML("""
<div style="text-align: center; padding: 20px;">
<h3>🔧 API Тестирование</h3>
<p>Проверьте работу API:</p>
<a href="/api/health" target="_blank" style="
display: inline-block;
padding: 10px 20px;
background: linear-gradient(45deg, #f093fb 0%, #f5576c 100%);
color: white;
text-decoration: none;
border-radius: 5px;
font-weight: bold;
">Проверить API</a>
</div>
""")
gr.Markdown("""
## 📖 Примеры использования
### Python
```python
import requests
import time
# Проверка здоровья
response = requests.get('https://naumxv-whisper-web-api.hf.space/api/health')
print(response.json())
# Асинхронная транскрипция файла
with open('audio.mp3', 'rb') as f:
files = {'audio': f}
data = {'language': 'ru', 'model': 'base'}
response = requests.post('https://naumxv-whisper-web-api.hf.space/api/transcribe',
files=files, data=data)
result = response.json()
task_id = result['taskId']
print(f"Задача создана: {task_id}")
# Проверка статуса
while True:
status_response = requests.get(f'https://naumxv-whisper-web-api.hf.space/api/status/{task_id}')
status_data = status_response.json()
if status_data['task']['status'] == 'completed':
break
elif status_data['task']['status'] == 'error':
print(f"Ошибка: {status_data['task']['error']}")
break
print(f"Прогресс: {status_data['task']['progress']}%")
time.sleep(5)
# Получение результата
result_response = requests.get(f'https://naumxv-whisper-web-api.hf.space/api/result/{task_id}')
final_result = result_response.json()
print(f"Транскрипция: {final_result['transcription']['text']}")
```
### JavaScript
```javascript
// Асинхронная транскрипция по URL
const response = await fetch('https://naumxv-whisper-web-api.hf.space/api/transcribe/url', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
url: 'https://example.com/audio.mp3',
language: 'ru',
model: 'base'
})
});
const result = await response.json();
const taskId = result.taskId;
// Ожидание завершения
while (true) {
const statusResponse = await fetch(`https://naumxv-whisper-web-api.hf.space/api/status/${taskId}`);
const statusData = await statusResponse.json();
if (statusData.task.status === 'completed') {
break;
} else if (statusData.task.status === 'error') {
console.error(`Ошибка: ${statusData.task.error}`);
break;
}
console.log(`Прогресс: ${statusData.task.progress}%`);
await new Promise(resolve => setTimeout(resolve, 5000));
}
// Получение результата
const finalResponse = await fetch(`https://naumxv-whisper-web-api.hf.space/api/result/${taskId}`);
const finalResult = await finalResponse.json();
console.log(`Транскрипция: ${finalResult.transcription.text}`);
```
""")
return demo
# Создаем Gradio интерфейс
demo = create_interface()
# Добавляем API endpoints с правильными именами
with demo:
# Health check endpoint
gr.Interface(
fn=lambda: {"status": "ok", "message": "Whisper API сервер работает", "timestamp": time.time()},
inputs=None,
outputs=gr.JSON(),
api_name="health",
title="Health Check"
)
# Models endpoint
gr.Interface(
fn=lambda: {
"models": [
{"id": "tiny", "name": "Tiny", "size": "39 MB", "languages": "multilingual"},
{"id": "base", "name": "Base", "size": "74 MB", "languages": "multilingual"},
{"id": "small", "name": "Small", "size": "244 MB", "languages": "multilingual"},
{"id": "medium", "name": "Medium", "size": "769 MB", "languages": "multilingual"},
{"id": "large", "name": "Large", "size": "1550 MB", "languages": "multilingual"}
],
"default": "base"
},
inputs=None,
outputs=gr.JSON(),
api_name="models",
title="Available Models"
)
# Languages endpoint
gr.Interface(
fn=lambda: {
"languages": [
{"code": "auto", "name": "Автоопределение"},
{"code": "ru", "name": "Русский"},
{"code": "en", "name": "English"},
{"code": "es", "name": "Español"},
{"code": "fr", "name": "Français"},
{"code": "de", "name": "Deutsch"},
{"code": "it", "name": "Italiano"},
{"code": "pt", "name": "Português"},
{"code": "pl", "name": "Polski"},
{"code": "tr", "name": "Türkçe"},
{"code": "ja", "name": "日本語"},
{"code": "ko", "name": "한국어"},
{"code": "zh", "name": "中文"}
],
"default": "auto"
},
inputs=None,
outputs=gr.JSON(),
api_name="languages",
title="Supported Languages"
)
# Transcribe file endpoint
def transcribe_file(audio_file, language: str = "auto", model: str = "base"):
"""Асинхронная транскрипция аудио файла"""
if audio_file is None:
return {"error": "Аудио файл не предоставлен"}
# Создаем задачу
task_id = generate_task_id()
filename = audio_file.name if hasattr(audio_file, 'name') else "unknown"
task = create_task(task_id, filename, language, model)
# Запускаем транскрипцию в отдельном потоке
thread = threading.Thread(
target=simulate_transcription,
args=(task_id, language, model),
daemon=True
)
thread.start()
return {
"success": True,
"taskId": task_id,
"status": "processing",
"message": "Транскрипция началась",
"timestamp": time.time()
}
gr.Interface(
fn=transcribe_file,
inputs=[
gr.Audio(type="filepath", label="Audio File"),
gr.Dropdown(choices=["auto", "ru", "en", "es", "fr", "de", "it", "pt", "pl", "tr", "ja", "ko", "zh"],
value="auto", label="Language"),
gr.Dropdown(choices=["tiny", "base", "small", "medium", "large"],
value="base", label="Model")
],
outputs=gr.JSON(),
api_name="transcribe",
title="Transcribe Audio File"
)
# Transcribe URL endpoint
def transcribe_url(url: str, language: str = "auto", model: str = "base"):
"""Асинхронная транскрипция по URL"""
if not url:
return {"error": "URL не предоставлен"}
# Создаем задачу
task_id = generate_task_id()
task = create_task(task_id, url, language, model)
# Запускаем транскрипцию в отдельном потоке
thread = threading.Thread(
target=simulate_transcription,
args=(task_id, language, model),
daemon=True
)
thread.start()
return {
"success": True,
"taskId": task_id,
"status": "processing",
"message": "Транскрипция по URL началась",
"timestamp": time.time()
}
gr.Interface(
fn=transcribe_url,
inputs=[
gr.Textbox(label="Audio URL"),
gr.Dropdown(choices=["auto", "ru", "en", "es", "fr", "de", "it", "pt", "pl", "tr", "ja", "ko", "zh"],
value="auto", label="Language"),
gr.Dropdown(choices=["tiny", "base", "small", "medium", "large"],
value="base", label="Model")
],
outputs=gr.JSON(),
api_name="transcribe_url",
title="Transcribe Audio URL"
)
# Task status endpoint
def get_task_status(task_id: str):
"""Получает статус задачи"""
task = transcription_tasks.get(task_id)
if not task:
return {"error": "Задача не найдена"}
return {
"success": True,
"task": task,
"timestamp": time.time()
}
gr.Interface(
fn=get_task_status,
inputs=gr.Textbox(label="Task ID"),
outputs=gr.JSON(),
api_name="status",
title="Get Task Status"
)
# Task result endpoint
def get_task_result(task_id: str):
"""Получает результат задачи"""
task = transcription_tasks.get(task_id)
if not task:
return {"error": "Задача не найдена"}
if task["status"] == "processing":
return {
"success": True,
"status": "processing",
"progress": task["progress"],
"message": "Транскрипция еще выполняется"
}
if task["status"] == "error":
return {
"success": False,
"status": "error",
"error": task["error"]
}
return {
"success": True,
"status": "completed",
"transcription": task["result"],
"timestamp": time.time()
}
gr.Interface(
fn=get_task_result,
inputs=gr.Textbox(label="Task ID"),
outputs=gr.JSON(),
api_name="result",
title="Get Task Result"
)
# Запускаем Gradio интерфейс
demo.launch(
server_name="0.0.0.0",
server_port=7860,
share=False,
show_error=True,
show_api=True # Показываем API документацию
)