diff --git a/AGENTS.md b/AGENTS.md index 785b7b2..4e08dbf 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -77,6 +77,18 @@ mytoolkit voice tts "文本" -v zh_male_wennuanahu_moon_bigtts --format mp3 --sp 语速:`-50` 到 `100`(0 = 正常) +### 语音识别 (STT) + +豆包录音文件识别极速版(flash ASR),与 TTS 共用同一组 Volcengine 凭证: + +```bash +mytoolkit voice stt -i clip.webm +mytoolkit voice stt -i clip.wav -o out.txt +mytoolkit voice stt -i clip.wav --no-correct # 跳过大模型纠错 +``` + +识别请求会带 ASR 原生热词(默认「小荷」)与可选对话上下文(`--history-file`)。`--correct` 可再开文本大模型兜底纠错(默认关闭)。webm / m4a 等会尽量经 `ffmpeg` 转 16 kHz mono wav。 + --- ## 4. LaTeX 脚手架与投稿模板 diff --git a/README.md b/README.md index 5ec9ad8..db8da98 100644 --- a/README.md +++ b/README.md @@ -56,7 +56,7 @@ mytoolkit --help | `latex` | `compile` / `count` | | `bib` | `to-markdown` / `cv-update` | | `video` | `avi-to-mp4` | -| `voice` | Text-to-speech via Volcengine/Doubao (`voice tts`, mp3/wav/pcm) | +| `voice` | Text-to-speech / speech-to-text via Volcengine/Doubao (`voice tts`, `voice stt`) | | `webpage` | Docusaurus build/list/rebuild/push | | `mail` | IMAP/SMTP client (read/draft/forward/reply, no auto-send) | | `templates` | Inspect template root; external override still possible | diff --git a/mytoolkit/api/__init__.py b/mytoolkit/api/__init__.py index 418d8ca..885ed56 100644 --- a/mytoolkit/api/__init__.py +++ b/mytoolkit/api/__init__.py @@ -1,6 +1,13 @@ """Pure API clients for external services (no CLI deps).""" +from .asr import transcribe_flash +from .asr_correct import correct_asr_text from .image import generate_image from .voice import synthesize_tts -__all__ = ["generate_image", "synthesize_tts"] +__all__ = [ + "generate_image", + "synthesize_tts", + "transcribe_flash", + "correct_asr_text", +] diff --git a/mytoolkit/api/asr.py b/mytoolkit/api/asr.py new file mode 100644 index 0000000..095a017 --- /dev/null +++ b/mytoolkit/api/asr.py @@ -0,0 +1,206 @@ +"""Doubao / Volcengine flash ASR (录音文件识别极速版).""" + +from __future__ import annotations + +import base64 +import json +import shutil +import subprocess +import tempfile +import uuid +from pathlib import Path +from typing import Any, Sequence + +import urllib.error +import urllib.request + +_FLASH_URL = "https://openspeech.bytedance.com/api/v3/auc/bigmodel/recognize/flash" +_RESOURCE_ID = "volc.bigasr.auc_turbo" +_OK_CODE = "20000000" + +_FORMAT_BY_SUFFIX = { + ".wav": "wav", + ".mp3": "mp3", + ".ogg": "ogg", + ".webm": "webm", + ".m4a": "mp4", + ".mp4": "mp4", + ".aac": "aac", + ".flac": "flac", +} + +# Product defaults — ASR-side boosting, not post-hoc string replace. +DEFAULT_HOTWORDS: tuple[str, ...] = ("小荷", "Xiaohe") +DEFAULT_SCENE_HINTS: tuple[str, ...] = ( + "我在使用个人助理小荷(Xiaohe)", +) + + +def _to_wav_if_needed(path: Path) -> tuple[bytes, str]: + """Return (audio_bytes, format). Convert unsupported containers via ffmpeg.""" + suf = path.suffix.lower() + fmt = _FORMAT_BY_SUFFIX.get(suf) + data = path.read_bytes() + if fmt in {"wav", "mp3", "ogg"}: + return data, fmt + ffmpeg = shutil.which("ffmpeg") + if not ffmpeg: + if fmt: + return data, fmt + raise RuntimeError( + f"unsupported audio format {suf or '(unknown)'}; install ffmpeg " + "or upload wav/mp3/ogg" + ) + with tempfile.TemporaryDirectory(prefix="mytoolkit-asr-") as tmp: + out = Path(tmp) / "out.wav" + proc = subprocess.run( + [ + ffmpeg, + "-y", + "-i", + str(path), + "-ac", + "1", + "-ar", + "16000", + str(out), + ], + capture_output=True, + text=True, + check=False, + ) + if proc.returncode != 0 or not out.is_file(): + detail = (proc.stderr or proc.stdout or "").strip()[:300] + raise RuntimeError(f"ffmpeg convert failed: {detail or 'unknown'}") + return out.read_bytes(), "wav" + + +def build_asr_corpus( + *, + hotwords: Sequence[str] | None = None, + history: Sequence[str] | None = None, + scene_hints: Sequence[str] | None = None, + max_history: int = 8, +) -> dict[str, Any] | None: + """Build ``request.corpus`` for Doubao ASR (hotwords + dialog context).""" + if hotwords is None: + hotwords = DEFAULT_HOTWORDS + if scene_hints is None: + scene_hints = DEFAULT_SCENE_HINTS + + corpus: dict[str, Any] = {} + + words = [w.strip() for w in hotwords if w and w.strip()] + if words: + corpus["context"] = json.dumps( + { + "hotwords": [ + {"word": w, "scale": 2.0} for w in words[:200] + ] + }, + ensure_ascii=False, + ) + + # Newest-first context snippets (ASR truncates old ones when over limit). + snippets: list[str] = [] + for h in scene_hints: + t = (h or "").strip() + if t: + snippets.append(t) + hist = [((h or "").strip()) for h in (history or ())] + hist = [h for h in hist if h] + # Caller may pass oldest→newest; reverse to newest-first for the API. + for h in reversed(hist[-max_history:]): + snippets.append(h[:200]) + if snippets: + corpus["context_data"] = json.dumps( + [{"text": s} for s in snippets[:20]], + ensure_ascii=False, + ) + + return corpus or None + + +def transcribe_flash( + appid: str, + token: str, + audio_path: str | Path, + *, + uid: str | None = None, + timeout_sec: float = 60.0, + hotwords: Sequence[str] | None = None, + history: Sequence[str] | None = None, + scene_hints: Sequence[str] | None = None, +) -> str: + """Transcribe a local audio file via Doubao flash ASR. Returns text.""" + path = Path(audio_path) + if not path.is_file(): + raise FileNotFoundError(str(path)) + + audio_bytes, fmt = _to_wav_if_needed(path) + if not audio_bytes: + raise RuntimeError("empty audio") + + headers = { + "Content-Type": "application/json", + "X-Api-App-Key": appid, + "X-Api-Access-Key": token, + "X-Api-Resource-Id": _RESOURCE_ID, + "X-Api-Request-Id": str(uuid.uuid4()), + "X-Api-Sequence": "-1", + } + request_body: dict[str, Any] = { + "model_name": "bigmodel", + "enable_itn": True, + "enable_punc": True, + "enable_ddc": True, + } + corpus = build_asr_corpus( + hotwords=hotwords, history=history, scene_hints=scene_hints + ) + if corpus: + request_body["corpus"] = corpus + + body = { + "user": {"uid": uid or appid}, + "audio": { + "data": base64.b64encode(audio_bytes).decode("ascii"), + "format": fmt, + }, + "request": request_body, + } + req = urllib.request.Request( + _FLASH_URL, + data=json.dumps(body).encode("utf-8"), + headers=headers, + method="POST", + ) + try: + with urllib.request.urlopen(req, timeout=timeout_sec) as resp: + status = resp.headers.get("X-Api-Status-Code", "") + message = resp.headers.get("X-Api-Message", "") + raw = resp.read().decode("utf-8", errors="replace") + except urllib.error.HTTPError as exc: + detail = exc.read().decode("utf-8", errors="replace")[:400] + raise RuntimeError( + f"ASR HTTP {exc.code}: {detail or exc.reason}" + ) from exc + except urllib.error.URLError as exc: + raise RuntimeError(f"ASR network error: {exc.reason}") from exc + + if status and status != _OK_CODE: + raise RuntimeError( + f"ASR failed: status={status} message={message or 'unknown'}" + ) + + try: + payload = json.loads(raw) if raw else {} + except json.JSONDecodeError as exc: + raise RuntimeError("ASR response is not JSON") from exc + + result = payload.get("result") if isinstance(payload, dict) else None + if isinstance(result, dict): + text = (result.get("text") or "").strip() + if text: + return text + raise RuntimeError("ASR returned empty text") diff --git a/mytoolkit/api/asr_correct.py b/mytoolkit/api/asr_correct.py new file mode 100644 index 0000000..96f0f14 --- /dev/null +++ b/mytoolkit/api/asr_correct.py @@ -0,0 +1,128 @@ +"""LLM post-correction for ASR transcripts (Ark / DeepSeek).""" + +from __future__ import annotations + +import os +from typing import Callable + +# Soft domain hints for the model — not string-replace rules. +DEFAULT_CONTEXT = ( + "这是个人助理「小荷」(Xiaohe)客户端的语音输入。" + "识别结果可能把专有名词听错成同音字(例如产品名)。" +) + +DEFAULT_ARK_MODEL = os.environ.get( + "MYTOOLKIT_ASR_CORRECT_MODEL", "doubao-seed-1-6-flash-250828" +) +DEFAULT_DEEPSEEK_MODEL = os.environ.get( + "MYTOOLKIT_ASR_CORRECT_DEEPSEEK_MODEL", "deepseek-chat" +) + + +def _system_prompt(context: str) -> str: + system = ( + "你是语音识别结果校对助手。根据上下文,轻量修正明显的识别错误" + "(同音字、专有名词、标点),保留原意与口语风格。" + "不要新增原句没有的信息,不要翻译,不要解释。" + "只输出修正后的完整句子。" + ) + if context.strip(): + system += f"\n背景:{context.strip()}" + return system + + +def _unwrap(text: str) -> str: + out = (text or "").strip() + if len(out) >= 2 and out[0] == out[-1] and out[0] in "\"'“”": + out = out[1:-1].strip() + return out + + +def _chat_correct( + *, + base_url: str, + api_key: str, + model: str, + text: str, + context: str, + timeout_sec: float, +) -> str: + from openai import OpenAI + + client = OpenAI(base_url=base_url, api_key=api_key, timeout=timeout_sec) + resp = client.chat.completions.create( + model=model, + temperature=0.1, + max_tokens=min(800, max(64, len(text) * 3)), + messages=[ + {"role": "system", "content": _system_prompt(context)}, + {"role": "user", "content": text}, + ], + ) + return _unwrap(resp.choices[0].message.content or "") or text + + +def correct_asr_text( + text: str, + *, + context: str = DEFAULT_CONTEXT, + ark_api_key: str | None = None, + deepseek_api_key: str | None = None, + ark_model: str | None = None, + deepseek_model: str | None = None, + timeout_sec: float = 20.0, + on_skip: Callable[[str], None] | None = None, +) -> str: + """Correct an ASR transcript with a chat model. Returns corrected text. + + Tries Ark first (if key present), then DeepSeek. On failure, returns the + original text. Domain hints are prompt context only — no hardcoded replaces. + """ + cleaned = (text or "").strip() + if not cleaned: + return "" + + attempts: list[tuple[str, str, str, str]] = [] + if ark_api_key: + attempts.append( + ( + "ark", + "https://ark.cn-beijing.volces.com/api/v3", + ark_api_key, + (ark_model or DEFAULT_ARK_MODEL).strip() or DEFAULT_ARK_MODEL, + ) + ) + if deepseek_api_key: + attempts.append( + ( + "deepseek", + "https://api.deepseek.com", + deepseek_api_key, + (deepseek_model or DEFAULT_DEEPSEEK_MODEL).strip() + or DEFAULT_DEEPSEEK_MODEL, + ) + ) + + if not attempts: + if on_skip: + on_skip("no LLM API key for ASR correction") + return cleaned + + last_err = "" + for name, base_url, key, model in attempts: + try: + return _chat_correct( + base_url=base_url, + api_key=key, + model=model, + text=cleaned, + context=context, + timeout_sec=timeout_sec, + ) + except Exception as exc: # noqa: BLE001 + last_err = f"{name}: {exc}" + continue + + if on_skip: + on_skip(last_err or "ASR correction failed") + return cleaned diff --git a/mytoolkit/commands/voice.py b/mytoolkit/commands/voice.py index 30be341..b4e2f66 100644 --- a/mytoolkit/commands/voice.py +++ b/mytoolkit/commands/voice.py @@ -1,4 +1,4 @@ -"""Voice / TTS utilities via Volcengine (ByteDance) WebSocket API.""" +"""Voice / TTS / STT utilities via Volcengine (ByteDance) APIs.""" import os import re @@ -6,6 +6,8 @@ from pathlib import Path import click +from mytoolkit.api.asr import transcribe_flash +from mytoolkit.api.asr_correct import correct_asr_text from mytoolkit.api.voice import synthesize_tts from mytoolkit.config import config from mytoolkit.utils import handle_errors @@ -59,7 +61,7 @@ def _synthesize_sync( @click.group() def voice(): - """Voice synthesis (TTS) commands.""" + """Voice synthesis (TTS) and recognition (STT) commands.""" pass @@ -97,6 +99,84 @@ def tts_cmd(text, file, speaker, fmt, speed, output): click.echo(os.path.abspath(result)) +@voice.command("stt") +@click.option( + "-i", + "--input", + "input_path", + type=click.Path(exists=True, dir_okay=False), + required=True, + help="输入音频文件", +) +@click.option( + "-o", + "--output", + type=click.Path(dir_okay=False), + help="写出识别文本(默认打印到 stdout)", +) +@click.option( + "--history-file", + type=click.Path(exists=True, dir_okay=False), + help="近期对话文本(每行一条,旧→新),传给 ASR corpus 上下文", +) +@click.option( + "--hotword", + multiple=True, + help="ASR 热词(可重复);默认含「小荷」", +) +@click.option( + "--correct/--no-correct", + default=False, + show_default=True, + help="识别后再用文本大模型纠错(可选兜底;需 ark/deepseek key)", +) +@handle_errors +def stt_cmd(input_path, output, history_file, hotword, correct): + """使用豆包/火山引擎极速 ASR 将语音转为文本。""" + appid, access_token = _get_credentials() + if not appid or not access_token: + raise click.UsageError( + "Volcengine credentials not set. Run:\n" + " mytoolkit env set secrets.volcengine.app_id \n" + " mytoolkit env set secrets.volcengine.access_token " + ) + history = None + if history_file: + history = [ + ln.strip() + for ln in Path(history_file).read_text(encoding="utf-8").splitlines() + if ln.strip() + ] + hotwords = list(hotword) if hotword else None + text = transcribe_flash( + appid, + access_token, + input_path, + history=history, + hotwords=hotwords, + ) + if correct: + ark_key = config.resolve_key("secrets.api_keys.ark", "ARK_API_KEY") + deepseek_key = config.resolve_key( + "secrets.api_keys.deepseek", "DEEPSEEK_API_KEY" + ) + ark_model = config.resolve_key( + "settings.asr_correct_ark_model", "MYTOOLKIT_ASR_CORRECT_MODEL" + ) + text = correct_asr_text( + text, + ark_api_key=ark_key, + deepseek_api_key=deepseek_key, + ark_model=ark_model, + on_skip=lambda msg: click.echo(f"ASR correct skipped: {msg}", err=True), + ) + if output: + Path(output).write_text(text + "\n", encoding="utf-8") + click.echo(os.path.abspath(output)) + else: + click.echo(text) + + def _natural_sort_key(path: Path): name = path.stem return [int(text) if text.isdigit() else text.lower()