|
|
|
import json |
|
import os |
|
from typing import Optional, Dict, Any |
|
|
|
import aiohttp |
|
|
|
|
|
async def _send_bytes_as_form_data( |
|
file_bytes: bytes, |
|
endpoint_url: str, |
|
field_name: str = "file", |
|
auth_token: Optional[str] = None, |
|
content_type: str = "application/octet-stream" |
|
) -> Dict[str, Any]: |
|
""" |
|
Send bytes as multipart form data POST request to an endpoint. |
|
|
|
Args: |
|
file_bytes: Bytes content to send |
|
endpoint_url: URL endpoint to send the POST request to |
|
field_name: Form field name for the file (default: "file") |
|
auth_token: Optional bearer token for authorization |
|
content_type: MIME type of the content (default: "application/octet-stream") |
|
|
|
Returns: |
|
Dictionary containing response status and data |
|
""" |
|
|
|
data = aiohttp.FormData() |
|
data.add_field( |
|
field_name, |
|
file_bytes, |
|
content_type=content_type |
|
) |
|
|
|
|
|
headers = {} |
|
if auth_token: |
|
headers['Authorization'] = f'Bearer {auth_token}' |
|
|
|
|
|
async with aiohttp.ClientSession() as session: |
|
async with session.post( |
|
endpoint_url, |
|
data=data, |
|
headers=headers if headers else None |
|
) as response: |
|
response_text = await response.text() |
|
|
|
return { |
|
'status': response.status, |
|
'success': response.status < 400, |
|
'response': response_text, |
|
'headers': dict(response.headers) |
|
} |
|
|
|
|
|
async def transcribe(_bytes: bytes) -> dict: |
|
""" |
|
Transcribe audio bytes using Modal endpoint |
|
|
|
Args: |
|
_bytes: Audio file bytes |
|
|
|
Returns: |
|
Dictionary containing transcription results |
|
""" |
|
auth_token = os.environ['MODAL_AUTH_TOKEN'] |
|
response = await _send_bytes_as_form_data( |
|
file_bytes=_bytes, |
|
endpoint_url='https://yigitsekerci6174--transcribe-audio.modal.run', |
|
auth_token=auth_token, |
|
field_name='file' |
|
) |
|
|
|
return json.loads(response['response']) |
|
|