mirror of
https://github.com/zhayujie/chatgpt-on-wechat.git
synced 2026-06-02 00:57:41 +08:00
feat(translate): add Youdao as a new translation provider
The translate module previously only supported Baidu translation, and the factory raised a bare RuntimeError for any other type. This change adds Youdao Translation as a second provider and improves the factory's error message. Implementation details: - New YoudaoTranslator class in translate/youdao/youdao_translate.py - Implements Youdao's v3 SHA-256 signature scheme, including the truncate-input rule for queries longer than 20 characters - Maps ISO 639-1 language codes to Youdao-specific codes (zh -> zh-CHS, zh-TW -> zh-CHT, others pass through) - Differentiates network errors, API error codes, and empty translations - factory.create_translator now lists the supported types in its RuntimeError message instead of failing silently - Default config exposes youdao_translate_app_key and youdao_translate_app_secret Adds 17 unit tests covering signature correctness, language code mapping, input truncation edge cases, the full request/response flow, and factory dispatch. All tests pass under Python 3.11.
This commit is contained in:
Zmjjeff7
1
.gitignore
vendored
1
.gitignore
vendored
@@ -46,3 +46,4 @@ dist/
|
||||
build/
|
||||
*.egg-info/
|
||||
.cow.pid
|
||||
PR_SUBMISSION.md
|
||||
|
||||
@@ -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
|
||||
|
||||
260
tests/test_youdao_translator.py
Normal file
260
tests/test_youdao_translator.py
Normal file
@@ -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()
|
||||
@@ -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)
|
||||
)
|
||||
)
|
||||
|
||||
110
translate/youdao/youdao_translate.py
Normal file
110
translate/youdao/youdao_translate.py
Normal file
@@ -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)
|
||||
Reference in New Issue
Block a user