文档首页 > > 开发指南> 代码样例> Python代码样例> AX模式

AX模式

分享
更新时间: 2019/10/08 GMT+08:00

样例

AX模式绑定接口

AX模式解绑接口

AX模式绑定信息修改接口

AX模式绑定信息查询接口

AX模式设置临时被叫接口

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

呼叫事件通知接口

话单通知接口

短信通知接口

环境要求

基于Python 3.7.0版本,要求Python 3.0及以上版本。

引用库

requests 2.18.1(仅“获取录音文件下载地址接口”引用)

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

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

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

  • 本文档所述Demo在提供服务的过程中,可能会涉及个人数据的使用,建议您遵从国家的相关法律采取足够的措施,以确保用户的个人数据受到充分的保护。
  • 本文档所述Demo仅用于功能演示,不允许客户直接进行商业使用。
  • 本文档信息仅供参考,不构成任何要约或承诺。

AX模式绑定接口

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
# -*- coding: utf-8 -*-
import time
import uuid
import hashlib
import base64
import json
import ssl
import urllib.request

# 必填,请参考"开发准备"获取如下数据,替换为实际值
realUrl = 'https://rtcapi.cn-north-1.myhuaweicloud.com:12543/rest/provision/caas/privatenumber/v1.0' #APP接入地址+接口访问URI
APP_KEY = "a1d1f50cad21415fbdd13d8f53d36d60" #APP_Key
APP_SECRET = "cfc881cc704c4fba8d8fef5788e03e6b" #APP_Secret
origNum = '+8618612345678' #A号码
privateNum = '+8617010000001' #X号码(隐私号码)

'''
选填,各参数要求请参考"AX模式绑定接口"
'''
privateNumType = 'mobile-virtual' #固定为mobile-virtual
# areaCode = '0755' #需要绑定的X号码对应的城市码
# recordFlag = 'true' #是否需要针对该绑定关系产生的所有通话录音
# recordHintTone = 'recordHintTone.wav' #设置录音提示音
calleeNumDisplay = '0' #设置非A用户呼叫X时,A接到呼叫时的主显号码
# privateSms = 'true' #设置该绑定关系是否支持短信功能

# callerHintTone = 'callerHintTone.wav' #设置A拨打X号码时的通话前等待音
# calleeHintTone = 'calleeHintTone.wav' #设置非A用户拨打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({
        'origNum':origNum,
        'privateNum':privateNum,
        'privateNumType':privateNumType,
#         'areaCode':areaCode,
#         'recordFlag':recordFlag,
#         'recordHintTone':recordHintTone,
        'calleeNumDisplay':calleeNumDisplay,
#         '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() 

AX模式解绑接口

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
# -*- coding: utf-8 -*-
import time
import uuid
import hashlib
import base64
import ssl
import urllib.request

# 必填,请参考"开发准备"获取如下数据,替换为实际值
realUrl = 'https://rtcapi.cn-north-1.myhuaweicloud.com:12543/rest/provision/caas/privatenumber/v1.0' #APP接入地址+接口访问URI
APP_KEY = "a1d1f50cad21415fbdd13d8f53d36d60" #APP_Key
APP_SECRET = "cfc881cc704c4fba8d8fef5788e03e6b" #APP_Secret

'''
选填,各参数要求请参考"AX模式解绑接口"
subscriptionId和(origNum+privateNum)为二选一关系,都携带时以subscriptionId为准
'''
origNum = '+8618612345678' #A号码
privateNum = '+8617010000001' #X号码
# subscriptionId = '0167ecc9-bfb6-4eec-b671-a7dab2ba78cf' #指定"AX模式绑定接口"返回的绑定ID进行解绑

'''
构造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({
        'origNum':origNum,
        'privateNum':privateNum,
#         'subscriptionId':subscriptionId
    })
    #完整请求地址
    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))
    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() 

AX模式绑定信息修改接口

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
# -*- coding: utf-8 -*-
import time
import uuid
import hashlib
import base64
import json
import ssl
import urllib.request

# 必填,请参考"开发准备"获取如下数据,替换为实际值
realUrl = 'https://rtcapi.cn-north-1.myhuaweicloud.com:12543/rest/provision/caas/privatenumber/v1.0' #APP接入地址+接口访问URI
APP_KEY = "a1d1f50cad21415fbdd13d8f53d36d60" #APP_Key
APP_SECRET = "cfc881cc704c4fba8d8fef5788e03e6b" #APP_Secret

'''
选填,各参数要求请参考"AX模式绑定信息修改接口"
subscriptionId和(origNum+privateNum)为二选一关系,都携带时以subscriptionId为准
'''
origNum = '+8618612345678' #A号码
privateNum = '+8617010000001' #X号码
# subscriptionId = '0167ecc9-bfb6-4eec-b671-a7dab2ba78cf' #指定"AX模式绑定接口"返回的绑定ID进行修改

privateSms = 'true' #必填,修改该绑定关系是否支持短信功能

'''
构造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({
        'origNum':origNum,
        'privateNum':privateNum,
#         'subscriptionId':subscriptionId,
        'privateSms':privateSms
    }).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))
    req.add_header('Content-Type', 'application/json;charset=UTF-8')

    # 为防止因HTTPS证书认证失败造成API调用失败,需要先忽略证书信任问题
    ssl._create_default_https_context = ssl._create_unverified_context

    try:
        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()

AX模式绑定信息查询接口

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
# -*- coding: utf-8 -*-
import time
import uuid
import hashlib
import base64
import ssl
import urllib.request

# 必填,请参考"开发准备"获取如下数据,替换为实际值
realUrl = 'https://rtcapi.cn-north-1.myhuaweicloud.com:12543/rest/provision/caas/privatenumber/v1.0' #APP接入地址+接口访问URI
APP_KEY = "a1d1f50cad21415fbdd13d8f53d36d60" #APP_Key
APP_SECRET = "cfc881cc704c4fba8d8fef5788e03e6b" #APP_Secret

'''
选填,各参数要求请参考"AX模式绑定信息查询接口"
subscriptionId和origNum为二选一关系,两者都携带时以subscriptionId为准
'''
origNum = '+8618612345678' #指定A号码进行查询
subscriptionId = '0167ecc9-bfb6-4eec-b671-a7dab2ba78cf' #指定"AX模式绑定接口"返回的绑定ID进行查询

'''
构造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({
        'origNum':origNum,
        'subscriptionId':subscriptionId
    })
    #完整请求地址
    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))
    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() 

AX模式设置临时被叫接口

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
# -*- coding: utf-8 -*-
import time
import uuid
import hashlib
import base64
import json
import ssl
import urllib.request

# 必填,请参考"开发准备"获取如下数据,替换为实际值
realUrl = 'https://rtcapi.cn-north-1.myhuaweicloud.com:12543/rest/caas/privatenumber/calleenumber/v1.0' #APP接入地址+接口访问URI
APP_KEY = "a1d1f50cad21415fbdd13d8f53d36d60" #APP_Key
APP_SECRET = "cfc881cc704c4fba8d8fef5788e03e6b" #APP_Secret

'''
选填,各参数要求请参考"AX模式设置临时被叫接口"
subscriptionId和(origNum+privateNum)为二选一关系,都携带时以subscriptionId为准
'''
origNum = '+8618612345678' #A号码
privateNum = '+8617010000001' #X号码
# subscriptionId = '0167ecc9-bfb6-4eec-b671-a7dab2ba78cf' #指定“AX模式绑定接口”返回的绑定ID设置临时被叫

calleeNum = '+8618612345679' #必填,本次呼叫的真实被叫号码

'''
构造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({
        'origNum':origNum,
        'privateNum':privateNum,
#         'subscriptionId':subscriptionId,
        'calleeNum':calleeNum
    }).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))
    req.add_header('Content-Type', 'application/json;charset=UTF-8')

    # 为防止因HTTPS证书认证失败造成API调用失败,需要先忽略证书信任问题
    ssl._create_default_https_context = ssl._create_unverified_context

    try:
        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() 

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
# -*- coding: utf-8 -*-
import time
import uuid
import hashlib
import base64
import urllib
import requests #需要先使用pip install requests命令安装依赖

# 必填,请参考"开发准备"获取如下数据,替换为实际值
realUrl = 'https://rtcapi.cn-north-1.myhuaweicloud.com:12543/rest/provision/voice/record/v1.0' #APP接入地址+接口访问URI
APP_KEY = "a1d1f50cad21415fbdd13d8f53d36d60" #APP_Key
APP_SECRET = "cfc881cc704c4fba8d8fef5788e03e6b" #APP_Secret

# 必填,通过"话单通知接口"获取
recordDomain = 'ostor.huawei.com' #录音文件存储的服务器域名
fileName = '1200_366_0_20161228102743.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() 

呼叫事件通知接口

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
# -*- coding: utf-8 -*-
'''
呼叫事件通知
客户平台收到隐私保护通话平台的呼叫事件通知的接口通知
'''
import json

#呼叫事件通知样例
jsonBody = json.dumps({
    'eventType': 'disconnect',
    'statusInfo': {
        'sessionId': '1200_1827_4294967295_20190124023003@callenabler246.huaweicaas.com',
        'timestamp': '2019-01-24 02:30:22',
        'caller': '+8613800000022',
        'called': '+8613866887021',
        'stateCode': 0,
        'stateDesc': 'The user releases the call.',
        'subscriptionId': '1d39eaef-9279-4d18-b806-72e43ab3f85c'
    }
}).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()

话单通知接口

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
# -*- coding: utf-8 -*-
'''
话单通知
客户平台收到隐私保护通话平台的话单通知的接口通知
'''
import json

#话单通知样例
jsonBody = json.dumps({
    'eventType': 'fee',
    'feeLst': [
        {
            'direction': 1,
            'spId': 'linlingoo_omp',
            'appKey': 'E783q8G7GfSV7SZoq5Vj8d4BxXQ7',
            'icid': 'ba171f34e6953fcd751edc77127748f4.3757285803.338666679.5',
            'bindNum': '+8613800000022',
            'sessionId': '1200_1827_4294967295_20190124023003@callenabler246.huaweicaas.com',
            'subscriptionId': '1d39eaef-9279-4d18-b806-72e43ab3f85c',
            'callerNum': '+8613800000021',
            'calleeNum': '+8613800000022',
            'fwdDisplayNum': '+8613800000022',
            'fwdDstNum': '+8613866887021',
            'callInTime': '2019-01-24 02:30:03',
            'fwdStartTime': '2019-01-24 02:30:03',
            'fwdAlertingTime': '2019-01-24 02:30:04',
            'fwdAnswerTime': '2019-01-24 02:30:06',
            'callEndTime': '2019-01-24 02:30:22',
            'fwdUnaswRsn': 0,
            'ulFailReason': 0,
            'sipStatusCode': 0,
            'callOutUnaswRsn': 0,
            'recordFlag': 1,
            'recordStartTime': '2019-01-24 02:30:06',
            'recordDomain': 'ostor.huawei.com',
            'recordBucketName': 'sp-4jynna8mi6trquij21t67h3sv9br',
            'recordObjectName': '19012407030312039967859.wav',
            'ttsPlayTimes': 0,
            'ttsTransDuration': 0,
            'mptyId': '1d39eaef-9279-4d18-b806-72e43ab3f85c',
            'serviceType': '003',
            '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
    '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
    '''
    #短时间内有多个通话结束时隐私保护通话平台会将话单合并推送,每条消息最多携带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()

短信通知接口

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
# -*- coding: utf-8 -*-
'''
短信通知
客户平台收到隐私保护通话平台的短信通知的接口通知
'''
import json

#短信通知样例
jsonBody = json.dumps({
    'appKey': 'S368k6Y31w4kNuI6X1gsne5n09gA',
    'smsEvent': {
        'smsIdentifier': 'db2e7744804f4c5aa7d923aeddb42443',
        'notificationMode': 'Block',
        'calling': '+8613800000001',
        'virtualNumber': '+8613800000000',
        'event': 'TextSMS',
        'timeStamp': '2018-09-13T09:46:16.023Z'
    }
}).encode('ascii')

print(jsonBody)

'''
短信通知
@see: 详细内容以接口文档为准
@param param: jsonBody
@return: 
'''
def onSmsEvent(jsonBody):
    jsonObj = json.loads(jsonBody) #将通知消息解析为jsonObj

    if ('smsEvent' not in jsonObj):
        print('param error: no smsEvent.')
        return

    #print(jsonObj['appKey']) #商户应用的AppKey

    smsEvent = jsonObj['smsEvent'] #短信通知信息

    '''
    Example: 此处已解析notificationMode为例,请按需解析所需参数并自行实现相关处理

    'smsIdentifier': 短信唯一标识
    'notificationMode': 通知模式
    'calling': 真实发送方号码
    'called': 真实接收方号码
    'virtualNumber': 隐私号码(X号码)
    'event': 短信状态事件
    'timeStamp': 短信事件发生的系统时间戳,UTC时间
    'subscriptionId': 绑定ID
    'smsContent': 用户发送的短信内容
    '''
    if ('notificationMode' in smsEvent):
        if ('Block' == smsEvent['notificationMode']):
            #收到隐私保护通话平台的短信通知,若为Block模式,请参考接口文档回消息指示下一步操作
            actions = {
                'operation': 'vNumberRoute', #操作类型:转发短信/丢弃短信
                'message': {
                    'called': '+8613866887022', #真实接收方号码
                    'calling': '+8613866887021' #真实发送方号码
                }
            }
            resp = json.dumps({'actions': [actions]}) #Block模式响应消息
            print('resp: ' + resp);
        elif ('Notify' == smsEvent['notificationMode']):
            #收到隐私保护通话平台的短信通知,若为Notify模式,请回HTTP状态码为200的空消息
            #statusCode = 200;
            print('This is AX sms Notify mode.');
        else:
            print('notificationMode param error.');

def main():
    onSmsEvent(jsonBody); #短信通知处理

if __name__ == '__main__':
    main()
分享:

    相关文档

    相关产品

文档是否有解决您的问题?

提交成功!

非常感谢您的反馈,我们会继续努力做到更好!

反馈提交失败,请稍后再试!

*必选

请至少选择或填写一项反馈信息

字符长度不能超过200

提交反馈 取消

如您有其它疑问,您也可以通过华为云社区问答频道来与我们联系探讨

跳转到云社区