atlas-iac/services/hermes/scripts/hermes_stt_client.py

77 lines
2.4 KiB
Python

#!/usr/bin/env python3
"""Send one Hermes local-command STT request to the private Whisper service."""
from __future__ import annotations
import argparse
import json
import mimetypes
import os
import secrets
from pathlib import Path
from urllib.request import Request, urlopen
def _multipart(audio: Path, language: str, model: str) -> tuple[bytes, str]:
boundary = f"atlas-hermes-{secrets.token_hex(12)}"
mime = mimetypes.guess_type(audio.name)[0] or "application/octet-stream"
chunks: list[bytes] = []
def field(name: str, value: str) -> None:
chunks.extend(
[
f"--{boundary}\r\n".encode(),
f'Content-Disposition: form-data; name="{name}"\r\n\r\n'.encode(),
value.encode(),
b"\r\n",
]
)
field("language", language)
field("model", model)
chunks.extend(
[
f"--{boundary}\r\n".encode(),
f'Content-Disposition: form-data; name="file"; filename="{audio.name}"\r\n'.encode(),
f"Content-Type: {mime}\r\n\r\n".encode(),
audio.read_bytes(),
b"\r\n",
f"--{boundary}--\r\n".encode(),
]
)
return b"".join(chunks), boundary
def main() -> None:
"""Transcribe one file and emit the .txt contract Hermes expects."""
parser = argparse.ArgumentParser()
parser.add_argument("input_path", type=Path)
parser.add_argument("--output-dir", required=True, type=Path)
parser.add_argument("--language", default="auto")
parser.add_argument("--model", default="small")
args = parser.parse_args()
body, boundary = _multipart(args.input_path, args.language, args.model)
request = Request(
os.getenv(
"HERMES_STT_URL",
"http://hermes-stt.hermes.svc.cluster.local:9000/v1/audio/transcriptions",
),
data=body,
headers={
"Content-Type": f"multipart/form-data; boundary={boundary}",
"Accept": "application/json",
},
method="POST",
)
with urlopen(request, timeout=120) as response:
result = json.loads(response.read().decode("utf-8"))
transcript = str(result.get("text") or "").strip()
args.output_dir.mkdir(parents=True, exist_ok=True)
output = args.output_dir / f"{args.input_path.stem}.txt"
output.write_text(transcript, encoding="utf-8")
if __name__ == "__main__":
main()