diff --git a/.gitignore b/.gitignore index 0612e1e3..57504ed9 100644 --- a/.gitignore +++ b/.gitignore @@ -46,3 +46,4 @@ dist/ build/ *.egg-info/ .cow.pid +PR_SUBMISSION.md diff --git a/config.py b/config.py index 955e93cd..5515b324 100644 --- a/config.py +++ b/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 diff --git a/tests/test_youdao_translator.py b/tests/test_youdao_translator.py new file mode 100644 index 00000000..5c50ffc7 --- /dev/null +++ b/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() diff --git a/translate/factory.py b/translate/factory.py index ba80aa59..7e7bea8e 100644 --- a/translate/factory.py +++ b/translate/factory.py @@ -1,6 +1,17 @@ -def create_translator(voice_type): - if voice_type == "baidu": +SUPPORTED_TRANSLATORS = ("baidu", "youdao") + + +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) + ) + ) diff --git a/translate/youdao/youdao_translate.py b/translate/youdao/youdao_translate.py new file mode 100644 index 00000000..4db9602c --- /dev/null +++ b/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)