Python对接TonaAi OpenAPI
- 请提前在main.py同目录放一个origin_file.docx和一个config.json
- 接口采用大厂同款标准化签名校验机制，以及时间戳验证，防止请求被恶意篡改，防止重放攻击，尽可能避免客户的账户余额损失，故代码实现方面稍复杂
requirements.txt
certifi==2025.8.3
cffi==1.17.1
charset-normalizer==3.4.3
cryptography==41.0.2
idna==3.10
pyasn1==0.6.1
pycparser==2.22
requests==2.32.4
urllib3==2.5.0
config.json配置文件
{
"base_url": "https://openapi.tonaai.cn",
"api_key": "您的ApiKey，你可以在https://api.tonaai.cn管理面板获取",
"tonaaiapi_public_key": "TonaAI生成的公钥，用于对TonaAI返回的信息进行签名校验，你可以在https://api.tonaai.cn管理面板获取秘钥",
"downstream_private_key": "您的私钥，用于在提交任务时进行签名，你可以用支付宝开放平台秘钥工具生成秘钥，并在https://api.tonaai.cn管理面板设置公钥，同时将私钥填写在这一行"
}
main.py
import os
import json
import requests
from datetime import datetime
from typing import List, Dict, Any, Optional
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.hazmat.backends import default_backend
import base64
import uuid
import logging
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class TonaAiClient:
def __init__(self, config: Dict[str, str]):
"""
:param config: 包含以下键的字典
- base_url: API 基地址
- api_key: ApiKey
- downstream_private_key: PEM 格式的私钥字符串
"""
self.base_url = config["base_url"]
self.api_key = config["api_key"]
self.app_id = "1001"
self.private_key = self._load_private_key(config["downstream_private_key"])
def _load_private_key(self, pem_key: str):
"""从 PEM 字符串加载私钥"""
pem_key = pem_key.strip()
if "-----BEGIN PRIVATE KEY-----" not in pem_key:
pem_key = "-----BEGIN PRIVATE KEY-----\n" + pem_key + "\n-----END PRIVATE KEY-----"
try:
private_key = serialization.load_pem_private_key(
pem_key.encode('utf-8'),
password=None,
backend=default_backend()
)
return private_key
except Exception as e:
raise ValueError(f"Failed to load private key: {e}")
def _sign(self, content: str) -> str:
"""使用私钥对内容进行 SHA256withRSA 签名，并返回 Base64 编码结果"""
try:
signature = self.private_key.sign(
content.encode('utf-8'),
padding.PKCS1v15(),
hashes.SHA256()
)
return base64.b64encode(signature).decode('utf-8')
except Exception as e:
logger.error(f"签名失败: {e}")
return None
def _build_sign_content(self, app_id: str, timestamp: str, nonce_str: str, data_json: str) -> str:
"""拼接待签名字符串（按字段字典序排序）"""
# 注意：实际应按字段名字典序排序！
parts = [
f"app_id={app_id}",
f"data={data_json}",
f"nonce_str={nonce_str}",
f"timestamp={timestamp}"
]
return "&".join(parts)
def _make_signed_request_body(self, data: Dict[str, Any]) -> str:
"""构造带签名的请求体 JSON 字符串"""
timestamp = datetime.utcnow().isoformat() + "Z" # 使用 UTC 时间，ISO 8601 格式
nonce_str = str(uuid.uuid4()).replace("-", "")[:16]
data_json = json.dumps(data, separators=(',', ':'), ensure_ascii=False) # 紧凑格式
sign_content = self._build_sign_content(self.app_id, timestamp, nonce_str, data_json)
logger.info(f"Sign Content: {sign_content}")
sign = self._sign(sign_content)
if not sign:
return None
request_body = {
"app_id": self.app_id,
"timestamp": timestamp,
"nonce_str": nonce_str,
"data": data,
"sign": sign
}
return json.dumps(request_body, ensure_ascii=False)
def upload_file(self, file_path: str, file_param_name: str = "file") -> Optional[str]:
"""上传文件，返回 fileId"""
if not os.path.exists(file_path):
logger.error(f"文件不存在: {file_path}")
return None
url = f"{self.base_url}/v1/file/upload"
headers = {"ApiKey": self.api_key}
try:
with open(file_path, 'rb') as f:
files = {file_param_name: (os.path.basename(file_path), f, 'application/octet-stream')}
response = requests.post(url, headers=headers, files=files, timeout=120)
if response.status_code == 200:
res_json = response.json()
if res_json.get("code") == 0:
return res_json["data"]["fileId"]
else:
logger.error(f"上传失败: {res_json.get('message')}")
else:
logger.error(f"请求失败，状态码: {response.status_code}")
except Exception as e:
logger.error(f"上传文件时发生错误: {e}")
return None
def post_deai_task(self, origin: str, type_: str, lang: str, mode: str) -> Optional[str]:
"""提交 AI 降重任务"""
data = {"origin": origin, "type": type_, "lang": lang, "mode": mode}
return self._post_task("/v1/task/reduce_ai_rate", data)
def post_rewrite_task(self, origin: str, type_: str, lang: str, mode: str) -> Optional[str]:
"""提交改写任务"""
data = {"origin": origin, "type": type_, "lang": lang, "mode": mode}
return self._post_task("/v1/task/reduce_rp_rate", data)
def post_aigc_check_task(self, title: str, author: str, file_id: int) -> Optional[str]:
"""提交 AIGC 检测任务"""
data = {"title": title, "author": author, "fileId": file_id}
return self._post_task("/v1/task/aigc_check", data)
def _post_task(self, endpoint: str, data: Dict[str, Any]) -> Optional[str]:
"""通用任务提交方法"""
body_str = self._make_signed_request_body(data)
if not body_str:
raise RuntimeError("提交任务失败")
url = f"{self.base_url}{endpoint}"
headers = {
"ApiKey": self.api_key,
"Content-Type": "application/json; charset=utf-8"
}
try:
response = requests.post(url, headers=headers, data=body_str.encode('utf-8'), timeout=10)
if response.status_code == 200:
res_json = response.json()
if res_json.get("code") == 0:
return res_json["data"]["taskId"]
else:
raise RuntimeError(f"提交任务失败: {res_json.get('message')}")
else:
raise RuntimeError(f"请求失败，状态码: {response.status_code}")
except Exception as e:
logger.error(f"请求异常: {e}")
raise RuntimeError("提交任务失败")
def get_task_result(self, task_ids: List[int]) -> List[Dict[str, Any]]:
"""批量获取任务结果"""
url = f"{self.base_url}/v1/use_record/batch_record"
body_str = json.dumps(task_ids, separators=(',', ':'))
headers = {
"ApiKey": self.api_key,
"Content-Type": "application/json; charset=utf-8"
}
try:
response = requests.post(url, headers=headers, data=body_str, timeout=10)
if response.status_code == 200:
res_json = response.json()
if res_json.get("code") == 0:
return res_json["data"]
else:
raise RuntimeError(f"获取任务结果失败: {res_json.get('message')}")
else:
raise RuntimeError(f"请求失败，状态码: {response.status_code}")
except Exception as e:
logger.error(f"请求异常: {e}")
raise RuntimeError("获取任务结果失败")
def get_balance(self) -> float:
"""获取余额"""
url = f"{self.base_url}/v1/user/balance"
headers = {"ApiKey": self.api_key}
try:
response = requests.get(url, headers=headers, timeout=10)
if response.status_code == 200:
res_json = response.json()
if res_json.get("code") == 0:
return float(res_json["data"])
else:
raise RuntimeError(f"获取余额失败: {res_json.get('message')}")
else:
raise RuntimeError(f"请求失败，状态码: {response.status_code}")
except Exception as e:
logger.error(f"请求异常: {e}")
raise RuntimeError("获取余额失败")
def aigcCheck(util):
fileId = util.upload_file("origin_file.docx")
print(f"成功上传文件：fileId:{fileId}")
taskId = util.post_aigc_check_task("标题", "作者", fileId)
print(f"成功提交任务：taskId:{taskId}")
def deaiText(util):
text = ("春日的序曲 清晨，阳光透过薄雾洒在大地上，万物仿佛被镀上了一层柔和的金色。微风轻拂，带着一丝凉意，却也夹杂着泥土与青草的清新气息。"
"远处的山峦在晨光中若隐若现，如同一幅淡雅的水墨画，静谧而悠远。田野间，露珠在嫩绿的草叶上闪烁，晶莹剔透，仿佛是大自然精心点缀的珍珠。"
"鸟儿在枝头欢快地鸣叫，清脆的歌声在空气中回荡，为这宁静的清晨增添了几分生机。")
taskId = util.post_deai_task(text, "text", "cn", "normal")
print(f"成功提交任务：taskId:{taskId}")
def deaiFile(util):
fileId = util.upload_file("origin_file.docx")
print(f"成功上传文件：fileId:{fileId}")
taskId = util.post_deai_task(str(fileId), "file", "cn", "normal")
print(f"成功提交任务：taskId:{taskId}")
def rewriteText(util):
text = ("春日的序曲 清晨，阳光透过薄雾洒在大地上，万物仿佛被镀上了一层柔和的金色。微风轻拂，带着一丝凉意，却也夹杂着泥土与青草的清新气息。"
"远处的山峦在晨光中若隐若现，如同一幅淡雅的水墨画，静谧而悠远。田野间，露珠在嫩绿的草叶上闪烁，晶莹剔透，仿佛是大自然精心点缀的珍珠。"
"鸟儿在枝头欢快地鸣叫，清脆的歌声在空气中回荡，为这宁静的清晨增添了几分生机。")
taskId = util.post_rewrite_task(text, "text", "cn", "normal")
print(f"成功提交任务：taskId:{taskId}")
def rewriteFile(util):
fileId = util.upload_file("origin_file.docx")
print(f"成功上传文件：fileId:{fileId}")
taskId = util.post_rewrite_task(str(fileId), "file", "cn", "normal")
print(f"成功提交任务：taskId:{taskId}")
def rewriteTextPowerful(util):
text = ("春日的序曲 清晨，阳光透过薄雾洒在大地上，万物仿佛被镀上了一层柔和的金色。微风轻拂，带着一丝凉意，却也夹杂着泥土与青草的清新气息。"
"远处的山峦在晨光中若隐若现，如同一幅淡雅的水墨画，静谧而悠远。田野间，露珠在嫩绿的草叶上闪烁，晶莹剔透，仿佛是大自然精心点缀的珍珠。"
"鸟儿在枝头欢快地鸣叫，清脆的歌声在空气中回荡，为这宁静的清晨增添了几分生机。")
taskId = util.post_rewrite_task(text, "text", "cn", "powerful")
print(f"成功提交任务：taskId:{taskId}")
def rewriteFilePowerful(util):
fileId = util.upload_file("origin_file.docx")
print(f"成功上传文件：fileId:{fileId}")
taskId = util.post_rewrite_task(str(fileId), "file", "cn", "powerful")
print(f"成功提交任务：taskId:{taskId}")
def getResult(util):
# task_ids里面的元素是之前的步骤中提交任务返回的taskId，每次最大查询100条
task_ids=[751826577758752768,751826610038116352,751826632611860480]
result_list=util.get_task_result(task_ids)
for result in result_list:
print(result)
if __name__ == '__main__':
with open('config.json', 'r', encoding='utf-8') as f:
config = json.load(f)
util = TonaAiClient(config)
'''
请提前在main.py同目录放一个origin_file.docx
1. AIGC检测：aigcCheck(util)
2. 文本降AI：deaiText(util)
3. 文件降AI：deaiFile(util)
4. 文本降重：rewriteText(util)
5. 文件降重：rewriteFile(util)
6. 文本强力降重：rewriteTextPowerful(util)
7. 文件强力降重：rewriteFilePowerful(util)
8. 查询任务结果：getResult(util)
'''