云服务器内容精选

  • 短信通知接口 1 2 3 4 5 6 7 8 910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576 # -*- coding: utf-8 -*-'''短信通知客户平台收到 隐私保护通话 平台的短信通知的接口通知'''import json#短信通知样例jsonBody = json.dumps({ 'appKey': '****', 'smsEvent': { 'smsIdentifier': '****', 'notificationMode': 'Block', 'calling': '+86138****0001', 'virtualNumber': '+86138****0000', 'event': 'Text SMS ', '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': '+86138****7022', #真实接收方号码 'calling': '+86138****7021' #真实发送方号码 } } resp = json.dumps({'actions': [actions]}) #Block模式响应消息 print(resp); elif ('Notify' == smsEvent['notificationMode']): #收到隐私保护通话平台的短信通知,若为Notify模式,请回HTTP状态码为200的空消息 #statusCode = 200; print('This is AXB sms Notify mode.'); else: print('notificationMode param error.');def main(): onSmsEvent(jsonBody); #短信通知处理if __name__ == '__main__': main()
  • 获取录音文件下载地址接口 1 2 3 4 5 6 7 8 91011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162 # -*- coding: utf-8 -*-import timeimport uuidimport hashlibimport base64import urllibimport requests #需要先使用pip install requests命令安装依赖import hmacfrom hashlib import sha256# 必填,请参考"开发准备"获取如下数据,替换为实际值realUrl = 'https://rtcpns.cn-north-1.myhuaweicloud.com/rest/provision/voice/record/v1.0' #APP接入地址+接口访问URIAPP_KEY = "a1********" #APP_KeyAPP_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()
  • 呼叫事件通知接口 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 99100101102103104105106107108109110111112113114115116117118119120121 # -*- 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']) returndef main(): onCallEvent(jsonBody) #呼叫事件处理if __name__ == '__main__': main()
  • AXB模式绑定信息查询接口 1 2 3 4 5 6 7 8 9101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566 # -*- coding: utf-8 -*-import timeimport uuidimport hashlibimport base64import sslimport urllib.requestimport hmacfrom hashlib import sha256# 必填,请参考"开发准备"获取如下数据,替换为实际值realUrl = 'https://rtcpns.cn-north-1.myhuaweicloud.com/rest/caas/relationnumber/partners/v1.0' #APP接入地址+接口访问URIAPP_KEY = "a1********" #APP_KeyAPP_SECRET = "cfc8********" #APP_Secret'''选填,各参数要求请参考"AXB模式绑定信息查询接口"subscriptionId和relationNum为二选一关系,两者都携带时以subscriptionId为准'''subscriptionId = '****' #指定"AXB模式绑定接口"返回的绑定ID进行查询relationNum = '+86170****0001' #指定X号码(隐私号码)进行查询# pageIndex = 1 #查询的分页索引,从1开始编号# pageSize = 20 #查询的分页大小,即每次查询返回多少条数据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, 'relationNum':relationNum,# 'pageIndex':pageIndex,# 'pageSize':pageSize }) #完整请求地址 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()
  • AXB模式绑定信息修改接口 1 2 3 4 5 6 7 8 9101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778 # -*- coding: utf-8 -*-import timeimport uuidimport hashlibimport base64import jsonimport sslimport urllib.requestimport hmacfrom hashlib import sha256# 必填,请参考"开发准备"获取如下数据,替换为实际值realUrl = 'https://rtcpns.cn-north-1.myhuaweicloud.com/rest/caas/relationnumber/partners/v1.0' #APP接入地址+接口访问URIAPP_KEY = "a1********" #APP_KeyAPP_SECRET = "cfc8********" #APP_SecretsubscriptionId = '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# };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({ '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', '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(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模式解绑接口 1 2 3 4 5 6 7 8 91011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859 # -*- coding: utf-8 -*-import timeimport uuidimport hashlibimport base64import sslimport urllib.requestimport hmacfrom hashlib import sha256# 必填,请参考"开发准备"获取如下数据,替换为实际值realUrl = 'https://rtcpns.cn-north-1.myhuaweicloud.com/rest/caas/relationnumber/partners/v1.0' #APP接入地址+接口访问URIAPP_KEY = "a1********" #APP_KeyAPP_SECRET = "cfc8********" #APP_Secret'''选填,各参数要求请参考"AXB模式解绑接口"subscriptionId和relationNum为二选一关系,两者都携带时以subscriptionId为准'''subscriptionId = '****' #指定"AXB模式绑定接口"返回的绑定ID进行解绑relationNum = '+86170****0001' #指定X号码(隐私号码)进行解绑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, 'relationNum':relationNum }) #完整请求地址 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()
  • AXB模式绑定接口 1 2 3 4 5 6 7 8 9101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687 # -*- coding: utf-8 -*-import timeimport uuidimport hashlibimport base64import jsonimport sslimport urllib.requestimport hmacfrom hashlib import sha256# 必填,请参考"开发准备"获取如下数据,替换为实际值realUrl = 'https://rtcpns.cn-north-1.myhuaweicloud.com/rest/caas/relationnumber/partners/v1.0' #APP接入地址+接口访问URIAPP_KEY = "a1********" #APP_KeyAPP_SECRET = "cfc8********" #APP_SecretrelationNum = '+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# };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({ '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', '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()