#!/usr/bin/env python3 """Unified ASR provider abstraction for local GPU (qwen3cam) and online models (Soniox, Gemini 3.5 Transcribe via LiteLLM, OpenAI Transcribe via LiteLLM, Grok). """ from __future__ import annotations import json import mimetypes import os import re import shutil import subprocess import tempfile import time import urllib.error import urllib.request import uuid from dataclasses import dataclass, field import http.client from pathlib import Path from typing import Any from .config import CFG from .log import log class EmptyAudioError(RuntimeError): """Raised when an audio file contains silence/noise or produces zero ASR speech tokens.""" pass PROVIDERS: dict[str, dict[str, str]] = { "qwen3cam": { "label": "内网 Qwen3-ASR 1.7B + CAM++ (分角色 + 声纹)", "default_model": "qwen3-asr-1.7b-cam++", "mode": "lan-diarized-voiceprint", }, "soniox": { "label": "Soniox Async v5 (线上分角色 + 词库热词注入)", "default_model": "stt-async-v5", "mode": "cloud-diarized", }, "gemini": { "label": "Gemini 3.5 Transcribe (LiteLLM 统一路由)", "default_model": "aistudio/gemini-3.5-transcribe", "mode": "litellm-segmented", }, "openai": { "label": "OpenAI GPT-4o Transcribe Diarize (LiteLLM 统一路由)", "default_model": "openai/gpt-4o-transcribe-diarize", "mode": "litellm-diarized", }, "grok": { "label": "Grok Voice Transcribe 2.0 (线上分角色)", "default_model": "grok-voice-transcribe-2.0", "mode": "cloud-diarized", }, } @dataclass class TranscribeResult: provider: str model: str transcript_text: str raw_response: Any diarized: bool = True voiceprint_session: str = "" duration_seconds: int = 0 meta: dict[str, Any] = field(default_factory=dict) def normalize_provider_name(name: str | None) -> str: raw = (name or CFG.asr_provider or "qwen3cam").strip().lower() if raw in ("paraformer", "qwen3", "qwen3-cam"): return "qwen3cam" if raw in ("gemini-3.5-transcribe", "aistudio/gemini-3.5-transcribe", "vertex/gemini-3.5-transcribe"): return "gemini" if raw in ("openai-diarize", "gpt-4o-transcribe-diarize", "gpt-4o-transcribe", "gpt-transcribe"): return "openai" if raw in ("xai", "grok-voice-transcribe-2.0"): return "grok" return raw if raw in PROVIDERS else "qwen3cam" def provider_status() -> list[dict[str, Any]]: """Return readiness metadata for each ASR provider without exposing secrets.""" has_litellm = False try: has_litellm = bool(CFG.api_key()) except Exception: has_litellm = False return [ { "id": "qwen3cam", "label": PROVIDERS["qwen3cam"]["label"], "model": PROVIDERS["qwen3cam"]["default_model"], "ready": bool(CFG.diarize_url), "endpoint": CFG.diarize_url, "via": "LAN GPU (:10099)", }, { "id": "soniox", "label": PROVIDERS["soniox"]["label"], "model": CFG.soniox_model, "ready": bool(CFG.soniox_api_key), "endpoint": CFG.soniox_api_base, "via": "SONIOX_API_KEY 直连", }, { "id": "gemini", "label": PROVIDERS["gemini"]["label"], "model": CFG.gemini_asr_model, "ready": has_litellm, "endpoint": CFG.transcriptions_endpoint, "via": "LiteLLM 统一网关", }, { "id": "openai", "label": PROVIDERS["openai"]["label"], "model": CFG.openai_asr_model, "ready": has_litellm, "endpoint": CFG.transcriptions_endpoint, "via": "LiteLLM 统一网关", }, { "id": "grok", "label": PROVIDERS["grok"]["label"], "model": CFG.grok_asr_model, "ready": bool(CFG.grok_api_key), "endpoint": CFG.grok_api_base + "/stt", "via": "XAI_API_KEY 直连", }, ] def collect_hotwords(limit: int = 120) -> list[str]: """Collect canonical entity names and glossary terms for ASR context biasing.""" terms: list[str] = [] seen: set[str] = set() def add(word: str) -> None: w = word.strip() if not w or len(w) < 2 or len(w) > 40 or w in seen: return if re.match(r"^\d{4}-\d{2}|^_|speaker_", w): return seen.add(w) terms.append(w) try: from . import lexicon con = lexicon.connect() for row in con.execute( "SELECT canonical FROM entity ORDER BY synced_at DESC LIMIT ?", (limit,)).fetchall(): add(str(row["canonical"])) for row in con.execute( "SELECT DISTINCT canonical FROM alias WHERE confirmed=1 LIMIT ?", (limit,)).fetchall(): add(str(row["canonical"])) con.close() except Exception: pass if CFG.glossary.exists(): for raw in CFG.glossary.read_text(errors="ignore").splitlines(): s = raw.strip() if not s.startswith(("-", "*")) or re.search(r"[→⇒]|->", s): continue s = re.split(r"\s[—–-]\s|[:：]", re.sub(r"^[-*+\d\.\)\s]+", "", s), maxsplit=1)[0].strip() s = re.sub(r"\[\[|\]\]", "", s).strip() add(s) if len(terms) >= limit: break return terms[:limit] def _normalize_speaker_label(raw: Any, mapping: dict[str, str]) -> str: s = str(raw if raw is not None else "").strip() if not s or s == "?": return "speaker_0" if re.match(r"^[\u4e00-\u9fff]{2,8}$", s): return s if re.match(r"^speaker_\d+$", s): return s m = re.match(r"^[sS](?:peaker)?[_ -]?(\d+)$", s) if m: return f"speaker_{m.group(1)}" if s.isdigit(): # Soniox often uses 1-based speaker numbers ("1", "2"); normalize to speaker_0..N-1 if s not in mapping: mapping[s] = f"speaker_{len(mapping)}" return mapping[s] if s not in mapping: mapping[s] = f"speaker_{len(mapping)}" return mapping[s] def format_diarized_transcript( segments: list[dict[str, Any]], *, provider: str, model: str, header_title: str, voiceprint_session: str = "", matched_names: list[str] | None = None, ) -> str: if not segments: raise EmptyAudioError( f"ASR ({provider}) 未识别到有效语音片段（音频可能为静音、设备未拾音或音量过低）" ) speaker_map: dict[str, str] = {} turns: list[list[str]] = [] speakers_seen: set[str] = set() for seg in segments: raw_spk = seg.get("speaker") text = str(seg.get("text") or "").strip() if not text: continue spk = (_normalize_speaker_label(raw_spk, speaker_map) if provider != "qwen3cam" else str(raw_spk if raw_spk is not None else "speaker_0")) speakers_seen.add(spk) if turns and turns[-1][0] == spk: turns[-1][1] += text else: turns.append([spk, text]) if not turns: raise EmptyAudioError( f"ASR ({provider}) 未识别到任何有效文字（音频可能为静音或底噪）" ) lines = [f""] if voiceprint_session or provider == "qwen3cam": lines.append( f"") hdr = f"# 转写({header_title})" if matched_names: hdr += " | 声纹命名: " + ", ".join(matched_names) lines.append(hdr) lines.append("") lines.extend(f"【{spk}】{txt}\n" for spk, txt in turns) return "\n".join(lines).strip() + "\n" def transcript_from_qwen3cam_response(payload: dict[str, Any]) -> str: segments = payload.get("segments") if not isinstance(segments, list) or not segments: raise EmptyAudioError("ASR (qwen3cam) 未识别到有效语音片段（音频可能为静音、设备未拾音或音量过低）") vp = payload.get("voiceprint") or {} sid = str(vp.get("session_id") or "") matched = [ f"{m.get('speaker_local')}={m.get('person_name')}" for m in vp.get("matches", []) if m.get("status") == "matched" ] return format_diarized_transcript( segments, provider="qwen3cam", model="qwen3-asr-1.7b-cam++", header_title="Qwen3-ASR 1.7B + CAM++ 分角色 + 声纹", voiceprint_session=sid, matched_names=matched, ) def transcript_from_soniox_response( payload: dict[str, Any], *, model: str = "stt-async-v5", ) -> str: """Convert Soniox token-level or segment-level diarized response into turns.""" tokens = payload.get("tokens") segments: list[dict[str, Any]] = [] if isinstance(tokens, list) and tokens: cur_speaker: Any = None cur_pieces: list[str] = [] for tok in tokens: txt = str(tok.get("text") or "") if not txt: continue spk = tok.get("speaker") if cur_speaker is not None and spk != cur_speaker and cur_pieces: segments.append({"speaker": cur_speaker, "text": "".join(cur_pieces).strip()}) cur_pieces = [] cur_speaker = spk if spk is not None else (cur_speaker or "1") cur_pieces.append(txt) if cur_pieces: segments.append({"speaker": cur_speaker or "1", "text": "".join(cur_pieces).strip()}) elif isinstance(payload.get("segments"), list) and payload["segments"]: segments = payload["segments"] elif isinstance(payload.get("text"), str) and payload["text"].strip(): segments = [{"speaker": "speaker_0", "text": payload["text"].strip()}] return format_diarized_transcript( segments, provider="soniox", model=model, header_title=f"Soniox {model} 分角色", ) def transcript_from_openai_response( payload: dict[str, Any], *, model: str, ) -> str: """Convert OpenAI / LiteLLM transcription response (diarized or plain) into text.""" segments = payload.get("segments") if isinstance(segments, list) and segments and any( s.get("speaker") is not None for s in segments if isinstance(s, dict)): return format_diarized_transcript( segments, provider="openai", model=model, header_title=f"OpenAI {model} 分角色", ) text = str(payload.get("text") or payload.get("transcript") or "").strip() if not text and isinstance(segments, list): text = "".join(str(s.get("text") or "") for s in segments).strip() if not text: raise RuntimeError(f"OpenAI ASR ({model}) returned empty transcript") if re.search(r"【[^】]+】", text): return ( f"\n" f"# 转写(OpenAI {model})\n\n{text}\n" ) return ( f"\n" f"--- 第1段 [00:00 - 结束] ---\n{text}\n" ) def transcript_from_grok_response( payload: dict[str, Any], *, model: str = "grok-voice-transcribe-2.0", ) -> str: segments = payload.get("segments") or payload.get("utterances") or payload.get("turns") if isinstance(segments, list) and segments: norm_segs = [] for s in segments: if not isinstance(s, dict): continue spk = s.get("speaker") if spk is None: spk = s.get("speaker_id") if spk is None: spk = "speaker_0" txt = str(s.get("text") or s.get("transcript") or "").strip() if txt: norm_segs.append({"speaker": spk, "text": txt}) if norm_segs: return format_diarized_transcript( norm_segs, provider="grok", model=model, header_title=f"Grok {model} 分角色", ) words = payload.get("words") if isinstance(words, list) and words and any( isinstance(w, dict) and w.get("speaker") is not None for w in words): norm_segs = [] cur_spk = None buf: list[str] = [] for w in words: if not isinstance(w, dict): continue txt = str(w.get("word") or w.get("text") or "") spk = w.get("speaker") if cur_spk is not None and spk != cur_spk and buf: norm_segs.append({"speaker": cur_spk, "text": "".join(buf).strip()}) buf = [] cur_spk = spk if spk is not None else (cur_spk or "0") buf.append(txt) if buf: norm_segs.append({"speaker": cur_spk or "0", "text": "".join(buf).strip()}) if norm_segs: return format_diarized_transcript( norm_segs, provider="grok", model=model, header_title=f"Grok {model} 分角色", ) text = str(payload.get("text") or payload.get("transcript") or "").strip() if not text: raise RuntimeError(f"Grok ASR ({model}) returned empty transcript") return format_diarized_transcript( [{"speaker": "speaker_0", "text": text}], provider="grok", model=model, header_title=f"Grok {model}", ) def probe_duration_seconds(audio: Path) -> int: try: out = subprocess.check_output([ "ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=nk=1:nw=1", str(audio), ], text=True, timeout=30).strip() return max(1, int(float(out) + 0.999)) if out else 0 except Exception: return 0 def _encode_multipart( fields: dict[str, str], file_field: str, file_path: Path, ) -> tuple[bytes, str]: boundary = f"----ASRBoundary{uuid.uuid4().hex}" body = bytearray() for name, val in fields.items(): body.extend(f"--{boundary}\r\n".encode()) body.extend( f'Content-Disposition: form-data; name="{name}"\r\n\r\n'.encode()) body.extend(str(val).encode("utf-8")) body.extend(b"\r\n") mime = mimetypes.guess_type(file_path.name)[0] or "application/octet-stream" body.extend(f"--{boundary}\r\n".encode()) body.extend( f'Content-Disposition: form-data; name="{file_field}"; ' f'filename="{file_path.name}"\r\n'.encode()) body.extend(f"Content-Type: {mime}\r\n\r\n".encode()) body.extend(file_path.read_bytes()) body.extend(f"\r\n--{boundary}--\r\n".encode()) return bytes(body), f"multipart/form-data; boundary={boundary}" def _http_multipart( url: str, *, fields: dict[str, str], file_field: str, file_path: Path, headers: dict[str, str] | None = None, timeout: int = 900, retries: int = 3, ) -> dict[str, Any]: data, content_type = _encode_multipart(fields, file_field, file_path) req_headers = {"Content-Type": content_type, **(headers or {})} last_exc: Exception | None = None for attempt in range(1, max(1, retries) + 1): req = urllib.request.Request(url, data=data, headers=req_headers, method="POST") try: with urllib.request.urlopen(req, timeout=timeout) as resp: raw = resp.read().decode("utf-8", errors="ignore") break except urllib.error.HTTPError as exc: detail = exc.read().decode("utf-8", errors="ignore")[:300] if exc.code in (502, 503, 504) and attempt < retries: last_exc = exc time.sleep(2 * attempt) continue raise RuntimeError(f"ASR HTTP {exc.code}: {detail}") from exc except (urllib.error.URLError, http.client.HTTPException, ConnectionError, TimeoutError, OSError) as exc: last_exc = exc if attempt < retries: log(f"multipart upload transient error (attempt {attempt}/{retries}): {exc}") time.sleep(2 * attempt) continue raise RuntimeError(f"ASR upload connection error after {retries} attempts: {exc}") from exc else: raise RuntimeError(f"ASR upload failed: {last_exc}") try: return json.loads(raw) except ValueError as exc: raise RuntimeError(f"ASR returned invalid JSON: {raw[:200]}") from exc def _http_json( method: str, url: str, *, payload: dict[str, Any] | None = None, headers: dict[str, str] | None = None, timeout: int = 120, retries: int = 3, ) -> dict[str, Any]: data = json.dumps(payload, ensure_ascii=False).encode("utf-8") if payload is not None else None req_headers = {"Accept": "application/json", **(headers or {})} if data is not None: req_headers["Content-Type"] = "application/json" last_exc: Exception | None = None for attempt in range(1, max(1, retries) + 1): req = urllib.request.Request(url, data=data, headers=req_headers, method=method) try: with urllib.request.urlopen(req, timeout=timeout) as resp: raw = resp.read().decode("utf-8", errors="ignore") break except urllib.error.HTTPError as exc: detail = exc.read().decode("utf-8", errors="ignore")[:300] if exc.code in (502, 503, 504) and attempt < retries: last_exc = exc time.sleep(1.5 * attempt) continue raise RuntimeError(f"ASR HTTP {exc.code}: {detail}") from exc except (urllib.error.URLError, http.client.HTTPException, ConnectionError, TimeoutError, OSError) as exc: last_exc = exc if attempt < retries: time.sleep(1.5 * attempt) continue raise RuntimeError(f"ASR HTTP request failed after {retries} attempts: {exc}") from exc else: raise RuntimeError(f"ASR HTTP request failed: {last_exc}") if not raw.strip(): return {} try: return json.loads(raw) except ValueError as exc: raise RuntimeError(f"ASR returned invalid JSON: {raw[:200]}") from exc def _compress_or_split_audio( audio: Path, tmpdir: Path, *, max_bytes: int = 23 * 1024 * 1024, segment_seconds: int = 2400, force_split: bool = False, ) -> list[tuple[Path, int, int]]: """Return [(segment_path, start_sec, end_sec), ...] suitable for upstream API.""" duration = probe_duration_seconds(audio) if not force_split and audio.stat().st_size <= max_bytes and audio.suffix.lower() in { ".mp3", ".m4a", ".wav", ".ogg", ".webm", ".flac"}: return [(audio, 0, duration)] if not shutil.which("ffmpeg"): if audio.stat().st_size <= max_bytes: return [(audio, 0, duration)] raise RuntimeError("audio exceeds size limit and ffmpeg is not installed") if not force_split: compressed = tmpdir / "compressed.mp3" subprocess.run([ "ffmpeg", "-hide_banner", "-loglevel", "error", "-y", "-i", str(audio), "-ar", "16000", "-ac", "1", "-b:a", "48k", str(compressed), ], check=True, timeout=600) if compressed.exists() and compressed.stat().st_size <= max_bytes: return [(compressed, 0, duration)] total = max(duration, 1) step = max(60, segment_seconds) n_segs = (total + step - 1) // step out: list[tuple[Path, int, int]] = [] for i in range(n_segs): start = i * step end = min(total, start + step) seg = tmpdir / f"segment_{i + 1:03d}.mp3" subprocess.run([ "ffmpeg", "-hide_banner", "-loglevel", "error", "-y", "-ss", str(start), "-t", str(step), "-i", str(audio), "-ar", "16000", "-ac", "1", "-b:a", "64k", str(seg), ], check=True, timeout=600) out.append((seg, start, end)) return out def transcribe_qwen3cam(audio: Path, *, model: str = "") -> TranscribeResult: result = subprocess.run([ "curl", "-sS", "--fail-with-body", "--max-time", "900", "-F", f"file=@{audio}", "-F", "language=Chinese", "-F", "save_voiceprint_session=true", CFG.diarize_url, ], capture_output=True, timeout=930) if result.returncode: detail = result.stderr.decode("utf-8", errors="ignore")[:240] raise RuntimeError(f"qwen3cam ASR HTTP failed ({result.returncode}): {detail}") try: payload = json.loads(result.stdout) except ValueError as exc: raise RuntimeError("qwen3cam returned invalid JSON") from exc text = transcript_from_qwen3cam_response(payload) vp = payload.get("voiceprint") or {} return TranscribeResult( provider="qwen3cam", model=model or PROVIDERS["qwen3cam"]["default_model"], transcript_text=text, raw_response=payload, diarized=True, voiceprint_session=str(vp.get("session_id") or ""), ) def build_soniox_context(title: str = "", user_context: str = "") -> dict[str, Any]: ctx: dict[str, Any] = {} general = [{"key": "domain", "value": "Biotech / Investment / Corporate Meeting"}] if title.strip(): general.append({"key": "title", "value": title.strip()[:120]}) ctx["general"] = general if user_context.strip(): ctx["text"] = user_context.strip()[:2000] terms = collect_hotwords(120) if terms: ctx["terms"] = terms return ctx def transcribe_soniox( audio: Path, *, model: str = "", title: str = "", user_context: str = "", poll_interval: float = 3.0, max_wait_seconds: float = 900.0, ) -> TranscribeResult: key = CFG.soniox_api_key if not key: raise RuntimeError("SONIOX_API_KEY is not configured") base = CFG.soniox_api_base chosen_model = model or CFG.soniox_model or "stt-async-v5" headers = {"Authorization": f"Bearer {key}"} with tempfile.TemporaryDirectory(prefix="asr-soniox-") as tmp: upload_path = audio if audio.exists() and audio.stat().st_size > 8 * 1024 * 1024 and shutil.which("ffmpeg"): compressed = Path(tmp) / "soniox_upload.mp3" try: subprocess.run([ "ffmpeg", "-hide_banner", "-loglevel", "error", "-y", "-i", str(audio), "-ar", "16000", "-ac", "1", "-b:a", "48k", str(compressed), ], check=True, timeout=600) if compressed.exists() and 0 < compressed.stat().st_size < audio.stat().st_size: upload_path = compressed except Exception as exc: log(f"soniox audio pre-compress skipped: {exc}") file_resp = _http_multipart( f"{base}/files", fields={}, file_field="file", file_path=upload_path, headers=headers, timeout=600, ) file_id = str(file_resp.get("id") or "") if not file_id: raise RuntimeError(f"Soniox file upload returned no id: {file_resp}") transcription_id = "" try: job_payload: dict[str, Any] = { "model": chosen_model, "file_id": file_id, "enable_speaker_diarization": True, "enable_language_identification": True, "language_hints": ["zh", "en"], "context": build_soniox_context(title, user_context), } job_resp = _http_json( "POST", f"{base}/transcriptions", payload=job_payload, headers=headers, timeout=60, ) transcription_id = str(job_resp.get("id") or "") if not transcription_id: raise RuntimeError(f"Soniox transcription create returned no id: {job_resp}") deadline = time.time() + max_wait_seconds status_resp: dict[str, Any] = {} while time.time() < deadline: status_resp = _http_json( "GET", f"{base}/transcriptions/{transcription_id}", headers=headers, timeout=60, ) st = str(status_resp.get("status") or "").lower() if st == "completed": break if st in ("error", "failed"): err = status_resp.get("error_message") or status_resp raise RuntimeError(f"Soniox transcription failed: {err}") time.sleep(poll_interval) else: raise RuntimeError("Soniox transcription timed out waiting for completion") if status_resp.get("tokens") or status_resp.get("segments"): transcript_payload = status_resp else: transcript_payload = _http_json( "GET", f"{base}/transcriptions/{transcription_id}/transcript", headers=headers, timeout=120, ) text = transcript_from_soniox_response( transcript_payload, model=chosen_model) return TranscribeResult( provider="soniox", model=chosen_model, transcript_text=text, raw_response=transcript_payload, diarized=True, ) finally: for cleanup_url in ( f"{base}/transcriptions/{transcription_id}" if transcription_id else "", f"{base}/files/{file_id}" if file_id else "", ): if cleanup_url: try: _http_json("DELETE", cleanup_url, headers=headers, timeout=15) except Exception: pass def transcribe_gemini( audio: Path, *, model: str = "", title: str = "", user_context: str = "", ) -> TranscribeResult: chosen_model = model or CFG.gemini_asr_model or "aistudio/gemini-3.5-transcribe" headers = {"Authorization": f"Bearer {CFG.api_key()}"} with tempfile.TemporaryDirectory(prefix="asr-gemini-") as tmp: segments = _compress_or_split_audio( audio, Path(tmp), max_bytes=18 * 1024 * 1024, segment_seconds=CFG.segment_seconds, force_split=probe_duration_seconds(audio) > CFG.segment_seconds + 60, ) responses: list[dict[str, Any]] = [] parts: list[str] = [f""] for idx, (seg_path, start, end) in enumerate(segments, 1): resp = _http_multipart( CFG.transcriptions_endpoint, fields={"model": chosen_model, "response_format": "json"}, file_field="file", file_path=seg_path, headers=headers, timeout=600, ) responses.append(resp) seg_text = str(resp.get("text") or resp.get("transcript") or "").strip() if not seg_text: raise RuntimeError(f"Gemini segment {idx} returned empty text") parts.append( f"--- 第{idx}段 [{start // 60:02d}:{start % 60:02d} - " f"{end // 60:02d}:{end % 60:02d}] ---\n{seg_text}\n") return TranscribeResult( provider="gemini", model=chosen_model, transcript_text="\n\n".join(parts).strip() + "\n", raw_response={"segments": responses}, diarized=False, ) def transcribe_openai( audio: Path, *, model: str = "", title: str = "", user_context: str = "", ) -> TranscribeResult: chosen_model = model or CFG.openai_asr_model or "openai/gpt-4o-transcribe-diarize" headers = {"Authorization": f"Bearer {CFG.api_key()}"} is_diarize = "diarize" in chosen_model.lower() with tempfile.TemporaryDirectory(prefix="asr-openai-") as tmp: segments = _compress_or_split_audio( audio, Path(tmp), max_bytes=23 * 1024 * 1024, segment_seconds=2400, force_split=False, ) responses: list[dict[str, Any]] = [] all_diarized_segs: list[dict[str, Any]] = [] plain_parts: list[str] = [f""] hotwords = collect_hotwords(80) prompt = "" if not is_diarize and hotwords: prompt = "会议专有名词参考：" + "、".join(hotwords)[:800] for idx, (seg_path, start, end) in enumerate(segments, 1): fields: dict[str, str] = { "model": chosen_model, "language": "zh", "response_format": "diarized_json" if is_diarize else "json", } if prompt: fields["prompt"] = prompt try: resp = _http_multipart( CFG.transcriptions_endpoint, fields=fields, file_field="file", file_path=seg_path, headers=headers, timeout=600, ) except RuntimeError: if is_diarize: # Fallback if gateway enforces response_format=json fields["response_format"] = "json" resp = _http_multipart( CFG.transcriptions_endpoint, fields=fields, file_field="file", file_path=seg_path, headers=headers, timeout=600, ) else: raise responses.append(resp) if isinstance(resp.get("segments"), list) and resp["segments"]: all_diarized_segs.extend(resp["segments"]) else: txt = str(resp.get("text") or resp.get("transcript") or "").strip() if txt: plain_parts.append( f"--- 第{idx}段 [{start // 60:02d}:{start % 60:02d} - " f"{end // 60:02d}:{end % 60:02d}] ---\n{txt}\n") if all_diarized_segs: text = transcript_from_openai_response( {"segments": all_diarized_segs}, model=chosen_model) diarized = True elif len(responses) == 1: text = transcript_from_openai_response(responses[0], model=chosen_model) diarized = "【" in text else: if len(plain_parts) <= 1: raise RuntimeError(f"OpenAI ASR ({chosen_model}) returned no text") text = "\n\n".join(plain_parts).strip() + "\n" diarized = False return TranscribeResult( provider="openai", model=chosen_model, transcript_text=text, raw_response={"segments": responses} if len(responses) > 1 else responses[0], diarized=diarized, ) def transcribe_grok( audio: Path, *, model: str = "", title: str = "", user_context: str = "", ) -> TranscribeResult: key = CFG.grok_api_key if not key: raise RuntimeError("XAI_API_KEY is not configured") chosen_model = model or CFG.grok_asr_model or "grok-voice-transcribe-2.0" headers = {"Authorization": f"Bearer {key}"} url = f"{CFG.grok_api_base}/stt" with tempfile.TemporaryDirectory(prefix="asr-grok-") as tmp: segments = _compress_or_split_audio( audio, Path(tmp), max_bytes=23 * 1024 * 1024, segment_seconds=2400) all_texts: list[str] = [] raw_list: list[dict[str, Any]] = [] for seg_path, _start, _end in segments: resp = _http_multipart( url, fields={"model": chosen_model, "diarize": "true", "language": "zh"}, file_field="file", file_path=seg_path, headers=headers, timeout=600, ) raw_list.append(resp) all_texts.append(transcript_from_grok_response(resp, model=chosen_model)) text = all_texts[0] if len(all_texts) == 1 else "\n\n".join(all_texts) return TranscribeResult( provider="grok", model=chosen_model, transcript_text=text, raw_response=raw_list[0] if len(raw_list) == 1 else {"chunks": raw_list}, diarized=True, ) def transcribe_audio( audio: Path, *, provider: str | None = None, model: str = "", title: str = "", user_context: str = "", ) -> TranscribeResult: """Dispatch transcription to the selected provider and return normalized result.""" prov = normalize_provider_name(provider) log(f"transcribe start provider={prov} model={model or 'default'} file={audio.name}") if prov == "qwen3cam": return transcribe_qwen3cam(audio, model=model) if prov == "soniox": return transcribe_soniox(audio, model=model, title=title, user_context=user_context) if prov == "gemini": return transcribe_gemini(audio, model=model, title=title, user_context=user_context) if prov == "openai": return transcribe_openai(audio, model=model, title=title, user_context=user_context) if prov == "grok": return transcribe_grok(audio, model=model, title=title, user_context=user_context) raise ValueError(f"unsupported ASR provider: {prov}")