Python 代码样例 - API错误码处理_隐私保护通话 PrivateNumber_开发指南

//短时间内有多个通话结束时隐私保护通话平台会将话单合并推送,每条消息最多携带50个话单 if (sizeof($feeLst) > 1) {

foreach ($feeLst as $loop){

if (array_key_exists('sessionId', $loop)) {

print_r('sessionId: ' . $loop['sessionId'] . PHP_EOL);

} }

} else if(sizeof($feeLst) == 1) {

if (array_key_exists('sessionId', $feeLst[0])) {

print_r('sessionId: ' . $feeLst[0]['sessionId'] . PHP_EOL);

} } else {

print_r('feeLst error: no element.');

} }?>

3.3 Python 代码样例

3.3.1 AXB 模式

样例

AXB模式绑定接口

AXB模式解绑接口

AXB模式绑定信息修改接口 AXB模式绑定信息查询接口

获取录音文件下载地址接口呼叫事件通知接口

话单通知接口短信通知接口

环境要求基于Python 3.7.0版本，要求Python 3.0及以上版本。

引用库 requests 2.18.1（仅“获取录音文件下载地址接口”引用）

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

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

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

须知

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

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

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

● 本文档接口携带参数只是用作参考，不可以直接复制使用，填写参数需要替换为实际值，请参考“开发准备”获取所需数据。

AXB 模式绑定接口

# coding: utf-8 -*-import time

import uuid import hashlib import base64 import json import ssl

import urllib.request

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

realUrl = 'https://rtcpns.cn-north-1.myhuaweicloud.com/rest/caas/relationnumber/partners/v1.0' #APP接入地址+接口访问URI

APP_KEY = "a1********" #APP_Key APP_SECRET = "cfc8********" #APP_Secret

relationNum = '+86170****0001' #X号码(隐私号码) callerNum = '+86186****5678' #A号码

calleeNum = '+86186****5679' #B号码 '''选填,各参数要求请参考"AXB模式绑定接口"

'''# areaCode = '0755' #需要绑定的X号码对应的城市码

# callDirection = 0 #允许呼叫的方向

# duration = 86400 #绑定关系保持时间,到期后会被系统自动解除绑定关系

# recordFlag = 'false' #是否需要针对该绑定关系产生的所有通话录音

# recordHintTone = 'recordHintTone.wav' #设置录音提示音

# maxDuration = 60 #设置允许单次通话进行的最长时间,通话时间从接通被叫的时刻开始计算

# lastMinVoice = 'lastMinVoice.wav' #设置通话剩余最后一分钟时的提示音

# privateSms = 'true' #设置该绑定关系是否支持短信功能

# callerHintTone = 'callerHintTone.wav' #设置A拨打X号码时的通话前等待音

# calleeHintTone = 'calleeHintTone.wav' #设置B拨打X号码时的通话前等待音

# preVoice = {

# 'callerHintTone': callerHintTone,

# 'calleeHintTone': calleeHintTone

# };

'''构造X-WSSE参数值

@param appKey: string

@param appSecret: string

@return: string

'''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 main():

# 请求Body,可按需删除选填参数

jsonData = json.dumps({

'relationNum':relationNum,

# 'areaCode':areaCode, 'callerNum':callerNum, 'calleeNum':calleeNum,

# 'callDirection':callDirection,

# 'duration':duration,

# 'recordFlag':recordFlag,

# 'recordHintTone':recordHintTone,

# 'maxDuration':maxDuration,

# 'lastMinVoice':lastMinVoice,

# 'privateSms':privateSms,

# 'preVoice':preVoice }).encode('ascii')

req = urllib.request.Request(url=realUrl, data=jsonData, method='POST') #请求方法为POST # 请求Headers参数

req.add_header('Authorization', 'WSSE realm="SDP",profile="UsernameToken",type="Appkey"') req.add_header('X-WSSE', buildWSSEHeader(APP_KEY, APP_SECRET))

req.add_header('Content-Type', 'application/json;charset=UTF-8') # 为防止因HTTPS证书认证失败造成API调用失败,需要先忽略证书信任问题 ssl._create_default_https_context = ssl._create_unverified_context try:

fo = open('bind_data.txt', 'a', encoding='utf-8') #打开本地文件

fo.write('绑定请求数据：' + jsonData.decode('utf-8') + '\n') #绑定请求参数记录到本地文件,方便定位问题 r = urllib.request.urlopen(req) #发送请求

print(r.read().decode('utf-8')) #打印响应结果

fo.write('绑定结果：' + str(r.read().decode('utf-8')) + '\n') #绑定ID很重要,请记录到本地文件,方便后续修改绑定关系及解绑

except urllib.error.HTTPError as e:

print(e.code)

print(e.read().decode('utf-8')) #打印错误信息 except urllib.error.URLError as e:

print(e.reason) finally:

fo.close() #关闭文件 if __name__ == '__main__':

main()

AXB 模式解绑接口

# coding: utf-8 -*-import time

import uuid import hashlib import base64 import ssl

import urllib.request

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

realUrl = 'https://rtcpns.cn-north-1.myhuaweicloud.com/rest/caas/relationnumber/partners/v1.0' #APP接入地址+接口访问URI

APP_KEY = "a1********" #APP_Key APP_SECRET = "cfc8********" #APP_Secret '''选填,各参数要求请参考"AXB模式解绑接口"

subscriptionId和relationNum为二选一关系,两者都携带时以subscriptionId为准 '''subscriptionId = '****' #指定"AXB模式绑定接口"返回的绑定ID进行解绑 relationNum = '+86170****0001' #指定X号码(隐私号码)进行解绑 '''构造X-WSSE参数值

@param appKey: string

@param appSecret: string

@return: string

'''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 main():

# 请求URL参数

formData = urllib.parse.urlencode({

'subscriptionId':subscriptionId, 'relationNum':relationNum })

#完整请求地址

fullUrl = realUrl + '?' + formData

req = urllib.request.Request(url=fullUrl, method='DELETE') #请求方法为DELETE # 请求Headers参数

req.add_header('Authorization', 'WSSE realm="SDP",profile="UsernameToken",type="Appkey"') req.add_header('X-WSSE', buildWSSEHeader(APP_KEY, APP_SECRET))

print(formData) #打印请求数据 r = urllib.request.urlopen(req) #发送请求 print(r.read().decode('utf-8')) #打印响应结果 except urllib.error.HTTPError as e:

print(e.code)

print(e.read().decode('utf-8')) #打印错误信息 except urllib.error.URLError as e:

print(e.reason) if __name__ == '__main__':

main()

AXB 模式绑定信息修改接口

# coding: utf-8 -*-import time

import uuid import hashlib import base64 import json import ssl

import urllib.request

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

realUrl = 'https://rtcpns.cn-north-1.myhuaweicloud.com/rest/caas/relationnumber/partners/v1.0' #APP接入地址+接口访问URI

APP_KEY = "a1********" #APP_Key APP_SECRET = "cfc8********" #APP_Secret

subscriptionId = '0167ecc9-bfb6-4eec-b671-a7dab2ba78c' #必填,指定"AXB模式绑定接口"返回的绑定ID进行修改

'''选填,各参数要求请参考"AXB模式绑定信息修改接口"

'''callerNum = '+86186****5678' #A号码 calleeNum = '+86186****5679' #B号码

# callDirection = 0 #允许呼叫的方向

# duration = 86400 #绑定关系保持时间,到期后会被系统自动解除绑定关系

# maxDuration = 60 #设置允许单次通话进行的最长时间,通话时间从接通被叫的时刻开始计算

# lastMinVoice = 'lastMinVoice.wav' #设置通话剩余最后一分钟时的提示音

# privateSms = 'true' #设置该绑定关系是否支持短信功能

# callerHintTone = 'callerHintTone.wav' #设置A拨打X号码时的通话前等待音

# calleeHintTone = 'calleeHintTone.wav' #设置B拨打X号码时的通话前等待音

# preVoice = {

# 'callerHintTone': callerHintTone,

# 'calleeHintTone': calleeHintTone

# };

'''构造X-WSSE参数值

@param appKey: string

@param appSecret: string

@return: string

'''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 main():

# 请求Body,可按需删除选填参数 jsonData = json.dumps({

'subscriptionId':subscriptionId, 'callerNum':callerNum, 'calleeNum':calleeNum,

# 'callDirection':callDirection,

# 'duration':duration,

# 'maxDuration':maxDuration,

# 'lastMinVoice':lastMinVoice,

# 'privateSms':privateSms,

# 'preVoice':preVoice }).encode('ascii')

req = urllib.request.Request(url=realUrl, data=jsonData, method='PUT') #请求方法为PUT # 请求Headers参数

req.add_header('Authorization', 'WSSE realm="SDP",profile="UsernameToken",type="Appkey"') req.add_header('X-WSSE', buildWSSEHeader(APP_KEY, APP_SECRET))

print(jsonData.decode('utf-8')) #打印请求数据 r = urllib.request.urlopen(req) #发送请求 print(r.read().decode('utf-8')) #打印响应结果 except urllib.error.HTTPError as e:

print(e.code)

print(e.read().decode('utf-8')) #打印错误信息 except urllib.error.URLError as e:

print(e.reason) if __name__ == '__main__':

main()

AXB 模式绑定信息查询接口

# coding: utf-8 -*-import time

import uuid import hashlib import base64 import ssl

import urllib.request

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

realUrl = 'https://rtcpns.cn-north-1.myhuaweicloud.com/rest/caas/relationnumber/partners/v1.0' #APP接入地址+接口访问URI

APP_KEY = "a1********" #APP_Key APP_SECRET = "cfc8********" #APP_Secret

'''选填,各参数要求请参考"AXB模式绑定信息查询接口"

subscriptionId和relationNum为二选一关系,两者都携带时以subscriptionId为准 '''subscriptionId = '****' #指定"AXB模式绑定接口"返回的绑定ID进行查询 relationNum = '+86170****0001' #指定X号码(隐私号码)进行查询

# pageIndex = 1 #查询的分页索引,从1开始编号

# pageSize = 20 #查询的分页大小,即每次查询返回多少条数据 '''构造X-WSSE参数值

@param appKey: string

@param appSecret: string

@return: string

'''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 main():

# 请求URL参数,可按需删除选填参数 formData = urllib.parse.urlencode({

'subscriptionId':subscriptionId, 'relationNum':relationNum,

# 'pageIndex':pageIndex,

# 'pageSize':pageSize })

#完整请求地址

fullUrl = realUrl + '?' + formData

req = urllib.request.Request(url=fullUrl, method='GET') #请求方法为GET # 请求Headers参数

req.add_header('Authorization', 'WSSE realm="SDP",profile="UsernameToken",type="Appkey"') req.add_header('X-WSSE', buildWSSEHeader(APP_KEY, APP_SECRET))

fo = open('bind_data.txt', 'a', encoding='utf-8') #打开本地文件 r = urllib.request.urlopen(req) #发送请求

print(r.read().decode('utf-8')) #打印响应结果

fo.write('绑定查询结果：' + str(r.read().decode('utf-8')) + '\n') #查询结果,记录到本地文件 except urllib.error.HTTPError as e:

print(e.code)

print(e.read().decode('utf-8')) #打印错误信息 except urllib.error.URLError as e:

print(e.reason) finally:

fo.close() #关闭文件 if __name__ == '__main__':

main()

获取录音文件下载地址接口

# coding: utf-8 -*-import time

import uuid import hashlib import base64 import urllib

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

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

realUrl = 'https://rtcpns.cn-north-1.myhuaweicloud.com/rest/provision/voice/record/v1.0' #APP接入地址+接口访问URI

APP_KEY = "a1********" #APP_Key APP_SECRET = "cfc8********" #APP_Secret

# 必填,通过"话单通知接口"获取

recordDomain = '****.com' #录音文件存储的服务器域名 fileName = '****.wav' #录音文件名

'''构造X-WSSE参数值

@param appKey: string

@param appSecret: string

@return: string

'''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 main():

# 请求URL参数

formData = urllib.parse.urlencode({

'recordDomain':recordDomain, 'fileName':fileName

})

#完整请求地址

fullUrl = realUrl + '?' + formData # 请求Headers参数

header = {

'Authorization': 'WSSE realm="SDP",profile="UsernameToken",type="Appkey"', 'X-WSSE': buildWSSEHeader(APP_KEY, APP_SECRET),

'Content-Type': 'application/json;charset=UTF-8' }

try:

fo = open('bind_data.txt', 'a', encoding='utf-8') #打开本地文件

r = requests.get(fullUrl, headers=header, allow_redirects=False, verify=False) #发送请求 if(301 == r.status_code):

print(r.status_code) #打印响应结果 print(r.headers['Location'])

fo.write('获取录音文件下载地址：' + r.headers['Location'] + '\n') else:

print(r.status_code) #打印响应结果 print(r.text)

except urllib.error.HTTPError as e:

print(e.code)

print(e.read().decode('utf-8')) #打印错误信息 except urllib.error.URLError as e:

print(e.reason) finally:

fo.close() #关闭文件 if __name__ == '__main__':

main()

呼叫事件通知接口

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

呼叫事件通知

客户平台收到隐私保护通话平台的呼叫事件通知的接口通知 '''import json

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

'eventType': 'disconnect', 'statusInfo': {

'sessionId': '1200_1029_4294967295_20190123091514@callenabler246.huaweicaas.com', 'timestamp': '2019-01-23 09:16:41',

'caller': '+86138****0021', 'called': '+86138****7021', 'stateCode': 0,

'stateDesc': 'The user releases the call.', 'subscriptionId': '****'

}

}).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) #打印通知事件类型 #callin：呼入事件

if ('callin' == eventType):

'''

Example: 此处以解析sessionId为例,请按需解析所需参数并自行实现相关处理 'timestamp': 呼叫事件发生时隐私保护通话平台的UNIX时间戳

'sessionId': 通话链路的标识ID 'caller': 主叫号码

'called': 被叫号码

'subscriptionId': 绑定关系ID '''

if ('sessionId' in statusInfo):

print('sessionId: ' + statusInfo['sessionId']) return

#callout：呼出事件 if ('callout' == eventType):

'''

Example: 此处以解析sessionId为例,请按需解析所需参数并自行实现相关处理 'timestamp': 呼叫事件发生时隐私保护通话平台的UNIX时间戳

'sessionId': 通话链路的标识ID 'caller': 主叫号码

'called': 被叫号码

'subscriptionId': 绑定关系ID '''

if ('sessionId' in statusInfo):

print('sessionId: ' + statusInfo['sessionId']) return

#alerting：振铃事件 if ('alerting' == eventType):

'''

Example: 此处以解析sessionId为例,请按需解析所需参数并自行实现相关处理 'timestamp': 呼叫事件发生时隐私保护通话平台的UNIX时间戳

'sessionId': 通话链路的标识ID 'caller': 主叫号码

'called': 被叫号码

'subscriptionId': 绑定关系ID '''

if ('sessionId' in statusInfo):

print('sessionId: ' + statusInfo['sessionId']) return

#answer：应答事件 if ('answer' == eventType):

'''

Example: 此处以解析sessionId为例,请按需解析所需参数并自行实现相关处理 'timestamp': 呼叫事件发生时隐私保护通话平台的UNIX时间戳

'sessionId': 通话链路的标识ID 'caller': 主叫号码

'called': 被叫号码

'subscriptionId': 绑定关系ID '''

if ('sessionId' in statusInfo):

print('sessionId: ' + statusInfo['sessionId']) return

#disconnect：挂机事件 if ('disconnect' == eventType):

'''

Example: 此处以解析sessionId为例,请按需解析所需参数并自行实现相关处理 'timestamp': 呼叫事件发生时隐私保护通话平台的UNIX时间戳

'sessionId': 通话链路的标识ID 'caller': 主叫号码

'called': 被叫号码

'stateCode': 通话挂机的原因值 'stateDesc': 通话挂机的原因值的描述 'subscriptionId': 绑定关系ID '''

if ('sessionId' in statusInfo):

print('sessionId: ' + statusInfo['sessionId']) return

def main():

onCallEvent(jsonBody) #呼叫事件处理 if __name__ == '__main__':

main()

话单通知接口

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

话单通知

客户平台收到隐私保护通话平台的话单通知的接口通知 '''import json

#话单通知样例

jsonBody = json.dumps({

'eventType': 'fee', 'feeLst': [ {

'direction': 1,

'spId': '****', 'appKey': '********',

'icid': 'ba171f34e6953fcd751edc77127748f4.3757223714.337238282.9', 'bindNum': '+86138****0022',

'sessionId': '1200_1029_4294967295_20190123091514@callenabler246.huaweicaas.com', 'subscriptionId': '****',

'callerNum': '+86138****0021', 'calleeNum': '+86138****0022', 'fwdDisplayNum': '+86138****0022', 'fwdDstNum': '+86138****7021', 'callInTime': '2019-01-23 09:15:14', 'fwdStartTime': '2019-01-23 09:15:15', 'fwdAlertingTime': '2019-01-23 09:15:21', 'fwdAnswerTime': '2019-01-23 09:15:36', 'callEndTime': '2019-01-23 09:16:41', 'fwdUnaswRsn': 0,

'ulFailReason': 0, 'sipStatusCode': 0, 'callOutUnaswRsn': 0, 'recordFlag': 1,

'recordStartTime': '2019-01-23 09:15:37', 'recordDomain': '****.com',

'recordBucketName': 'sp-********', 'recordObjectName': '********.wav', 'ttsPlayTimes': 0,

'ttsTransDuration': 0, 'mptyId': '****', 'serviceType': '004',

'hostName': 'callenabler246.huaweicaas.com' }

]

}).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为例,请按需解析所需参数并自行实现相关处理 'direction': 通话的呼叫方向

'spId': 客户的云服务账号 'appKey': 商户应用的AppKey

在文檔中 API错误码处理_隐私保护通话 PrivateNumber_开发指南_华为云 (頁 53-93)