| import base64 |
| import io |
| import json |
| import os |
| import shutil |
| import subprocess |
| import sys |
| import tempfile |
| import threading |
| import time |
| from pathlib import Path |
| from typing import Optional |
|
|
| import gradio as gr |
| import soundfile as sf |
| import spaces |
| from huggingface_hub import snapshot_download |
|
|
| ASR_REPO = os.getenv("ASR_REPO", "Daumee/Qwen3-ASR-0.6B-ONNX-CPU") |
| ASR_THREADS = int(os.getenv("ASR_THREADS", "2")) |
| VOICE_API_TOKEN = os.getenv("VOICE_API_TOKEN", "").strip() |
|
|
| started_at = time.time() |
| asr_lock = threading.Lock() |
| asr_pipeline = None |
| tts_lock = threading.Lock() |
| tts_pipelines = {} |
|
|
|
|
| def check_key(api_key: str) -> None: |
| if VOICE_API_TOKEN and api_key != VOICE_API_TOKEN: |
| raise gr.Error("invalid api_key") |
|
|
|
|
| def normalize_language(language: Optional[str]) -> Optional[str]: |
| if language is None: |
| return None |
| value = language.strip() |
| if not value or value.lower() in {"auto", "none", "null"}: |
| return None |
| aliases = { |
| "zh": "Chinese", |
| "cn": "Chinese", |
| "zh-cn": "Chinese", |
| "chinese": "Chinese", |
| "en": "English", |
| "en-us": "English", |
| "english": "English", |
| "yue": "Cantonese", |
| "cantonese": "Cantonese", |
| } |
| return aliases.get(value.lower(), value) |
|
|
|
|
| def tts_lang_from_voice(voice: str) -> str: |
| if voice.startswith("zf_") or voice.startswith("zm_"): |
| return "z" |
| if voice.startswith("jf_") or voice.startswith("jm_"): |
| return "j" |
| if voice.startswith("bf_") or voice.startswith("bm_"): |
| return "b" |
| return "a" |
|
|
|
|
| def decode_audio_b64(audio_base64: str, filename: str) -> Path: |
| if "," in audio_base64[:128]: |
| audio_base64 = audio_base64.split(",", 1)[1] |
| suffix = Path(filename or "audio.wav").suffix or ".wav" |
| tmp = tempfile.NamedTemporaryFile(delete=False, suffix=suffix) |
| tmp_path = Path(tmp.name) |
| try: |
| tmp.write(base64.b64decode(audio_base64)) |
| tmp.close() |
| return tmp_path |
| except Exception: |
| tmp.close() |
| tmp_path.unlink(missing_ok=True) |
| raise gr.Error("invalid base64 audio") |
|
|
|
|
| def audio_bytes_to_b64(audio, sample_rate: int = 24000, audio_format: str = "wav") -> tuple[str, str]: |
| audio_format = (audio_format or "wav").lower().strip() |
| if audio_format not in {"wav", "mp3"}: |
| raise gr.Error("audio_format must be wav or mp3") |
|
|
| wav_buf = io.BytesIO() |
| sf.write(wav_buf, audio, sample_rate, format="WAV") |
| wav_bytes = wav_buf.getvalue() |
| if audio_format == "wav": |
| return base64.b64encode(wav_bytes).decode("ascii"), "audio/wav" |
|
|
| with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as wav_file: |
| wav_file.write(wav_bytes) |
| wav_path = Path(wav_file.name) |
| mp3_path = wav_path.with_suffix(".mp3") |
| try: |
| subprocess.run( |
| [ |
| "ffmpeg", |
| "-y", |
| "-hide_banner", |
| "-loglevel", |
| "error", |
| "-i", |
| str(wav_path), |
| "-codec:a", |
| "libmp3lame", |
| "-b:a", |
| "128k", |
| str(mp3_path), |
| ], |
| check=True, |
| ) |
| return base64.b64encode(mp3_path.read_bytes()).decode("ascii"), "audio/mpeg" |
| except Exception as exc: |
| raise gr.Error(f"mp3 encoding failed: {exc}") from exc |
| finally: |
| wav_path.unlink(missing_ok=True) |
| mp3_path.unlink(missing_ok=True) |
|
|
|
|
| def get_asr_pipeline(): |
| global asr_pipeline |
| with asr_lock: |
| if asr_pipeline is not None: |
| return asr_pipeline |
| model_root = Path(snapshot_download(repo_id=ASR_REPO)) |
| onnx_dir = model_root / "onnx_models" |
| tokenizer_link = onnx_dir / "tokenizer.json" |
| root_tokenizer = model_root / "tokenizer.json" |
| if root_tokenizer.exists() and not tokenizer_link.exists(): |
| tokenizer_link.symlink_to(root_tokenizer) |
| if str(model_root) not in sys.path: |
| sys.path.insert(0, str(model_root)) |
| from onnx_inference import OnnxAsrPipeline |
|
|
| asr_pipeline = OnnxAsrPipeline( |
| onnx_dir=str(onnx_dir), |
| num_threads=ASR_THREADS, |
| quantize="int8", |
| ) |
| return asr_pipeline |
|
|
|
|
| def get_tts_pipeline(voice: str): |
| from kokoro import KPipeline |
|
|
| lang_code = tts_lang_from_voice(voice) |
| with tts_lock: |
| if lang_code not in tts_pipelines: |
| tts_pipelines[lang_code] = KPipeline(lang_code=lang_code, device="cpu") |
| return tts_pipelines[lang_code] |
|
|
|
|
| def health(api_key: str = "") -> str: |
| check_key(api_key) |
| return json.dumps( |
| { |
| "ok": True, |
| "uptime_seconds": round(time.time() - started_at, 3), |
| "asr_loaded": asr_pipeline is not None, |
| "tts_loaded_languages": sorted(tts_pipelines.keys()), |
| }, |
| ensure_ascii=False, |
| ) |
|
|
|
|
| @spaces.GPU(duration=10) |
| def zerogpu_probe() -> str: |
| return "ok" |
|
|
|
|
| def transcribe_b64( |
| api_key: str, |
| audio_base64: str, |
| filename: str = "audio.wav", |
| language: str = "auto", |
| max_new_tokens: int = 512, |
| chunk_sec: int = 30, |
| ) -> str: |
| check_key(api_key) |
| max_new_tokens = int(max_new_tokens) |
| chunk_sec = int(chunk_sec) |
| if max_new_tokens < 32 or max_new_tokens > 2048: |
| raise gr.Error("max_new_tokens must be between 32 and 2048") |
| if chunk_sec < 10 or chunk_sec > 60: |
| raise gr.Error("chunk_sec must be between 10 and 60") |
|
|
| path = decode_audio_b64(audio_base64, filename) |
| try: |
| pipeline = get_asr_pipeline() |
| result = pipeline.transcribe( |
| str(path), |
| language=normalize_language(language), |
| max_new_tokens=max_new_tokens, |
| chunk_sec=chunk_sec, |
| ) |
| finally: |
| path.unlink(missing_ok=True) |
|
|
| timing = result.get("timing", {}) |
| return json.dumps( |
| { |
| "language": result.get("language") or normalize_language(language) or "", |
| "text": result.get("text", ""), |
| "duration_seconds": timing.get("audio_duration_s"), |
| "processing_seconds": timing.get("total_s"), |
| "rtf": timing.get("rtf"), |
| "chunks": timing.get("sub_chunks"), |
| "timing": timing, |
| }, |
| ensure_ascii=False, |
| ) |
|
|
|
|
| def speech_b64( |
| api_key: str, |
| text: str, |
| voice: str = "zf_xiaoxiao", |
| speed: float = 1.0, |
| audio_format: str = "wav", |
| ) -> str: |
| check_key(api_key) |
| speed = float(speed) |
| if not text.strip(): |
| raise gr.Error("text is required") |
| if len(text) > 5000: |
| raise gr.Error("text is too long; max 5000 characters") |
| if speed < 0.5 or speed > 2.0: |
| raise gr.Error("speed must be between 0.5 and 2.0") |
|
|
| pipeline = get_tts_pipeline(voice) |
| chunks = [] |
| for _, _, audio in pipeline(text, voice=voice, speed=speed): |
| chunks.append(audio) |
| if not chunks: |
| raise gr.Error("speech synthesis returned no audio") |
|
|
| import numpy as np |
|
|
| audio = np.concatenate(chunks) |
| audio_base64, mime_type = audio_bytes_to_b64(audio, 24000, audio_format) |
| return json.dumps( |
| { |
| "audio_base64": audio_base64, |
| "mime_type": mime_type, |
| "sample_rate": 24000, |
| "voice": voice, |
| "audio_format": audio_format, |
| }, |
| ensure_ascii=False, |
| ) |
|
|
|
|
| with gr.Blocks(title="Voice API") as demo: |
| gr.Markdown("# Voice API") |
| gr.Markdown("Use the named API endpoints from n8n. Audio input/output is base64 JSON for simple automation.") |
| api_key = gr.Textbox(label="api_key", type="password") |
| with gr.Tab("Health"): |
| health_out = gr.Textbox(label="result") |
| gr.Button("Check").click(health, inputs=[api_key], outputs=[health_out], api_name="health") |
| gr.Button("ZeroGPU probe", visible=False).click( |
| zerogpu_probe, |
| inputs=[], |
| outputs=[health_out], |
| api_name="zerogpu_probe", |
| ) |
| with gr.Tab("ASR"): |
| audio_b64 = gr.Textbox(label="audio_base64", lines=5) |
| filename = gr.Textbox(label="filename", value="audio.wav") |
| language = gr.Textbox(label="language", value="auto") |
| max_tokens = gr.Number(label="max_new_tokens", value=512, precision=0) |
| chunk_sec = gr.Number(label="chunk_sec", value=30, precision=0) |
| asr_out = gr.Textbox(label="result", lines=8) |
| gr.Button("Transcribe").click( |
| transcribe_b64, |
| inputs=[api_key, audio_b64, filename, language, max_tokens, chunk_sec], |
| outputs=[asr_out], |
| api_name="transcribe_b64", |
| ) |
| with gr.Tab("TTS"): |
| text = gr.Textbox(label="text", lines=4) |
| voice = gr.Textbox(label="voice", value="zf_xiaoxiao") |
| speed = gr.Number(label="speed", value=1.0) |
| audio_format = gr.Textbox(label="audio_format", value="wav") |
| tts_out = gr.Textbox(label="result", lines=8) |
| gr.Button("Synthesize").click( |
| speech_b64, |
| inputs=[api_key, text, voice, speed, audio_format], |
| outputs=[tts_out], |
| api_name="speech_b64", |
| ) |
|
|
| demo.queue(default_concurrency_limit=1).launch() |
|
|
