一、接口概览
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 results6.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.76.3 多引擎融合翻译
class MultiEngineTranslation:
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
"""多引擎融合翻译""" 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 texts7.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" )通过本攻略，您应该能够：
理解翻译接口的完整功能和参数配置
实现多种翻译服务的标准化对接
处理各种错误情况和性能优化
构建高性能的翻译应用系统
在实际业务中灵活应用翻译接口
建议根据实际业务需求选择合适的实现方案，并遵循最佳实践确保系统的稳定性和可维护性。
