一、接口概览
1.1 接口简介
t_text接口是通用翻译服务的核心接口，支持多语言文本互译，涵盖文本翻译、文档翻译、术语定制、语音翻译等场景。支持主流翻译引擎（百度、腾讯、阿里、有道、Google等）的标准化对接。
1.2 核心功能
✅ 多语言支持：支持200+种语言互译
✅ 智能翻译：神经网络翻译、语境理解、术语优化
✅ 批量处理：支持大规模文本批量翻译
✅ 格式保持：HTML、Markdown、JSON等格式翻译
✅ 领域定制：金融、医疗、法律、技术等专业领域
✅ 质量优化：术语库、翻译记忆库、后编辑优化
二、准备工作
2.1 环境配置
requirements.txt
requests>=2.28.0
python-dotenv>=1.0.0
pydantic>=2.0.0
aiohttp>=3.8.0
redis>=4.5.0
openai>=1.0.0
google-cloud-translate>=3.0.0
2.2 认证配置
config.py
import os
from dotenv import load_dotenv
from typing import Dict, Any
load_dotenv()
class TranslationConfig:
# 翻译服务配置
TRANSLATION_SERVICES = {
'baidu': {
'app_id': os.getenv('BAIDU_TRANSLATE_APP_ID'),
'app_key': os.getenv('BAIDU_TRANSLATE_APP_KEY'),
'api_base': 'https://fanyi-api.baidu.com/api/trans/vip/translate'
},
'tencent': {
'secret_id': os.getenv('TENCENT_TRANSLATE_SECRET_ID'),
'secret_key': os.getenv('TENCENT_TRANSLATE_SECRET_KEY'),
'api_base': 'https://tmt.tencentcloudapi.com'
},
'aliyun': {
'access_key_id': os.getenv('ALIYUN_TRANSLATE_ACCESS_KEY_ID'),
'access_key_secret': os.getenv('ALIYUN_TRANSLATE_ACCESS_KEY_SECRET'),
'api_base': 'https://mt.cn-hangzhou.aliyuncs.com'
},
'youdao': {
'app_key': os.getenv('YOUDAO_TRANSLATE_APP_KEY'),
'app_secret': os.getenv('YOUDAO_TRANSLATE_APP_SECRET'),
'api_base': 'https://openapi.youdao.com/api'
},
'google': {
'api_key': os.getenv('GOOGLE_TRANSLATE_API_KEY'),
'api_base': 'https://translation.googleapis.com/language/translate/v2'
},
'openai': {
'api_key': os.getenv('OPENAI_API_KEY'),
'api_base': 'https://api.openai.com/v1/chat/completions'
}
}
# 请求配置
REQUEST_TIMEOUT = 30
MAX_TEXT_LENGTH = 5000 # 单次请求最大文本长度
BATCH_SIZE = 50 # 批量翻译批次大小
# 缓存配置
CACHE_TTL = 86400 # 24小时
REDIS_URL = os.getenv('REDIS_URL', 'redis://localhost:6379/0')
三、接口详解
3.1 接口地址
POST /translate
3.2 请求参数详解
基础参数
参数名
类型
必填
说明
示例
q
string/array
是
待翻译文本
"Hello World"
source
string
是
源语言
"en"
target
string
是
目标语言
"zh"
format
string
否
文本格式
"text/html/json"
domain
string
否
专业领域
"finance/medical/legal"
glossary_id
string
否
术语库ID
"glossary_001"
高级参数
参数名
类型
必填
说明
示例
model
string
否
翻译模型
"nmt/llm/custom"
preserve_formatting
bool
否
保持格式
true
fallback
bool
否
降级处理
true
retry_count
int
否
重试次数
3
3.3 语言代码表
语言
代码
说明
中文
zh/zh-CN/zh-TW
简体/繁体中文
英语
en
英语
日语
ja
日语
韩语
ko
韩语
法语
fr
法语
德语
de
德语
俄语
ru
俄语
西班牙语
es
西班牙语
阿拉伯语
ar
阿拉伯语
葡萄牙语
pt
葡萄牙语
四、完整代码实现
4.1 Python完整实现
import requests
import time
import hashlib
import hmac
import json
import random
from typing import Dict, Any, List, Optional, Union
from datetime import datetime, timedelta
from dataclasses import dataclass
from urllib.parse import urlencode
import redis
import asyncio
import aiohttp
from openai import OpenAI
from google.cloud import translate_v2 as google_translate
@dataclass
class TranslationRequest:
"""翻译请求"""
text: Union[str, List[str]]
source_lang: str
target_lang: str
format: str = "text"
domain: Optional[str] = None
glossary_id: Optional[str] = None
model: str = "nmt"
preserve_formatting: bool = True
fallback: bool = True
retry_count: int = 3
@dataclass
class TranslationResult:
"""翻译结果"""
success: bool
code: int
message: str
data: Dict[str, Any]
source_text: Union[str, List[str]]
translated_text: Union[str, List[str]]
source_lang: str
target_lang: str
service: str
translation_time: float
confidence: Optional[float] = None
glossary_matches: Optional[List[str]] = None
@dataclass
class BatchTranslationResult:
"""批量翻译结果"""
success: bool
code: int
message: str
data: Dict[str, Any]
results: List[TranslationResult]
total_count: int
success_count: int
failed_count: int
total_time: float
class TranslationAPI:
"""通用翻译API客户端"""
def __init__(self, service_config: Dict[str, Any], redis_client=None):
self.config = service_config
self.redis = redis_client
self.session = requests.Session()
self.session.headers.update({
'User-Agent': 'Translation-API-Client/1.0',
'Accept': 'application/json'
})
self.openai_client = OpenAI(api_key=self.config.get('openai', {}).get('api_key'))
self.google_client = google_translate.Client()
def translate(
self,
text: Union[str, List[str]],
source_lang: str,
target_lang: str,
service: str = "baidu",
**kwargs
) -> TranslationResult:
"""
翻译文本
Args:
text: 待翻译文本（字符串或列表）
source_lang: 源语言代码
target_lang: 目标语言代码
service: 翻译服务（baidu/tencent/aliyun/youdao/google/openai）
**kwargs: 其他参数
Returns:
翻译结果
"""
# 构建请求
request = TranslationRequest(
text=text,
source_lang=source_lang,
target_lang=target_lang,
**kwargs
)
# 检查缓存
cache_key = self._get_cache_key(request, service)
if self.redis:
cached = self.redis.get(cache_key)
if cached:
data = json.loads(cached)
return TranslationResult(**data)
# 执行翻译
start_time = time.time()
try:
if service == "baidu":
result = self._translate_baidu(request)
elif service == "tencent":
result = self._translate_tencent(request)
elif service == "aliyun":
result = self._translate_aliyun(request)
elif service == "youdao":
result = self._translate_youdao(request)
elif service == "google":
result = self._translate_google(request)
elif service == "openai":
result = self._translate_openai(request)
else:
raise ValueError(f"不支持的翻译服务: {service}")
result.translation_time = time.time() - start_time
# 缓存结果
if self.redis and result.success:
self.redis.setex(
cache_key,
self.config.get('CACHE_TTL', 86400),
json.dumps(result.__dict__)
)
return result
except Exception as e:
translation_time = time.time() - start_time
return TranslationResult(
success=False,
code=500,
message=f"翻译失败: {str(e)}",
data={},
source_text=text,
translated_text=[] if isinstance(text, list) else "",
source_lang=source_lang,
target_lang=target_lang,
service=service,
translation_time=translation_time
)
def _translate_baidu(self, request: TranslationRequest) -> TranslationResult:
"""百度翻译"""
config = self.config.get('baidu', {})
app_id = config.get('app_id')
app_key = config.get('app_key')
api_base = config.get('api_base')
# 构建请求参数
params = {
'q': request.text if isinstance(request.text, str) else '\n'.join(request.text),
'from': request.source_lang,
'to': request.target_lang,
'appid': app_id
}
# 生成签名
salt = str(random.randint(32768, 65536))
sign_str = app_id + params['q'] + salt + app_key
sign = hashlib.md5(sign_str.encode('utf-8')).hexdigest()
params['salt'] = salt
params['sign'] = sign
# 发送请求
response = self.session.post(api_base, data=params, timeout=30)
if response.status_code == 200:
data = response.json()
if 'trans_result' in data:
trans_result = data['trans_result']
if isinstance(request.text, str):
translated_text = trans_result[0]['dst']
else:
translated_text = [item['dst'] for item in trans_result]
return TranslationResult(
success=True,
code=200,
message="成功",
data=data,
source_text=request.text,
translated_text=translated_text,
source_lang=request.source_lang,
target_lang=request.target_lang,
service="baidu",
translation_time=0
)
else:
error_code = data.get('error_code', 0)
error_msg = data.get('error_msg', '未知错误')
return TranslationResult(
success=False,
code=error_code,
message=error_msg,
data=data,
source_text=request.text,
translated_text=[] if isinstance(request.text, list) else "",
source_lang=request.source_lang,
target_lang=request.target_lang,
service="baidu",
translation_time=0
)
else:
return TranslationResult(
success=False,
code=response.status_code,
message=f"HTTP {response.status_code}",
data={},
source_text=request.text,
translated_text=[] if isinstance(request.text, list) else "",
source_lang=request.source_lang,
target_lang=request.target_lang,
service="baidu",
translation_time=0
)
def _translate_tencent(self, request: TranslationRequest) -> TranslationResult:
"""腾讯翻译"""
config = self.config.get('tencent', {})
secret_id = config.get('secret_id')
secret_key = config.get('secret_key')
api_base = config.get('api_base')
# 构建请求
import hashlib
import hmac
import base64
# 腾讯云API需要复杂的签名计算
# 这里简化实现，实际使用时需要完整的腾讯云SDK
try:
from tencentcloud.common import credential
from tencentcloud.common.profile.client_profile import ClientProfile
from tencentcloud.common.profile.http_profile import HttpProfile
from tencentcloud.tmt.v20180321 import tmt_client, models
cred = credential.Credential(secret_id, secret_key)
http_profile = HttpProfile()
http_profile.endpoint = "tmt.tencentcloudapi.com"
client_profile = ClientProfile()
client_profile.httpProfile = http_profile
client = tmt_client.TmtClient(cred, "ap-beijing", client_profile)
req = models.TextTranslateRequest()
req.SourceText = request.text if isinstance(request.text, str) else ' '.join(request.text)
req.Source = request.source_lang
req.Target = request.target_lang
req.ProjectId = 0
resp = client.TextTranslate(req)
if isinstance(request.text, str):
translated_text = resp.TargetText
else:
# 腾讯云批量翻译需要特殊处理
translated_text = resp.TargetText.split('\n')
return TranslationResult(
success=True,
code=200,
message="成功",
data=resp.to_json_string(),
source_text=request.text,
translated_text=translated_text,
source_lang=request.source_lang,
target_lang=request.target_lang,
service="tencent",
translation_time=0
)
except Exception as e:
return TranslationResult(
success=False,
code=500,
message=f"腾讯翻译失败: {str(e)}",
data={},
source_text=request.text,
translated_text=[] if isinstance(request.text, list) else "",
source_lang=request.source_lang,
target_lang=request.target_lang,
service="tencent",
translation_time=0
)
def _translate_google(self, request: TranslationRequest) -> TranslationResult:
"""Google翻译"""
try:
if isinstance(request.text, str):
result = self.google_client.translate(
request.text,
source_language=request.source_lang,
target_language=request.target_lang,
format_=request.format
)
translated_text = result['translatedText']
else:
results = self.google_client.translate(
request.text,
source_language=request.source_lang,
target_language=request.target_lang
)
translated_text = [result['translatedText'] for result in results]
return TranslationResult(
success=True,
code=200,
message="成功",
data={},
source_text=request.text,
translated_text=translated_text,
source_lang=request.source_lang,
target_lang=request.target_lang,
service="google",
translation_time=0
)
except Exception as e:
return TranslationResult(
success=False,
code=500,
message=f"Google翻译失败: {str(e)}",
data={},
source_text=request.text,
translated_text=[] if isinstance(request.text, list) else "",
source_lang=request.source_lang,
target_lang=request.target_lang,
service="google",
translation_time=0
)
def _translate_openai(self, request: TranslationRequest) -> TranslationResult:
"""OpenAI翻译"""
try:
# 构建提示词
system_prompt = f"""
你是一位专业的翻译专家，请将{request.source_lang}翻译成{request.target_lang}。
保持原文格式和语义准确，专业术语要翻译正确。
"""
if isinstance(request.text, str):
user_content = f"请翻译以下文本：\n\n{request.text}"
else:
user_content = f"请翻译以下文本列表：\n\n" + '\n'.join([f"{i+1}. {text}" for i, text in enumerate(request.text)])
# 调用OpenAI API
response = self.openai_client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_content}
],
temperature=0.1
)
translated_content = response.choices[0].message.content
if isinstance(request.text, str):
translated_text = translated_content.strip()
else:
# 解析批量翻译结果
translated_text = []
lines = translated_content.strip().split('\n')
for line in lines:
if '. ' in line:
text = line.split('. ', 1)[1]
translated_text.append(text)
else:
translated_text.append(line)
return TranslationResult(
success=True,
code=200,
message="成功",
data=response.to_dict(),
source_text=request.text,
translated_text=translated_text,
source_lang=request.source_lang,
target_lang=request.target_lang,
service="openai",
translation_time=0
)
except Exception as e:
return TranslationResult(
success=False,
code=500,
message=f"OpenAI翻译失败: {str(e)}",
data={},
source_text=request.text,
translated_text=[] if isinstance(request.text, list) else "",
source_lang=request.source_lang,
target_lang=request.target_lang,
service="openai",
translation_time=0
)
def batch_translate(
self,
texts: List[str],
source_lang: str,
target_lang: str,
service: str = "baidu",
batch_size: int = 50,
**kwargs
) -> BatchTranslationResult:
"""
批量翻译文本
Args:
texts: 待翻译文本列表
source_lang: 源语言代码
target_lang: 目标语言代码
service: 翻译服务
batch_size: 批次大小
**kwargs: 其他参数
Returns:
批量翻译结果
"""
start_time = time.time()
results = []
success_count = 0
failed_count = 0
# 分批处理
for i in range(0, len(texts), batch_size):
batch_texts = texts[i:i+batch_size]
try:
result = self.translate(
text=batch_texts,
source_lang=source_lang,
target_lang=target_lang,
service=service,
**kwargs
)
results.append(result)
if result.success:
success_count += len(batch_texts)
else:
failed_count += len(batch_texts)
# 避免请求过于频繁
time.sleep(0.1)
except Exception as e:
failed_count += len(batch_texts)
error_result = TranslationResult(
success=False,
code=500,
message=str(e),
data={},
source_text=batch_texts,
translated_text=[],
source_lang=source_lang,
target_lang=target_lang,
service=service,
translation_time=0
)
results.append(error_result)
total_time = time.time() - start_time
return BatchTranslationResult(
success=success_count > 0,
code=200 if success_count > 0 else 500,
message=f"批量翻译完成，成功{success_count}条，失败{failed_count}条",
data={},
results=results,
total_count=len(texts),
success_count=success_count,
failed_count=failed_count,
total_time=total_time
)
def _get_cache_key(self, request: TranslationRequest, service: str) -> str:
"""生成缓存键"""
import hashlib
# 构建唯一标识
text_str = request.text if isinstance(request.text, str) else '|'.join(request.text)
key_data = f"{service}:{text_str}:{request.source_lang}:{request.target_lang}:{request.domain or ''}"
return f"translation:{hashlib.md5(key_data.encode()).hexdigest()}"
def get_supported_languages(self, service: str) -> List[Dict[str, str]]:
"""获取支持的语言列表"""
# 各服务支持的语言配置
language_configs = {
'baidu': [
{'code': 'zh', 'name': '中文'},
{'code': 'en', 'name': '英语'},
{'code': 'ja', 'name': '日语'},
{'code': 'ko', 'name': '韩语'},
{'code': 'fr', 'name': '法语'},
{'code': 'de', 'name': '德语'},
{'code': 'ru', 'name': '俄语'},
{'code': 'es', 'name': '西班牙语'},
{'code': 'ar', 'name': '阿拉伯语'},
{'code': 'pt', 'name': '葡萄牙语'}
],
'google': [
{'code': 'zh', 'name': '中文'},
{'code': 'en', 'name': '英语'},
{'code': 'ja', 'name': '日语'},
{'code': 'ko', 'name': '韩语'},
{'code': 'fr', 'name': '法语'},
{'code': 'de', 'name': '德语'},
{'code': 'ru', 'name': '俄语'},
{'code': 'es', 'name': '西班牙语'},
{'code': 'ar', 'name': '阿拉伯语'},
{'code': 'pt', 'name': '葡萄牙语'},
# Google支持更多语言...
]
}
return language_configs.get(service, [])
使用示例
def demo_translation_api():
"""翻译API使用演示"""
# 初始化客户端
api = TranslationAPI(TranslationConfig.TRANSLATION_SERVICES)
print("=== 示例1：单条文本翻译 ===")
result = api.translate(
text="Hello, world! This is a translation example.",
source_lang="en",
target_lang="zh",
service="baidu"
)
if result.success:
print(f"原文: {result.source_text}")
print(f"译文: {result.translated_text}")
print(f"耗时: {result.translation_time:.3f}s")
else:
print(f"翻译失败: {result.message}")
print("\n=== 示例2：批量翻译 ===")
texts = [
"Good morning!",
"How are you today?",
"Have a nice day!",
"Thank you very much!"
]
batch_result = api.batch_translate(
texts=texts,
source_lang="en",
target_lang="zh",
service="baidu",
batch_size=2
)
if batch_result.success:
for i, result in enumerate(batch_result.results):
if result.success:
print(f"批次{i+1}: 成功翻译{len(result.translated_text)}条")
else:
print(f"批次{i+1}: 翻译失败 - {result.message}")
print("\n=== 示例3：多服务对比 ===")
services = ['baidu', 'google', 'openai']
text = "The quick brown fox jumps over the lazy dog."
for service in services:
result = api.translate(
text=text,
source_lang="en",
target_lang="zh",
service=service
)
if result.success:
print(f"{service}: {result.translated_text}")
if name == "main":
demo_translation_api()
4.2 异步实现
import aiohttp
import asyncio
from typing import List, Dict, Any
class AsyncTranslationAPI:
"""异步翻译API客户端"""
def __init__(self, service_config: Dict[str, Any]):
self.config = service_config
self.session = None
self.openai_client = OpenAI(api_key=self.config.get('openai', {}).get('api_key'))
async def __aenter__(self):
self.session = aiohttp.ClientSession()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
await self.session.close()
async def translate_async(
self,
text: Union[str, List[str]],
source_lang: str,
target_lang: str,
service: str = "baidu",
**kwargs
) -> TranslationResult:
"""
异步翻译文本
"""
start_time = time.time()
try:
if service == "baidu":
result = await self._translate_baidu_async(text, source_lang, target_lang)
elif service == "google":
result = await self._translate_google_async(text, source_lang, target_lang)
elif service == "openai":
result = await self._translate_openai_async(text, source_lang, target_lang)
else:
raise ValueError(f"不支持的异步翻译服务: {service}")
result.translation_time = time.time() - start_time
return result
except Exception as e:
translation_time = time.time() - start_time
return TranslationResult(
success=False,
code=500,
message=f"翻译失败: {str(e)}",
data={},
source_text=text,
translated_text=[] if isinstance(text, list) else "",
source_lang=source_lang,
target_lang=target_lang,
service=service,
translation_time=translation_time
)
async def _translate_baidu_async(
self,
text: Union[str, List[str]],
source_lang: str,
target_lang: str
) -> TranslationResult:
"""异步百度翻译"""
config = self.config.get('baidu', {})
app_id = config.get('app_id')
app_key = config.get('app_key')
api_base = config.get('api_base')
# 构建请求参数
params = {
'q': text if isinstance(text, str) else '\n'.join(text),
'from': source_lang,
'to': target_lang,
'appid': app_id
}
# 生成签名
salt = str(random.randint(32768, 65536))
sign_str = app_id + params['q'] + salt + app_key
sign = hashlib.md5(sign_str.encode('utf-8')).hexdigest()
params['salt'] = salt
params['sign'] = sign
# 发送请求
async with self.session.post(api_base, data=params) as response:
if response.status == 200:
data = await response.json()
if 'trans_result' in data:
trans_result = data['trans_result']
if isinstance(text, str):
translated_text = trans_result[0]['dst']
else:
translated_text = [item['dst'] for item in trans_result]
return TranslationResult(
success=True,
code=200,
message="成功",
data=data,
source_text=text,
translated_text=translated_text,
source_lang=source_lang,
target_lang=target_lang,
service="baidu",
translation_time=0
)
else:
error_code = data.get('error_code', 0)
error_msg = data.get('error_msg', '未知错误')
return TranslationResult(
success=False,
code=error_code,
message=error_msg,
data=data,
source_text=text,
translated_text=[] if isinstance(text, list) else "",
source_lang=source_lang,
target_lang=target_lang,
service="baidu",
translation_time=0
)
else:
return TranslationResult(
success=False,
code=response.status,
message=f"HTTP {response.status}",
data={},
source_text=text,
translated_text=[] if isinstance(text, list) else "",
source_lang=source_lang,
target_lang=target_lang,
service="baidu",
translation_time=0
)
async def _translate_openai_async(
self,
text: Union[str, List[str]],
source_lang: str,
target_lang: str
) -> TranslationResult:
"""异步OpenAI翻译"""
try:
# 构建提示词
system_prompt = f"""
你是一位专业的翻译专家，请将{source_lang}翻译成{target_lang}。
保持原文格式和语义准确，专业术语要翻译正确。
"""
if isinstance(text, str):
user_content = f"请翻译以下文本：\n\n{text}"
else:
user_content = f"请翻译以下文本列表：\n\n" + '\n'.join([f"{i+1}. {text}" for i, text in enumerate(text)])
# 调用OpenAI API
response = await self.openai_client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_content}
],
temperature=0.1
)
translated_content = response.choices[0].message.content
if isinstance(text, str):
translated_text = translated_content.strip()
else:
# 解析批量翻译结果
translated_text = []
lines = translated_content.strip().split('\n')
for line in lines:
if '. ' in line:
text = line.split('. ', 1)[1]
translated_text.append(text)
else:
translated_text.append(line)
return TranslationResult(
success=True,
code=200,
message="成功",
data=response.to_dict(),
source_text=text,
translated_text=translated_text,
source_lang=source_lang,
target_lang=target_lang,
service="openai",
translation_time=0
)
except Exception as e:
return TranslationResult(
success=False,
code=500,
message=f"OpenAI翻译失败: {str(e)}",
data={},
source_text=text,
translated_text=[] if isinstance(text, list) else "",
source_lang=source_lang,
target_lang=target_lang,
service="openai",
translation_time=0
)
异步批量翻译示例
async def async_batch_translation():
"""异步批量翻译示例"""
async with AsyncTranslationAPI(TranslationConfig.TRANSLATION_SERVICES) as api:
texts = [
"Hello, how are you?",
"What time is it now?",
"Have a nice day!",
"Thank you very much!",
"Good morning everyone!"
]
# 并发执行多个翻译任务
tasks = []
for text in texts:
task = api.translate_async(text, "en", "zh", "baidu")
tasks.append(task)
results = await asyncio.gather(*tasks)
for i, result in enumerate(results):
if result.success:
print(f"文本{i+1}: {result.translated_text}")
else:
print(f"文本{i+1}: 翻译失败 - {result.message}")
4.3 智能翻译服务选择
class IntelligentTranslationService:
"""智能翻译服务选择器"""
def __init__(self, api_client):
self.client = api_client
self.service_metrics = {} # 服务性能指标
self.service_costs = { # 服务成本估算
'baidu': 0.001, # 元/字符
'tencent': 0.002,
'aliyun': 0.0015,
'youdao': 0.002,
'google': 0.003,
'openai': 0.01
}
def smart_translate(
self,
text: Union[str, List[str]],
source_lang: str,
target_lang: str,
strategy: str = "auto",
**kwargs
) -> TranslationResult:
"""
智能翻译：根据策略选择最佳服务
Args:
text: 待翻译文本
source_lang: 源语言
target_lang: 目标语言
strategy: 选择策略
- auto: 自动选择
- cost: 成本优先
- quality: 质量优先
- speed: 速度优先
**kwargs: 其他参数
Returns:
翻译结果
"""
# 根据策略选择服务
if strategy == "auto":
service = self._select_service_auto(source_lang, target_lang, len(str(text)))
elif strategy == "cost":
service = self._select_service_cost(source_lang, target_lang)
elif strategy == "quality":
service = self._select_service_quality(source_lang, target_lang)
elif strategy == "speed":
service = self._select_service_speed(source_lang, target_lang)
else:
service = "baidu" # 默认服务
# 执行翻译
result = self.client.translate(text, source_lang, target_lang, service, **kwargs)
# 更新服务指标
self._update_service_metrics(service, result)
return result
def _select_service_auto(
self,
source_lang: str,
target_lang: str,
text_length: int
) -> str:
"""自动选择翻译服务"""
# 语言对支持度评估
language_pairs = {
('zh', 'en'): ['baidu', 'tencent', 'google', 'openai'],
('en', 'zh'): ['baidu', 'tencent', 'google', 'openai'],
('ja', 'zh'): ['baidu', 'google'],
('ko', 'zh'): ['baidu', 'google'],
('fr', 'zh'): ['google', 'openai'],
('de', 'zh'): ['google', 'openai']
}
# 检查特定语言对
key = (source_lang, target_lang)
if key in language_pairs:
services = language_pairs[key]
# 根据文本长度和成本选择
if text_length < 100:
return services[0] # 短文本用第一个服务
elif text_length < 1000:
return services[1] if len(services) > 1 else services[0]
else:
return services[-1] # 长文本用最后一个服务
# 默认选择
if source_lang == 'zh' or target_lang == 'zh':
return 'baidu' # 中文相关用百度
else:
return 'google' # 其他用Google
def _update_service_metrics(self, service: str, result: TranslationResult):
"""更新服务性能指标"""
if service not in self.service_metrics:
self.service_metrics[service] = {
'total_requests': 0,
'success_requests': 0,
'total_time': 0,
'avg_time': 0,
'success_rate': 0
}
metrics = self.service_metrics[service]
metrics['total_requests'] += 1
if result.success:
metrics['success_requests'] += 1
metrics['total_time'] += result.translation_time
metrics['avg_time'] = metrics['total_time'] / metrics['success_requests']
metrics['success_rate'] = metrics['success_requests'] / metrics['total_requests'] * 100
五、返回结果解析
5.1 成功响应示例
{
"success": true,
"code": 200,
"message": "成功",
"data": {
"from": "en",
"to": "zh",
"trans_result": [
{
"src": "Hello, world!",
"dst": "你好，世界！"
}
]
},
"source_text": "Hello, world!",
"translated_text": "你好，世界！",
"source_lang": "en",
"target_lang": "zh",
"service": "baidu",
"translation_time": 0.345,
"confidence": 0.95
}
5.2 批量响应示例
{
"success": true,
"code": 200,
"message": "批量翻译完成，成功2条，失败0条",
"data": {},
"results": [
{
"success": true,
"code": 200,
"message": "成功",
"data": {
"trans_result": [
{
"src": "Good morning!",
"dst": "早上好！"
}
]
},
"source_text": "Good morning!",
"translated_text": "早上好！",
"source_lang": "en",
"target_lang": "zh",
"service": "baidu",
"translation_time": 0.234
},
{
"success": true,
"code": 200,
"message": "成功",
"data": {
"trans_result": [
{
"src": "Thank you!",
"dst": "谢谢你！"
}
]
},
"source_text": "Thank you!",
"translated_text": "谢谢你！",
"source_lang": "en",
"target_lang": "zh",
"service": "baidu",
"translation_time": 0.256
}
],
"total_count": 2,
"success_count": 2,
"failed_count": 0,
"total_time": 0.512
}
5.3 错误响应示例
{
"success": false,
"code": 54003,
"message": "访问频率受限",
"data": {},
"source_text": "Hello, world!",
"translated_text": "",
"source_lang": "en",
"target_lang": "zh",
"service": "baidu",
"translation_time": 0.123
}
5.4 错误码说明
错误码
说明
处理建议
200
成功
-
52001
请求超时
重试请求
52002
系统错误
稍后重试
52003
未授权用户
检查API密钥
54000
必填参数为空
检查请求参数
54001
签名错误
检查签名算法
54003
访问频率受限
降低请求频率
54004
账户余额不足
充值账户
54005
长查询请求频繁
减少长文本请求
58000
客户端IP非法
检查IP白名单
58001
语言方向错误
检查语言代码
90107
认证未通过
重新认证
六、高级功能实现
6.1 术语库管理
class GlossaryManager:
"""术语库管理器"""
def __init__(self, redis_client):
self.redis = redis_client
self.glossary_prefix = "glossary:"
def create_glossary(self, glossary_id: str, terms: Dict[str, str]) -> bool:
"""创建术语库"""
try:
glossary_key = f"{self.glossary_prefix}{glossary_id}"
self.redis.hset(glossary_key, mapping=terms)
return True
except Exception as e:
print(f"创建术语库失败: {e}")
return False
def get_glossary_terms(self, glossary_id: str) -> Dict[str, str]:
"""获取术语库"""
glossary_key = f"{self.glossary_prefix}{glossary_id}"
terms = self.redis.hgetall(glossary_key)
return {k.decode(): v.decode() for k, v in terms.items()} if terms else {}
def apply_glossary(self, text: str, glossary_id: str) -> str:
"""应用术语库"""
terms = self.get_glossary_terms(glossary_id)
for source, target in terms.items():
text = text.replace(source, target)
return text
def batch_apply_glossary(self, texts: List[str], glossary_id: str) -> List[str]:
"""批量应用术语库"""
terms = self.get_glossary_terms(glossary_id)
results = []
for text in texts:
for source, target in terms.items():
text = text.replace(source, target)
results.append(text)
return results
6.2 翻译质量评估
class TranslationQualityEvaluator:
"""翻译质量评估器"""
def __init__(self):
self.metrics = ['accuracy', 'fluency', 'terminology', 'style']
def evaluate_quality(
self,
source_text: str,
translated_text: str,
source_lang: str,
target_lang: str
) -> Dict[str, float]:
"""
评估翻译质量
Args:
source_text: 源文本
translated_text: 译文
source_lang: 源语言
target_lang: 目标语言
Returns:
质量评分字典
"""
scores = {}
# 准确性评估（基于BLEU评分原理）
scores['accuracy'] = self._calculate_accuracy_score(source_text, translated_text)
# 流畅性评估（基于语言模型）
scores['fluency'] = self._calculate_fluency_score(translated_text, target_lang)
# 术语一致性评估
scores['terminology'] = self._calculate_terminology_score(source_text, translated_text)
# 风格匹配度评估
scores['style'] = self._calculate_style_score(source_text, translated_text)
# 综合评分
scores['overall'] = sum(scores.values()) / len(scores)
return scores
def _calculate_accuracy_score(self, source: str, target: str) -> float:
"""计算准确性评分"""
# 简化实现，实际应用中可使用BLEU、METEOR等算法
source_words = set(source.lower().split())
target_words = set(target.lower().split())
if len(source_words) == 0:
return 1.0
# 计算Jaccard相似度
intersection = len(source_words.intersection(target_words))
union = len(source_words.union(target_words))
return intersection / union if union > 0 else 0.0
def _calculate_fluency_score(self, text: str, lang: str) -> float:
"""计算流畅性评分"""
# 简化实现，实际应用中可使用语言模型
if len(text) == 0:
return 0.0
# 基于句子长度和标点符号的简单评估
sentence_endings = ['.', '!', '?', '。', '！', '？']
has_ending = any(text.endswith(char) for char in sentence_endings)
word_count = len(text.split())
# 流畅性评分规则
if word_count < 5:
return 0.7 if has_ending else 0.5
elif word_count < 20:
return 0.8 if has_ending else 0.6
else:
return 0.9 if has_ending else 0.7
6.3 多引擎融合翻译
class MultiEngineTranslation:
"""多引擎融合翻译"""
def __init__(self, api_client):
self.client = api_client
self.engine_weights = {
'baidu': 0.3,
'google': 0.4,
'openai': 0.3
}
def ensemble_translate(
self,
text: str,
source_lang: str,
target_lang: str
) -> TranslationResult:
"""
多引擎融合翻译
Args:
text: 待翻译文本
source_lang: 源语言
target_lang: 目标语言
Returns:
融合翻译结果
"""
# 并行调用多个翻译引擎
results = {}
for service in self.engine_weights.keys():
result = self.client.translate(text, source_lang, target_lang, service)
if result.success:
results[service] = result
if len(results) == 0:
return TranslationResult(
success=False,
code=500,
message="所有翻译引擎都失败了",
data={},
source_text=text,
translated_text="",
source_lang=source_lang,
target_lang=target_lang,
service="ensemble",
translation_time=0
)
# 融合翻译结果
if len(results) == 1:
return list(results.values())[0]
# 使用加权投票算法
translated_texts = {}
for service, result in results.items():
translated_text = result.translated_text
if translated_text not in translated_texts:
translated_texts[translated_text] = 0
translated_texts[translated_text] += self.engine_weights[service]
# 选择权重最高的译文
best_text = max(translated_texts.items(), key=lambda x: x[1])[0]
return TranslationResult(
success=True,
code=200,
message="融合翻译成功",
data={'ensemble_results': {k: v.translated_text for k, v in results.items()}},
source_text=text,
translated_text=best_text,
source_lang=source_lang,
target_lang=target_lang,
service="ensemble",
translation_time=sum(r.translation_time for r in results.values()) / len(results)
)
七、实战应用场景
7.1 国际化网站翻译
class WebsiteTranslator:
"""网站翻译器"""
def __init__(self, api_client, glossary_manager):
self.client = api_client
self.glossary = glossary_manager
self.page_cache = {}
def translate_webpage(
self,
content: str,
source_lang: str,
target_lang: str,
content_type: str = "html"
) -> str:
"""
翻译网页内容
Args:
content: 网页内容
source_lang: 源语言
target_lang: 目标语言
content_type: 内容类型（html/markdown/text）
Returns:
翻译后的内容
"""
# 检查缓存
cache_key = f"webpage:{hashlib.md5(content.encode()).hexdigest()}:{target_lang}"
if cache_key in self.page_cache:
return self.page_cache[cache_key]
# 提取文本内容
if content_type == "html":
texts = self._extract_text_from_html(content)
elif content_type == "markdown":
texts = self._extract_text_from_markdown(content)
else:
texts = [content]
# 批量翻译
translated_texts = []
for text in texts:
if len(text.strip()) > 0:
# 应用术语库
text = self.glossary.apply_glossary(text, f"{source_lang}_{target_lang}")
# 翻译文本
result = self.client.translate(text, source_lang, target_lang, "baidu")
if result.success:
translated_texts.append(result.translated_text)
else:
translated_texts.append(text) # 翻译失败保留原文
else:
translated_texts.append(text)
# 重建内容
if content_type == "html":
translated_content = self._rebuild_html_content(content, texts, translated_texts)
elif content_type == "markdown":
translated_content = self._rebuild_markdown_content(content, texts, translated_texts)
else:
translated_content = '\n'.join(translated_texts)
# 缓存结果
self.page_cache[cache_key] = translated_content
return translated_content
def _extract_text_from_html(self, html: str) -> List[str]:
"""从HTML中提取文本"""
import re
# 移除script和style标签
clean_html = re.sub(r'<script[^>]*>.*?</script>', '', html, flags=re.DOTALL)
clean_html = re.sub(r'<style[^>]*>.*?</style>', '', clean_html, flags=re.DOTALL)
# 提取文本内容
texts = []
# 匹配标签内的文本
pattern = r'>([^<]+)<'
matches = re.findall(pattern, clean_html)
for match in matches:
text = match.strip()
if text and len(text) > 0:
texts.append(text)
return texts
7.2 文档翻译系统
class DocumentTranslator:
"""文档翻译系统"""
def __init__(self, api_client, glossary_manager):
self.client = api_client
self.glossary = glossary_manager
def translate_document(
self,
file_path: str,
source_lang: str,
target_lang: str,
output_path: str = None
) -> bool:
"""
翻译文档
Args:
file_path: 文档路径
source_lang: 源语言
target_lang: 目标语言
output_path: 输出路径
Returns:
是否成功
"""
import os
from pathlib import Path
file_ext = Path(file_path).suffix.lower()
try:
if file_ext == '.txt':
return self._translate_text_file(file_path, source_lang, target_lang, output_path)
elif file_ext == '.md':
return self._translate_markdown_file(file_path, source_lang, target_lang, output_path)
elif file_ext in ['.docx', '.doc']:
return self._translate_word_file(file_path, source_lang, target_lang, output_path)
else:
print(f"不支持的文件格式: {file_ext}")
return False
except Exception as e:
print(f"文档翻译失败: {e}")
return False
def _translate_text_file(
self,
file_path: str,
source_lang: str,
target_lang: str,
output_path: str
) -> bool:
"""翻译文本文件"""
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
# 按段落分割
paragraphs = content.split('\n\n')
translated_paragraphs = []
for paragraph in paragraphs:
if paragraph.strip():
result = self.client.translate(paragraph, source_lang, target_lang, "baidu")
if result.success:
translated_paragraphs.append(result.translated_text)
else:
translated_paragraphs.append(paragraph) # 翻译失败保留原文
else:
translated_paragraphs.append('')
# 写入文件
output_path = output_path or file_path.replace('.txt', f'_{target_lang}.txt')
with open(output_path, 'w', encoding='utf-8') as f:
f.write('\n\n'.join(translated_paragraphs))
return True
八、故障排查与优化
8.1 常见问题解决
问题1：API限流处理
def translate_with_retry(
self,
text: str,
source_lang: str,
target_lang: str,
service: str,
max_retries: int = 3
) -> TranslationResult:
"""带重试的翻译"""
for attempt in range(max_retries):
try:
result = self.translate(text, source_lang, target_lang, service)
if result.success:
return result
elif result.code in [52001, 54003]: # 超时或限流
if attempt < max_retries - 1:
wait_time = 2 ** attempt # 指数退避
time.sleep(wait_time)
continue
else:
return result
except Exception as e:
if attempt < max_retries - 1:
wait_time = 2 ** attempt
time.sleep(wait_time)
continue
else:
return TranslationResult(
success=False,
code=500,
message=str(e),
data={},
source_text=text,
translated_text="",
source_lang=source_lang,
target_lang=target_lang,
service=service,
translation_time=0
)
return TranslationResult(
success=False,
code=500,
message="翻译重试次数超限",
data={},
source_text=text,
translated_text="",
source_lang=source_lang,
target_lang=target_lang,
service=service,
translation_time=0
)
问题2：长文本处理
def translate_long_text(
self,
text: str,
source_lang: str,
target_lang: str,
service: str,
max_length: int = 2000
) -> TranslationResult:
"""翻译长文本（自动分段）"""
if len(text) <= max_length:
return self.translate(text, source_lang, target_lang, service)
# 按句子分割
import re
sentences = re.split(r'[.!?。！？]+', text)
# 合并短句子
segments = []
current_segment = ""
for sentence in sentences:
if len(current_segment) + len(sentence) < max_length:
current_segment += sentence + ". "
else:
if current_segment:
segments.append(current_segment.strip())
current_segment = sentence + ". "
if current_segment:
segments.append(current_segment.strip())
# 批量翻译
batch_result = self.batch_translate(segments, source_lang, target_lang, service)
if batch_result.success:
translated_text = ' '.join([
result.translated_text for result in batch_result.results
if result.success
])
return TranslationResult(
success=True,
code=200,
message="长文本翻译成功",
data={'segments': len(segments)},
source_text=text,
translated_text=translated_text,
source_lang=source_lang,
target_lang=target_lang,
service=service,
translation_time=batch_result.total_time
)
else:
return TranslationResult(
success=False,
code=500,
message="长文本翻译失败",
data={},
source_text=text,
translated_text="",
source_lang=source_lang,
target_lang=target_lang,
service=service,
translation_time=0
)
8.2 性能优化建议
连接池管理
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
def create_optimized_session():
"""创建优化的会话"""
session = requests.Session()
# 配置重试策略
retry_strategy = Retry(
total=3,
backoff_factor=0.5,
status_forcelist=[429, 500, 502, 503, 504],
)
adapter = HTTPAdapter(
max_retries=retry_strategy,
pool_connections=10,
pool_maxsize=100,
pool_block=False
)
session.mount("https://", adapter)
session.mount("http://", adapter)
return session
智能缓存策略
class SmartTranslationCache:
"""智能翻译缓存"""
def __init__(self, redis_client):
self.redis = redis_client
self.memory_cache = {}
self.memory_ttl = 300 # 5分钟
def get_translation(self, cache_key: str):
"""获取翻译缓存"""
# 检查内存缓存
if cache_key in self.memory_cache:
data, expire_time = self.memory_cache[cache_key]
if time.time() < expire_time:
return data
# 检查Redis缓存
if self.redis:
cached = self.redis.get(cache_key)
if cached:
data = json.loads(cached)
# 更新内存缓存
self.memory_cache[cache_key] = (data, time.time() + self.memory_ttl)
return data
return None
def set_translation(self, cache_key: str, data: Dict[str, Any], ttl: int = 86400):
"""设置翻译缓存"""
# 设置内存缓存
self.memory_cache[cache_key] = (data, time.time() + self.memory_ttl)
# 设置Redis缓存
if self.redis:
self.redis.setex(cache_key, ttl, json.dumps(data))
九、最佳实践总结
9.1 安全实践
密钥管理：使用环境变量存储API密钥
HTTPS强制：确保所有请求使用HTTPS
输入验证：验证所有输入参数
错误处理：不暴露敏感错误信息
9.2 性能实践
缓存策略：根据文本内容设置合适的缓存时间
批量操作：合并多个请求减少API调用次数
异步处理：使用异步IO提高并发性能
连接复用：使用连接池减少连接建立开销
9.3 业务实践
术语库管理：维护专业术语库提高翻译质量
质量评估：定期评估翻译质量并优化策略
多引擎备用：准备多个翻译引擎应对服务故障
成本控制：根据文本类型选择合适的翻译服务
附录：快速开始模板
quick_start.py
from translation_api import TranslationAPI, TranslationConfig
1. 初始化客户端
api = TranslationAPI(TranslationConfig.TRANSLATION_SERVICES)
2. 简单翻译
result = api.translate(
text="Hello, world! This is a translation example.",
source_lang="en",
target_lang="zh",
service="baidu"
)
if result.success:
print(f"原文: {result.source_text}")
print(f"译文: {result.translated_text}")
3. 批量翻译
texts = [
"Good morning!",
"How are you?",
"Have a nice day!"
]
batch_result = api.batch_translate(
texts=texts,
source_lang="en",
target_lang="zh",
service="baidu"
)
if batch_result.success:
for i, result in enumerate(batch_result.results):
if result.success:
print(f"文本{i+1}: {result.translated_text}")
4. 智能翻译
from intelligent_translation import IntelligentTranslationService
smart_api = IntelligentTranslationService(api)
result = smart_api.smart_translate(
text="The quick brown fox jumps over the lazy dog.",
source_lang="en",
target_lang="zh",
strategy="auto"
)
通过本攻略，您应该能够：
理解翻译接口的完整功能和参数配置
实现多种翻译服务的标准化对接
处理各种错误情况和性能优化
构建高性能的翻译应用系统
在实际业务中灵活应用翻译接口
建议根据实际业务需求选择合适的实现方案，并遵循最佳实践确保系统的稳定性和可维护性。
