Add whisper remote backend and CLI
This commit is contained in:
13
cli/README.md
Normal file
13
cli/README.md
Normal file
@@ -0,0 +1,13 @@
|
||||
# whisper-remote
|
||||
|
||||
Local CLI that forwards media files to a remote `whisper-remote-backend` server.
|
||||
|
||||
## Run
|
||||
|
||||
```bash
|
||||
pip install -e .
|
||||
export WHISPER_REMOTE=http://127.0.0.1:8000
|
||||
whisper-remote ./audio.mp3 --model base --language en --output-format txt
|
||||
```
|
||||
|
||||
Use `--to-file` to save the returned transcript locally.
|
||||
27
cli/pyproject.toml
Normal file
27
cli/pyproject.toml
Normal file
@@ -0,0 +1,27 @@
|
||||
[build-system]
|
||||
requires = ["setuptools>=68", "wheel"]
|
||||
build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "whisper-remote-cli"
|
||||
version = "0.1.0"
|
||||
description = "CLI that forwards transcription requests to whisper-remote-backend"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.11"
|
||||
dependencies = [
|
||||
"httpx>=0.28.0,<1.0.0",
|
||||
]
|
||||
|
||||
[project.optional-dependencies]
|
||||
dev = [
|
||||
"pytest>=8.3.0,<9.0.0",
|
||||
]
|
||||
|
||||
[project.scripts]
|
||||
whisper-remote = "whisper_remote_cli.main:main"
|
||||
|
||||
[tool.setuptools]
|
||||
package-dir = {"" = "src"}
|
||||
|
||||
[tool.setuptools.packages.find]
|
||||
where = ["src"]
|
||||
1
cli/src/whisper_remote_cli/__init__.py
Normal file
1
cli/src/whisper_remote_cli/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
"""whisper-remote CLI package."""
|
||||
101
cli/src/whisper_remote_cli/main.py
Normal file
101
cli/src/whisper_remote_cli/main.py
Normal file
@@ -0,0 +1,101 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import httpx
|
||||
|
||||
SUPPORTED_FORMATS = ("txt", "vtt", "srt", "tsv", "json")
|
||||
|
||||
|
||||
def build_parser() -> argparse.ArgumentParser:
|
||||
parser = argparse.ArgumentParser(description="Send transcription jobs to a remote Whisper backend.")
|
||||
parser.add_argument("file", type=Path, help="Path to the local media file to upload.")
|
||||
parser.add_argument("--model", required=True, help="Whisper model name to use on the backend.")
|
||||
parser.add_argument("--language", help="Optional language code to pass through to Whisper.")
|
||||
parser.add_argument(
|
||||
"--output-format",
|
||||
default="txt",
|
||||
choices=SUPPORTED_FORMATS,
|
||||
help="Transcript artifact format returned by the backend.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--server",
|
||||
help="Override the backend base URL. Defaults to the WHISPER_REMOTE environment variable.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--to-file",
|
||||
type=Path,
|
||||
help="Optional local file path or directory to save the returned transcript artifact.",
|
||||
)
|
||||
return parser
|
||||
|
||||
|
||||
def resolve_server(args: argparse.Namespace) -> str:
|
||||
server = args.server or os.environ.get("WHISPER_REMOTE")
|
||||
if not server:
|
||||
raise SystemExit("WHISPER_REMOTE is not set and --server was not provided.")
|
||||
return server.rstrip("/")
|
||||
|
||||
|
||||
def infer_output_path(target: Path, input_file: Path, output_format: str) -> Path:
|
||||
if target.exists() and target.is_dir():
|
||||
return target / f"{input_file.stem}.{output_format}"
|
||||
if target.suffix:
|
||||
return target
|
||||
return target / f"{input_file.stem}.{output_format}"
|
||||
|
||||
|
||||
def print_response(response: httpx.Response) -> None:
|
||||
sys.stdout.write(response.text)
|
||||
if response.text and not response.text.endswith("\n"):
|
||||
sys.stdout.write("\n")
|
||||
|
||||
|
||||
def save_response(response: httpx.Response, destination: Path) -> None:
|
||||
destination.parent.mkdir(parents=True, exist_ok=True)
|
||||
destination.write_bytes(response.content)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = build_parser()
|
||||
args = parser.parse_args()
|
||||
|
||||
input_file = args.file.expanduser().resolve()
|
||||
if not input_file.is_file():
|
||||
parser.error(f"Input file does not exist: {input_file}")
|
||||
|
||||
server = resolve_server(args)
|
||||
endpoint = f"{server}/transcriptions"
|
||||
|
||||
with input_file.open("rb") as handle, httpx.Client(timeout=300.0) as client:
|
||||
response = client.post(
|
||||
endpoint,
|
||||
data={
|
||||
"model": args.model,
|
||||
"language": args.language or "",
|
||||
"output_format": args.output_format,
|
||||
},
|
||||
files={"file": (input_file.name, handle, "application/octet-stream")},
|
||||
)
|
||||
|
||||
try:
|
||||
response.raise_for_status()
|
||||
except httpx.HTTPStatusError as exc:
|
||||
message = exc.response.text.strip() or str(exc)
|
||||
parser.exit(1, f"{message}\n")
|
||||
|
||||
if args.to_file:
|
||||
destination = infer_output_path(args.to_file.expanduser(), input_file, args.output_format)
|
||||
save_response(response, destination)
|
||||
sys.stdout.write(f"{destination}\n")
|
||||
else:
|
||||
print_response(response)
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
11
cli/tests/conftest.py
Normal file
11
cli/tests/conftest.py
Normal file
@@ -0,0 +1,11 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
SRC = ROOT / "src"
|
||||
|
||||
if str(SRC) not in sys.path:
|
||||
sys.path.insert(0, str(SRC))
|
||||
28
cli/tests/test_main.py
Normal file
28
cli/tests/test_main.py
Normal file
@@ -0,0 +1,28 @@
|
||||
import os
|
||||
from argparse import Namespace
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from whisper_remote_cli import main
|
||||
|
||||
|
||||
def test_resolve_server_from_env(monkeypatch) -> None:
|
||||
monkeypatch.setenv("WHISPER_REMOTE", "http://localhost:8000/")
|
||||
assert main.resolve_server(Namespace(server=None)) == "http://localhost:8000"
|
||||
|
||||
|
||||
def test_resolve_server_requires_value(monkeypatch) -> None:
|
||||
monkeypatch.delenv("WHISPER_REMOTE", raising=False)
|
||||
with pytest.raises(SystemExit):
|
||||
main.resolve_server(Namespace(server=None))
|
||||
|
||||
|
||||
def test_infer_output_path_for_directory(tmp_path: Path) -> None:
|
||||
destination = main.infer_output_path(tmp_path, Path("clip.wav"), "srt")
|
||||
assert destination == tmp_path / "clip.srt"
|
||||
|
||||
|
||||
def test_infer_output_path_for_explicit_file(tmp_path: Path) -> None:
|
||||
destination = main.infer_output_path(tmp_path / "custom-name.txt", Path("clip.wav"), "txt")
|
||||
assert destination == tmp_path / "custom-name.txt"
|
||||
Reference in New Issue
Block a user