Spaces:
Sleeping
Sleeping
File size: 3,529 Bytes
1fe2f2f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 | from models.whisper import model
import modules.audio as audio
import os
from modules.r2 import upload_to_s3,get_url
import json
import os
import redis
redis_host = os.getenv('REDIS_HOST', 'localhost')
redis_pass = os.getenv('REDIS_PASS', 'localhost')
redis_port = os.getenv('REDIS_PORT', 6379)
cache = redis.Redis(host=redis_host, port=redis_port, password=redis_pass)
isOnline = os.environ.get("IS_ONLINE")
isOnline = "True" if isOnline == "True" else "False"
def prepare_audio(audio_path, key):
"""Prepara o áudio"""
audio_duration = audio.get_audio_duration(audio_path)
audio_duration = round(audio_duration, 2)
cuted = False
if audio_duration > 15:
audio_path = audio.cut_audio(audio_path, start_time="0", end_time="12", output_path=f"{key}_cut.wav")
cuted = True
if audio_duration < 8:
raise ValueError("audio_too_short")
audio_path = audio.add_silence_to_audio(audio_path, start_silence=0.5, end_silence=0.8, output_path=f"{key}.wav")
if cuted:
os.remove(f"{key}_cut.wav")
return audio_path
def get_audio_transcription(key):
"""Obtém a transcrição do áudio usando o Redis"""
if(isOnline == "True"):
data = cache.get(f"mimic:audio:translation:{key}")
return data
with open(f"{key}.txt", "r") as f:
transcription = f.read()
return transcription
def get_audio_path(key):
"""Obtém o caminho do áudio usando o Redis"""
if(isOnline == "True"):
data = cache.get(f"mimic:audio:url:{key}")
if(data):
return data
new_url = get_url(f"{key}.wav")
cache.set(f"mimic:audio:url:{key}", new_url)
cache.expire(f"mimic:audio:url:{key}", 600000)
return new_url
return f"{key}.wav"
def get_audio(key):
"""Obtém o áudio usando o Redis"""
transcription = get_audio_transcription(key)
audio_path = get_audio_path(key)
return {
"transcription": transcription,
"audio_path": audio_path,
"isOnline": isOnline
}
def process_audio(audio_path, key):
"""Processa o áudio usando o Whisper"""
if(isOnline == "True"):
audio_exists = cache.exists(f"mimic:audio_exists:{key}")
if audio_exists:
return get_audio(key)
audio_path = prepare_audio(audio_path, key)
transcription, info = model.transcribe(audio_path, beam_size=5)
content = ""
for segment in transcription:
content = f"{content} {segment.text}"
url = upload_to_s3(audio_path, f"{key}.wav", "wav")
cache.set(f"mimic:audio:translation:{key}", content)
cache.set(f"mimic:audio:url:{key}", url)
cache.expire(f"mimic:audio:url:{key}", 600000)
cache.set(f"mimic:audio_exists:{key}", "true")
return {"transcription": content, "audio_path": url, "isOnline": isOnline}
audio_exists = os.path.exists(f"{key}.txt")
if audio_exists:
return get_audio(key)
if os.path.exists(f"{key}.wav"):
os.remove(f"{key}.wav")
audio_path = prepare_audio(audio_path, key)
transcription, info = model.transcribe(audio_path, beam_size=5)
content = ""
for segment in transcription:
content = f"{content} {segment.text}"
with open(f"{key}.txt", "w") as f:
f.write(content)
return {"transcription": content, "audio_path": audio_path, "isOnline": isOnline} |