File size: 1,981 Bytes
bbfb0cc | 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 | import requests
import os
import base64
from typing import Optional
from . import APIProvider, register
MIME_MAP = {".wav": "audio/wav", ".mp3": "audio/mpeg", ".m4a": "audio/mp4", ".flac": "audio/flac"}
@register("zoom")
class ZoomProvider(APIProvider):
ENDPOINT = "https://api.zoom.us/v2/aiservices/scribe/transcribe"
def transcribe(
self,
model_variant: str,
audio_file_path: Optional[str],
sample: dict,
use_url: bool = False,
language: str = "en",
) -> str:
api_key = os.getenv("ZOOM_API_KEY")
if not api_key or api_key == "your_api_key":
raise ValueError("ZOOM_API_KEY environment variable not set")
if use_url:
file_payload = sample["row"]["audio"][0]["src"]
audio_duration = sample["row"]["audio_length_s"]
else:
audio_duration = (
len(sample["audio"]["array"]) / sample["audio"]["sampling_rate"]
)
with open(audio_file_path, "rb") as f:
audio_bytes = f.read()
mime = MIME_MAP.get(os.path.splitext(audio_file_path)[1].lower(), "audio/wav")
file_payload = f"data:{mime};base64,{base64.b64encode(audio_bytes).decode('ascii')}"
if audio_duration <= 29.9:
segmentation_mode = "none"
else:
segmentation_mode = "auto"
resp = requests.post(
self.ENDPOINT,
headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
json={"file": file_payload,
"config": {
"language": "en-US",
"segmentation_mode": segmentation_mode,
"experimental_feature": {"model_pro": True},
"timestamps": True,
}},
timeout=300,
)
resp.raise_for_status()
return resp.json().get("result", {}).get("text_display", "") or "."
|