更新时间：2024-09-24 GMT+08:00
Python
|
样例 |
|
|---|---|
|
环境要求 |
Python 3.0及以上版本。 |
|
引用库 |
requests 2.18.1 |
- 本文档所述Demo在提供服务的过程中，可能会涉及个人数据的使用，建议您遵从国家的相关法律采取足够的措施，以确保用户的个人数据受到充分的保护。
- 本文档所述Demo仅用于功能演示，不允许客户直接进行商业使用。
- 本文档信息仅供参考，不构成任何要约或承诺。
“语音回呼场景API”代码样例
# -*- coding: utf-8 -*-
import base64
import hashlib
import requests #需要先使用pip install requests命令安装依赖
import time
import uuid
import hmac
from hashlib import sha256
#必填,请参考"开发准备-申请资源"获取如下数据,替换为实际值
base_url = 'https://{domain}:{port}'#APP接入地址,购买服务时下发,请替换为实际值
appKey = '***appKey***'#语音回呼应用的appKey,购买服务时下发,请替换为实际值
appSecret = '***appSecret***'#语音回呼应用的appSecret,购买服务时下发,请替换为实际值
def buildAKSKHeader(appKey, appSecret):
now = time.strftime('%Y-%m-%dT%H:%M:%SZ') #Created
nonce = str(uuid.uuid4()).replace('-','') #Nonce
digist = hmac.new(appSecret.encode(), (nonce + now).encode(), digestmod=sha256).digest()
digestBase64 = base64.b64encode(digist).decode() #PasswordDigest
return 'UsernameToken Username="{}",PasswordDigest="{}",Nonce="{}",Created="{}"'.format(appKey, digestBase64, nonce, now);
def voiceCallAPI(displayNbr, callerNbr, displayCalleeNbr, calleeNbr):
if len(displayNbr) < 1 or len(callerNbr) < 1 or len(displayCalleeNbr) < 1 or len(calleeNbr) < 1:
return
apiUri = '/rest/httpsessions/click2Call/v2.0'
requestUrl = base_url + apiUri
header = {
'Content-Type': 'application/json;charset=UTF-8',
'Authorization': 'AKSK realm="SDP",profile="UsernameToken",type="Appkey"',
'X-AKSK': buildAKSKHeader(appKey, appSecret)
}
jsonData = {
# 必填参数
'displayNbr': displayNbr,#主叫用户手机终端的来电显示号码。
'callerNbr': callerNbr,#发起呼叫时所使用的主叫号码。
'displayCalleeNbr': displayCalleeNbr,#被叫用户终端的来电显示号码。
'calleeNbr': calleeNbr#发起呼叫时所拨打的被叫号码。
# 选填参数
# 'maxDuration': 0, #允许单次通话进行的最长时间
# 'lastMinVoice': 'lastmin_voice1.wav', #最后一分钟放音提示音
# 'lastMinToUE': 'both', #最后一分钟放音的播放对象
# 'playPreVoice': 'false', #设置主叫（callerNbr）应答语音回呼后,呼叫被叫（calleeNbr）前,是否向主叫（callerNbr）播放提示音
# 'preVoice': 'pre_voice1.wav', #设置主叫（callerNbr）应答语音回呼后,呼叫被叫（calleeNbr）前向主叫播放的提示音
# 'waitVoice': 'wait_voice1.wav', #设置主叫应答语音回呼后的等待音
# 'calleeMedia': 'all' , #指定被叫的媒体音播放方式
# 'statusUrl': '', #设置SP接收状态上报的URL,要求使用BASE64编码
# 'feeUrl': '', #设置SP接收话单上报的URL,要求使用BASE64编码
# 'recordFlag': 'false', #设置语音回呼通话过程是否录音
# 'recordHintTone': 'recordhint_voice1.wav', #设置使用录音功能的提示音
# 'partyTypeRequiredInDisconnect': 'false', #disconnect状态是否需要携带通话主动挂机的用户类型
# 'returnIdlePort': 'false', #指示是否需要返回平台空闲呼叫端口数量
# 'userData': 'customerId123' #设置用户的附属信息
}
try:
r = requests.post(requestUrl, json=jsonData, headers=header, verify=False)
print(r.text) #打印响应结果
except requests.exceptions.HTTPError as e:
print(e.code)
print(e.read().decode('utf-8')) #打印错误信息
pass
if __name__ == '__main__':
voiceCallAPI('+86531*******4', '+86135*******1', '+86531*******4', '+86135*******2')
“获取录音文件下载地址API”代码样例
# -*- coding: utf-8 -*-
import base64
import hashlib
import requests #需要先使用pip install requests命令安装依赖
import time
import urllib
import uuid
import hmac
from hashlib import sha256
#必填,请参考"开发准备-申请资源"获取如下数据,替换为实际值
base_url = 'https://{domain}:{port}'#APP接入地址,购买服务时下发，请替换为实际值
def buildAKSKHeader(appKey, appSecret):
now = time.strftime('%Y-%m-%dT%H:%M:%SZ') #Created
nonce = str(uuid.uuid4()).replace('-','') #Nonce
digist = hmac.new(appSecret.encode(), (nonce + now).encode(), digestmod=sha256).digest()
digestBase64 = base64.b64encode(digist).decode() #PasswordDigest
return 'UsernameToken Username="{}",PasswordDigest="{}",Nonce="{}",Created="{}"'.format(appKey, digestBase64, nonce, now);
def getRecordLinkAPI(fileName, recordDomain, xaksk):
if len(fileName) < 1 or len(recordDomain) < 1 or len(xaksk) < 1:
return
location = ''
apiUri = '/rest/provision/voice/record/v1.0'
queryParams = urllib.parse.urlencode({'fileName':fileName, 'recordDomain':recordDomain})
requestUrl = base_url + apiUri + '?' + queryParams
header = {
'Content-Type': 'application/json;charset=UTF-8',
'Authorization': 'AKSK realm="SDP",profile="UsernameToken",type="Appkey"',
'X-AKSK': xaksk
}
try:
r = requests.get(requestUrl, headers=header, allow_redirects=False, verify=False) #发送请求
if(301 == r.status_code):
print(r.status_code) #打印响应结果
location = r.headers['Location']
else:
print(r.status_code)
print(r.text)#打印响应结果
except requests.exceptions.HTTPError as e:
print(e.code)
print(e.read().decode('utf-8')) #打印错误信息
return location
if __name__ == '__main__':
aksk = buildAKSKHeader('***appKey***', '***appSecret***') #语音回呼应用的appKey和appSecret
location = getRecordLinkAPI('1200_366_0_20161228102743.wav', 'ostor.huawei.com', aksk)
print('The record file download link is: ' + location)
“呼叫状态通知API”代码样例
# -*- coding: utf-8 -*-
'''
呼叫事件通知
客户平台收到语音通话平台的呼叫事件通知的接口通知
'''
import json
#呼叫事件通知样例
jsonBody = json.dumps({
'eventType': 'callout',
'statusInfo': {
'sessionId': '1201_612_4294967295_20190124030424@callenabler245.huaweicaas.com',
'timestamp': '2019-01-24 03:04:24',
'caller': '+86138*******2',
'called': '+86138*******1',
'userData': 'customerId123'
}
}).encode('ascii')
print(jsonBody)
'''
呼叫事件通知
@see: 详细内容以接口文档为准
@param param: jsonBody
@return:
'''
def onCallEvent(jsonBody):
jsonObj = json.loads(jsonBody) #将通知消息解析为jsonObj
eventType = jsonObj['eventType'] #通知事件类型
if ('fee' == eventType):
print('EventType error: ' + eventType)
return
if ('statusInfo' not in jsonObj):
print('param error: no statusInfo.')
return
statusInfo = jsonObj['statusInfo'] #呼叫状态事件信息
print('eventType: ' + eventType) #打印通知事件类型
#callout：呼出事件
if ('callout' == eventType):
'''
Example: 此处以解析sessionId为例,请按需解析所需参数并自行实现相关处理
'timestamp': 该呼叫事件发生时语音通话平台的UNIX时间戳
'userData': 用户附属信息
'sessionId': 通话链路的标识ID
'caller': 主叫号码
'called': 被叫号码
'''
if ('sessionId' in statusInfo):
print('sessionId: ' + statusInfo['sessionId'])
return
#alerting：振铃事件
if ('alerting' == eventType):
'''
Example: 此处以解析sessionId为例,请按需解析所需参数并自行实现相关处理
'timestamp': 该呼叫事件发生时语音通话平台的UNIX时间戳
'userData': 用户附属信息
'sessionId': 通话链路的标识ID
'caller': 主叫号码
'called': 被叫号码
'''
if ('sessionId' in statusInfo):
print('sessionId: ' + statusInfo['sessionId'])
return
#answer：应答事件
if ('answer' == eventType):
'''
Example: 此处以解析sessionId为例,请按需解析所需参数并自行实现相关处理
'timestamp': 该呼叫事件发生时语音通话平台的UNIX时间戳
'userData': 用户附属信息
'sessionId': 通话链路的标识ID
'caller': 主叫号码
'called': 被叫号码
'''
if ('sessionId' in statusInfo):
print('sessionId: ' + statusInfo['sessionId'])
return
#collectInfo：放音收号结果事件,仅应用于语音通知场景
if ('collectInfo' == eventType):
'''
Example: 此处以解析digitInfo为例,请按需解析所需参数并自行实现相关处理
'timestamp': 该呼叫事件发生时语音通话平台的UNIX时间戳
'sessionId': 通话链路的标识ID
'digitInfo': 放音收号场景中,语音通话平台对开发者进行放音收号操作的结果描述
'''
if ('digitInfo' in statusInfo):
print('digitInfo: ' + statusInfo['digitInfo'])
return
#disconnect：挂机事件
if ('disconnect' == eventType):
'''
Example: 此处以解析sessionId为例,请按需解析所需参数并自行实现相关处理
'timestamp': 该呼叫事件发生时语音通话平台的UNIX时间戳
'userData': 用户附属信息
'sessionId': 通话链路的标识ID
'caller': 主叫号码
'called': 被叫号码
'partyType': 挂机的用户类型,仅在语音回呼场景携带
'stateCode': 通话挂机的原因值
'stateDesc': 通话挂机的原因值的描述
'''
if ('sessionId' in statusInfo):
print('sessionId: ' + statusInfo['sessionId'])
return
if __name__ == '__main__':
onCallEvent(jsonBody) #呼叫事件处理
“话单通知API”代码样例
# -*- coding: utf-8 -*-
'''
话单通知
客户平台收到语音通话平台的话单通知的接口通知
'''
import json
#话单通知样例
jsonBody = json.dumps({
'eventType': 'fee',
'feeLst': [
{
'direction': 0, #通话的呼叫方向,以语音通话平台为基准
'spId': 'CaaS_Test_01', #客户的云服务账号
'appKey': 'ka4k******9YnEm2', #应用的app_key
'icid': 'CAE-20190124110424-12019775', #呼叫记录的唯一标识
'bindNum': '+86755*******8', #发起此次呼叫的CallEnabler业务号码,即绑定号码
'sessionId': '1201_612_4294967295_20190124030424@callenabler245.huaweicaas.com', #通话链路的唯一标识
'callerNum': '+86138*******2', #主叫号码
'calleeNum': '+86138*******1', #被叫号码
'fwdDisplayNum': '+86138*******2', #转接呼叫时的显示号码(仅语音回呼场景携带)
'fwdDstNum': '+86138*******1', #转接呼叫时的转接号码(仅语音回呼场景携带)
'fwdStartTime': '2019-01-24 03:04:31', #转接呼叫操作的开始时间(仅语音回呼场景携带)
'fwdAlertingTime': '2019-01-24 03:04:36', #转接呼叫操作后的振铃时间(仅语音回呼场景携带)
'fwdAnswerTime': '2019-01-24 03:04:38', #转接呼叫操作后的应答时间(仅语音回呼场景携带)
'callEndTime': '2019-01-24 03:04:49', #呼叫结束时间
'fwdUnaswRsn': 0, #转接呼叫操作失败的Q850原因值
'failTime': '', #呼入,呼出的失败时间
'ulFailReason': 0, #通话失败的拆线点
'sipStatusCode': 0, #呼入,呼出的失败SIP状态码
'callOutStartTime': '2019-01-24 03:04:24', #Initcall的呼出开始时间
'callOutAlertingTime': '2019-01-24 03:04:27', #Initcall的呼出振铃时间
'callOutAnswerTime': '2019-01-24 03:04:31', #Initcall的呼出应答时间
'callOutUnaswRsn': 0, #Initcall的呼出失败的Q850原因值
'dynIVRStartTime': '', #自定义动态IVR开始时间(仅语音通知场景携带)
'dynIVRPath': '', #自定义动态IVR按键路径(仅语音通知场景携带)
'recordFlag': 0, #录音标识
'recordStartTime': '', #录音开始时间(仅语音回呼场景携带)
'recordObjectName': '', #录音文件名(仅语音回呼场景携带)
'recordBucketName': '', #录音文件所在的目录名(仅语音回呼场景携带)
'recordDomain': '', #存放录音文件的域名(仅语音回呼场景携带)
'recordFileDownloadUrl': '', #录音文件下载地址(仅语音回呼场景携带)
'ttsPlayTimes': 0, #应用TTS功能时,使用TTS的总次数
'ttsTransDuration': 0, #应用TTS功能时,TTS Server进行TTS转换的总时长(单位为秒)
'serviceType': '002', #携带呼叫的业务类型信息
'hostName': 'callenabler245.huaweicaas.com', #话单生成的服务器设备对应的主机名
'userData': '' #用户附属信息
}
]
}).encode('ascii')
print(jsonBody)
'''
话单通知
@see: 详细内容以接口文档为准
@param param: jsonBody
@return:
'''
def onFeeEvent(jsonBody):
jsonObj = json.loads(jsonBody) #将通知消息解析为jsonObj
eventType = jsonObj['eventType'] #通知事件类型
if ('fee' != eventType):
print('EventType error: ' + eventType)
return
if ('feeLst' not in jsonObj):
print('param error: no feeLst.');
return
feeLst = jsonObj['feeLst']
#Example: 此处以解析sessionId为例,请按需解析所需参数并自行实现相关处理
#短时间内有多个通话结束时语音通话平台会将话单合并推送,每条消息最多携带50个话单
if (len(feeLst) > 1):
for loop in feeLst:
if ('sessionId' in loop):
print('sessionId: ' + loop['sessionId'])
elif(len(feeLst) == 1):
if ('sessionId' in feeLst[0]):
print('sessionId: ' + feeLst[0]['sessionId'])
else:
print('feeLst error: no element.');
if __name__ == '__main__':
onFeeEvent(jsonBody) #话单处理
父主题： 语音回呼代码样例
