translator.py

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

import requests

DEFAULT_TENCENT_SECRET_ID  = os.environ.get("TENCENT_SECRET_ID")
DEFAULT_TENCENT_SECRET_KEY = os.environ.get("TENCENT_SECRET_KEY")
DEFAULT_TENCENT_URL = os.environ.get("TENCENT_TRANSLATE_URL", "https://tmt.tencentcloudapi.com")

DEFAULT_BAIDU_URL = os.environ.get("BAIDU_TRANSLATE_URL", "https://fanyi-api.baidu.com/api/trans/vip/translate")
try:
    DEFAULT_BAIDU_CREDENTIALS = json.loads(os.environ.get("BAIDU_CREDENTIALS_JSON", "[]"))
except json.JSONDecodeError:
    DEFAULT_BAIDU_CREDENTIALS = []

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_dynamic(
    texts: Sequence[str],
    secret_id: str,
    secret_key: str,
    url: str,
    src="auto",
    tgt="zh"
) -> 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 = int(time.time())
    date = datetime.utcfromtimestamp(ts).strftime("%Y-%m-%d")
    algorithm = "TC3-HMAC-SHA256"

    payload = {"SourceText": "\n".join(texts), "Source": src, "Target": tgt, "ProjectId": 0}
    payload_str = json.dumps(payload, ensure_ascii=False)

    canonical_request = "\n".join([
        "POST", "/", "", f"content-type:application/json; charset=utf-8\nhost:{host}\nx-tc-action:{action.lower()}\n",
        "content-type;host;x-tc-action", hashlib.sha256(payload_str.encode()).hexdigest(),
    ])
    credential_scope = f"{date}/{service}/tc3_request"
    string_to_sign = "\n".join([
        algorithm, str(ts), credential_scope, hashlib.sha256(canonical_request.encode()).hexdigest(),
    ])
    signature = _tc3_signature(secret_key, date, service, string_to_sign)
    authorization = (
        f"{algorithm} 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(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] Dynamic Tencent API error → {e}")
        return None

def _translate_with_baidu_dynamic(
    texts: Sequence[str],
    baidu_creds_list: List[Dict[str, str]],
    url: str,
    src="auto",
    tgt="zh"
) -> Optional[List[str]]:
    if not baidu_creds_list:
        return None
    
    cred = random.choice(baidu_creds_list)
    app_id, secret_key = cred.get("app_id"), cred.get("secret_key")
    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}
    try:
        resp = requests.get(url, params=params, timeout=8)
        resp.raise_for_status()
        data = resp.json()
        if "trans_result" in data:
            return [item["dst"] for item in data["trans_result"]]
        else:
            print(f"[translator] Baidu API returned error: {data.get('error_msg', 'Unknown error')}")
            return None
    except Exception as e:
        print(f"[translator] Dynamic Baidu API error → {e}")
        return None

def translate_texts_with_dynamic_keys(
    texts: Sequence[str],
    tencent_id: str,
    tencent_key: str,
    baidu_json_str: str,
    src_lang: str = "auto",
    tgt_lang: str = "zh"
) -> List[str]:
    if not texts:
        return []

    if tencent_id and tencent_key:
        out = _translate_with_tencent_dynamic(texts, tencent_id, tencent_key, DEFAULT_TENCENT_URL, src_lang, tgt_lang)
        if out is not None:
            return out
    
    baidu_creds = []
    if baidu_json_str:
        try:
            baidu_creds = json.loads(baidu_json_str)
            if not isinstance(baidu_creds, list): baidu_creds = []
        except json.JSONDecodeError:
            print("[translator] Invalid Baidu JSON provided by user.")
            baidu_creds = []

    if baidu_creds:
        out = _translate_with_baidu_dynamic(texts, baidu_creds, DEFAULT_BAIDU_URL, src_lang, tgt_lang)
        if out is not None:
            return out

    return list(texts)

def translate_texts(texts: Sequence[str],
                    src_lang: str = "auto",
                    tgt_lang: str = "zh") -> List[str]:
    return translate_texts_with_dynamic_keys(
        texts,
        DEFAULT_TENCENT_SECRET_ID,
        DEFAULT_TENCENT_SECRET_KEY,
        json.dumps(DEFAULT_BAIDU_CREDENTIALS), # 转换回 JSON 字符串以匹配接口
        src_lang,
        tgt_lang
    )