-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathmain.py
54 lines (40 loc) · 1.59 KB
/
main.py
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
from io import BytesIO
import httpx
from rich.console import Console
from starlette.applications import Starlette
from starlette.config import Config
from starlette.datastructures import Secret
from starlette.requests import Request
from starlette.responses import JSONResponse, PlainTextResponse
from starlette.routing import Route
config = Config(".env")
console = Console()
HOST: str = config("HOST", cast=str, default="0.0.0.0")
PORT: int = config("PORT", cast=int, default=8000)
DEBUG: bool = config("DEBUG", cast=bool, default=False)
OPENAI_API_KEY = Secret = config("OPENAI_API_KEY", cast=Secret)
async def transcribe(data: bytes) -> str:
file = ("audio.m4a", BytesIO(data), "audio/m4a")
async with httpx.AsyncClient() as client:
r = await client.post(
"https://api.openai.com/v1/audio/transcriptions",
files={"file": file},
headers={"Authorization": f"Bearer {OPENAI_API_KEY}"},
data={"model": "whisper-1"},
)
transcript = r.json()
return transcript["text"]
async def handle_index(request: Request) -> JSONResponse:
rv = {"hello": "world"}
return JSONResponse(rv)
async def handle_transcribe(request: Request) -> PlainTextResponse:
upload: bytes = await request.body()
console.log(f"Received {len(upload)} bytes")
transcription = await transcribe(upload)
console.log(f'Transcription: "{transcription}"')
return PlainTextResponse(transcription)
routes = [
Route("/", handle_index),
Route("/transcribe", handle_transcribe, methods=["POST"]),
]
app = Starlette(routes=routes, debug=DEBUG)