Update translator.py

translator.py

@@ -1,76 +1,83 @@
-"""
-translator.py
-"""
 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
 
+TRANSLATOR_ACCESS_KEY = os.environ.get("TRANSLATOR_ACCESS_KEY")
+
+TENCENT_SECRET_ID = os.environ.get("TENCENT_SECRET_ID")
+TENCENT_SECRET_KEY = os.environ.get("TENCENT_SECRET_KEY")
 TENCENT_TRANSLATE_URL = os.environ.get("TENCENT_TRANSLATE_URL", "https://tmt.tencentcloudapi.com")
-BAIDU_TRANSLATE_URL = os.environ.get("BAIDU_TRANSLATE_URL", "https://fanyi-api.baidu.com/api/trans/vip/translate")
 
-_baidu_idx: int = 0
-def _next_baidu_cred(creds_list):
-    global _baidu_idx
-    if not creds_list: return None
-    cred = creds_list[_baidu_idx]
-    _baidu_idx = (_baidu_idx + 1) % len(creds_list)
-    return cred
+BAIDU_TRANSLATE_URL = os.environ.get("BAIDU_TRANSLATE_URL", "https://fanyi-api.baidu.com/api/trans/vip/translate")
+try:
+    BAIDU_CREDENTIALS_DEFAULT = json.loads(os.environ.get("BAIDU_CREDENTIALS_JSON", "[]"))
+except json.JSONDecodeError:
+    BAIDU_CREDENTIALS_DEFAULT = []
 
 def _sign(key: bytes, msg: str) -> bytes:
     return hmac.new(key, msg.encode("utf-8"), hashlib.sha256).digest()
 
 def _tc3_signature(secret_key: str, date: str, service: str, string_to_sign: str) -> str:
-    secret_date     = _sign(("TC3" + secret_key).encode(), date)
-    secret_service  = _sign(secret_date, service)
-    secret_signing  = _sign(secret_service, "tc3_request")
+    secret_date = _sign(("TC3" + secret_key).encode(), date)
+    secret_service = _sign(secret_date, service)
+    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: str, tgt: str,
-                           secret_id: Optional[str],
-                           secret_key: Optional[str]) -> Optional[List[str]]:
-    """使用传入的密钥执行腾讯云翻译。如果密钥无效则失败。"""
-    if not (secret_id and secret_key):
-        return None
+def _translate_with_tencent(texts: Sequence[str], src: str, tgt: str, secret_id: str, secret_key: str) -> Optional[List[str]]:
+    if not (secret_id and secret_key): return None
 
     service, host, action, version, region = "tmt", "tmt.tencentcloudapi.com", "TextTranslate", "2018-03-21", "ap-beijing"
-    ts, date = int(time.time()), datetime.utcfromtimestamp(int(time.time())).strftime("%Y-%m-%d")
-    algorithm = "TC3-HMAC-SHA256"
-    payload_str = json.dumps({"SourceText": "\n".join(texts), "Source": src, "Target": tgt, "ProjectId": 0}, ensure_ascii=False)
+    ts = int(time.time())
+    date = datetime.utcfromtimestamp(ts).strftime("%Y-%m-%d")
+    
+    payload = {"SourceText": "\n".join(texts), "Source": src, "Target": tgt, "ProjectId": 0}
+    payload_str = json.dumps(payload, ensure_ascii=False)
+    
+    canonical_request = "POST\n/\n\n" + \
+                        f"content-type:application/json; charset=utf-8\nhost:{host}\nx-tc-action:{action.lower()}\n\n" + \
+                        "content-type;host;x-tc-action\n" + \
+                        hashlib.sha256(payload_str.encode()).hexdigest()
     
-    canonical_request = f"POST\n/\n\ncontent-type:application/json; charset=utf-8\nhost:{host}\nx-tc-action:{action.lower()}\n\ncontent-type;host;x-tc-action\n{hashlib.sha256(payload_str.encode()).hexdigest()}"
     credential_scope = f"{date}/{service}/tc3_request"
-    string_to_sign = f"{algorithm}\n{ts}\n{credential_scope}\n{hashlib.sha256(canonical_request.encode()).hexdigest()}"
+    string_to_sign = "TC3-HMAC-SHA256\n" + str(ts) + "\n" + credential_scope + "\n" + \
+                     hashlib.sha256(canonical_request.encode()).hexdigest()
+    
     signature = _tc3_signature(secret_key, date, service, string_to_sign)
     
-    authorization = f"{algorithm} Credential={secret_id}/{credential_scope}, SignedHeaders=content-type;host;x-tc-action, Signature={signature}"
-    headers = {"Authorization": authorization, "Content-Type": "application/json; charset=utf-8", "Host": host, "X-TC-Action": action, "X-TC-Timestamp": str(ts), "X-TC-Version": version, "X-TC-Region": region}
+    authorization = f"TC3-HMAC-SHA256 Credential={secret_id}/{credential_scope}, " + \
+                    f"SignedHeaders=content-type;host;x-tc-action, Signature={signature}"
+    
+    headers = {
+        "Authorization": authorization, "Content-Type": "application/json; charset=utf-8",
+        "Host": host, "X-TC-Action": action, "X-TC-Timestamp": str(ts),
+        "X-TC-Version": version, "X-TC-Region": region,
+    }
 
     try:
         resp = requests.post(TENCENT_TRANSLATE_URL, headers=headers, data=payload_str, timeout=8)
         resp.raise_for_status()
         data = resp.json()
-        if "Error" in data.get("Response", {}):
-            print(f"[translator] Tencent API error: {data['Response']['Error']['Message']}")
+        if "Response" in data and "TargetText" in data["Response"]:
+            return data["Response"]["TargetText"].split("\n")
+        else:
+            print(f"[translator] Tencent API abnormal response: {data}")
             return None
-        return data.get("Response", {}).get("TargetText", "").split("\n")
     except Exception as e:
-        print(f"[translator] Tencent request error: {e}")
+        print(f"[translator] Tencent API error → {e}")
         return None
 
-def _translate_with_baidu(texts: Sequence[str], src: str, tgt: str,
-                         baidu_credentials_list: Optional[list]) -> Optional[List[str]]:
-    if not baidu_credentials_list:
-        return None
+def _translate_with_baidu(texts: Sequence[str], src: str, tgt: str, baidu_credentials: List[Dict[str, str]]) -> Optional[List[str]]:
+    if not baidu_credentials: return None
 
-    creds = _next_baidu_cred(baidu_credentials_list)
-    if not creds or not creds.get("app_id") or not creds.get("secret_key"):
-        print("[translator] Baidu credentials format error.")
-        return None
+    cred = random.choice(baidu_credentials)
+    app_id, secret_key = cred.get("app_id"), cred.get("secret_key")
 
-    app_id, secret_key = creds["app_id"], creds["secret_key"]
-    salt, query = random.randint(32768, 65536), "\n".join(texts)
+    if not (app_id and secret_key): return None
+    
+    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}
 
@@ -78,25 +85,57 @@ def _translate_with_baidu(texts: Sequence[str], src: str, tgt: str,
         resp = requests.get(BAIDU_TRANSLATE_URL, params=params, timeout=8)
         resp.raise_for_status()
         data = resp.json()
-        if "error_code" in data:
-            print(f"[translator] Baidu API error: {data.get('error_msg', 'Unknown error')}")
+        if "trans_result" in data:
+            return [item["dst"] for item in data["trans_result"]]
+        else:
+            print(f"[translator] Baidu API abnormal response: {data}")
             return None
-        return [item["dst"] for item in data.get("trans_result", [])]
     except Exception as e:
-        print(f"[translator] Baidu request error: {e}")
+        print(f"[translator] Baidu API error → {e}")
         return None
 
+
 def translate_texts(texts: Sequence[str],
                     src_lang: str = "auto",
                     tgt_lang: str = "zh",
-                    tencent_secret_id: Optional[str] = None,
-                    tencent_secret_key: Optional[str] = None,
-                    baidu_credentials_list: Optional[list] = None) -> List[str]:
+                    system_key_input: Optional[str] = None,
+                    tencent_id: Optional[str] = None,
+                    tencent_key: Optional[str] = None,
+                    baidu_creds_json_str: Optional[str] = None) -> List[str]:
     if not texts:
         return []
 
-    out = _translate_with_tencent(texts, src_lang, tgt_lang, tencent_secret_id, tencent_secret_key)
-    if out is None:
-        out = _translate_with_baidu(texts, src_lang, tgt_lang, baidu_credentials_list)
+    use_tencent_id, use_tencent_key = None, None
+    use_baidu_creds = []
+
+    if tencent_id and tencent_key:
+        use_tencent_id, use_tencent_key = tencent_id, tencent_key
+        print("[translator] Using custom Tencent API key.")
+    if baidu_creds_json_str:
+        try:
+            creds = json.loads(baidu_creds_json_str)
+            if isinstance(creds, list) and all(isinstance(d, dict) for d in creds):
+                use_baidu_creds = creds
+                print("[translator] Using custom Baidu API key(s).")
+            else:
+                 print("[translator] Warning: Custom Baidu credentials format is incorrect.")
+        except json.JSONDecodeError:
+            print("[translator] Warning: Failed to parse custom Baidu credentials JSON.")
+
+    elif TRANSLATOR_ACCESS_KEY and system_key_input == TRANSLATOR_ACCESS_KEY:
+        print("[translator] System access key validated. Using system-configured API keys.")
+        use_tencent_id, use_tencent_key = TENCENT_SECRET_ID, TENCENT_SECRET_KEY
+        use_baidu_creds = BAIDU_CREDENTIALS_DEFAULT
+    
+    else:
+        print("[translator] Translation disabled: No valid API keys or system key provided.")
+        return list(texts)
+
+    translated_texts = None
+    if use_tencent_id and use_tencent_key:
+        translated_texts = _translate_with_tencent(texts, src_lang, tgt_lang, use_tencent_id, use_tencent_key)
     
-    return out or list(texts)
+    if translated_texts is None and use_baidu_creds:
+        translated_texts = _translate_with_baidu(texts, src_lang, tgt_lang, use_baidu_creds)
+
+    return translated_texts or list(texts)