blog/scripts/translate/tencent_tmt.py

#!/usr/bin/env python3
"""腾讯云 TMT (机器翻译) TC3-HMAC-SHA256 签名客户端。

零第三方依赖（hashlib/hmac/urllib），供 translate_posts.py 编排调用；
--selftest 用官方签名文档 (cloud.tencent.com/document/api/213/30654) 公布的
cvm 示例中间量做逐字节断言，并打印 tmt 请求的完整签名链中间量（固定假密钥，
不触网），供与官方 API Explorer「签名串生成」交叉比对。

已知约束：单请求 SourceText 上限 2000 字符，分块由编排器负责。
"""

from __future__ import annotations

import hashlib
import hmac
import json
import sys
import time
import urllib.error
import urllib.request
from datetime import datetime, timezone

HOST = "tmt.tencentcloudapi.com"
SERVICE = "tmt"
ACTION = "TextTranslate"
VERSION = "2018-03-21"
REGION = "ap-guangzhou"  # TMT 为全球服务；该头不参与签名，仅路由用
CT = "application/json; charset=utf-8"
SIGNED_HEADERS = "content-type;host"  # 官方要求的最低签名头集合
SOURCE_TEXT_LIMIT = 2000
TIMEOUT = 30


def _sha256_hex(s: str) -> str:
    return hashlib.sha256(s.encode("utf-8")).hexdigest()


def _hmac(key: bytes, msg: str) -> bytes:
    return hmac.new(key, msg.encode("utf-8"), hashlib.sha256).digest()


def sign(secret_id: str, secret_key: str, action: str, service: str, host: str,
         payload: str, timestamp: int,
         extra_signed_headers: dict[str, str] | None = None) -> tuple[dict, dict]:
    """按 TC3 规范签名，返回 (请求头, 中间量字典)。

    extra_signed_headers: 除 content-type/host 外追加参与签名的头（key 小写），
        仅供 selftest 复现官方文档示例；正常调用不传。
    中间量字典含 canonical_request / string_to_sign / signature 等，
    供 selftest 展示；正常调用方可忽略。
    """
    date = datetime.fromtimestamp(timestamp, tz=timezone.utc).strftime("%Y-%m-%d")
    # Date 必须由时间戳换算 UTC 日期：带本地时区会在跨日时段必然签名失败（官方文档陷阱）
    credential_scope = f"{date}/{service}/tc3_request"

    signed = {"content-type": CT, "host": host}
    if extra_signed_headers:
        signed.update({k.lower(): v for k, v in extra_signed_headers.items()})
    signed_headers = ";".join(sorted(signed))
    # 头按 key 的 ASCII 升序拼接，每行 key:value\n 结尾（官方拼接规则）
    canonical_headers = "".join(f"{k}:{v}\n" for k, v in sorted(signed.items()))
    hashed_payload = _sha256_hex(payload)
    canonical_request = (
        f"POST\n/\n\n{canonical_headers}\n{signed_headers}\n{hashed_payload}"
    )

    hashed_canonical_request = _sha256_hex(canonical_request)
    string_to_sign = (
        f"TC3-HMAC-SHA256\n{timestamp}\n{credential_scope}\n{hashed_canonical_request}"
    )

    secret_date = _hmac(f"TC3{secret_key}".encode("utf-8"), date)
    secret_service = _hmac(secret_date, service)
    secret_signing = _hmac(secret_service, "tc3_request")
    signature = hmac.new(secret_signing, string_to_sign.encode("utf-8"),
                         hashlib.sha256).hexdigest()

    authorization = (
        f"TC3-HMAC-SHA256 Credential={secret_id}/{credential_scope}, "
        f"SignedHeaders={SIGNED_HEADERS}, Signature={signature}"
    )
    headers = {
        "Authorization": authorization,
        "Content-Type": CT,
        "Host": host,
        "X-TC-Action": action,
        "X-TC-Timestamp": str(timestamp),
        "X-TC-Version": VERSION,
        "X-TC-Region": REGION,
    }
    debug = {
        "timestamp": timestamp,
        "date": date,
        "credential_scope": credential_scope,
        "canonical_request": canonical_request,
        "hashed_request_payload": hashed_payload,
        "hashed_canonical_request": hashed_canonical_request,
        "string_to_sign": string_to_sign,
        "secret_date_hex": secret_date.hex(),
        "secret_service_hex": secret_service.hex(),
        "secret_signing_hex": secret_signing.hex(),
        "signature": signature,
        "authorization": authorization,
    }
    return headers, debug


def translate(secret_id: str, secret_key: str, text: str,
              source: str = "zh", target: str = "en",
              retries: int = 3) -> str:
    """调用 TextTranslate 并重试（指数退避 1/2/4s），返回 TargetText。

    网络异常、HTTP 非 2xx、响应含 Response.Error 均触发重试；
    耗尽后抛 RuntimeError（调用方按篇捕获，不阻断其它篇）。
    """
    if len(text) > SOURCE_TEXT_LIMIT:
        raise ValueError(f"SourceText 超限: {len(text)} > {SOURCE_TEXT_LIMIT}")

    payload = json.dumps({"Source": source, "Target": target,
                          "ProjectId": 0, "SourceText": text})
    last_err: Exception | None = None
    for attempt in range(retries):
        if attempt:
            time.sleep(2 ** (attempt - 1))
        try:
            headers, _ = sign(secret_id, secret_key, ACTION, SERVICE, HOST,
                              payload, int(time.time()))
            req = urllib.request.Request(f"https://{HOST}/", data=payload.encode("utf-8"),
                                         headers=headers, method="POST")
            with urllib.request.urlopen(req, timeout=TIMEOUT) as resp:
                body = json.loads(resp.read().decode("utf-8"))
            if "Error" in body.get("Response", {}):
                err = body["Response"]["Error"]
                raise RuntimeError(f"TMT {err.get('Code')}: {err.get('Message')}")
            return body["Response"]["TargetText"]
        except (urllib.error.URLError, urllib.error.HTTPError, OSError,
                RuntimeError, KeyError, ValueError) as e:
            last_err = e
    raise RuntimeError(f"重试 {retries} 次仍失败: {last_err}")


# ---- selftest：官方文档公布值断言 + 假密钥全链路展示（不触网） ----

# 官方签名文档 cvm 示例的输入与公布中间量（密钥打码故仅断言无密钥依赖的三组量）
_DOC_PAYLOAD = ('{"Limit": 1, "Filters": [{"Values": ["\\u672a\\u547d\\u540d"], '
                '"Name": "instance-name"}]}')
_DOC_HASHED_PAYLOAD = "35e9c5b0e3ae67532d3c9f17ead6c90222632e5b1ff7f6e89887f1398934f064"
_DOC_CANONICAL_REQUEST = (
    "POST\n/\n\n"
    "content-type:application/json; charset=utf-8\n"
    "host:cvm.tencentcloudapi.com\n"
    "x-tc-action:describeinstances\n"
    "\n"
    "content-type;host;x-tc-action\n"
    + _DOC_HASHED_PAYLOAD
)
_DOC_HASHED_CANONICAL = "7019a55be8395899b900fb5564e4200d984910f34794a27cb3fb7d10ff6a1e84"
_DOC_STRING_TO_SIGN = (
    "TC3-HMAC-SHA256\n1551113065\n2019-02-25/cvm/tc3_request\n" + _DOC_HASHED_CANONICAL
)


def selftest() -> int:
    """断言签名拼接与官方文档示例一致；失败 exit 1，全过 exit 0。"""
    failures = []

    # 官方签名文档 cvm 示例（三头规格）直接走 sign() 代码路径，逐字节复现规范串
    _, doc_debug = sign("AKIDDemo", "DemoKey", "DescribeInstances", "cvm",
                        "cvm.tencentcloudapi.com", _DOC_PAYLOAD, 1551113065,
                        extra_signed_headers={"x-tc-action": "describeinstances"})
    if doc_debug["canonical_request"] != _DOC_CANONICAL_REQUEST:
        failures.append(f"sign() 生成的 CanonicalRequest 与文档不符:\n{doc_debug['canonical_request']!r}")

    hp = _sha256_hex(_DOC_PAYLOAD)
    if hp != _DOC_HASHED_PAYLOAD:
        failures.append(f"HashedRequestPayload 不符: {hp}")

    hcr = _sha256_hex(doc_debug["canonical_request"])
    if hcr != _DOC_HASHED_CANONICAL:
        failures.append(f"HashedCanonicalRequest 不符: {hcr}")

    sts = f"TC3-HMAC-SHA256\n1551113065\n2019-02-25/cvm/tc3_request\n{hcr}"
    if sts != _DOC_STRING_TO_SIGN:
        failures.append(f"StringToSign 不符: {sts!r}")

    if failures:
        print("[SELFTEST-FAIL]")
        for f in failures:
            print("  -", f)
        return 1

    print("[SELFTEST-PASS] 官方文档 cvm 示例中间量逐字节一致："
          "HashedRequestPayload / HashedCanonicalRequest / StringToSign")
    print()

    # 固定假密钥 + tmt 真实参数，打印全链路中间量（可与官方 API Explorer 比对）
    fake_id, fake_key = "AKIDFAKEexample0000000000000000000000", "FakeSecretKeyDoNotUse00000"
    payload = json.dumps({"Source": "zh", "Target": "en", "ProjectId": 0,
                          "SourceText": "你好，世界"}, ensure_ascii=False)
    headers, debug = sign(fake_id, fake_key, ACTION, SERVICE, HOST, payload, 1551113065)
    print("[TMT-REQUEST 签名链中间量]（假密钥，供与官方 API Explorer 比对）")
    for k in ("timestamp", "date", "credential_scope", "canonical_request",
              "hashed_request_payload", "hashed_canonical_request", "string_to_sign",
              "secret_date_hex", "secret_service_hex", "secret_signing_hex",
              "signature", "authorization"):
        print(f"--- {k} ---")
        print(debug[k])
    print()
    print("--- 实际发送 HTTP 头 ---")
    for k, v in headers.items():
        print(f"{k}: {v}")
    print(f"(body) {payload}")
    return 0


if __name__ == "__main__":
    if "--selftest" in sys.argv:
        sys.exit(selftest())
    print("用法: python3 tencent_tmt.py --selftest；正常使用经 translate_posts.py 编排",
          file=sys.stderr)
    sys.exit(2)