79 lines
2.5 KiB
Python
79 lines
2.5 KiB
Python
# -*- coding: utf-8 -*-
|
|
import os
|
|
import tempfile
|
|
from flask import Flask, request, jsonify
|
|
from faster_whisper import WhisperModel
|
|
|
|
# ==========================================================
|
|
# Configuration
|
|
# ==========================================================
|
|
MODEL_PATH = os.environ.get("WHISPER_MODEL", "/models/whisper")
|
|
DEVICE = os.environ.get("WHISPER_DEVICE", "cuda")
|
|
COMPUTE_TYPE = "float16" if DEVICE == "cuda" else "int8"
|
|
PORT = int(os.environ.get("PORT", 5001))
|
|
|
|
app = Flask(__name__)
|
|
|
|
# ==========================================================
|
|
# Load Whisper Model
|
|
# ==========================================================
|
|
print(f"[stt] Loading model from {MODEL_PATH} on {DEVICE} ({COMPUTE_TYPE})")
|
|
try:
|
|
model = WhisperModel(MODEL_PATH, device=DEVICE, compute_type=COMPUTE_TYPE)
|
|
print("[stt] Whisper model loaded successfully")
|
|
except Exception as e:
|
|
print(f"[stt] Fatal error loading model: {e}")
|
|
exit(1)
|
|
|
|
# ==========================================================
|
|
# Speech To Text Endpoint
|
|
# ==========================================================
|
|
@app.route("/transcribe", methods=["POST"])
|
|
def transcribe():
|
|
if "audio" not in request.files:
|
|
return jsonify({"error": "audio file missing"}), 400
|
|
|
|
audio_file = request.files["audio"]
|
|
|
|
# استفاده از NamedTemporaryFile با مدیریت بهتر برای محیط داکر
|
|
tmp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".wav")
|
|
audio_path = tmp_file.name
|
|
tmp_file.close()
|
|
|
|
try:
|
|
audio_file.save(audio_path)
|
|
|
|
# اجرای اینفرنس با تنظیمات بهینه شده
|
|
segments, info = model.transcribe(
|
|
audio_path,
|
|
language="fa",
|
|
beam_size=10,
|
|
vad_filter=True,
|
|
vad_parameters=dict(min_silence_duration_ms=500) # بهبود تشخیص سکوت
|
|
)
|
|
|
|
text = " ".join([segment.text.strip() for segment in segments])
|
|
|
|
return jsonify({
|
|
"text": text.strip(),
|
|
"language": info.language
|
|
})
|
|
|
|
except Exception as e:
|
|
print(f"[stt] Error during transcription: {e}")
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
finally:
|
|
if os.path.exists(audio_path):
|
|
os.remove(audio_path)
|
|
|
|
# ==========================================================
|
|
# Health Check
|
|
# ==========================================================
|
|
@app.route("/health", methods=["GET"])
|
|
def health():
|
|
return jsonify({"status": "ok"})
|
|
|
|
if __name__ == "__main__":
|
|
app.run(host="0.0.0.0", port=PORT)
|