Update translator.py

translator.py

@@ -1,48 +1,50 @@
 """
 translator.py
-腾讯云 + 百度翻译 API 轮询封装
-⚠️ 需在 HF 空间的 “Variables” 页设置以下环境变量
+腾讯云 (批量接口+动态分包+多账号轮询) + 百度翻译 API 封装
+⚠️ 需在 HF 空间的 “Variables” 页设置以下环境变量
 ------------------------------------------------------------------
-TENCENT_SECRET_ID        腾讯云 SecretId
-TENCENT_SECRET_KEY       腾讯云 SecretKey
-TENCENT_TRANSLATE_URL    (可选) 默认 https://tmt.tencentcloudapi.com
-BAIDU_TRANSLATE_URL      (可选) 默认 https://fanyi-api.baidu.com/api/trans/vip/translate
-BAIDU_CREDENTIALS_JSON   形如:
+TENCENT_CREDENTIALS_JSON  形如:
 [
-  {"app_id": "xxxx", "secret_key": "yyyy"},
-  {"app_id": "aaaa", "secret_key": "bbbb"}
+  {"secret_id": "AKIDxxxx", "secret_key": "yyyy"},
+  {"secret_id": "AKIDaaaa", "secret_key": "bbbb"}
+]
+TENCENT_SECRET_ID         (兼容旧配置) 单个 SecretId
+TENCENT_SECRET_KEY        (兼容旧配置) 单个 SecretKey
+------------------------------------------------------------------
+BAIDU_CREDENTIALS_JSON    形如:
+[
+  {"app_id": "xxxx", "secret_key": "yyyy"}
 ]
 ------------------------------------------------------------------
 """
 import hashlib, hmac, json, os, random, time
 from datetime import datetime
-from typing import List, Sequence, Optional
+from typing import List, Sequence, Optional, Dict, Any
 
 import requests
 
-# ------------------------------------------------------------------
-# 读取环境变量
-# ------------------------------------------------------------------
-TENCENT_SECRET_ID  = os.environ.get("TENCENT_SECRET_ID")
-TENCENT_SECRET_KEY = os.environ.get("TENCENT_SECRET_KEY")
+_tencent_creds_list = json.loads(os.environ.get("TENCENT_CREDENTIALS_JSON", "[]"))
+
+# 兼容旧的单账号配置
+_legacy_id = os.environ.get("TENCENT_SECRET_ID")
+_legacy_key = os.environ.get("TENCENT_SECRET_KEY")
+if _legacy_id and _legacy_key:
+    if not any(c.get("secret_id") == _legacy_id for c in _tencent_creds_list):
+        _tencent_creds_list.append({"secret_id": _legacy_id, "secret_key": _legacy_key})
+
 TENCENT_TRANSLATE_URL = os.environ.get("TENCENT_TRANSLATE_URL", "https://tmt.tencentcloudapi.com")
 
+_tencent_idx: int = 0
+
+
 BAIDU_TRANSLATE_URL = os.environ.get("BAIDU_TRANSLATE_URL", "https://fanyi-api.baidu.com/api/trans/vip/translate")
-BAIDU_CREDENTIALS   = json.loads(os.environ.get("BAIDU_CREDENTIALS_JSON", "[]"))
+_baidu_creds_list = json.loads(os.environ.get("BAIDU_CREDENTIALS_JSON", "[]"))
 
-# 内部轮询索引
+# 全局索引，用于轮询
 _baidu_idx: int = 0
-def _next_baidu_cred():
-    global _baidu_idx
-    if not BAIDU_CREDENTIALS:
-        return None
-    cred = BAIDU_CREDENTIALS[_baidu_idx]
-    _baidu_idx = (_baidu_idx + 1) % len(BAIDU_CREDENTIALS)
-    return cred
 
-# ------------------------------------------------------------------
-# 腾讯翻译
-# ------------------------------------------------------------------
+
+# 腾讯云翻译逻辑 (批量接口 TextTranslateBatch)
 def _sign(key: bytes, msg: str) -> bytes:
     return hmac.new(key, msg.encode("utf-8"), hashlib.sha256).digest()
 
@@ -52,13 +54,16 @@ def _tc3_signature(secret_key: str, date: str, service: str, string_to_sign: str
     secret_signing  = _sign(secret_service, "tc3_request")
     return hmac.new(secret_signing, string_to_sign.encode("utf-8"), hashlib.sha256).hexdigest()
 
-def _translate_with_tencent(texts: Sequence[str], src="auto", tgt="zh") -> Optional[List[str]]:
-    """优先使用腾讯云翻译。失败返回 None"""
-    if not (TENCENT_SECRET_ID and TENCENT_SECRET_KEY):
-        return None                                           # 未配置凭证
+def _call_tencent_batch_once(cred: Dict[str, str], text_list: List[str], src: str, tgt: str) -> List[str]:
+    """
+    调用腾讯云 TextTranslateBatch 接口
+    """
+    secret_id = cred["secret_id"]
+    secret_key = cred["secret_key"]
+    
     service  = "tmt"
     host     = "tmt.tencentcloudapi.com"
-    action   = "TextTranslate"
+    action   = "TextTranslateBatch" 
     version  = "2018-03-21"
     region   = "ap-beijing"
     ts       = int(time.time())
@@ -66,14 +71,14 @@ def _translate_with_tencent(texts: Sequence[str], src="auto", tgt="zh") -> Optio
     algorithm = "TC3-HMAC-SHA256"
 
     payload = {
-        "SourceText": "\n".join(texts),
+        "SourceTextList": text_list, 
         "Source":     src,
         "Target":     tgt,
         "ProjectId":  0,
     }
+    
     payload_str = json.dumps(payload, ensure_ascii=False)
 
-    # ---------- step‑1 canonical request ----------
     canonical_request = "\n".join([
         "POST",
         "/",
@@ -83,19 +88,16 @@ def _translate_with_tencent(texts: Sequence[str], src="auto", tgt="zh") -> Optio
         hashlib.sha256(payload_str.encode()).hexdigest(),
     ])
 
-    # ---------- step‑2 string to sign ----------
-    credential_scope   = f"{date}/{service}/tc3_request"
-    string_to_sign     = "\n".join([
+    credential_scope    = f"{date}/{service}/tc3_request"
+    string_to_sign      = "\n".join([
         algorithm, str(ts), credential_scope,
         hashlib.sha256(canonical_request.encode()).hexdigest(),
     ])
 
-    # ---------- step‑3 signature ----------
-    signature = _tc3_signature(TENCENT_SECRET_KEY, date, service, string_to_sign)
+    signature = _tc3_signature(secret_key, date, service, string_to_sign)
 
-    # ---------- step‑4 headers ----------
     authorization = (
-        f"{algorithm} Credential={TENCENT_SECRET_ID}/{credential_scope}, "
+        f"{algorithm} Credential={secret_id}/{credential_scope}, "
         f"SignedHeaders=content-type;host;x-tc-action, Signature={signature}"
     )
     headers = {
@@ -108,27 +110,103 @@ def _translate_with_tencent(texts: Sequence[str], src="auto", tgt="zh") -> Optio
         "X-TC-Region":    region,
     }
 
-    # ---------- request ----------
-    try:
-        resp = requests.post(TENCENT_TRANSLATE_URL, headers=headers, data=payload_str, timeout=8)
-        resp.raise_for_status()
-        data = resp.json()
-        return data["Response"]["TargetText"].split("\n")
-    except Exception as e:
-        print(f"[translator] Tencent API error → {e}")
+    resp = requests.post(TENCENT_TRANSLATE_URL, headers=headers, data=payload_str, timeout=8)
+    resp.raise_for_status()
+    data = resp.json()
+    
+    if "Response" in data and "Error" in data["Response"]:
+        err_code = data["Response"]["Error"].get("Code", "")
+        err_msg = data["Response"]["Error"].get("Message", "")
+        raise Exception(f"Tencent Biz Error: {err_code} - {err_msg}")
+
+    return data["Response"]["TargetTextList"]
+
+
+def _translate_with_tencent_pool(texts: Sequence[str], src="auto", tgt="zh") -> Optional[List[str]]:
+    """
+    腾讯云入口：
+    1. 动态分包：同时考虑字符数限制 (<6000) 和 条数限制。
+       - 累计字符数 < 5000 (安全阈值)
+       - 单批次条数 < 50 (安全阈值)
+    2. 账号轮询 (Polling)：每组请求如果失败，会自动换号重试。
+    """
+    global _tencent_idx, _tencent_creds_list
+    
+    if not _tencent_creds_list:
         return None
 
-# ------------------------------------------------------------------
-# 百度翻译
-# ------------------------------------------------------------------
-def _translate_with_baidu(texts: Sequence[str], src="auto", tgt="zh") -> Optional[List[str]]:
-    creds = _next_baidu_cred()
-    if creds is None:
-        return None                                           # 未配置凭证
-    app_id, secret_key = creds["app_id"], creds["secret_key"]
+    # 配置安全阈值
+    MAX_CHARS_PER_BATCH = 5000  # 官方限制 6000，留 1000 buffer
+    MAX_ITEMS_PER_BATCH = 50    # 避免单次数组过大
+
+    chunks = []
+    current_chunk = []
+    current_char_count = 0
+    
+    for text in texts:
+        text_len = len(text)
+        
+        # 检查加入当前文本是否会超限
+        if current_chunk and (
+            (current_char_count + text_len > MAX_CHARS_PER_BATCH) or 
+            (len(current_chunk) >= MAX_ITEMS_PER_BATCH)
+        ):
+            # 结算当前块
+            chunks.append(current_chunk)
+            current_chunk = []
+            current_char_count = 0
+            
+        current_chunk.append(text)
+        current_char_count += text_len
+        
+    # 处理剩余的最后一块
+    if current_chunk:
+        chunks.append(current_chunk)
+
+    all_results = []
+    
+    for chunk in chunks:
+        chunk_success = False
+        
+        attempts = len(_tencent_creds_list)
+        for _ in range(attempts):
+            cred = _tencent_creds_list[_tencent_idx]
+            _tencent_idx = (_tencent_idx + 1) % len(_tencent_creds_list)
+            
+            try:
+                res = _call_tencent_batch_once(cred, list(chunk), src, tgt)
+                all_results.extend(res)
+                chunk_success = True
+                break # 成功则跳出重试
+            except Exception as e:
+                safe_id = cred['secret_id'][:4] + "****"
+                print(f"[translator] Tencent ID {safe_id} failed on batch: {e}. Switching...")
+                continue
+        
+        if not chunk_success:
+            print("[translator] All Tencent credentials failed for a batch. Falling back to Baidu.")
+            return None # 只要有一个分片失败，整体降级，保证一致性
+
+    return all_results
+
+
+# 百度翻译逻辑
+def _translate_with_baidu_pool(texts: Sequence[str], src="auto", tgt="zh") -> Optional[List[str]]:
+    global _baidu_idx, _baidu_creds_list
+    
+    if not _baidu_creds_list:
+        return None
+        
+    cred = _baidu_creds_list[_baidu_idx]
+    _baidu_idx = (_baidu_idx + 1) % len(_baidu_creds_list)
+    
+    app_id, secret_key = cred["app_id"], cred["secret_key"]
     salt  = random.randint(32768, 65536)
+    
     query = "\n".join(texts)
+    
     sign  = hashlib.md5((app_id + query + str(salt) + secret_key).encode()).hexdigest()
+    
     params = {
         "q": query, "from": src, "to": tgt,
         "appid": app_id, "salt": salt, "sign": sign,
@@ -137,24 +215,32 @@ def _translate_with_baidu(texts: Sequence[str], src="auto", tgt="zh") -> Optiona
         resp = requests.get(BAIDU_TRANSLATE_URL, params=params, timeout=8)
         resp.raise_for_status()
         data = resp.json()
+        if "error_code" in data:
+             raise Exception(f"Baidu Biz Error: {data['error_code']} - {data.get('error_msg')}")
         return [item["dst"] for item in data["trans_result"]]
     except Exception as e:
         print(f"[translator] Baidu API error → {e}")
         return None
 
-# ------------------------------------------------------------------
-# 对外统一函数
-# ------------------------------------------------------------------
+
+# 对外统一入口
 def translate_texts(texts: Sequence[str],
                     src_lang: str = "auto",
                     tgt_lang: str = "zh") -> List[str]:
     """
-    优先 Tencent → 失败再 Baidu → 如果都失败，返回原文。
+    逻辑：
+    1. 尝试腾讯云 (批量接口 + 多账号轮询)
+    2. 失败降级到百度云
+    3. 还失败返回原文
     """
     if not texts:
         return []
 
-    out = _translate_with_tencent(texts, src_lang, tgt_lang)
+    # 1. 优先尝试腾讯云
+    out = _translate_with_tencent_pool(texts, src_lang, tgt_lang)
+    
+    # 2. 失败降级到百度
     if out is None:
-        out = _translate_with_baidu(texts, src_lang, tgt_lang)
-    return out or list(texts)
+        out = _translate_with_baidu_pool(texts, src_lang, tgt_lang)
+        
+    return out or list(texts)