diff --git a/custom_components/hass_stt_doubao/doubaoime_asr/__init__.py b/custom_components/hass_stt_doubao/doubaoime_asr/__init__.py index 4adbd21..282d09f 100644 --- a/custom_components/hass_stt_doubao/doubaoime_asr/__init__.py +++ b/custom_components/hass_stt_doubao/doubaoime_asr/__init__.py @@ -1,6 +1,11 @@ from .asr import ( DoubaoASR, ASRResponse, + ASRResult, + ASRAlternative, + ASRWord, + ASRExtra, + OIDecodingInfo, ASRError, ResponseType, AudioChunk, @@ -9,15 +14,26 @@ from .asr import ( transcribe_realtime, ) from .config import ASRConfig +from .ner import NerResponse, NerResult, NerWord, ner, async_ner __all__ = [ "DoubaoASR", "ASRConfig", "ASRResponse", + "ASRResult", + "ASRAlternative", + "ASRWord", + "ASRExtra", + "OIDecodingInfo", "ASRError", "ResponseType", "AudioChunk", "transcribe", "transcribe_stream", "transcribe_realtime", + "NerResponse", + "NerResult", + "NerWord", + "ner", + "async_ner", ] diff --git a/custom_components/hass_stt_doubao/doubaoime_asr/asr.py b/custom_components/hass_stt_doubao/doubaoime_asr/asr.py index af11c57..bd224be 100644 --- a/custom_components/hass_stt_doubao/doubaoime_asr/asr.py +++ b/custom_components/hass_stt_doubao/doubaoime_asr/asr.py @@ -2,7 +2,7 @@ from __future__ import annotations import asyncio import contextlib -from dataclasses import dataclass +from dataclasses import dataclass, field from enum import Enum, auto import json from pathlib import Path @@ -37,6 +37,59 @@ class ResponseType(Enum): UNKNOWN = auto() +@dataclass +class ASRWord: + """单词级别的识别结果""" + word: str + start_time: float + end_time: float + + +@dataclass +class OIDecodingInfo: + """OI 解码信息""" + oi_former_word_num: int = 0 + oi_latter_word_num: int = 0 + oi_words: Optional[List] = None + + +@dataclass +class ASRAlternative: + """识别候选结果""" + text: str + start_time: float + end_time: float + words: List[ASRWord] = field(default_factory=list) + semantic_related_to_prev: Optional[bool] = None + oi_decoding_info: Optional[OIDecodingInfo] = None + + +@dataclass +class ASRResult: + """单条识别结果""" + text: str + start_time: float + end_time: float + confidence: float = 0.0 + alternatives: List[ASRAlternative] = field(default_factory=list) + is_interim: bool = True + is_vad_finished: bool = False + index: int = 0 + + +@dataclass +class ASRExtra: + """响应附加信息""" + audio_duration: Optional[int] = None + model_avg_rtf: Optional[float] = None + model_send_first_response: Optional[int] = None + speech_adaptation_version: Optional[str] = None + model_total_process_time: Optional[int] = None + packet_number: Optional[int] = None + vad_start: Optional[bool] = None + req_payload: Optional[dict] = None + + @dataclass class ASRResponse: """ @@ -50,6 +103,8 @@ class ASRResponse: packet_number: int = -1 error_msg: str = "" raw_json: Optional[dict] = None + results: List[ASRResult] = field(default_factory=list) + extra: Optional[ASRExtra] = None class ASRError(Exception): @@ -71,6 +126,14 @@ class _SessionState(BaseModel): error: Optional[ASRResponse] = None +def _create_ssl_context() -> ssl.SSLContext: + """ + 在线程中创建并初始化 SSL 上下文，避免阻塞事件循环 + """ + ctx = ssl.create_default_context() + return ctx + + class DoubaoASR: """ 豆包输入法 ASR 客户端 @@ -86,13 +149,13 @@ class DoubaoASR: async def __aexit__(self, exc_type, exc_val, exc_tb) -> None: pass - async def _ensure_ssl_context(self): + async def _get_ssl_context(self) -> ssl.SSLContext: """ - 确保 SSL 上下文已创建（在 executor 中运行避免阻塞） + 获取预初始化的 SSL 上下文（懒加载，在线程中创建） """ if self._ssl_context is None: - loop = asyncio.get_event_loop() - self._ssl_context = await loop.run_in_executor(None, ssl.create_default_context) + self._ssl_context = await asyncio.to_thread(_create_ssl_context) + return self._ssl_context async def transcribe(self, audio: Union[str, Path, bytes], *, realtime = False, on_interim: Callable[[str], None] = None) -> str: """ @@ -127,21 +190,18 @@ class DoubaoASR: else: pcm_data = audio - opus_frames = await self._encoder.pcm_to_opus_frames(pcm_data) + opus_frames = self._encoder.pcm_to_opus_frames(pcm_data) state = _SessionState() - # 预创建 SSL 上下文避免阻塞 - await self._ensure_ssl_context() - - # 获取 WebSocket URL - ws_url = await self.config.get_ws_url() + ws_url = await self.config.async_ws_url() + ssl_context = await self._get_ssl_context() try: async with websockets.connect( ws_url, additional_headers=self.config.headers, open_timeout=self.config.connect_timeout, - ssl=self._ssl_context, + ssl=ssl_context, ) as ws: # 初始化会话 async for resp in self._initialize_session(ws, state): @@ -207,18 +267,15 @@ class DoubaoASR: """ state = _SessionState() - # 预创建 SSL 上下文避免阻塞 - await self._ensure_ssl_context() - - # 获取 WebSocket URL - ws_url = await self.config.get_ws_url() + ws_url = await self.config.async_ws_url() + ssl_context = await self._get_ssl_context() try: async with websockets.connect( ws_url, additional_headers=self.config.headers, open_timeout=self.config.connect_timeout, - ssl=self._ssl_context, + ssl=ssl_context, ) as ws: # 初始化会话 async for resp in self._initialize_session(ws, state): @@ -270,9 +327,6 @@ class DoubaoASR: """ 从异步迭代器读取 PCM 数据并实时发送 """ - # 预先获取编码器，避免在循环中触发阻塞导入 - encoder = await self._encoder.get_encoder() - timestamp_ms = int(time.time() * 1000) frame_index = 0 pcm_buffer = b"" @@ -294,7 +348,7 @@ class DoubaoASR: pcm_buffer = pcm_buffer[bytes_per_frame:] # 编码为 Opus - opus_frame = encoder.encode(pcm_frame, samples_per_frame) + opus_frame = self._encoder.encoder.encode(pcm_frame, samples_per_frame) # 确定帧状态（实时模式下不知道最后一帧，使用 FIRST/MIDDLE） if frame_index == 0: @@ -317,7 +371,7 @@ class DoubaoASR: if len(pcm_buffer) < bytes_per_frame: pcm_buffer += b"\x00" * (bytes_per_frame - len(pcm_buffer)) - opus_frame = encoder.encode(pcm_buffer, samples_per_frame) + opus_frame = self._encoder.encoder.encode(pcm_buffer, samples_per_frame) msg = _build_asr_request( opus_frame, @@ -330,7 +384,7 @@ class DoubaoASR: # 没有剩余数据，但需要发送一个 LAST 帧标记 # 发送一个空的 LAST 帧（静音） silent_frame = b"\x00" * bytes_per_frame - opus_frame = encoder.encode(silent_frame, samples_per_frame) + opus_frame = self._encoder.encoder.encode(silent_frame, samples_per_frame) msg = _build_asr_request( opus_frame, @@ -342,14 +396,14 @@ class DoubaoASR: # FinishSession if not state.is_finished: - token = await self.config.get_token() + token = await self.config.async_get_token() await ws.send(_build_finish_session(state.request_id, token)) async def _initialize_session(self, ws: ClientConnection, state: _SessionState) -> AsyncIterator[ASRResponse]: """ 初始化 ASR 会话 """ - token = await self.config.get_token() + token = await self.config.async_get_token() # StartTask await ws.send(_build_start_task(state.request_id, token)) @@ -360,9 +414,9 @@ class DoubaoASR: yield parsed # StartSession - session_config = await self.config.get_session_config() + session_cfg = await self.config.async_session_config() await ws.send( - _build_start_session(state.request_id, token, session_config) + _build_start_session(state.request_id, token, session_cfg) ) resp = await ws.recv() parsed = _parse_response(resp) @@ -406,7 +460,7 @@ class DoubaoASR: await asyncio.sleep(frame_interval) # FinishSession - token = await self.config.get_token() + token = await self.config.async_get_token() await ws.send(_build_finish_session(state.request_id, token)) async def _receive_responses( @@ -499,6 +553,68 @@ def _build_asr_request( +def _parse_word(data: dict) -> ASRWord: + """解析单词数据""" + return ASRWord( + word=data.get("word", ""), + start_time=data.get("start_time", 0.0), + end_time=data.get("end_time", 0.0), + ) + + +def _parse_oi_decoding_info(data: Optional[dict]) -> Optional[OIDecodingInfo]: + """解析 OI 解码信息""" + if data is None: + return None + return OIDecodingInfo( + oi_former_word_num=data.get("oi_former_word_num", 0), + oi_latter_word_num=data.get("oi_latter_word_num", 0), + oi_words=data.get("oi_words"), + ) + + +def _parse_alternative(data: dict) -> ASRAlternative: + """解析候选结果""" + words = [_parse_word(w) for w in data.get("words", [])] + return ASRAlternative( + text=data.get("text", ""), + start_time=data.get("start_time", 0.0), + end_time=data.get("end_time", 0.0), + words=words, + semantic_related_to_prev=data.get("semantic_related_to_prev"), + oi_decoding_info=_parse_oi_decoding_info(data.get("oi_decoding_info")), + ) + + +def _parse_result(data: dict) -> ASRResult: + """解析单条识别结果""" + alternatives = [_parse_alternative(a) for a in data.get("alternatives", [])] + return ASRResult( + text=data.get("text", ""), + start_time=data.get("start_time", 0.0), + end_time=data.get("end_time", 0.0), + confidence=data.get("confidence", 0.0), + alternatives=alternatives, + is_interim=data.get("is_interim", True), + is_vad_finished=data.get("is_vad_finished", False), + index=data.get("index", 0), + ) + + +def _parse_extra(data: dict) -> ASRExtra: + """解析附加信息""" + return ASRExtra( + audio_duration=data.get("audio_duration"), + model_avg_rtf=data.get("model_avg_rtf"), + model_send_first_response=data.get("model_send_first_response"), + speech_adaptation_version=data.get("speech_adaptation_version"), + model_total_process_time=data.get("model_total_process_time"), + packet_number=data.get("packet_number"), + vad_start=data.get("vad_start"), + req_payload=data.get("req_payload"), + ) + + def _parse_response(data: bytes) -> ASRResponse: """解析 ASR 响应 (使用 protobuf)""" pb = AsrResponsePb() @@ -530,20 +646,33 @@ def _parse_response(data: bytes) -> ASRResponse: except json.JSONDecodeError: return ASRResponse(type=ResponseType.UNKNOWN) - results = json_data.get("results") - extra = json_data.get("extra", {}) + results_raw = json_data.get("results") + extra_raw = json_data.get("extra", {}) + + # 解析为强类型 + parsed_extra = _parse_extra(extra_raw) # 无 results，可能是心跳包 - if results is None: + if results_raw is None: return ASRResponse( type=ResponseType.HEARTBEAT, - packet_number=extra.get("packet_number", -1), + packet_number=extra_raw.get("packet_number", -1), raw_json=json_data, + extra=parsed_extra, ) + # 解析 results + parsed_results = [_parse_result(r) for r in results_raw] + # VAD 开始 - if extra.get("vad_start"): - return ASRResponse(type=ResponseType.VAD_START, vad_start=True, raw_json=json_data) + if extra_raw.get("vad_start"): + return ASRResponse( + type=ResponseType.VAD_START, + vad_start=True, + raw_json=json_data, + results=parsed_results, + extra=parsed_extra, + ) # 解析识别结果 text = "" @@ -551,7 +680,7 @@ def _parse_response(data: bytes) -> ASRResponse: vad_finished = False nonstream_result = False - for r in results: + for r in results_raw: if r.get("text"): text = r.get("text") if r.get("is_interim") is False: @@ -569,6 +698,8 @@ def _parse_response(data: bytes) -> ASRResponse: is_final=True, vad_finished=vad_finished, raw_json=json_data, + results=parsed_results, + extra=parsed_extra, ) # 中间结果 @@ -577,6 +708,8 @@ def _parse_response(data: bytes) -> ASRResponse: text=text, is_final=False, raw_json=json_data, + results=parsed_results, + extra=parsed_extra, ) @@ -653,4 +786,4 @@ async def transcribe_realtime( """ async with DoubaoASR(config) as asr: async for resp in asr.transcribe_realtime(audio_source): - yield resp + yield resp \ No newline at end of file diff --git a/custom_components/hass_stt_doubao/doubaoime_asr/audio.py b/custom_components/hass_stt_doubao/doubaoime_asr/audio.py index d6650f5..9361784 100644 --- a/custom_components/hass_stt_doubao/doubaoime_asr/audio.py +++ b/custom_components/hass_stt_doubao/doubaoime_asr/audio.py @@ -1,13 +1,11 @@ -from typing import Optional, List, Union, TYPE_CHECKING +from typing import Optional, List, Union from pathlib import Path -import asyncio +import opuslib import miniaudio from .config import ASRConfig -if TYPE_CHECKING: - import opuslib class AudioEncoder: @@ -16,36 +14,19 @@ class AudioEncoder: """ def __init__(self, config: ASRConfig) -> None: self.config = config - self._encoder: Optional["opuslib.Encoder"] = None - self._opuslib_module = None + self._encoder: Optional[opuslib.Encoder] = None - async def _ensure_opuslib(self): - """在 executor 中导入 opuslib，避免阻塞事件循环""" - if self._opuslib_module is None: - loop = asyncio.get_event_loop() - self._opuslib_module = await loop.run_in_executor(None, self._import_opuslib) - - @staticmethod - def _import_opuslib(): - """在独立线程中导入 opuslib""" - import opuslib - return opuslib - - async def get_encoder(self) -> "opuslib.Encoder": - """异步获取编码器""" + @property + def encoder(self) -> opuslib.Encoder: if self._encoder is None: - await self._ensure_opuslib() - self._encoder = self._opuslib_module.Encoder( + self._encoder = opuslib.Encoder( self.config.sample_rate, self.config.channels, - self._opuslib_module.APPLICATION_AUDIO, + opuslib.APPLICATION_AUDIO, ) return self._encoder - async def pcm_to_opus_frames(self, pcm_data: bytes) -> List[bytes]: - """将 PCM 数据转换为 Opus 帧（异步）""" - encoder = await self.get_encoder() - + def pcm_to_opus_frames(self, pcm_data: bytes) -> List[bytes]: samples_per_frame = ( self.config.sample_rate * self.config.frame_duration_ms // 1000 ) @@ -57,7 +38,7 @@ class AudioEncoder: if len(chunk) < bytes_per_frame: chunk = chunk + b"\x00" * (bytes_per_frame - len(chunk)) - opus_frame = encoder.encode(chunk, samples_per_frame) + opus_frame = self.encoder.encode(chunk, samples_per_frame) frames.append(opus_frame) return frames diff --git a/custom_components/hass_stt_doubao/doubaoime_asr/config.py b/custom_components/hass_stt_doubao/doubaoime_asr/config.py index 2134d53..1f9f083 100644 --- a/custom_components/hass_stt_doubao/doubaoime_asr/config.py +++ b/custom_components/hass_stt_doubao/doubaoime_asr/config.py @@ -1,13 +1,34 @@ -from dataclasses import dataclass, field import asyncio +import base64 +from dataclasses import dataclass, field import json from pathlib import Path +import time from typing import Optional, Union from pydantic import BaseModel -import aiofiles from .constants import WEBSOCKET_URL, USER_AGENT, AID from .device import DeviceCredentials, register_device, get_asr_token +from .sami import get_sami_token + + +def _jwt_is_expired(token: str, margin: int = 60) -> bool: + """ + 检查 JWT token 是否已过期（提前 margin 秒视为过期） + """ + try: + payload_b64 = token.split(".")[1] + # JWT base64url 需要补齐 padding + padding = 4 - len(payload_b64) % 4 + if padding != 4: + payload_b64 += "=" * padding + payload = json.loads(base64.urlsafe_b64decode(payload_b64)) + exp = payload.get("exp") + if exp is None: + return False + return time.time() >= exp - margin + except (IndexError, ValueError, json.JSONDecodeError): + return False class _AudioInfo(BaseModel): @@ -92,10 +113,11 @@ class ASRConfig: # 内部状态 _credentials: Optional[DeviceCredentials] = field(default=None, repr=None) _initialized: bool = field(default=False, repr=False) + _wave_client: Optional[object] = field(default=None, repr=False) - async def _load_credentials_from_file(self) -> Optional[DeviceCredentials]: + def _load_credentials_from_file_sync(self) -> Optional[DeviceCredentials]: """ - 从缓存文件中加载凭据信息（异步） + 同步的文件读取实现（在线程中调用） """ if self.credential_path is None: return None @@ -105,34 +127,88 @@ class ASRConfig: return None try: - async with aiofiles.open(path, 'r', encoding='utf-8') as f: - content = await f.read() - data = json.loads(content) + with open(path, 'r', encoding='utf-8') as f: + data = json.loads(f.read()) return DeviceCredentials(**data) except (json.JSONDecodeError, OSError): return None - async def _save_credentials_to_file(self, creds: DeviceCredentials): + def _save_credentials_to_file_sync(self, creds: DeviceCredentials): """ - 保存凭据至缓存文件（异步） + 同步的文件写入实现（在线程中调用） """ if self.credential_path is None: return path = Path(self.credential_path).expanduser() - - # 使用 run_in_executor 来执行同步的 mkdir 操作 - loop = asyncio.get_event_loop() - await loop.run_in_executor(None, lambda: path.parent.mkdir(parents=True, exist_ok=True)) + path.parent.mkdir(parents=True, exist_ok=True) - async with aiofiles.open(path, 'w', encoding='utf-8') as f: - content = json.dumps(creds.model_dump(), indent=2, ensure_ascii=False) - await f.write(content) + with open(path, 'w', encoding='utf-8') as f: + json.dump(creds.model_dump(), f, indent=2, ensure_ascii=False) + + async def _load_credentials_from_file(self) -> Optional[DeviceCredentials]: + """ + 从缓存文件中加载凭据信息（异步，不阻塞事件循环） + """ + return await asyncio.to_thread(self._load_credentials_from_file_sync) + + async def _save_credentials_to_file(self, creds: DeviceCredentials): + """ + 保存凭据至缓存文件（异步，不阻塞事件循环） + """ + await asyncio.to_thread(self._save_credentials_to_file_sync, creds) + + def ensure_credentials(self): + """ + 确保凭据已初始化（同步版本，仅用于非 async 上下文） + """ + if self._initialized: + return + + # 保存直接通过参数传入的凭据，用于进行覆盖 + user_device_id = self.device_id + user_token = self.token + + # 尝试从文件中加载凭据 + file_creds = self._load_credentials_from_file_sync() + if file_creds: + self._credentials = file_creds + # 使用文件中的值作为默认 + if self.device_id is None: + self.device_id = file_creds.device_id + if self.token is None: + self.token = file_creds.token + + # 如果 device_id 仍为 None, 则注册设备 + need_save = False + if self.device_id is None: + self._credentials = register_device() + self.device_id = self._credentials.device_id + need_save = True + + # 如果 token 仍为 None, 则获取 token + if self.token is None: + cdid = self._credentials.cdid if self._credentials else None + self.token = get_asr_token(self.device_id, cdid) + + # 如果指定了 credential_path 且有新注册的凭据，则保存至文件 + if self.credential_path and need_save and self._credentials: + self._credentials.token = self.token + self._save_credentials_to_file_sync(self._credentials) + + # 覆盖用户传入的参数 + if user_device_id is not None: + self.device_id = user_device_id + + if user_token is not None: + self.token = user_token + + self._initialized = True async def async_ensure_credentials(self): """ - 确保凭据已初始化（异步） + 确保凭据已初始化（异步版本，不阻塞事件循环） 优先级： 1. 直接传入的 device_id/token 参数（最高优先级） @@ -148,7 +224,7 @@ class ASRConfig: user_device_id = self.device_id user_token = self.token - # 尝试从文件中加载凭据 + # 尝试从文件中加载凭据（异步） file_creds = await self._load_credentials_from_file() if file_creds: self._credentials = file_creds @@ -158,23 +234,19 @@ class ASRConfig: if self.token is None: self.token = file_creds.token - # 如果 device_id 仍为 None, 则注册设备 + # 如果 device_id 仍为 None, 则注册设备（网络 I/O 在线程中执行） need_save = False if self.device_id is None: - # 在 executor 中运行同步的 register_device - loop = asyncio.get_event_loop() - self._credentials = await loop.run_in_executor(None, register_device) + self._credentials = await asyncio.to_thread(register_device) self.device_id = self._credentials.device_id need_save = True - # 如果 token 仍为 None, 则获取 token + # 如果 token 仍为 None, 则获取 token（网络 I/O 在线程中执行） if self.token is None: cdid = self._credentials.cdid if self._credentials else None - # 在 executor 中运行同步的 get_asr_token - loop = asyncio.get_event_loop() - self.token = await loop.run_in_executor(None, get_asr_token, self.device_id, cdid) + self.token = await asyncio.to_thread(get_asr_token, self.device_id, cdid) - # 如果指定了 credential_path 且有新注册的凭据，则保存至文件 + # 如果指定了 credential_path 且有新注册的凭据，则保存至文件（异步） if self.credential_path and need_save and self._credentials: self._credentials.token = self.token await self._save_credentials_to_file(self._credentials) @@ -188,8 +260,18 @@ class ASRConfig: self._initialized = True - async def get_ws_url(self) -> str: - """获取 WebSocket URL（异步）""" + @property + def ws_url(self) -> str: + """ + 获取 WebSocket URL（同步版本，需确保凭据已初始化） + """ + self.ensure_credentials() + return f'{self.url}?aid={self.aid}&device_id={self.device_id}' + + async def async_ws_url(self) -> str: + """ + 获取 WebSocket URL（异步版本，不阻塞事件循环） + """ await self.async_ensure_credentials() return f'{self.url}?aid={self.aid}&device_id={self.device_id}' @@ -201,9 +283,24 @@ class ASRConfig: "x-custom-keepalive": "true" } - async def get_session_config(self) -> SessionConfig: - """获取会话配置（异步）""" + def session_config(self) -> SessionConfig: + """ + 获取会话配置（同步版本，需确保凭据已初始化） + """ + self.ensure_credentials() + return self._build_session_config() + + async def async_session_config(self) -> SessionConfig: + """ + 获取会话配置（异步版本，不阻塞事件循环） + """ await self.async_ensure_credentials() + return self._build_session_config() + + def _build_session_config(self) -> SessionConfig: + """ + 构建会话配置（内部方法，需确保凭据已初始化后调用） + """ audio_info = _AudioInfo( channel=self.channels, format="speech_opus", @@ -225,7 +322,114 @@ class ASRConfig: extra=extra, ) - async def get_token(self) -> str: - """获取 token（异步）""" + def get_token(self) -> str: + """ + 获取 token（同步版本，需确保凭据已初始化） + """ + self.ensure_credentials() + return self.token + + async def async_get_token(self) -> str: + """ + 获取 token（异步版本，不阻塞事件循环） + """ await self.async_ensure_credentials() - return self.token \ No newline at end of file + return self.token + + def _on_wave_session_update(self, session) -> None: + """ + Wave 会话更新时的回调，将新会话同步到凭据缓存 + """ + if self._credentials: + self._credentials.wave_session = session.to_dict() + self._save_credentials_to_file_sync(self._credentials) + + def get_wave_client(self): + """ + 获取 WaveClient 实例（同步版本，懒加载，自动管理握手） + """ + from .wave_client import WaveClient, WaveSession + + self.ensure_credentials() + if self._wave_client is None: + cached_session = None + if self._credentials and self._credentials.wave_session: + try: + session = WaveSession.from_dict(self._credentials.wave_session) + if not session.is_expired(): + cached_session = session + except (KeyError, ValueError): + pass + + self._wave_client = WaveClient( + self.device_id, + self.aid, + session=cached_session, + on_session_update=self._on_wave_session_update, + ) + return self._wave_client + + async def async_get_wave_client(self): + """ + 获取 WaveClient 实例（异步版本，不阻塞事件循环） + """ + from .wave_client import WaveClient, WaveSession + + await self.async_ensure_credentials() + if self._wave_client is None: + cached_session = None + if self._credentials and self._credentials.wave_session: + try: + session = WaveSession.from_dict(self._credentials.wave_session) + if not session.is_expired(): + cached_session = session + except (KeyError, ValueError): + pass + + self._wave_client = WaveClient( + self.device_id, + self.aid, + session=cached_session, + on_session_update=self._on_wave_session_update, + ) + return self._wave_client + + def get_sami_token(self) -> str: + """ + 获取 SAMI token（同步版本，需确保凭据已初始化） + """ + self.ensure_credentials() + + if (self._credentials + and self._credentials.sami_token + and not _jwt_is_expired(self._credentials.sami_token)): + return self._credentials.sami_token + + cdid = self._credentials.cdid if self._credentials else None + sami_token = get_sami_token(cdid) + + if self._credentials: + self._credentials.sami_token = sami_token + self._save_credentials_to_file_sync(self._credentials) + + return sami_token + + async def async_get_sami_token(self) -> str: + """ + 获取 SAMI token（异步版本，不阻塞事件循环） + """ + await self.async_ensure_credentials() + + if (self._credentials + and self._credentials.sami_token + and not _jwt_is_expired(self._credentials.sami_token)): + return self._credentials.sami_token + + cdid = self._credentials.cdid if self._credentials else None + sami_token = await asyncio.to_thread(get_sami_token, cdid) + + if self._credentials: + self._credentials.sami_token = sami_token + await self._save_credentials_to_file(self._credentials) + + return sami_token diff --git a/custom_components/hass_stt_doubao/doubaoime_asr/constants.py b/custom_components/hass_stt_doubao/doubaoime_asr/constants.py index 5dea4ba..fc855f1 100644 --- a/custom_components/hass_stt_doubao/doubaoime_asr/constants.py +++ b/custom_components/hass_stt_doubao/doubaoime_asr/constants.py @@ -7,9 +7,24 @@ SETTINGS_URL = "https://is.snssdk.com/service/settings/v3/" # ASR WebSocket URL WEBSOCKET_URL = "wss://frontier-audio-ime-ws.doubao.com/ocean/api/v1/ws" +# 需要通过该接口获取 SAMI TOKEN 来访问 NER 接口 +SAMI_CONFIG_URL = "https://ime.oceancloudapi.com/api/v1/user/get_config" + +# 加密协议握手 URL +HANDSHAKE_URL = "https://keyhub.zijieapi.com/handshake" + +# NER 接口 URL +NER_URL = "https://speech.bytedance.com/api/v3/context/ime/ner" + # 豆包输入法的 APP ID AID = 401734 +# SAMI 语音相关的服务 APP KEY (写死在客户端的) +SAMI_APP_KEY = "SYlxZr6LnvBaIVmF" + +# HKDF info 字符串 (写死在客户端的) +HKDF_INFO = b"4e30514609050cd3" + # 应用配置 (豆包输入法) APP_CONFIG = { "aid": AID, diff --git a/custom_components/hass_stt_doubao/doubaoime_asr/device.py b/custom_components/hass_stt_doubao/doubaoime_asr/device.py index ac2446f..22962c9 100644 --- a/custom_components/hass_stt_doubao/doubaoime_asr/device.py +++ b/custom_components/hass_stt_doubao/doubaoime_asr/device.py @@ -29,6 +29,14 @@ class DeviceCredentials(BaseModel): """ 用于 ASR 的 token """ + sami_token: Optional[str] = None + """ + 用于 NER 等 SAMI 服务的 token + """ + wave_session: Optional[dict] = None + """ + Wave 加密会话缓存（序列化后的 WaveSession） + """ class DeviceRegisterHeaderField(BaseModel): @@ -273,7 +281,6 @@ def register_device() -> DeviceCredentials: params=params.model_dump(), json=body.model_dump(), headers=headers, - verify=False, # 禁用SSL证书验证以兼容某些环境 ) response.raise_for_status() @@ -311,7 +318,6 @@ def get_asr_token(device_id: str, cdid: str) -> str: params=params, data=body_str, headers=headers, - verify=False, # 禁用SSL证书验证以兼容某些环境 ) response.raise_for_status() diff --git a/custom_components/hass_stt_doubao/doubaoime_asr/ner.py b/custom_components/hass_stt_doubao/doubaoime_asr/ner.py new file mode 100644 index 0000000..d5d4481 --- /dev/null +++ b/custom_components/hass_stt_doubao/doubaoime_asr/ner.py @@ -0,0 +1,126 @@ +from __future__ import annotations + +import base64 +import json +import uuid +from typing import TYPE_CHECKING, List + +import requests +from pydantic import BaseModel, Field + +from .wave_client import WaveClient +from .constants import AID, APP_CONFIG, NER_URL, SAMI_APP_KEY + +if TYPE_CHECKING: + from .config import ASRConfig + + +class NerUserInfo(BaseModel): + uid: str = "0" + did: str + app_name: str + app_version: str + sdk_version: str = "" + platform: str = "android" + experience_improve: bool = False + + @classmethod + def new(cls, did: str, app_name: str): + app_version = APP_CONFIG.get("version_name", "") + return cls(did=did, app_name=app_name, app_version=app_version) + + +class NerRequest(BaseModel): + """ + ner 接口请求体 + """ + user: NerUserInfo + text: str + additions: dict = Field(default_factory=dict) + + @classmethod + def new(cls, text: str, did: str, app_name: str = "", addiction: dict = None): + return cls(user=NerUserInfo.new(did, app_name), text=text, additions=addiction or {}) + + +class NerWord(BaseModel): + freq: int + word: str + + +class NerResult(BaseModel): + text: str + words: List[NerWord] + + +class NerResponse(BaseModel): + """ + ner 接口响应体 + """ + results: List[NerResult] + + +def get_ner_results(wave_client: WaveClient, sami_token: str, text: str, did: str, app_name: str = "") -> NerResponse: + """ + 调用 ner 接口获取结果 + """ + request = NerRequest.new(text, did, app_name) + + headers = { + 'app_version': APP_CONFIG.get('version_name', ''), + 'app_id': str(AID), + 'os_type': 'android', + 'x-api-resource-id': 'asr.user.context', + 'x-api-app-key': SAMI_APP_KEY, + 'x-api-token': sami_token, + 'x-api-request-id': str(uuid.uuid4()), + } + req_data = request.model_dump_json().encode() + + payload, headers = wave_client.prepare_request(req_data, headers) + + response = requests.post(NER_URL, data=payload, headers=headers) + + resp_headers = response.headers + nonce = base64.b64decode(resp_headers.get('x-tt-e-p')) + + decoded = wave_client.decrypt(response.content, nonce=nonce) + + return NerResponse(**json.loads(decoded.decode())) + + +def ner(config: ASRConfig, text: str, app_name: str = "") -> NerResponse: + """ + 通过 ASRConfig 调用 NER 接口（同步便捷函数） + + 自动管理 WaveClient、sami_token 和 device_id。 + + :param config: ASR 配置（会自动初始化凭据） + :param text: 需要进行 NER 的文本 + :param app_name: 应用名称（可选），可能会根据不同应用的使用场景适配不同的识别策略？" + :return: NER 响应 + """ + config.ensure_credentials() + wave_client = config.get_wave_client() + sami_token = config.get_sami_token() + return get_ner_results(wave_client, sami_token, text, config.device_id, app_name) + + +async def async_ner(config: ASRConfig, text: str, app_name: str = "") -> NerResponse: + """ + 通过 ASRConfig 调用 NER 接口（异步便捷函数，不阻塞事件循环） + + 自动管理 WaveClient、sami_token 和 device_id。 + + :param config: ASR 配置（会自动初始化凭据） + :param text: 需要进行 NER 的文本 + :param app_name: 应用名称（可选） + :return: NER 响应 + """ + import asyncio + await config.async_ensure_credentials() + wave_client = await config.async_get_wave_client() + sami_token = await config.async_get_sami_token() + return await asyncio.to_thread( + get_ner_results, wave_client, sami_token, text, config.device_id, app_name + ) diff --git a/custom_components/hass_stt_doubao/doubaoime_asr/sami.py b/custom_components/hass_stt_doubao/doubaoime_asr/sami.py new file mode 100644 index 0000000..b1bfd9b --- /dev/null +++ b/custom_components/hass_stt_doubao/doubaoime_asr/sami.py @@ -0,0 +1,137 @@ +""" +SAMI 服务相关 +""" + +import hashlib +import time +import uuid +from typing import Optional + +import requests +from pydantic import BaseModel, ConfigDict, Field + +from .constants import SAMI_CONFIG_URL, SAMI_APP_KEY, USER_AGENT, APP_CONFIG, DEFAULT_DEVICE_CONFIG + + +class _SamiConfigParams(BaseModel): + """ + SAMI 配置接口的 URL Params + """ + model_config = ConfigDict(populate_by_name=True, serialize_by_alias=True) + + device_platform: str = "android" + os: str = "android" + ssmix: str = "a" + rticket: str = Field(default_factory=lambda: str(int(time.time() * 1000)), serialization_alias="_rticket") + cdid: str + channel: str + aid: str + app_name: str + version_code: str + version_name: str + manifest_version_code: str + update_version_code: str + resolution: str + dpi: str + device_type: str + device_brand: str + language: str + os_api: str + os_version: str + ac: str = "wifi" + use_olympus_account: str = Field(default="1", serialization_alias="use-olympus-account") + + @classmethod + def default(cls, cdid: str) -> "_SamiConfigParams": + """ + 使用默认配置构建 SAMI 配置 Params + """ + app_config = { + **{k: APP_CONFIG[k] for k in ("channel", "app_name", "version_name")}, + "aid": str(APP_CONFIG["aid"]), + "version_code": str(APP_CONFIG["version_code"]), + "manifest_version_code": str(APP_CONFIG["manifest_version_code"]), + "update_version_code": str(APP_CONFIG["update_version_code"]), + } + + device_keys = ("device_platform", "os", "resolution", "dpi", "device_type", + "device_brand", "language", "os_api", "os_version") + device_config = {k: DEFAULT_DEVICE_CONFIG[k] for k in device_keys} + + return cls(cdid=cdid, **app_config, **device_config) + + +class _SamiConfigRequest(BaseModel): + """ + SAMI 配置接口的请求体 + """ + sami_app_key: str = SAMI_APP_KEY + + +class _SamiConfigData(BaseModel): + """SAMI 配置数据""" + sami_token: str + + +class _SamiConfigResponse(BaseModel): + """ + SAMI 配置接口的响应 + """ + code: int + msg: str + data: _SamiConfigData + + @property + def sami_token(self) -> str: + return self.data.sami_token + + +def get_sami_config(cdid: str) -> requests.Response: + """ + 获取 SAMI 配置 (包含 token) + + Args: + cdid: 客户端设备 ID + + Returns: + 响应对象 + """ + params = _SamiConfigParams.default(cdid) + body = _SamiConfigRequest() + body_json = body.model_dump_json() + x_ss_stub = hashlib.md5(body_json.encode()).hexdigest().upper() + + headers = { + "User-Agent": USER_AGENT, + "Content-Type": "application/json", + "app_version": APP_CONFIG["version_name"], + "app_id": str(APP_CONFIG["aid"]), + "os_type": "Android", + "x-ss-stub": x_ss_stub, + } + + response = requests.post( + SAMI_CONFIG_URL, + params=params.model_dump(by_alias=True), + data=body_json, + headers=headers, + ) + + return response + + +def get_sami_token(cdid: Optional[str] = None) -> str: + """ + 获取 SAMI token + + :param cdid: 客户端设备 ID，如果为 None 则自动生成 + :return: SAMI token 字符串 + """ + if cdid is None: + cdid = str(uuid.uuid4()) + + response = get_sami_config(cdid) + response.raise_for_status() + + data = _SamiConfigResponse(**response.json()) + return data.sami_token diff --git a/custom_components/hass_stt_doubao/doubaoime_asr/wave_client.py b/custom_components/hass_stt_doubao/doubaoime_asr/wave_client.py new file mode 100644 index 0000000..6da2556 --- /dev/null +++ b/custom_components/hass_stt_doubao/doubaoime_asr/wave_client.py @@ -0,0 +1,259 @@ +""" +ByteDance Wave 加密协议客户端 +""" + +import base64 +import hashlib +import secrets +import time +from typing import Callable, Optional, Union + +import requests +from cryptography.hazmat.backends import default_backend +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import ec +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms +from cryptography.hazmat.primitives.kdf.hkdf import HKDF +from pydantic import BaseModel + +from .constants import HANDSHAKE_URL, HKDF_INFO, USER_AGENT + + +class _KeyShare(BaseModel): + """ + 密钥交换信息 + """ + curve: str + pubkey: str + + +class _HandshakeRequest(BaseModel): + """ + 握手请求 + """ + version: int = 2 + random: str + app_id: str + did: str + key_shares: list[_KeyShare] + cipher_suites: list[int] = [4097] # ChaCha20 + + +class _HandshakeResponse(BaseModel): + """ + 握手响应 + """ + version: int + random: str + key_share: _KeyShare + cipher_suite: int + cert: str + ticket: str + ticket_exp: int + ticket_long: str + ticket_long_exp: int + + +class WaveSession(BaseModel): + """ + Wave 加密会话 + """ + ticket: str + ticket_long: str + encryption_key: bytes + client_random: bytes + server_random: bytes + shared_key: bytes + ticket_exp: int + ticket_long_exp: int + expires_at: float + + class Config: + arbitrary_types_allowed = True + + def is_expired(self) -> bool: + """检查会话是否已过期""" + return time.time() >= self.expires_at + + def to_dict(self) -> dict: + """序列化为可 JSON 存储的 dict（bytes 字段用 base64 编码）""" + return { + "ticket": self.ticket, + "ticket_long": self.ticket_long, + "encryption_key": base64.b64encode(self.encryption_key).decode(), + "client_random": base64.b64encode(self.client_random).decode(), + "server_random": base64.b64encode(self.server_random).decode(), + "shared_key": base64.b64encode(self.shared_key).decode(), + "ticket_exp": self.ticket_exp, + "ticket_long_exp": self.ticket_long_exp, + "expires_at": self.expires_at, + } + + @classmethod + def from_dict(cls, data: dict) -> "WaveSession": + """从 dict 反序列化（base64 解码 bytes 字段）""" + return cls( + ticket=data["ticket"], + ticket_long=data["ticket_long"], + encryption_key=base64.b64decode(data["encryption_key"]), + client_random=base64.b64decode(data["client_random"]), + server_random=base64.b64decode(data["server_random"]), + shared_key=base64.b64decode(data["shared_key"]), + ticket_exp=data["ticket_exp"], + ticket_long_exp=data["ticket_long_exp"], + expires_at=data["expires_at"], + ) + + +class WaveClient: + """ + Wave 协议客户端 + """ + + def __init__( + self, + device_id: str, + app_id: Union[str, int], + session: Optional[WaveSession] = None, + on_session_update: Optional[Callable[[WaveSession], None]] = None, + ): + self.device_id = device_id + self.app_id = str(app_id) + self.session = session + self._on_session_update = on_session_update + + @staticmethod + def _chacha20_crypt(key: bytes, nonce: bytes, data: bytes) -> bytes: + """ChaCha20 加密/解密""" + if len(nonce) == 12: + nonce_16 = b'\x00\x00\x00\x00' + nonce + else: + nonce_16 = nonce + cipher = Cipher(algorithms.ChaCha20(key, nonce_16), mode=None, backend=default_backend()) + encryptor = cipher.encryptor() + return encryptor.update(data) + encryptor.finalize() + + @staticmethod + def _derive_key(shared_key: bytes, salt: bytes, info: bytes) -> bytes: + """HKDF 密钥派生""" + return HKDF( + algorithm=hashes.SHA256(), + length=32, + salt=salt, + info=info, + backend=default_backend() + ).derive(shared_key) + + def handshake(self) -> bool: + """ + 执行 Wave 握手，建立加密会话 + + :return: 握手是否成功 + """ + # 生成 ECDH 密钥对 + private_key = ec.generate_private_key(ec.SECP256R1(), default_backend()) + client_random = secrets.token_bytes(32) + + pubkey_bytes = private_key.public_key().public_bytes( + encoding=serialization.Encoding.X962, + format=serialization.PublicFormat.UncompressedPoint + ) + + request = _HandshakeRequest( + random=base64.b64encode(client_random).decode(), + app_id=self.app_id, + did=self.device_id, + key_shares=[_KeyShare(curve="secp256r1", pubkey=base64.b64encode(pubkey_bytes).decode())], + ) + + request_json = request.model_dump_json(by_alias=True) + + # ECDSA 签名 + signature = private_key.sign(request_json.encode(), ec.ECDSA(hashes.SHA256())) + + headers = { + "Content-Type": "application/json", + "x-tt-s-sign": base64.b64encode(signature).decode(), + "User-Agent": USER_AGENT, + } + + response = requests.post(HANDSHAKE_URL, data=request_json, headers=headers) + + if response.status_code != 200: + return False + + resp = _HandshakeResponse(**response.json()) + + # 计算共享密钥 + server_pubkey = base64.b64decode(resp.key_share.pubkey) + server_public_key = ec.EllipticCurvePublicKey.from_encoded_point(ec.SECP256R1(), server_pubkey) + shared_key = private_key.exchange(ec.ECDH(), server_public_key) + server_random = base64.b64decode(resp.random) + + # 派生加密密钥 + salt = client_random + server_random + encryption_key = self._derive_key(shared_key, salt, HKDF_INFO) + + # 保存会话，计算绝对过期时间（提前 60 秒刷新） + self.session = WaveSession( + ticket=resp.ticket, + ticket_long=resp.ticket_long, + encryption_key=encryption_key, + client_random=client_random, + server_random=server_random, + shared_key=shared_key, + ticket_exp=resp.ticket_exp, + ticket_long_exp=resp.ticket_long_exp, + expires_at=time.time() + resp.ticket_exp - 60, + ) + + if self._on_session_update: + self._on_session_update(self.session) + + return True + + def _ensure_session(self) -> None: + """确保会话有效，如果过期则自动刷新""" + if self.session is None or self.session.is_expired(): + if not self.handshake(): + raise RuntimeError("Failed to establish/refresh session") + + def prepare_request(self, plaintext: bytes, extra_headers: Optional[dict] = None) -> tuple[bytes, dict]: + """ + 准备加密请求 + + :param plaintext: 明文数据 + :param extra_headers: 额外的请求头 + :return: (密文, headers) 元组 + """ + self._ensure_session() + + nonce = secrets.token_bytes(12) + ciphertext = self._chacha20_crypt(self.session.encryption_key, nonce, plaintext) + stub = hashlib.md5(ciphertext).hexdigest().upper() + + headers = { + "Content-Type": "application/json", + "x-tt-e-b": "1", + "x-tt-e-t": self.session.ticket, + "x-tt-e-p": base64.b64encode(nonce).decode(), + "x-ss-stub": stub, + } + + if extra_headers: + headers.update(extra_headers) + + return ciphertext, headers + + def decrypt(self, ciphertext: bytes, nonce: bytes) -> bytes: + """ + 解密数据 + + :param ciphertext: 密文数据 + :param nonce: 12 字节 nonce + :return: 明文数据 + """ + if not self.session: + raise RuntimeError("No active session. Call handshake() first.") + + return self._chacha20_crypt(self.session.encryption_key, nonce, ciphertext)