Add Tencent Cloud ASR transcription provider

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)

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"