#!/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)