MahdiandCursor
7757c0837d
Ship assistant answers, contact moods, voice dictation, and the related CRM UI so these can be deployed to production. Co-authored-by: Cursor <cursoragent@cursor.com>
146 lines
4.2 KiB
Python
146 lines
4.2 KiB
Python
"""Speech-to-text sidecar. Uses Pipecat OpenAISTTService only — no TTS, no voice agent."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
from dotenv import load_dotenv
|
|
from fastapi import FastAPI, File, Header, HTTPException, UploadFile
|
|
from fastapi.responses import JSONResponse
|
|
from pipecat.services.openai.stt import OpenAISTTService
|
|
from pipecat.transcriptions.language import Language
|
|
|
|
load_dotenv(Path(__file__).resolve().parent / ".env")
|
|
|
|
MAX_BYTES = 8 * 1024 * 1024
|
|
WAV_MAGIC = (b"RIFF", b"WAVE")
|
|
|
|
app = FastAPI(title="Mento STT", docs_url=None, redoc_url=None)
|
|
_stt: OpenAISTTService | None = None
|
|
|
|
|
|
def _env(name: str, default: str = "") -> str:
|
|
return (os.getenv(name) or default).strip()
|
|
|
|
|
|
def _require_token(header_token: str | None) -> None:
|
|
expected = _env("STT_INTERNAL_TOKEN")
|
|
if not expected:
|
|
return
|
|
if (header_token or "").strip() != expected:
|
|
raise HTTPException(status_code=401, detail="توکن سرویس تبدیل صوت نامعتبر است")
|
|
|
|
|
|
def _stt_service() -> OpenAISTTService:
|
|
global _stt
|
|
if _stt is not None:
|
|
return _stt
|
|
|
|
api_key = _env("OPENAI_API_KEY")
|
|
if not api_key:
|
|
raise HTTPException(status_code=503, detail="کلید تبدیل صوت تنظیم نشده است")
|
|
|
|
model = _env("STT_MODEL", "whisper-1")
|
|
prompt = _env(
|
|
"STT_PROMPT",
|
|
"یادداشت و مکالمهٔ فروش به زبان فارسی. اسامی مشتریان و اصطلاحات کسب‌وکار را دقیق بنویس.",
|
|
)
|
|
base_url = _env("OPENAI_BASE_URL") or None
|
|
|
|
_stt = OpenAISTTService(
|
|
api_key=api_key,
|
|
base_url=base_url,
|
|
settings=OpenAISTTService.Settings(
|
|
model=model,
|
|
language=Language.FA,
|
|
prompt=prompt,
|
|
),
|
|
)
|
|
return _stt
|
|
|
|
|
|
def _is_wav(data: bytes) -> bool:
|
|
return len(data) >= 12 and data[:4] == WAV_MAGIC[0] and data[8:12] == WAV_MAGIC[1]
|
|
|
|
|
|
def _to_wav(data: bytes, filename: str) -> bytes:
|
|
if _is_wav(data):
|
|
return data
|
|
|
|
ffmpeg = shutil.which("ffmpeg")
|
|
if not ffmpeg:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="فرمت صوت پشتیبانی نشد. فایل WAV ارسال کنید یا ffmpeg را نصب کنید.",
|
|
)
|
|
|
|
suffix = Path(filename or "audio.webm").suffix or ".webm"
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
src = Path(tmp) / f"in{suffix}"
|
|
dst = Path(tmp) / "out.wav"
|
|
src.write_bytes(data)
|
|
result = subprocess.run(
|
|
[
|
|
ffmpeg,
|
|
"-y",
|
|
"-i",
|
|
str(src),
|
|
"-ac",
|
|
"1",
|
|
"-ar",
|
|
"16000",
|
|
"-f",
|
|
"wav",
|
|
str(dst),
|
|
],
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
if result.returncode != 0 or not dst.exists():
|
|
raise HTTPException(status_code=400, detail="تبدیل فایل صوتی ناموفق بود")
|
|
return dst.read_bytes()
|
|
|
|
|
|
@app.get("/health")
|
|
async def health() -> dict[str, object]:
|
|
return {
|
|
"ok": True,
|
|
"engine": "pipecat",
|
|
"service": "OpenAISTTService",
|
|
"configured": bool(_env("OPENAI_API_KEY")),
|
|
}
|
|
|
|
|
|
@app.post("/transcribe")
|
|
async def transcribe(
|
|
audio: UploadFile = File(...),
|
|
x_stt_token: str | None = Header(default=None, alias="X-STT-Token"),
|
|
) -> JSONResponse:
|
|
_require_token(x_stt_token)
|
|
|
|
data = await audio.read()
|
|
if not data:
|
|
raise HTTPException(status_code=400, detail="فایل صوتی خالی است")
|
|
if len(data) > MAX_BYTES:
|
|
raise HTTPException(status_code=400, detail="حجم فایل صوتی بیش از حد مجاز است")
|
|
|
|
wav = _to_wav(data, audio.filename or "audio.wav")
|
|
service = _stt_service()
|
|
|
|
try:
|
|
result = await service._transcribe(wav)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as exc:
|
|
raise HTTPException(
|
|
status_code=502,
|
|
detail=f"تبدیل صوت به متن ناموفق بود: {exc}",
|
|
) from exc
|
|
|
|
text = (getattr(result, "text", None) or "").strip()
|
|
return JSONResponse({"text": text})
|