From c353080470a779722e6a9ed627c55326ed718b14 Mon Sep 17 00:00:00 2001 From: jordansilly77-stack Date: Sat, 20 Jun 2026 00:02:17 +0800 Subject: [PATCH] feat: add Tencent Cloud ASR provider --- sapat/providers/tencentcloud.py | 239 +++++++++++++++++++++++++++ tests/providers/test_tencentcloud.py | 134 +++++++++++++++ 2 files changed, 373 insertions(+) create mode 100644 sapat/providers/tencentcloud.py create mode 100644 tests/providers/test_tencentcloud.py diff --git a/sapat/providers/tencentcloud.py b/sapat/providers/tencentcloud.py new file mode 100644 index 0000000..0950354 --- /dev/null +++ b/sapat/providers/tencentcloud.py @@ -0,0 +1,239 @@ +# ABOUTME: Tencent Cloud ASR transcription provider +# ABOUTME: Uses signed SentenceRecognition requests for short local audio files + +import base64 +import datetime as dt +import hashlib +import hmac +import json +import os +import time +from typing import Dict, Optional + +import requests + +from sapat.providers import register +from sapat.providers.base import ( + AudioFormat, + ProviderConfig, + TranscriptionProvider, + TranscriptionResult, +) + +SERVICE = "asr" +VERSION = "2019-06-14" +ACTION = "SentenceRecognition" +DEFAULT_ENDPOINT = "asr.tencentcloudapi.com" +LANGUAGE_MODELS: Dict[str, str] = { + "zh": "16k_zh", + "zh-cn": "16k_zh", + "zh_cn": "16k_zh", + "cmn": "16k_zh", + "en": "16k_en", + "en-us": "16k_en", + "en_us": "16k_en", + "ja": "16k_ja", + "ko": "16k_ko", + "yue": "16k_yue", + "vi": "16k_vi", + "ms": "16k_ms", + "id": "16k_id", + "fil": "16k_fil", + "th": "16k_th", + "pt": "16k_pt", + "tr": "16k_tr", + "ar": "16k_ar", + "es": "16k_es", + "hi": "16k_hi", + "fr": "16k_fr", + "de": "16k_de", +} + + +@register +class TencentCloudProvider(TranscriptionProvider): + name = "tencentcloud" + config = ProviderConfig( + required_env_vars=["TENCENTCLOUD_SECRET_ID", "TENCENTCLOUD_SECRET_KEY"], + max_file_size_mb=3.0, + preferred_format=AudioFormat.MP3, + supports_correction=False, + default_model="16k_zh", + ) + + def transcribe( + self, + audio_file: str, + model: str, + language: str = "en", + prompt: Optional[str] = None, + temperature: float = 0, + **kwargs, + ) -> TranscriptionResult: + with open(audio_file, "rb") as f: + audio_data = f.read() + + voice_format = kwargs.get("voice_format") or self._detect_voice_format( + audio_file + ) + payload = { + "EngSerViceType": self.resolve_model( + model or self._model_for_language(language) + ), + "SourceType": 1, + "VoiceFormat": voice_format, + "ProjectId": 0, + "SubServiceType": 2, + "UsrAudioKey": kwargs.get("usr_audio_key", "sapat-local-audio"), + "Data": base64.b64encode(audio_data).decode("utf-8"), + "DataLen": len(audio_data), + } + + for key in ( + "WordInfo", + "FilterDirty", + "FilterModal", + "FilterPunc", + "ConvertNumMode", + "HotwordId", + "CustomizationId", + "ReinforceHotword", + "HotwordList", + "InputSampleRate", + "ReplaceTextId", + ): + value = kwargs.get(self._snake_case(key)) + if value is not None: + payload[key] = value + + response = requests.post( + self._endpoint_url(), + headers=self._build_headers(payload), + data=json.dumps(payload, separators=(",", ":")), + timeout=60, + ) + + result = response.json() + if response.status_code == 200 and "Response" in result: + inner = result["Response"] + if "Error" not in inner: + return TranscriptionResult( + text=inner.get("Result", ""), + duration=self._duration_seconds(inner.get("AudioDuration")), + segments=inner.get("WordList"), + raw_response=result, + ) + error = inner["Error"] + message = error.get("Message") or error.get("Code") or response.text + else: + message = response.text + + raise RuntimeError(f"Tencent Cloud ASR transcription failed: {message}") + + def resolve_model(self, model_alias: str) -> str: + aliases = { + "default": "16k_zh", + "zh": "16k_zh", + "mandarin": "16k_zh", + "en": "16k_en", + "english": "16k_en", + "cantonese": "16k_yue", + "yue": "16k_yue", + "multi": "16k_zh-PY", + } + return aliases.get((model_alias or "").lower(), model_alias) + + def _build_headers(self, payload: dict) -> dict: + secret_id = os.getenv("TENCENTCLOUD_SECRET_ID", "") + secret_key = os.getenv("TENCENTCLOUD_SECRET_KEY", "") + region = os.getenv("TENCENTCLOUD_REGION", "ap-guangzhou") + host = self._host() + timestamp = int(os.getenv("TENCENTCLOUD_TIMESTAMP", str(int(time.time())))) + date = dt.datetime.fromtimestamp(timestamp, dt.UTC).strftime("%Y-%m-%d") + body = json.dumps(payload, separators=(",", ":")) + hashed_payload = hashlib.sha256(body.encode("utf-8")).hexdigest() + + canonical_request = "\n".join( + [ + "POST", + "/", + "", + f"content-type:application/json; charset=utf-8\nhost:{host}\n", + "content-type;host", + hashed_payload, + ] + ) + credential_scope = f"{date}/{SERVICE}/tc3_request" + string_to_sign = "\n".join( + [ + "TC3-HMAC-SHA256", + str(timestamp), + credential_scope, + hashlib.sha256(canonical_request.encode("utf-8")).hexdigest(), + ] + ) + signature = self._sign(secret_key, date, string_to_sign) + authorization = ( + "TC3-HMAC-SHA256 " + f"Credential={secret_id}/{credential_scope}, " + "SignedHeaders=content-type;host, " + f"Signature={signature}" + ) + + return { + "Authorization": authorization, + "Content-Type": "application/json; charset=utf-8", + "Host": host, + "X-TC-Action": ACTION, + "X-TC-Timestamp": str(timestamp), + "X-TC-Version": VERSION, + "X-TC-Region": region, + } + + def _sign(self, secret_key: str, date: str, string_to_sign: str) -> str: + secret_date = hmac.new( + ("TC3" + secret_key).encode("utf-8"), + date.encode("utf-8"), + hashlib.sha256, + ).digest() + secret_service = hmac.new( + secret_date, SERVICE.encode("utf-8"), hashlib.sha256 + ).digest() + secret_signing = hmac.new( + secret_service, b"tc3_request", hashlib.sha256 + ).digest() + return hmac.new( + secret_signing, string_to_sign.encode("utf-8"), hashlib.sha256 + ).hexdigest() + + def _model_for_language(self, language: Optional[str]) -> str: + if not language: + return self.config.default_model + normalized = language.strip().lower() + return LANGUAGE_MODELS.get( + normalized, + LANGUAGE_MODELS.get(normalized.split("-")[0], self.config.default_model), + ) + + def _endpoint_url(self) -> str: + return f"https://{self._host()}" + + def _host(self) -> str: + return os.getenv("TENCENTCLOUD_ASR_ENDPOINT", DEFAULT_ENDPOINT) + + def _detect_voice_format(self, audio_file: str) -> str: + suffix = os.path.splitext(audio_file)[1].lstrip(".").lower() + return suffix or self.config.preferred_format.value + + def _duration_seconds(self, duration_ms): + if duration_ms is None: + return None + return float(duration_ms) / 1000.0 + + def _snake_case(self, value: str) -> str: + chars = [] + for index, char in enumerate(value): + if char.isupper() and index: + chars.append("_") + chars.append(char.lower()) + return "".join(chars) diff --git a/tests/providers/test_tencentcloud.py b/tests/providers/test_tencentcloud.py new file mode 100644 index 0000000..812fd0b --- /dev/null +++ b/tests/providers/test_tencentcloud.py @@ -0,0 +1,134 @@ +# ABOUTME: Mock-based tests for Tencent Cloud ASR provider +# ABOUTME: Verifies signed request payloads, response parsing, and errors + +import base64 +import json +from unittest.mock import Mock, patch + +import pytest + + +@pytest.fixture +def audio_file(tmp_path): + path = tmp_path / "sample.mp3" + path.write_bytes(b"fake audio bytes") + return str(path) + + +@pytest.fixture +def tencent_env(): + return { + "TENCENTCLOUD_SECRET_ID": "test-secret-id", + "TENCENTCLOUD_SECRET_KEY": "test-secret-key", + "TENCENTCLOUD_REGION": "ap-guangzhou", + "TENCENTCLOUD_TIMESTAMP": "1700000000", + } + + +class TestTencentCloudProvider: + def test_config_values(self): + from sapat.providers.tencentcloud import TencentCloudProvider + + cfg = TencentCloudProvider.config + assert cfg.required_env_vars == [ + "TENCENTCLOUD_SECRET_ID", + "TENCENTCLOUD_SECRET_KEY", + ] + assert cfg.preferred_format.value == "mp3" + assert cfg.max_file_size_mb == 3.0 + assert cfg.default_model == "16k_zh" + + def test_is_available_when_credentials_are_set(self, tencent_env): + from sapat.providers.tencentcloud import TencentCloudProvider + + with patch.dict("os.environ", tencent_env, clear=True): + assert TencentCloudProvider.is_available() is True + + @patch("sapat.providers.tencentcloud.requests.post") + def test_transcribe_posts_signed_sentence_recognition_request( + self, mock_post, audio_file, tencent_env + ): + from sapat.providers.tencentcloud import TencentCloudProvider + + mock_post.return_value = Mock( + status_code=200, + json=lambda: { + "Response": { + "Result": "hello from tencent", + "AudioDuration": 2500, + "WordList": [{"Word": "hello", "StartTime": 0, "EndTime": 500}], + "RequestId": "req-1", + } + }, + text="ok", + ) + + with patch.dict("os.environ", tencent_env, clear=True): + provider = TencentCloudProvider() + result = provider.transcribe( + audio_file, + "english", + language="en", + word_info=1, + hotword_list="Sapat|10,Daytona|8", + ) + + assert result.text == "hello from tencent" + assert result.duration == 2.5 + assert result.segments == [{"Word": "hello", "StartTime": 0, "EndTime": 500}] + + url = mock_post.call_args.args[0] + headers = mock_post.call_args.kwargs["headers"] + payload = json.loads(mock_post.call_args.kwargs["data"]) + + assert url == "https://asr.tencentcloudapi.com" + assert headers["X-TC-Action"] == "SentenceRecognition" + assert headers["X-TC-Version"] == "2019-06-14" + assert headers["X-TC-Region"] == "ap-guangzhou" + assert headers["X-TC-Timestamp"] == "1700000000" + assert headers["Authorization"].startswith( + "TC3-HMAC-SHA256 Credential=test-secret-id/" + ) + assert payload["EngSerViceType"] == "16k_en" + assert payload["SourceType"] == 1 + assert payload["VoiceFormat"] == "mp3" + assert payload["ProjectId"] == 0 + assert payload["SubServiceType"] == 2 + assert payload["Data"] == base64.b64encode(b"fake audio bytes").decode("utf-8") + assert payload["DataLen"] == len(b"fake audio bytes") + assert payload["WordInfo"] == 1 + assert payload["HotwordList"] == "Sapat|10,Daytona|8" + + @patch("sapat.providers.tencentcloud.requests.post") + def test_transcribe_raises_on_tencent_error( + self, mock_post, audio_file, tencent_env + ): + from sapat.providers.tencentcloud import TencentCloudProvider + + mock_post.return_value = Mock( + status_code=200, + json=lambda: { + "Response": { + "Error": { + "Code": "AuthFailure.SignatureFailure", + "Message": "bad signature", + }, + "RequestId": "req-2", + } + }, + text="bad signature", + ) + + with patch.dict("os.environ", tencent_env, clear=True): + provider = TencentCloudProvider() + with pytest.raises(RuntimeError, match="bad signature"): + provider.transcribe(audio_file, "default") + + def test_resolve_model_aliases_and_language_default(self): + from sapat.providers.tencentcloud import TencentCloudProvider + + provider = TencentCloudProvider.__new__(TencentCloudProvider) + assert provider.resolve_model("english") == "16k_en" + assert provider.resolve_model("cantonese") == "16k_yue" + assert provider._model_for_language("ja") == "16k_ja" + assert provider._model_for_language("unknown") == "16k_zh"