File size: 2,627 Bytes
bbfb0cc 51419fe | 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 | import requests
import os
import io
import json
from typing import Optional
from . import APIProvider, register
MIME_MAP = {".wav": "audio/wav", ".mp3": "audio/mpeg", ".m4a": "audio/mp4", ".flac": "audio/flac"}
@register("microsoft")
class MicrosoftAzureProvider(APIProvider):
ENDPOINT = "https://northeurope.api.cognitive.microsoft.com/speechtotext/transcriptions:transcribe?api-version=2025-10-15"
# support 26 languages, list of locales in ML benchmark
# It is Multi-lingual model, can use without specifying the language.
LOCALE_DICT = {
"en": "en-US",
"es": "es-ES",
"fr": "fr-FR",
"de": "de-DE",
"it": "it-IT",
"pt": "pt-PT",
}
def transcribe(
self,
model_variant: str,
audio_file_path: Optional[str],
sample: dict,
use_url: bool = False,
language: str = "en",
prompt: Optional[str] = None,
) -> str:
api_key = os.getenv("AZURE_API_KEY")
if not api_key or api_key == "your_api_key":
raise ValueError("AZURE_API_KEY environment variable not set")
locale = self.LOCALE_DICT.get(language, "")
definition = {
"locales": [locale],
"profanityFilterMode": "None",
"enhancedMode": {
"enabled": True,
"task": "transcribe",
},
}
if prompt is not None:
# E.g., prompt = "Output must be in lexical format."
definition["enhancedMode"]["prompt"] = [prompt]
if use_url:
file_url = sample["row"]["audio"][0]["src"]
audio_resp = requests.get(file_url, timeout=120)
audio_resp.raise_for_status()
audio_data = io.BytesIO(audio_resp.content)
files = [
("definition", (None, json.dumps(definition))),
("audio", ("audio.wav", audio_data, "audio/wav")),
]
else:
mime = MIME_MAP.get(os.path.splitext(audio_file_path)[1].lower(), "audio/wav")
files = [
("definition", (None, json.dumps(definition))),
("audio", (audio_file_path, open(audio_file_path, "rb"), mime)),
]
resp = requests.post(
self.ENDPOINT,
headers={"Ocp-Apim-Subscription-Key": api_key},
files=files,
timeout=300,
)
if not resp.ok:
print(f"Azure API error {resp.status_code}: {resp.text}")
resp.raise_for_status()
return resp.json().get("combinedPhrases", [{}])[0].get("text", "") or "."
|