# AXE模式 | 样例 | [AXE模式绑定接口] [AXE模式解绑接口] [AXE模式绑定信息查询接口] [获取录音文件下载地址接口] [呼叫事件通知接口] [话单通知接口] | |---|---| | **环境要求** | 基于Python 3.7.0版本，要求Python 3.0及以上版本。 | | **引用库** | requests 2.18.1（仅"获取录音文件下载地址接口"引用） 1. 请自行下载安装Python 3.x，并完成环境配置。 2. 打开命令行窗口，执行pip install requests命令。 3. 执行pip list查看安装结果。 | ![](https://support.huaweicloud.com/devg-PrivateNumber/public_sys-resources/notice_3.0-zh-cn.png) - 本文档所述Demo在提供服务的过程中，可能会涉及个人数据的使用，建议您遵从国家的相关法律采取足够的措施，以确保用户的个人数据受到充分的保护。 - 本文档所述Demo仅用于功能演示，不允许客户直接进行商业使用。 - 本文档信息仅供参考，不构成任何要约或承诺。 - 本文档接口携带参数只是用作参考，不可以直接复制使用，填写参数需要替换为实际值，请参考"[开发准备](https://support.huaweicloud.com/devg-PrivateNumber/privatenumber_01_0005.html)"获取所需数据。 #### AXE模式绑定接口 ``` # -*- coding: utf-8 -*- import time import uuid import hashlib import base64 import json import ssl import urllib.request import hmac from hashlib import sha256 # 必填,请参考"开发准备"获取如下数据,替换为实际值 realUrl = 'https://rtcpns.cn-north-1.myhuaweicloud.com/rest/caas/extendnumber/v1.0' #APP接入地址+接口访问URI APP_KEY = "a1********" #APP_Key APP_SECRET = "cfc8********" #APP_Secret virtualNum = '+86180****0001' #AXE中的X号码 bindNum = '+86186****5678' #AXE中的A号码 ''' 选填,各参数要求请参考"AXE模式绑定接口" ''' # areaCode = '0755' #需要绑定的X号码对应的城市码 # displayNumMode = '0' #非A用户呼叫X号码时，A看到的主显号码 # recordFlag = 'false' #是否需要针对该绑定关系产生的所有通话录音 # recordHintTone = 'recordHintTone.wav' #设置录音提示音 # callbackTone = 'callbackTone.wav' #A呼叫X不存在回呼记录的提示音 # callbackNum = '+86180****0021' #A呼叫X不存在回呼记录的转接号码 # bindExpiredTime = 24 #绑定关系的有效时间 # callbackExpiredTime = 24 #回呼记录有效时间 # userData = 'abcdefg' #用户自定义数据 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 main(): # 请求Body,可按需删除选填参数 jsonData = json.dumps({ 'virtualNum':virtualNum, 'bindNum':bindNum, # 'areaCode':areaCode, # 'displayNumMode':displayNumMode, # 'recordFlag':recordFlag, # 'recordHintTone':recordHintTone, # 'callbackTone':callbackTone, # 'callbackNum':callbackNum, # 'bindExpiredTime':bindExpiredTime, # 'callbackExpiredTime':callbackExpiredTime, # 'userData':userData }).encode('ascii') req = urllib.request.Request(url=realUrl, data=jsonData, method='POST') #请求方法为POST # 请求Headers参数 req.add_header('Authorization', 'AKSK realm="SDP",profile="UsernameToken",type="Appkey"') req.add_header('X-AKSK', buildAKSKHeader(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() ``` #### AXE模式解绑接口 ``` # -*- coding: utf-8 -*- import time import uuid import hashlib import base64 import ssl import urllib.request import hmac from hashlib import sha256 # 必填,请参考"开发准备"获取如下数据,替换为实际值 realUrl = 'https://rtcpns.cn-north-1.myhuaweicloud.com/rest/caas/extendnumber/v1.0' #APP接入地址+接口访问URI APP_KEY = "a1********" #APP_Key APP_SECRET = "cfc8********" #APP_Secret ''' 选填,各参数要求请参考"AXE模式解绑接口" subscriptionId和(virtualNum+extendNum)二选一即可,当都传入时,优先选用subscriptionId ''' subscriptionId = '****' #指定"AXE模式绑定接口"返回的绑定ID进行解绑 virtualNum = '+86180****0001' #AXE中的X号码 extendNum = '1234' #AXE中的E号码 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 main(): # 请求URL参数 formData = urllib.parse.urlencode({ 'subscriptionId':subscriptionId, 'virtualNum':virtualNum, 'extendNum':extendNum }) #完整请求地址 fullUrl = realUrl + '?' + formData req = urllib.request.Request(url=fullUrl, method='DELETE') #请求方法为DELETE # 请求Headers参数 req.add_header('Authorization', 'AKSK realm="SDP",profile="UsernameToken",type="Appkey"') req.add_header('X-AKSK', buildAKSKHeader(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: 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() ``` #### AXE模式绑定信息查询接口 ``` # -*- coding: utf-8 -*- import time import uuid import hashlib import base64 import ssl import urllib.request import hmac from hashlib import sha256 # 必填,请参考"开发准备"获取如下数据,替换为实际值 realUrl = 'https://rtcpns.cn-north-1.myhuaweicloud.com/rest/caas/extendnumber/v1.0' #APP接入地址+接口访问URI APP_KEY = "a1********" #APP_Key APP_SECRET = "cfc8********" #APP_Secret ''' 选填,各参数要求请参考"AXE模式绑定信息查询接口" subscriptionId和(virtualNum+extendNum)二选一即可,当都传入时,优先选用subscriptionId ''' subscriptionId = '****' #指定"AXE模式绑定接口"返回的绑定ID进行查询 virtualNum = '+86180****0001' #AXE中的X号码 extendNum = '1234' #AXE中的E号码 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 main(): # 请求URL参数,可按需删除选填参数 formData = urllib.parse.urlencode({ 'subscriptionId':subscriptionId, 'virtualNum':virtualNum, 'extendNum':extendNum, }) #完整请求地址 fullUrl = realUrl + '?' + formData req = urllib.request.Request(url=fullUrl, method='GET') #请求方法为GET # 请求Headers参数 req.add_header('Authorization', 'AKSK realm="SDP",profile="UsernameToken",type="Appkey"') req.add_header('X-AKSK', buildAKSKHeader(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') #打开本地文件 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命令安装依赖 import hmac from hashlib import sha256 # 必填,请参考"开发准备"获取如下数据,替换为实际值 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' #录音文件名 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 main(): # 请求URL参数 formData = urllib.parse.urlencode({ 'recordDomain':recordDomain, 'fileName':fileName }) #完整请求地址 fullUrl = realUrl + '?' + formData # 请求Headers参数 header = { 'Authorization': 'AKSK realm="SDP",profile="UsernameToken",type="Appkey"', 'X-AKSK': buildAKSKHeader(appKey, appSecret), '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': '1202_1051_4294967295_20190124070250@callenabler246.huaweicaas.com', 'timestamp': '2019-01-24 07:03:28', 'caller': '+86138****0022', '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': 被叫号码 ''' if ('sessionId' in statusInfo): print('sessionId: ' + statusInfo['sessionId']) return #collectInfo：放音收号结果事件,仅AXE模式下的A被叫场景携带 if ('collectInfo' == eventType): ''' Example: 此处以解析digitInfo为例,请按需解析所需参数并自行实现相关处理 'timestamp': 呼叫事件发生时隐私保护通话平台的UNIX时间戳 'sessionId': 通话链路的标识ID 'digitInfo': AXE场景中携带收号结果(即用户输入的数字) ''' if ('digitInfo' in statusInfo): print('digitInfo: ' + statusInfo['digitInfo']) 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': 0, 'spId': '****', 'appKey': '********', 'icid': 'ba171f34e6953fcd751edc77127748f4.3757289590.338833305.5', 'bindNum': '+86138****0022', 'sessionId': '1201_11275_4294967295_20190124033310@callenabler246.huaweicaas.com', 'subscriptionId': '********', 'callerNum': '+86138****7021', 'calleeNum': '+86138****0022', 'fwdDisplayNum': '+86138****0022', 'fwdDstNum': '+86138****0021', 'callInTime': '2019-01-24 03:33:10', 'fwdStartTime': '2019-01-24 03:33:16', 'fwdAlertingTime': '2019-01-24 03:33:19', 'fwdAnswerTime': '2019-01-24 03:33:28', 'callEndTime': '2019-01-24 03:33:57', 'fwdUnaswRsn': 0, 'ulFailReason': 0, 'sipStatusCode': 0, 'callOutUnaswRsn': 0, 'recordFlag': 1, 'recordStartTime': '2019-01-24 03:33:28', 'recordDomain': '****.com', 'recordBucketName': '****', 'recordObjectName': '********.wav', 'ttsPlayTimes': 0, 'ttsTransDuration': 0, 'mptyId': '********', 'serviceType': '005', 'hostName': 'callenabler246.huaweicaas.com', 'extendNumber': '02' } ] }).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 'icid': 呼叫记录的唯一标识 'bindNum': 隐私保护号码 'sessionId': 通话链路的唯一标识 'callerNum': 主叫号码 'calleeNum': 被叫号码 'fwdDisplayNum': 转接呼叫时的显示号码 'fwdDstNum': 转接呼叫时的转接号码 'callInTime': 呼入的开始时间 'fwdStartTime': 转接呼叫操作的开始时间 'fwdAlertingTime': 转接呼叫操作后的振铃时间 'fwdAnswerTime': 转接呼叫操作后的应答时间 'callEndTime': 呼叫结束时间 'fwdUnaswRsn': 转接呼叫操作失败的Q850原因值 'failTime': 呼入,呼出的失败时间 'ulFailReason': 通话失败的拆线点 'sipStatusCode': 呼入,呼出的失败SIP状态码 'recordFlag': 录音标识 'recordStartTime': 录音开始时间 'recordObjectName': 录音文件名 'recordBucketName': 录音文件所在的目录名 'recordDomain': 存放录音文件的域名 'serviceType': 携带呼叫的业务类型信息 'hostName': 话单生成的服务器设备对应的主机名 'subscriptionId': 绑定关系ID 'extendNum': 分机号E ''' #短时间内有多个通话结束时隐私保护通话平台会将话单合并推送,每条消息最多携带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.'); def main(): onFeeEvent(jsonBody) #话单处理 if __name__ == '__main__': main() ```