translator.py

"""
translator.py
"""
import hashlib, hmac, json, os, random, time
from datetime import datetime
from typing import List, Sequence, Optional

import requests

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

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")
    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

    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)
    
    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()}"
    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}

    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']}")
            return None
        return data.get("Response", {}).get("TargetText", "").split("\n")
    except Exception as e:
        print(f"[translator] Tencent request 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

    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

    app_id, secret_key = creds["app_id"], creds["secret_key"]
    salt, query = random.randint(32768, 65536), "\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}

    try:
        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')}")
            return None
        return [item["dst"] for item in data.get("trans_result", [])]
    except Exception as e:
        print(f"[translator] Baidu request 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]:
    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)
    
    return out or list(texts)