用 Groq Whisper 转写前校验音频文件

Groq 的语音转写走 OpenAI 兼容的 /openai/v1/audio/transcriptions。官方限制文件 ≤100MB，并限定扩展名。 这篇在发请求前先做本地校验，再用 multipart 上传；mock 返回假转写文本。

先看清成品

• hello.wav 经 mock 返回「转写自 hello.wav」。
• .txt 被本地拒绝（不支持的格式）。
• 2 项 pytest 通过。未向 Groq 上传音频。

合规提醒：使用自己的 GroqCloud 账号与 API Key；语音内容需你有权处理。模型与价格以 groq.com/pricing 与控制台为准，本文不写价格。遵守各模型许可。

运行说明：没有可用的 Groq 密钥。真实运行过：流式聊天 / Whisper / tool-calling 客户端对本地 mock 的请求，以及对 https://api.groq.com 使用无效密钥得到 HTTP 401（2026-10-23，UTC+8）。未上传真实用户音频，未产生计费。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

94-groqcloud-B-src/
├── transcribe.py
├── mock_whisper.py
└── test_94b.py

第 1 步：客户端

#!/usr/bin/env python3
"""Groq Whisper transcription client with local file checks (size/format)."""
from __future__ import annotations

import argparse
import json
import mimetypes
import os
import sys
import uuid
import urllib.error
import urllib.request
from pathlib import Path

MAX_BYTES = 100 * 1024 * 1024
ALLOWED = {".flac", ".mp3", ".m4a", ".mpeg", ".mpga", ".ogg", ".wav", ".webm"}


def validate(path: Path) -> None:
    if not path.is_file():
        raise SystemExit(f"文件不存在: {path}")
    if path.suffix.lower() not in ALLOWED:
        raise SystemExit(f"不支持的格式 {path.suffix}（官方允许: {', '.join(sorted(ALLOWED))}）")
    size = path.stat().st_size
    if size > MAX_BYTES:
        raise SystemExit(f"文件 {size} 字节超过官方 100 MB 上限")
    if size == 0:
        raise SystemExit("空文件")


def transcribe(*, base_url: str, api_key: str, model: str, path: Path, language: str | None = None) -> dict:
    validate(path)
    boundary = f"----binbho{uuid.uuid4().hex}"
    mime = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
    chunks: list[bytes] = []

    def field(name: str, value: str):
        chunks.append(f"--{boundary}\r\nContent-Disposition: form-data; name=\"{name}\"\r\n\r\n{value}\r\n".encode())

    field("model", model)
    if language:
        field("language", language)
    data = path.read_bytes()
    chunks.append(
        f"--{boundary}\r\nContent-Disposition: form-data; name=\"file\"; filename=\"{path.name}\"\r\n"
        f"Content-Type: {mime}\r\n\r\n".encode()
        + data
        + b"\r\n"
    )
    chunks.append(f"--{boundary}--\r\n".encode())
    body = b"".join(chunks)
    url = f"{base_url.rstrip('/')}/openai/v1/audio/transcriptions"
    req = urllib.request.Request(
        url,
        data=body,
        headers={
            "Authorization": f"Bearer {api_key}",
            "Content-Type": f"multipart/form-data; boundary={boundary}",
        },
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=120) as resp:
            return json.loads(resp.read().decode())
    except urllib.error.HTTPError as e:
        raise SystemExit(f"HTTP {e.code}: {e.read().decode(errors='replace')}") from e


def main(argv=None) -> int:
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument("file", type=Path)
    p.add_argument("--base", default=os.environ.get("GROQ_BASE_URL", "https://api.groq.com"))
    p.add_argument("--key", default=os.environ.get("GROQ_API_KEY", ""))
    p.add_argument("--model", default="whisper-large-v3")
    p.add_argument("--language", default=None)
    args = p.parse_args(argv)
    if not args.key:
        print("设置 GROQ_API_KEY", file=sys.stderr)
        return 2
    r = transcribe(base_url=args.base, api_key=args.key, model=args.model, path=args.file, language=args.language)
    print(r.get("text", r))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

第 2 步：Mock

#!/usr/bin/env python3
from __future__ import annotations
import json
from http.server import BaseHTTPRequestHandler, HTTPServer

API_KEY = "gsk-test"
MAX_BYTES = 100 * 1024 * 1024  # docs: 100 MB
ALLOWED = {".flac", ".mp3", ".m4a", ".mpeg", ".mpga", ".ogg", ".wav", ".webm"}


class H(BaseHTTPRequestHandler):
    def log_message(self, *a):
        pass

    def do_POST(self):
        if self.path != "/openai/v1/audio/transcriptions":
            self.send_error(404); return
        if self.headers.get("Authorization") != f"Bearer {API_KEY}":
            self.send_response(401); self.end_headers(); return
        ctype = self.headers.get("Content-Type", "")
        n = int(self.headers.get("Content-Length", 0))
        raw = self.rfile.read(n)
        if n > MAX_BYTES:
            self.send_response(413); self.end_headers(); return
        # very small multipart parse: look for filename=
        if "filename=" in raw.decode("latin1", errors="ignore"):
            import re
            m = re.search(r'filename="([^"]+)"', raw.decode("latin1", errors="ignore"))
            name = m.group(1) if m else "x.wav"
        else:
            name = "x.wav"
        from pathlib import Path
        ext = Path(name).suffix.lower()
        if ext not in ALLOWED:
            self.send_response(400)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            self.wfile.write(json.dumps({"error": {"message": f"unsupported format {ext}"}}).encode())
            return
        resp = {"text": f"转写自 {name}", "model": "whisper-large-v3"}
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(json.dumps(resp, ensure_ascii=False).encode())


if __name__ == "__main__":
    import os
    HTTPServer(("127.0.0.1", int(os.environ.get("PORT", "3096"))), H).serve_forever()

第 3 步：测试

import threading
from http.server import HTTPServer
from pathlib import Path
import mock_whisper
from transcribe import validate, transcribe

def serve():
    httpd = HTTPServer(("127.0.0.1", 0), mock_whisper.H)
    threading.Thread(target=httpd.serve_forever, daemon=True).start()
    return httpd, httpd.server_address[1]

def test_validate_bad_ext(tmp_path: Path):
    f = tmp_path / "a.txt"
    f.write_text("x")
    try:
        validate(f)
        assert False
    except SystemExit as e:
        assert "不支持" in str(e)

def test_ok(tmp_path: Path):
    f = tmp_path / "hello.wav"
    f.write_bytes(b"RIFF" + b"\0" * 100)
    httpd, port = serve()
    try:
        r = transcribe(base_url=f"http://127.0.0.1:{port}", api_key=mock_whisper.API_KEY,
                       model="whisper-large-v3", path=f)
        assert "hello.wav" in r["text"]
    finally:
        httpd.shutdown()

第 4 步：运行

# 2026-10-23 12:37 UTC+8
{
  "text": "转写自 hello.wav",
  "model": "whisper-large-v3"
}
validate: 文件不存在: /tmp/a.txt

$ python -c "..."  # validate .txt
不支持的格式 .txt（官方允许: .flac, .m4a, .mp3, .mpeg, .mpga, .ogg, .wav, .webm）

$ python -m pytest -q
2 passed

怎么检查结果？

• 超大/错误扩展名在本地失败，不产生网络请求。
• 有 Key 时设 GROQ_API_KEY 与真实音频路径。

调试示例

空文件与假 wav。 测试用 RIFF + padding 只为通过扩展名与非空检查；真实转写需要合法音频。上线前用短录音验收模型与语言参数。

常见问题

模板规则

• 先本地校验再上传。
• 只处理有权处理的音频。
• 密钥放环境变量。

项目资源

文件门禁能挡住大部分无效请求和意外超大上传。