""" 通义多模态翻译客户端 基于阿里云 AnyTrans API (2025-07-07版本) 文档：https://help.aliyun.com/zh/model-studio/api-anytrans-2025-07-07 """ import logging from typing import Optional, List, Dict, Any from alibabacloud_anytrans20250707.client import Client as AnyTransClient from alibabacloud_tea_openapi import models as open_api_models from alibabacloud_anytrans20250707 import models as anytrans_models from alibabacloud_tea_util import models as util_models logger = logging.getLogger(__name__) class TongyiAnyTransClient: """通义多模态翻译客户端""" def __init__(self, access_key_id: str, access_key_secret: str, workspace_id: str): """ 初始化客户端 Args: access_key_id: 阿里云 AccessKey ID access_key_secret: 阿里云 AccessKey Secret workspace_id: 百炼工作空间 ID """ if not all([access_key_id, access_key_secret, workspace_id]): raise ValueError("access_key_id, access_key_secret and workspace_id are required") self.workspace_id = workspace_id # 创建配置 config = open_api_models.Config( access_key_id=access_key_id, access_key_secret=access_key_secret, endpoint='anytrans.cn-beijing.aliyuncs.com' ) # 创建客户端 self.client = AnyTransClient(config) logger.info("TongyiAnyTransClient initialized successfully") def text_translate( self, source_text: str, target_language: str, source_language: str = "auto", scene: str = "mt-turbo", terminologies: Optional[List[Dict[str, str]]] = None, domain_hint: Optional[str] = None ) -> Dict[str, Any]: """ 文本翻译 Args: source_text: 源文本 target_language: 目标语言（如：en, zh, ja） source_language: 源语言（默认auto自动检测） scene: 模型类型（mt-plus专业版/mt-turbo轻量版） terminologies: 术语干预列表 domain_hint: 领域提示 Returns: 翻译结果字典 """ try: # 构建基础请求 request = anytrans_models.TextTranslateRequest( workspace_id=self.workspace_id, source_language=source_language, target_language=target_language, text=source_text, scene=scene ) # 添加扩展参数 if terminologies or domain_hint: ext = anytrans_models.TextTranslateRequestExt() if terminologies: # 转换术语格式 term_objects = [] for term in terminologies: term_obj = anytrans_models.TextTranslateRequestExtTerminologies() term_obj.src = term["src"] term_obj.tgt = term["tgt"] term_objects.append(term_obj) ext.terminologies = term_objects if domain_hint: ext.domain_hint = domain_hint request.ext = ext # 直接调用 response = self.client.text_translate(request) if response.status_code != 200: raise Exception(f"API调用失败: {response.status_code}") body = response.body return { "translated_text": body.data.translation, "detected_language": getattr(body.data, 'detected_language', None), "usage": { "input_tokens": body.data.usage.input_tokens if hasattr(body.data, 'usage') else 0, "output_tokens": body.data.usage.output_tokens if hasattr(body.data, 'usage') else 0 } } except Exception as e: logger.error(f"文本翻译失败: {str(e)}") raise Exception(f"文本翻译失败: {str(e)}") def batch_translate( self, texts: List[str], target_language: str, source_language: str = "auto", scene: str = "mt-turbo" ) -> List[Dict[str, Any]]: """ 批量文本翻译 Args: texts: 文本列表（最多100条） target_language: 目标语言 source_language: 源语言 scene: 模型类型 Returns: 翻译结果列表 """ try: request = anytrans_models.BatchTranslateRequest( workspace_id=self.workspace_id, source_language=source_language, target_language=target_language, source_texts=texts, scene=scene ) runtime = util_models.RuntimeOptions() response = self.client.batch_translate_with_options(request, runtime) if response.status_code != 200: raise Exception(f"API调用失败: {response.status_code}") body = response.body results = [] for idx, item in enumerate(body.data.translations): results.append({ "index": idx, "source_text": texts[idx], "translated_text": item.translated_text, "status": "completed", "usage": { "input_tokens": item.usage.input_tokens if hasattr(item, 'usage') else 0, "output_tokens": item.usage.output_tokens if hasattr(item, 'usage') else 0 } }) return results except Exception as e: logger.error(f"批量翻译失败: {str(e)}") raise Exception(f"批量翻译失败: {str(e)}") def submit_long_text_task( self, source_text: str, target_language: str, source_language: str = "auto", scene: str = "mt-turbo" ) -> str: """ 提交长文本翻译任务 Args: source_text: 源文本（最大100万字符） target_language: 目标语言 source_language: 源语言 scene: 模型类型 Returns: 任务ID """ try: request = anytrans_models.SubmitLongTextTranslateTaskRequest( workspace_id=self.workspace_id, source_language=source_language, target_language=target_language, source_text=source_text, scene=scene ) runtime = util_models.RuntimeOptions() response = self.client.submit_long_text_translate_task_with_options(request, runtime) if response.status_code != 200: raise Exception(f"API调用失败: {response.status_code}") return response.body.data.task_id except Exception as e: logger.error(f"提交长文本任务失败: {str(e)}") raise Exception(f"提交长文本任务失败: {str(e)}") def get_long_text_result(self, task_id: str) -> Dict[str, Any]: """ 获取长文本翻译结果 Args: task_id: 任务ID Returns: 任务结果 """ try: request = anytrans_models.GetLongTextTranslateTaskRequest( workspace_id=self.workspace_id, task_id=task_id ) runtime = util_models.RuntimeOptions() response = self.client.get_long_text_translate_task_with_options(request, runtime) if response.status_code != 200: raise Exception(f"API调用失败: {response.status_code}") body = response.body data = body.data return { "task_id": task_id, "status": data.status, "translated_text": data.translated_text if hasattr(data, 'translated_text') else None, "progress": data.progress if hasattr(data, 'progress') else 0, "error_message": data.error_message if hasattr(data, 'error_message') else None, "usage": { "input_tokens": data.usage.input_tokens if hasattr(data, 'usage') else 0, "output_tokens": data.usage.output_tokens if hasattr(data, 'usage') else 0 } if hasattr(data, 'usage') else None } except Exception as e: logger.error(f"获取长文本结果失败: {str(e)}") raise Exception(f"获取长文本结果失败: {str(e)}") def submit_image_task( self, image_url: str, target_languages: List[str], source_language: str = "auto", scene: str = "flash", terminologies: Optional[List[Dict[str, str]]] = None, domain_hint: Optional[str] = None ) -> str: """ 提交图片翻译任务 Args: image_url: 图片URL target_languages: 目标语言列表 source_language: 源语言 scene: 模型类型（general专业版/flash轻量版） terminologies: 术语干预列表 domain_hint: 领域提示 Returns: 任务ID """ try: # 构建基础请求 request = anytrans_models.SubmitImageTranslateTaskRequest( workspace_id=self.workspace_id, source_language=source_language, target_language=target_languages, text=image_url, scene=scene ) # 添加扩展参数 if terminologies or domain_hint: ext = anytrans_models.SubmitImageTranslateTaskRequestExt() if terminologies: term_objects = [] for term in terminologies: term_obj = anytrans_models.SubmitImageTranslateTaskRequestExtTerminologies() term_obj.src = term["src"] term_obj.tgt = term["tgt"] term_objects.append(term_obj) ext.terminologies = term_objects if domain_hint: ext.domain_hint = domain_hint request.ext = ext response = self.client.submit_image_translate_task(request) if response.status_code != 200: raise Exception(f"API调用失败: {response.status_code}") return response.body.data.task_id except Exception as e: logger.error(f"提交图片翻译任务失败: {str(e)}") raise Exception(f"提交图片翻译任务失败: {str(e)}") def get_image_result(self, task_id: str) -> Dict[str, Any]: """ 获取图片翻译结果 Args: task_id: 任务ID Returns: 任务结果 """ try: request = anytrans_models.GetImageTranslateTaskRequest( workspace_id=self.workspace_id, task_id=task_id ) response = self.client.get_image_translate_task(request) if response.status_code != 200: raise Exception(f"API调用失败: {response.status_code}") body = response.body data = body.data # 解析翻译结果 translation = data.translation if hasattr(data, 'translation') else None result = { "task_id": task_id, "status": "completed" if translation else "processing", "translation": None } if translation: result["translation"] = { "angle": translation.angle if hasattr(translation, 'angle') else 0, "width": translation.width if hasattr(translation, 'width') else 0, "height": translation.height if hasattr(translation, 'height') else 0, "boxes_count": translation.boxes_count if hasattr(translation, 'boxes_count') else 0, "bounding_boxes": [] } if hasattr(translation, 'bounding_boxes'): for box in translation.bounding_boxes: box_data = { "text": box.text if hasattr(box, 'text') else "", "translation": box.translation if hasattr(box, 'translation') else {}, "confidence": box.confidence if hasattr(box, 'confidence') else 0, "up_left": {"x": box.up_left.x, "y": box.up_left.y} if hasattr(box, 'up_left') else None, "up_right": {"x": box.up_right.x, "y": box.up_right.y} if hasattr(box, 'up_right') else None, "down_left": {"x": box.down_left.x, "y": box.down_left.y} if hasattr(box, 'down_left') else None, "down_right": {"x": box.down_right.x, "y": box.down_right.y} if hasattr(box, 'down_right') else None } result["translation"]["bounding_boxes"].append(box_data) return result except Exception as e: logger.error(f"获取图片翻译结果失败: {str(e)}") raise Exception(f"获取图片翻译结果失败: {str(e)}") def submit_doc_task( self, doc_url: str, target_language: str, source_language: str = "auto", scene: str = "mt-turbo", terminologies: Optional[List[Dict[str, str]]] = None, domain_hint: Optional[str] = None, skip_img_trans: bool = False ) -> str: """ 提交文档翻译任务 Args: doc_url: 文档URL target_language: 目标语言 source_language: 源语言 scene: 模型类型 terminologies: 术语干预列表 domain_hint: 领域提示 skip_img_trans: 是否跳过PDF中的图片翻译 Returns: 任务ID """ try: # 构建基础请求 request = anytrans_models.SubmitDocTranslateTaskRequest( workspace_id=self.workspace_id, source_language=source_language, target_language=target_language, text=doc_url, scene=scene ) # 添加扩展参数 if terminologies or domain_hint or skip_img_trans: ext = anytrans_models.SubmitDocTranslateTaskRequestExt() if terminologies: term_objects = [] for term in terminologies: term_obj = anytrans_models.SubmitDocTranslateTaskRequestExtTerminologies() term_obj.src = term["src"] term_obj.tgt = term["tgt"] term_objects.append(term_obj) ext.terminologies = term_objects if domain_hint: ext.domain_hint = domain_hint # 配置对象 config = anytrans_models.SubmitDocTranslateTaskRequestExtConfig() config.skip_img_trans = skip_img_trans ext.config = config request.ext = ext response = self.client.submit_doc_translate_task(request) if response.status_code != 200: raise Exception(f"API调用失败: {response.status_code}") return response.body.data.task_id except Exception as e: logger.error(f"提交文档翻译任务失败: {str(e)}") raise Exception(f"提交文档翻译任务失败: {str(e)}") def get_doc_result(self, task_id: str) -> Dict[str, Any]: """ 获取文档翻译结果 Args: task_id: 任务ID Returns: 任务结果 """ try: request = anytrans_models.GetDocTranslateTaskRequest( workspace_id=self.workspace_id, task_id=task_id ) response = self.client.get_doc_translate_task(request) if response.status_code != 200: raise Exception(f"API调用失败: {response.status_code}") body = response.body data = body.data result = { "task_id": task_id, "status": data.status if hasattr(data, 'status') else "processing" } # 如果翻译完成，添加结果信息 if hasattr(data, 'translate_file_url') and data.translate_file_url: result["translate_file_url"] = data.translate_file_url result["characters_count"] = data.characters_count if hasattr(data, 'characters_count') else 0 result["page_count"] = data.page_count if hasattr(data, 'page_count') else 0 return result except Exception as e: logger.error(f"获取文档翻译结果失败: {str(e)}") raise Exception(f"获取文档翻译结果失败: {str(e)}") def submit_html_task( self, html_content: str, target_language: str, source_language: str = "auto", scene: str = "mt-turbo", terminologies: Optional[List[Dict[str, str]]] = None, domain_hint: Optional[str] = None ) -> str: """ 提交HTML翻译任务 Args: html_content: HTML内容 target_language: 目标语言 source_language: 源语言 scene: 模型类型 terminologies: 术语干预列表 domain_hint: 领域提示 Returns: 任务ID """ try: # 构建基础请求 request = anytrans_models.SubmitHtmlTranslateTaskRequest( workspace_id=self.workspace_id, source_language=source_language, target_language=target_language, text=html_content, scene=scene ) # 添加扩展参数 if terminologies or domain_hint: ext = anytrans_models.SubmitHtmlTranslateTaskRequestExt() if terminologies: term_objects = [] for term in terminologies: term_obj = anytrans_models.SubmitHtmlTranslateTaskRequestExtTerminologies() term_obj.src = term["src"] term_obj.tgt = term["tgt"] term_objects.append(term_obj) ext.terminologies = term_objects if domain_hint: ext.domain_hint = domain_hint request.ext = ext response = self.client.submit_html_translate_task(request) if response.status_code != 200: raise Exception(f"API调用失败: {response.status_code}") return response.body.data.task_id except Exception as e: logger.error(f"提交HTML任务失败: {str(e)}") raise Exception(f"提交HTML任务失败: {str(e)}") def get_html_result(self, task_id: str) -> Dict[str, Any]: """ 获取HTML翻译结果 Args: task_id: 任务ID Returns: 任务结果 """ try: request = anytrans_models.GetHtmlTranslateTaskRequest( workspace_id=self.workspace_id, task_id=task_id ) response = self.client.get_html_translate_task(request) if response.status_code != 200: raise Exception(f"API调用失败: {response.status_code}") body = response.body # 检查是否有data字段 if not hasattr(body, 'data'): logger.warning(f"HTML翻译任务 {task_id} 暂无结果数据") return { "task_id": task_id, "status": "processing" } data = body.data # 检查是否有translation字段（翻译完成） if hasattr(data, 'translation') and data.translation: result = { "task_id": task_id, "status": "completed", "translated_html": data.translation } # 添加token使用统计 if hasattr(data, 'usage'): result["usage"] = { "input_tokens": data.usage.input_tokens if hasattr(data.usage, 'input_tokens') else 0, "output_tokens": data.usage.output_tokens if hasattr(data.usage, 'output_tokens') else 0, "total_tokens": data.usage.total_tokens if hasattr(data.usage, 'total_tokens') else 0 } return result else: # 任务还在处理中 return { "task_id": task_id, "status": "processing" } except Exception as e: error_msg = str(e) logger.error(f"获取HTML结果失败: {error_msg}") # 如果是服务内部异常，可能是任务还在处理中 if "Server.Internal.Error" in error_msg or "服务内部异常" in error_msg: logger.info(f"HTML翻译任务 {task_id} 可能还在处理中") return { "task_id": task_id, "status": "processing" } raise Exception(f"获取HTML结果失败: {error_msg}") def term_query( self, source_language: str, target_language: str, scene: str = "mt-turbo", text: Optional[str] = None ) -> List[Dict[str, str]]: """ 查询术语库 Args: source_language: 源语言 target_language: 目标语言 scene: 模型类型 text: 包含术语的句子（可选） Returns: 术语列表 """ try: request = anytrans_models.TermQueryRequest( workspace_id=self.workspace_id, source_language=source_language, target_language=target_language, scene=scene, text=text ) runtime = util_models.RuntimeOptions() response = self.client.term_query_with_options(request, runtime) if response.status_code != 200: raise Exception(f"API调用失败: {response.status_code}") body = response.body terms = [] if hasattr(body.data, 'terms'): for term in body.data.terms: terms.append({ "term_id": term.term_id, "src": term.src, "tgt": term.tgt }) return terms except Exception as e: logger.error(f"查询术语失败: {str(e)}") raise Exception(f"查询术语失败: {str(e)}") def term_edit( self, action: str, source_language: str, target_language: str, scene: str, terms: List[Dict[str, str]] ) -> List[Dict[str, str]]: """ 编辑术语库 Args: action: 操作类型（ADD/DELETE/MODIFY） source_language: 源语言 target_language: 目标语言 scene: 模型类型 terms: 术语列表 [{"src": "源文本", "tgt": "目标文本", "term_id": "术语ID"}] Returns: 操作后的术语列表 """ try: # 构建术语对象 term_objects = [] for term in terms: term_obj = anytrans_models.TermEditRequestTerms() term_obj.src = term["src"] term_obj.tgt = term["tgt"] if "term_id" in term: term_obj.term_id = term["term_id"] term_objects.append(term_obj) request = anytrans_models.TermEditRequest( workspace_id=self.workspace_id, action=action, source_language=source_language, target_language=target_language, scene=scene, terms=term_objects ) runtime = util_models.RuntimeOptions() response = self.client.term_edit_with_options(request, runtime) if response.status_code != 200: raise Exception(f"API调用失败: {response.status_code}") body = response.body result_terms = [] if hasattr(body.data, 'terms'): for term in body.data.terms: result_terms.append({ "term_id": term.term_id, "src": term.src, "tgt": term.tgt }) return result_terms except Exception as e: logger.error(f"编辑术语失败: {str(e)}") raise Exception(f"编辑术语失败: {str(e)}")