222 lines
8.1 KiB
Python
222 lines
8.1 KiB
Python
# coding:utf-8
|
||
"""
|
||
Copyright (year) Beijing Volcano Engine Technology Ltd.
|
||
|
||
Licensed under the Apache License, Version 2.0 (the "License");
|
||
you may not use this file except in compliance with the License.
|
||
You may obtain a copy of the License at
|
||
|
||
http://www.apache.org/licenses/LICENSE-2.0
|
||
|
||
Unless required by applicable law or agreed to in writing, software
|
||
distributed under the License is distributed on an "AS IS" BASIS,
|
||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||
See the License for the specific language governing permissions and
|
||
limitations under the License.
|
||
"""
|
||
|
||
import datetime
|
||
import hashlib
|
||
import hmac
|
||
import json
|
||
from urllib.parse import quote
|
||
|
||
import requests
|
||
|
||
# 以下参数视服务不同而不同，一个服务内通常是一致的
|
||
Service = "rtc" # 服务名称
|
||
Version = "2024-12-01" # UpdateVoiceChat API版本
|
||
Region = "cn-north-1"
|
||
Host = "rtc.volcengineapi.com"
|
||
ContentType = "application/json"
|
||
|
||
# 请求的凭证，从IAM或者STS服务中获取
|
||
AK = "AKLTMGZmMGIxNTU4MjBlNDU2OWE0M2VmMmY2MjBlODkxNzQ"
|
||
SK = "WldJeU9XSXlNVFprWWpNd05HTmtNVGczTkdNMll6QXpNakEwTm1JM05qSQ=="
|
||
# 当使用临时凭证时，需要使用到SessionToken传入Header，并计算进SignedHeader中，请自行在header参数中添加X-Security-Token头
|
||
# SessionToken = ""
|
||
|
||
|
||
def norm_query(params):
|
||
query = ""
|
||
for key in sorted(params.keys()):
|
||
if type(params[key]) == list:
|
||
for k in params[key]:
|
||
query = (
|
||
query + quote(key, safe="-_.~") + "=" + quote(k, safe="-_.~") + "&"
|
||
)
|
||
else:
|
||
query = (query + quote(key, safe="-_.~") + "=" + quote(params[key], safe="-_.~") + "&")
|
||
query = query[:-1]
|
||
return query.replace("+", "%20")
|
||
|
||
|
||
# 第一步：准备辅助函数。
|
||
# sha256 非对称加密
|
||
def hmac_sha256(key: bytes, content: str):
|
||
return hmac.new(key, content.encode("utf-8"), hashlib.sha256).digest()
|
||
|
||
|
||
# sha256 hash算法
|
||
def hash_sha256(content: str):
|
||
return hashlib.sha256(content.encode("utf-8")).hexdigest()
|
||
|
||
|
||
# 第二步：签名请求函数
|
||
def request(method, date, query, header, ak, sk, action, body):
|
||
# 第三步：创建身份证明。其中的 Service 和 Region 字段是固定的。ak 和 sk 分别代表
|
||
# AccessKeyID 和 SecretAccessKey。同时需要初始化签名结构体。一些签名计算时需要的属性也在这里处理。
|
||
# 初始化身份证明结构体
|
||
credential = {
|
||
"access_key_id": ak,
|
||
"secret_access_key": sk,
|
||
"service": Service,
|
||
"region": Region,
|
||
}
|
||
# 初始化签名结构体
|
||
request_param = {
|
||
"body": body,
|
||
"host": Host,
|
||
"path": "/",
|
||
"method": method,
|
||
"content_type": ContentType,
|
||
"date": date,
|
||
"query": {"Action": action, "Version": Version, **query},
|
||
}
|
||
if body is None:
|
||
request_param["body"] = ""
|
||
# 第四步：接下来开始计算签名。在计算签名前，先准备好用于接收签算结果的 signResult 变量，并设置一些参数。
|
||
# 初始化签名结果的结构体
|
||
x_date = request_param["date"].strftime("%Y%m%dT%H%M%SZ")
|
||
short_x_date = x_date[:8]
|
||
x_content_sha256 = hash_sha256(request_param["body"])
|
||
sign_result = {
|
||
"Host": request_param["host"],
|
||
"X-Content-Sha256": x_content_sha256,
|
||
"X-Date": x_date,
|
||
"Content-Type": request_param["content_type"],
|
||
}
|
||
# 第五步：计算 Signature 签名。
|
||
signed_headers_str = ";".join(
|
||
["content-type", "host", "x-content-sha256", "x-date"]
|
||
)
|
||
# signed_headers_str = signed_headers_str + ";x-security-token"
|
||
canonical_request_str = "\n".join(
|
||
[request_param["method"].upper(),
|
||
request_param["path"],
|
||
norm_query(request_param["query"]),
|
||
"\n".join(
|
||
[
|
||
"content-type:" + request_param["content_type"],
|
||
"host:" + request_param["host"],
|
||
"x-content-sha256:" + x_content_sha256,
|
||
"x-date:" + x_date,
|
||
]
|
||
),
|
||
"",
|
||
signed_headers_str,
|
||
x_content_sha256,
|
||
]
|
||
)
|
||
|
||
# 打印正规化的请求用于调试比对
|
||
print(canonical_request_str)
|
||
hashed_canonical_request = hash_sha256(canonical_request_str)
|
||
|
||
# 打印hash值用于调试比对
|
||
print(hashed_canonical_request)
|
||
credential_scope = "/".join([short_x_date, credential["region"], credential["service"], "request"])
|
||
string_to_sign = "\n".join(["HMAC-SHA256", x_date, credential_scope, hashed_canonical_request])
|
||
|
||
# 打印最终计算的签名字符串用于调试比对
|
||
print("------------------------------")
|
||
print("string_to_sign:")
|
||
print(string_to_sign)
|
||
print("------------------------------")
|
||
k_date = hmac_sha256(credential["secret_access_key"].encode("utf-8"), short_x_date)
|
||
k_region = hmac_sha256(k_date, credential["region"])
|
||
k_service = hmac_sha256(k_region, credential["service"])
|
||
k_signing = hmac_sha256(k_service, "request")
|
||
signature = hmac_sha256(k_signing, string_to_sign).hex()
|
||
|
||
sign_result["Authorization"] = "HMAC-SHA256 Credential={}, SignedHeaders={}, Signature={}".format(
|
||
credential["access_key_id"] + "/" + credential_scope,
|
||
signed_headers_str,
|
||
signature,
|
||
)
|
||
header = {**header, **sign_result}
|
||
print("------------------------------")
|
||
print(header)
|
||
print("------------------------------")
|
||
# header = {**header, **{"X-Security-Token": SessionToken}}
|
||
# 第六步：将 Signature 签名写入 HTTP Header 中，并发送 HTTP 请求。
|
||
r = requests.request(method=method,
|
||
url="https://{}{}".format(request_param["host"], request_param["path"]),
|
||
headers=header,
|
||
params=request_param["query"],
|
||
data=request_param["body"],
|
||
)
|
||
return r.json()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# response_body = request("Get", datetime.datetime.utcnow(), {}, {}, AK, SK, "ListUsers", None)
|
||
# print(response_body)
|
||
|
||
now = datetime.datetime.utcnow()
|
||
# now = datetime.datetime.strptime('20201230T081805Z', '%Y%m%dT%H%M%SZ')
|
||
# Body的格式需要配合Content-Type，API使用的类型请阅读具体的官方文档，如:json格式需要json.dumps(obj)
|
||
# response_body = request("GET", now, {"Limit": "2"}, {}, AK, SK, "ListUsers", None)
|
||
# print(response_body)
|
||
|
||
# response_body = request("POST", now, {"Limit": "10"}, {}, AK, SK, "ListUsers", "UnUseParam=ASDF")
|
||
# print(response_body)
|
||
|
||
# UpdateVoiceChat API调用示例
|
||
# 构造请求体
|
||
body = {
|
||
"AppId": "67b59a4586198401ec980bf9", # 替换为你的AppId
|
||
"RoomId": "Room888",
|
||
"TaskId": "e5b1c872-3c4c-48be-8d6e-4ab80e561b6b",
|
||
# "UserId": "User1",
|
||
"Command": "function",
|
||
"Message": json.dumps({
|
||
"ToolCallID": "call_oidylcvnjcd6676oq2wn01fg",
|
||
"Content": "下中雨"
|
||
})
|
||
}
|
||
|
||
# 发送请求
|
||
response_body = request(
|
||
method="POST",
|
||
date=now,
|
||
query={}, # Query参数在URL中已包含Action和Version
|
||
header={},
|
||
ak=AK,
|
||
sk=SK,
|
||
action="UpdateVoiceChat",
|
||
body=json.dumps(body) # 将body转换为JSON字符串
|
||
)
|
||
print("\nUpdateVoiceChat Response:")
|
||
print(json.dumps(response_body, indent=2, ensure_ascii=False))
|
||
|
||
# # 打断智能体示例
|
||
# interrupt_body = {
|
||
# "AppId": "661e****543cf", # 替换为你的AppId
|
||
# "RoomId": "Room1",
|
||
# "UserId": "User1",
|
||
# "Command": "interrupt"
|
||
# }
|
||
|
||
# response_body = request(
|
||
# method="POST",
|
||
# date=now,
|
||
# query={},
|
||
# header={},
|
||
# ak=AK,
|
||
# sk=SK,
|
||
# action="UpdateVoiceChat",
|
||
# body=json.dumps(interrupt_body)
|
||
# )
|
||
# print("\nInterrupt Bot Response:")
|
||
# print(json.dumps(response_body, indent=2, ensure_ascii=False)) |