Python

环境要求 Python 3.0及以上版本。

引用库 requests 2.18.1

1. 请自行下载安装Python 3.x，并完成环境配置。

2. 打开命令行窗口，执行pip install requests命令。

3. 执行pip list查看安装结果。

须知

● 本文档所述Demo在提供服务的过程中，可能会涉及个人数据的使用，建议您遵从 国家的相关法律采取足够的措施，以确保用户的个人数据受到充分的保护。

● 本文档所述Demo仅用于功能演示，不允许客户直接进行商业使用。

● 本文档信息仅供参考，不构成任何要约或承诺。

“语音回呼场景 API”代码样例

# coding: utf-8 -*-import base64 import hashlib

import requests #需要先使用pip install requests命令安装依赖 import time

import uuid

#必填,请参考"开发准备-申请资源"获取如下数据,替换为实际值

base_url = 'https://{domain}:{port}'#APP接入地址,购买服务时下发,请替换为实际值 appKey = '***appKey***'#语音回呼应用的appKey,购买服务时下发,请替换为实际值 appSecret = '***appSecret***'#语音回呼应用的appSecret,购买服务时下发,请替换为实际值 def buildWSSEHeader(appKey, appSecret):

now = time.strftime('%Y-%m-%dT%H:%M:%SZ') #Created nonce = str(uuid.uuid4()).replace('-', '') #Nonce

digest = hashlib.sha256((nonce + now + appSecret).encode()).hexdigest() digestBase64 = base64.b64encode(digest.encode()).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': 'WSSE realm="SDP",profile="UsernameToken",type="Appkey"', 'X-WSSE': buildWSSEHeader(appKey, appSecret)

}

jsonData = { # 必填参数

'displayNbr': displayNbr,#主叫用户手机终端的来电显示号码。

'callerNbr': callerNbr,#发起呼叫时所使用的主叫号码。

'displayCalleeNbr': displayCalleeNbr,#被叫用户终端的来电显示号码。

'calleeNbr': calleeNbr#发起呼叫时所拨打的被叫号码。

# 'preVoice': 'pre_voice1.wav', #设置主叫（callerNbr）应答语音回呼后,呼叫被叫（calleeNbr）前向主叫播 放的提示音

# '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

now = time.strftime('%Y-%m-%dT%H:%M:%SZ') #Created nonce = str(uuid.uuid4()).replace('-', '') #Nonce

digest = hashlib.sha256((nonce + now + appSecret).encode()).hexdigest() digestBase64 = base64.b64encode(digest.encode()).decode() #PasswordDigest

return 'UsernameToken Username="{}",PasswordDigest="{}",Nonce="{}",Created="{}"'.format(appKey, digestBase64, nonce, now);

def getRecordLinkAPI(fileName, recordDomain, xwsse):

if len(fileName) < 1 or len(recordDomain) < 1 or len(xwsse) < 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': 'WSSE realm="SDP",profile="UsernameToken",type="Appkey"', 'X-WSSE': xwsse

} 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__':

wsse = buildWSSEHeader('***appKey***', '***appSecret***') #语音回呼应用的appKey和appSecret location = getRecordLinkAPI('1200_366_0_20161228102743.wav', 'ostor.huawei.com', wsse) print('The record file download link is: ' + location)

“呼叫状态通知 API”代码样例

# coding: utf-8 -*-'''

呼叫事件通知

客户平台收到语音通话平台的呼叫事件通知的接口通知 '''import json

#呼叫事件通知样例 jsonBody = json.dumps({

'eventType': 'callout', 'statusInfo': {

'sessionId': '[email protected]', '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) #打印通知事件类型

'called': 被叫号码

onCallEvent(jsonBody) #呼叫事件处理

“话单通知 API”代码样例

'sessionId': '[email protected]', #通话链路的 唯一标识 'callerNum': '+86138*******2', #主叫号码

'''

话单通知@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) #话单处理