Spaces:
Running
Running
import os | |
import tempfile | |
import wave | |
import numpy as np | |
import soundfile as sf | |
from flask import Flask, request, jsonify, send_file, send_from_directory | |
from flask_cors import CORS | |
from werkzeug.utils import secure_filename | |
from gtts import gTTS, lang | |
from kokoro import KPipeline | |
import google.generativeai as genai | |
from google.generativeai.types import GenerateContentConfig, SpeechConfig, VoiceConfig, PrebuiltVoiceConfig | |
# Load API key | |
GEMINI_API_KEY = os.getenv("GEMINI_API_KEY") | |
if not GEMINI_API_KEY: | |
raise ValueError("GEMINI_API_KEY environment variable not set") | |
genai.configure(api_key=GEMINI_API_KEY) | |
# Flask app setup | |
app = Flask(__name__, static_folder='static') | |
CORS(app) | |
# Supported languages | |
KOKORO_LANGUAGES = { | |
"American English": "a", "British English": "b", "Mandarin Chinese": "z", | |
"Spanish": "e", "French": "f", "Hindi": "h", "Italian": "i", "Brazilian Portuguese": "p" | |
} | |
GTTS_LANGUAGES = lang.tts_langs() | |
GTTS_LANGUAGES['ja'] = 'Japanese' | |
SUPPORTED_LANGUAGES = sorted(set(KOKORO_LANGUAGES.keys()) | set(GTTS_LANGUAGES.values())) | |
# Voice name for Gemini TTS | |
DEFAULT_GEMINI_VOICE = "Kore" | |
def wave_file(filename, pcm, channels=1, rate=24000, sample_width=2): | |
with wave.open(filename, "wb") as wf: | |
wf.setnchannels(channels) | |
wf.setsampwidth(sample_width) | |
wf.setframerate(rate) | |
wf.writeframes(pcm) | |
def serve_index(): | |
return send_from_directory(app.static_folder, 'index.html') | |
def get_languages(): | |
return jsonify(SUPPORTED_LANGUAGES) | |
def translate_audio(): | |
try: | |
if 'audio' not in request.files: | |
return jsonify({'error': 'No audio file uploaded'}), 400 | |
audio_file = request.files['audio'] | |
target_language = request.form.get('language', 'English') | |
if not audio_file or audio_file.filename == '': | |
return jsonify({'error': 'Invalid audio file'}), 400 | |
allowed_mime_types = ['audio/wav', 'audio/mpeg', 'audio/mp4', 'audio/webm'] | |
if audio_file.mimetype not in allowed_mime_types: | |
return jsonify({'error': f'Unsupported file type: {audio_file.mimetype}'}), 400 | |
# Transcribe audio with Gemini | |
model = genai.GenerativeModel("models/gemini-1.5-flash") | |
audio_blob = { | |
'mime_type': audio_file.mimetype, | |
'data': audio_file.read() | |
} | |
convo = model.start_chat() | |
convo.send_message("You are a professional transcriber. Transcribe this audio accurately and verbatim.") | |
response = convo.send_message(audio_blob) | |
transcription = response.text.strip() | |
# Translate | |
prompt = f"Translate the following text to {target_language}:\n\n{transcription}" | |
translation_response = model.generate_content(prompt) | |
translated_text = translation_response.text.strip() | |
# Try Gemini 2.5 TTS | |
try: | |
tts_response = genai.generate_content( | |
model="models/gemini-2.5-flash-preview-tts", | |
contents=translated_text, | |
generation_config=GenerateContentConfig( | |
response_mime_type="audio/wav" | |
), | |
speech_config=SpeechConfig( | |
voice_config=VoiceConfig( | |
prebuilt_voice=PrebuiltVoiceConfig(voice_name=DEFAULT_GEMINI_VOICE) | |
) | |
) | |
) | |
data = tts_response.candidates[0].content.parts[0].inline_data.data | |
temp_output_path = os.path.join(tempfile.gettempdir(), "tts_gemini.wav") | |
wave_file(temp_output_path, data) | |
except Exception as gemini_tts_error: | |
app.logger.warning(f"Gemini TTS failed: {gemini_tts_error}") | |
# Fallback to Kokoro or gTTS | |
if target_language in KOKORO_LANGUAGES: | |
lang_code = KOKORO_LANGUAGES[target_language] | |
pipeline = KPipeline(lang_code=lang_code) | |
generator = pipeline(translated_text, voice="af_heart", speed=1) | |
audio_segments = [audio for _, _, audio in generator if audio is not None] | |
if audio_segments: | |
audio_data = np.concatenate(audio_segments) | |
temp_output_path = os.path.join(tempfile.gettempdir(), "tts_kokoro.wav") | |
sf.write(temp_output_path, audio_data, 24000) | |
else: | |
raise ValueError("No audio generated by Kokoro.") | |
else: | |
lang_code = next((k for k, v in GTTS_LANGUAGES.items() if v == target_language), 'en') | |
tts = gTTS(translated_text, lang=lang_code) | |
temp_output_path = os.path.join(tempfile.gettempdir(), "tts_gtts.mp3") | |
tts.save(temp_output_path) | |
return jsonify({ | |
'transcription': transcription, | |
'translation': translated_text, | |
'audio_url': f'/download/{os.path.basename(temp_output_path)}' | |
}) | |
except Exception as e: | |
app.logger.error(f"Processing error: {str(e)}") | |
return jsonify({'error': str(e)}), 500 | |
def download_file(filename): | |
try: | |
return send_file( | |
os.path.join(tempfile.gettempdir(), filename), | |
mimetype="audio/mpeg", | |
as_attachment=True, | |
download_name=f"translated_{filename}" | |
) | |
except FileNotFoundError: | |
return jsonify({'error': 'File not found'}), 404 | |
if __name__ == '__main__': | |
app.run(host="0.0.0.0", port=7860) |