157 lines
5.0 KiB
Python
157 lines
5.0 KiB
Python
import datetime
|
|
import json
|
|
import hashlib
|
|
import hmac
|
|
from urllib.parse import quote
|
|
import requests
|
|
from django.conf import settings
|
|
|
|
# 火山引擎API配置
|
|
VOLCENGINE_SERVICE = "rtc"
|
|
VOLCENGINE_VERSION = "2024-12-01"
|
|
VOLCENGINE_REGION = "cn-north-1"
|
|
VOLCENGINE_HOST = "rtc.volcengineapi.com"
|
|
VOLCENGINE_CONTENT_TYPE = "application/json"
|
|
|
|
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")
|
|
|
|
def hmac_sha256(key: bytes, content: str):
|
|
return hmac.new(key, content.encode("utf-8"), hashlib.sha256).digest()
|
|
|
|
def hash_sha256(content: str):
|
|
return hashlib.sha256(content.encode("utf-8")).hexdigest()
|
|
|
|
def volcengine_request(method, date, query, header, ak, sk, action, body):
|
|
credential = {
|
|
"access_key_id": ak,
|
|
"secret_access_key": sk,
|
|
"service": VOLCENGINE_SERVICE,
|
|
"region": VOLCENGINE_REGION,
|
|
}
|
|
request_param = {
|
|
"body": body,
|
|
"host": VOLCENGINE_HOST,
|
|
"path": "/",
|
|
"method": method,
|
|
"content_type": VOLCENGINE_CONTENT_TYPE,
|
|
"date": date,
|
|
"query": {"Action": action, "Version": VOLCENGINE_VERSION, **query},
|
|
}
|
|
if body is None:
|
|
request_param["body"] = ""
|
|
|
|
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"],
|
|
}
|
|
|
|
signed_headers_str = ";".join(
|
|
["content-type", "host", "x-content-sha256", "x-date"]
|
|
)
|
|
|
|
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,
|
|
]
|
|
)
|
|
|
|
hashed_canonical_request = hash_sha256(canonical_request_str)
|
|
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])
|
|
|
|
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}
|
|
|
|
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()
|
|
|
|
def update_voice_chat(app_id, room_id, task_id, call_id, content):
|
|
"""
|
|
调用火山引擎UpdateVoiceChat API
|
|
|
|
Args:
|
|
app_id: 应用ID
|
|
room_id: 房间ID
|
|
task_id: 任务ID
|
|
call_id: 调用ID
|
|
content: 消息内容
|
|
|
|
Returns:
|
|
API响应结果
|
|
"""
|
|
try:
|
|
# 构建请求参数
|
|
date = datetime.datetime.utcnow()
|
|
query = {}
|
|
header = {}
|
|
body = {
|
|
"AppId": app_id,
|
|
"RoomId": room_id,
|
|
"TaskId": task_id,
|
|
"Command": "function",
|
|
"Message": json.dumps({
|
|
"ToolCallID": call_id,
|
|
"Content": content
|
|
})
|
|
}
|
|
|
|
# 调用API
|
|
response = volcengine_request(
|
|
method="POST",
|
|
date=date,
|
|
query=query,
|
|
header=header,
|
|
ak=settings.VOLCENGINE_ACCESS_KEY,
|
|
sk=settings.VOLCENGINE_SECRET_KEY,
|
|
action="UpdateVoiceChat",
|
|
body=json.dumps(body)
|
|
)
|
|
|
|
return response
|
|
|
|
except Exception as e:
|
|
print(f"调用火山引擎API失败: {str(e)}")
|
|
raise |