Merge pull request #2797 from Zmjjeff7/feat-translate-youdao

feat(translate): add Youdao as a new translation provider

config.py

@@ -123,10 +123,13 @@ available_setting = {
     "chat_start_time": "00:00",  # 服务开始时间
     "chat_stop_time": "24:00",  # 服务结束时间
     # 翻译api
-    "translate": "baidu",  # 翻译api，支持baidu
+    "translate": "baidu",  # 翻译api，支持baidu, youdao
     # baidu翻译api的配置
     "baidu_translate_app_id": "",  # 百度翻译api的appid
     "baidu_translate_app_key": "",  # 百度翻译api的秘钥
+    # youdao翻译api的配置
+    "youdao_translate_app_key": "",  # 有道翻译api的应用ID
+    "youdao_translate_app_secret": "",  # 有道翻译api的应用密钥
     # wechatmp的配置
     "wechatmp_token": "",  # 微信公众平台的Token
     "wechatmp_port": 8080,  # 微信公众平台的端口,需要端口转发到80或443

tests/test_youdao_translator.py

@@ -0,0 +1,260 @@
+# encoding:utf-8
+"""
+Unit tests for the Youdao translator integration:
+  - YoudaoTranslator class behavior (signature, language code mapping,
+    request/response handling, error handling).
+  - translate.factory.create_translator dispatch and error message.
+"""
+import os
+import sys
+import unittest
+from hashlib import sha256
+from unittest.mock import MagicMock, patch
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+
+def _mock_conf(**values):
+    """Build a callable that mimics config.conf() returning the provided dict."""
+    cfg = MagicMock()
+    cfg.get = MagicMock(side_effect=lambda key, default=None: values.get(key, default))
+    return MagicMock(return_value=cfg)
+
+
+class TestYoudaoTranslatorInit(unittest.TestCase):
+    def test_init_success(self):
+        with patch(
+            "translate.youdao.youdao_translate.conf",
+            _mock_conf(
+                youdao_translate_app_key="key123",
+                youdao_translate_app_secret="secret456",
+            ),
+        ):
+            from translate.youdao.youdao_translate import YoudaoTranslator
+
+            translator = YoudaoTranslator()
+            self.assertEqual(translator.app_key, "key123")
+            self.assertEqual(translator.app_secret, "secret456")
+
+    def test_init_missing_credentials_raises(self):
+        with patch(
+            "translate.youdao.youdao_translate.conf",
+            _mock_conf(youdao_translate_app_key="", youdao_translate_app_secret=""),
+        ):
+            from translate.youdao.youdao_translate import YoudaoTranslator
+
+            with self.assertRaises(Exception) as ctx:
+                YoudaoTranslator()
+            self.assertIn("youdao", str(ctx.exception).lower())
+
+
+class TestYoudaoTranslatorHelpers(unittest.TestCase):
+    def test_truncate_input_short(self):
+        from translate.youdao.youdao_translate import YoudaoTranslator
+
+        # length <= 20 -> returned as-is
+        self.assertEqual(YoudaoTranslator._truncate_input("hello"), "hello")
+        self.assertEqual(YoudaoTranslator._truncate_input("a" * 20), "a" * 20)
+
+    def test_truncate_input_long(self):
+        from translate.youdao.youdao_translate import YoudaoTranslator
+
+        # length > 20 -> first 10 + len + last 10
+        text = "abcdefghij" + "X" * 5 + "1234567890"  # 25 chars
+        result = YoudaoTranslator._truncate_input(text)
+        self.assertEqual(result, "abcdefghij" + "25" + "1234567890")
+
+    def test_truncate_input_exactly_21(self):
+        from translate.youdao.youdao_translate import YoudaoTranslator
+
+        text = "a" * 21
+        result = YoudaoTranslator._truncate_input(text)
+        # first 10 'a' + "21" + last 10 'a'
+        self.assertEqual(result, "a" * 10 + "21" + "a" * 10)
+
+    def test_convert_lang_known_codes(self):
+        from translate.youdao.youdao_translate import YoudaoTranslator
+
+        self.assertEqual(YoudaoTranslator._convert_lang(""), "auto")
+        self.assertEqual(YoudaoTranslator._convert_lang("auto"), "auto")
+        self.assertEqual(YoudaoTranslator._convert_lang("zh"), "zh-CHS")
+        self.assertEqual(YoudaoTranslator._convert_lang("zh-CN"), "zh-CHS")
+        self.assertEqual(YoudaoTranslator._convert_lang("zh-TW"), "zh-CHT")
+
+    def test_convert_lang_passthrough(self):
+        from translate.youdao.youdao_translate import YoudaoTranslator
+
+        # unknown codes pass through unchanged (Youdao accepts ISO codes for many langs)
+        self.assertEqual(YoudaoTranslator._convert_lang("en"), "en")
+        self.assertEqual(YoudaoTranslator._convert_lang("ja"), "ja")
+        self.assertEqual(YoudaoTranslator._convert_lang("fr"), "fr")
+
+    def test_convert_lang_none(self):
+        from translate.youdao.youdao_translate import YoudaoTranslator
+
+        self.assertEqual(YoudaoTranslator._convert_lang(None), "auto")
+
+    def test_build_sign_matches_v3_spec(self):
+        with patch(
+            "translate.youdao.youdao_translate.conf",
+            _mock_conf(
+                youdao_translate_app_key="appKey",
+                youdao_translate_app_secret="appSecret",
+            ),
+        ):
+            from translate.youdao.youdao_translate import YoudaoTranslator
+
+            translator = YoudaoTranslator()
+            query = "hello"
+            salt = "saltvalue"
+            curtime = "1700000000"
+            expected = sha256(
+                ("appKey" + "hello" + "saltvalue" + "1700000000" + "appSecret").encode("utf-8")
+            ).hexdigest()
+            self.assertEqual(translator._build_sign(query, salt, curtime), expected)
+
+
+class TestYoudaoTranslatorTranslate(unittest.TestCase):
+    def _make_translator(self):
+        with patch(
+            "translate.youdao.youdao_translate.conf",
+            _mock_conf(
+                youdao_translate_app_key="appKey",
+                youdao_translate_app_secret="appSecret",
+            ),
+        ):
+            from translate.youdao.youdao_translate import YoudaoTranslator
+
+            return YoudaoTranslator()
+
+    def test_translate_success(self):
+        translator = self._make_translator()
+
+        mock_response = MagicMock()
+        mock_response.json.return_value = {
+            "errorCode": "0",
+            "translation": ["你好"],
+            "query": "hello",
+            "l": "en2zh-CHS",
+        }
+        mock_response.raise_for_status = MagicMock()
+
+        with patch(
+            "translate.youdao.youdao_translate.requests.post",
+            return_value=mock_response,
+        ) as mock_post:
+            result = translator.translate("hello", from_lang="en", to_lang="zh")
+
+        self.assertEqual(result, "你好")
+        mock_post.assert_called_once()
+        # Check posted payload contains the right language codes
+        call_kwargs = mock_post.call_args.kwargs
+        payload = call_kwargs["data"]
+        self.assertEqual(payload["q"], "hello")
+        self.assertEqual(payload["from"], "en")
+        self.assertEqual(payload["to"], "zh-CHS")
+        self.assertEqual(payload["appKey"], "appKey")
+        self.assertEqual(payload["signType"], "v3")
+        self.assertIn("salt", payload)
+        self.assertIn("sign", payload)
+        self.assertIn("curtime", payload)
+
+    def test_translate_multiline_joins_with_newlines(self):
+        translator = self._make_translator()
+        mock_response = MagicMock()
+        mock_response.json.return_value = {
+            "errorCode": "0",
+            "translation": ["line one", "line two"],
+        }
+        mock_response.raise_for_status = MagicMock()
+
+        with patch(
+            "translate.youdao.youdao_translate.requests.post",
+            return_value=mock_response,
+        ):
+            result = translator.translate("multi\nline")
+        self.assertEqual(result, "line one\nline two")
+
+    def test_translate_empty_query_returns_empty(self):
+        translator = self._make_translator()
+        # Should not even hit the network for an empty query
+        with patch("translate.youdao.youdao_translate.requests.post") as mock_post:
+            self.assertEqual(translator.translate(""), "")
+            mock_post.assert_not_called()
+
+    def test_translate_error_code_raises(self):
+        translator = self._make_translator()
+        mock_response = MagicMock()
+        mock_response.json.return_value = {
+            "errorCode": "108",
+            "msg": "appKey无效",
+        }
+        mock_response.raise_for_status = MagicMock()
+
+        with patch(
+            "translate.youdao.youdao_translate.requests.post",
+            return_value=mock_response,
+        ):
+            with self.assertRaises(Exception) as ctx:
+                translator.translate("hello")
+        msg = str(ctx.exception)
+        self.assertIn("108", msg)
+
+    def test_translate_empty_translation_raises(self):
+        translator = self._make_translator()
+        mock_response = MagicMock()
+        mock_response.json.return_value = {"errorCode": "0", "translation": []}
+        mock_response.raise_for_status = MagicMock()
+
+        with patch(
+            "translate.youdao.youdao_translate.requests.post",
+            return_value=mock_response,
+        ):
+            with self.assertRaises(Exception):
+                translator.translate("hello")
+
+    def test_translate_default_target_language(self):
+        translator = self._make_translator()
+        mock_response = MagicMock()
+        mock_response.json.return_value = {"errorCode": "0", "translation": ["hello"]}
+        mock_response.raise_for_status = MagicMock()
+
+        with patch(
+            "translate.youdao.youdao_translate.requests.post",
+            return_value=mock_response,
+        ) as mock_post:
+            translator.translate("你好")  # no from/to provided
+
+        payload = mock_post.call_args.kwargs["data"]
+        self.assertEqual(payload["from"], "auto")
+        self.assertEqual(payload["to"], "en")
+
+
+class TestTranslatorFactory(unittest.TestCase):
+    def test_factory_creates_youdao(self):
+        with patch(
+            "translate.youdao.youdao_translate.conf",
+            _mock_conf(
+                youdao_translate_app_key="k",
+                youdao_translate_app_secret="s",
+            ),
+        ):
+            from translate.factory import create_translator
+            from translate.youdao.youdao_translate import YoudaoTranslator
+
+            translator = create_translator("youdao")
+            self.assertIsInstance(translator, YoudaoTranslator)
+
+    def test_factory_unknown_type_message(self):
+        from translate.factory import create_translator
+
+        with self.assertRaises(RuntimeError) as ctx:
+            create_translator("nonexistent")
+        msg = str(ctx.exception)
+        self.assertIn("nonexistent", msg)
+        self.assertIn("baidu", msg)
+        self.assertIn("youdao", msg)
+
+
+if __name__ == "__main__":
+    unittest.main()

translate/factory.py

@@ -1,6 +1,17 @@
-def create_translator(voice_type):
+SUPPORTED_TRANSLATORS = ("baidu", "youdao")
-    if voice_type == "baidu":
+
+
+def create_translator(translator_type):
+    if translator_type == "baidu":
         from translate.baidu.baidu_translate import BaiduTranslator
 
         return BaiduTranslator()
-    raise RuntimeError
+    if translator_type == "youdao":
+        from translate.youdao.youdao_translate import YoudaoTranslator
+
+        return YoudaoTranslator()
+    raise RuntimeError(
+        "unsupported translator type: {}, supported: {}".format(
+            translator_type, ", ".join(SUPPORTED_TRANSLATORS)
+        )
+    )

translate/youdao/youdao_translate.py

@@ -0,0 +1,110 @@
+# -*- coding: utf-8 -*-
+"""
+Youdao translator implementation.
+
+Youdao Translation API v3 documentation:
+https://ai.youdao.com/DOCSIRMA/html/trans/api/wbfy/index.html
+
+Configuration keys (in config.json):
+    youdao_translate_app_key:    Application key from Youdao AI platform.
+    youdao_translate_app_secret: Application secret from Youdao AI platform.
+"""
+
+import time
+import uuid
+from hashlib import sha256
+
+import requests
+
+from config import conf
+from translate.translator import Translator
+
+
+class YoudaoTranslator(Translator):
+    """Youdao translator using the v3 signature scheme."""
+
+    API_URL = "https://openapi.youdao.com/api"
+
+    # Mapping from ISO 639-1 codes (used by the Translator interface)
+    # to Youdao-specific language codes.
+    # Reference: https://ai.youdao.com/DOCSIRMA/html/trans/api/wbfy/index.html
+    LANG_CODE_MAP = {
+        "": "auto",
+        "auto": "auto",
+        "zh": "zh-CHS",
+        "zh-CN": "zh-CHS",
+        "zh-TW": "zh-CHT",
+        "yue": "yue",  # Cantonese
+    }
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.app_key = conf().get("youdao_translate_app_key")
+        self.app_secret = conf().get("youdao_translate_app_secret")
+        if not self.app_key or not self.app_secret:
+            raise Exception("youdao translate app_key or app_secret not set")
+
+    def translate(self, query: str, from_lang: str = "", to_lang: str = "en") -> str:
+        if not query:
+            return ""
+
+        from_lang_code = self._convert_lang(from_lang) or "auto"
+        to_lang_code = self._convert_lang(to_lang) or "en"
+
+        salt = str(uuid.uuid4())
+        curtime = str(int(time.time()))
+        sign = self._build_sign(query, salt, curtime)
+
+        payload = {
+            "q": query,
+            "from": from_lang_code,
+            "to": to_lang_code,
+            "appKey": self.app_key,
+            "salt": salt,
+            "sign": sign,
+            "signType": "v3",
+            "curtime": curtime,
+        }
+        headers = {"Content-Type": "application/x-www-form-urlencoded"}
+
+        response = requests.post(self.API_URL, data=payload, headers=headers, timeout=10)
+        response.raise_for_status()
+        result = response.json()
+
+        error_code = result.get("errorCode", "0")
+        if error_code != "0":
+            raise Exception(
+                "youdao translate error: code={}, msg={}".format(
+                    error_code, result.get("msg", "")
+                )
+            )
+
+        translations = result.get("translation") or []
+        if not translations:
+            raise Exception("youdao translate returned empty translation")
+        return "\n".join(translations)
+
+    def _build_sign(self, query: str, salt: str, curtime: str) -> str:
+        """
+        Build the v3 signature.
+
+        sign = sha256(appKey + input + salt + curtime + appSecret),
+        where input = q if len(q) <= 20 else q[:10] + str(len(q)) + q[-10:].
+        """
+        input_str = self._truncate_input(query)
+        sign_str = self.app_key + input_str + salt + curtime + self.app_secret
+        return sha256(sign_str.encode("utf-8")).hexdigest()
+
+    @staticmethod
+    def _truncate_input(query: str) -> str:
+        length = len(query)
+        if length <= 20:
+            return query
+        return query[:10] + str(length) + query[-10:]
+
+    @classmethod
+    def _convert_lang(cls, lang: str) -> str:
+        """Convert ISO 639-1 language code to Youdao-specific code."""
+        if lang is None:
+            return "auto"
+        return cls.LANG_CODE_MAP.get(lang, lang)