1. 適用版本
適用于RPA2020.4以及以上版本,
2. 接口 API 手冊
調用方式及字段,請參考論壇手冊:
暫未在論壇公開,請向當地廠商技術索取
3. Python 調用代碼
如果想用機器人來調 API 接口,可參考以下代碼,其它語言邏輯一樣:
import json
import time
import requests
from urllib import parse
import hmac
import base64
from hashlib import sha256
def rpa_rest_2020_4(host='',rest_type='',data_json=None,mode='',port=443,accesstoken=None,retry=2):
????'''
????host:地址,str型,示例:'https://192.168.202.11'
????rest_type:str型,接口類型,示例:'/oapi/v1/job/create'
????data_json:字典型,發送的報文數據json格式
????mode:接口請求方式(get、post、delete及put)
????port:int型,https端口
????retry:int型,重試次數
????返回值:get_field_json:字典型,
????'''
????get_field_json={'code': 40, 'msg': 'fail,意外情況','result':None}
????#url參數轉換
????def json2Params(data_json):
????????get_field_json={'code': 41, 'msg': 'fail,轉換URL參數失??!','result':None}
????????try:
????????????data_json=json.loads(data_json)
????????????url_str = ''
????????????nums = 0
????????????max_nums = len(data_json)
????????????for key in data_json:
????????????????nums += 1
????????????????if nums == max_nums:
????????????????????url_str += str(key) + '=' + str(data_json[key])
????????????????else:????????????
????????????????????url_str += str(key) + '=' + str(data_json[key]) + '&'
????????except Exception as e:
????????????print('參數轉化失敗:',e)
????????????url_str=''
????????finally:
????????????return url_str
????if mode != 'get' and mode != 'put' and mode != 'post' and mode != 'delete':
????????get_field_json={'code': 42, 'msg': 'fail,mode錯誤,只能為get、put、post或delete!','result':None}
????????return get_field_json
????#獲得簽名sign
????sign = ''
????sign_yc = ''
????url = host
????if port != 443:
????????url = url + ':' + str(port)
????url += rest_type
????timestamp = None
????#獲取簽名值sign:
????try:
????????get_field_json={'code': 41, 'msg': 'fail,獲取簽名失?。?,'result':None}
????????#獲得毫秒級時間戳(時間出入不能大于10分鐘)
????????timestamp=str(int(round(time.time() * 1000)))
????????#處理json數據不為str類型的情況
????????for key in data_json:
????????????data_json[key]=str(data_json[key])
????????#json排序
????????data_json=json.dumps(data_json,sort_keys=True)
????????#構造源串sign_yc
????????if mode == 'get':
????????????json_str=json2Params(data_json)
????????????sign_yc=json_str + '&token=' + accesstoken + '×tamp=' + timestamp
????????????url = url+'?'+json_str
????????else:
????????????sign_yc=str(data_json) + '&token=' + accesstoken + '×tamp=' + timestamp
????????print('源串:',sign_yc)
????????print('URL:',url)
????????#URL編碼
????????url_bm = parse.quote(sign_yc, encoding="utf-8")
????????#sha256加密密碼
????????byte_key = bytes(accesstoken, encoding="utf-8")
????????byte_url_bm = bytes(url_bm, encoding="utf-8")
????????hn256 = hmac.new(byte_key, byte_url_bm, digestmod=sha256)
????????hh256 = hn256.hexdigest()
????????#Base64編碼
????????bb64 = base64.b64encode(bytes(hh256, encoding="utf-8"))
????????#獲取sign
????????sign = str(bb64, "utf-8")
????????#替換rn
????????sign=sign.replace('rn', '')
????????print('簽名值sign:',sign)
????except Exception as e:
????????print(e)
????????return get_field_json
????#開始嘗試發送API
????print('開始嘗試發送API')
????for count in range(retry):
????????print('嘗試第',count+1,'次API請求任務。')
????????try:
????????????print('開始嘗試:',mode)
????????????#post、delete、put
????????????if mode!='get':
????????????????header_dict = {'Content-Type': 'application/json;charset=UTF-8','accesstoken':accesstoken,'timestamp':timestamp,'signature':sign}
????????????????if mode=='post':
????????????????????if rest_type == '/oapi/v1/job/create':
????????????????????????header_dict = {'Content-Type': 'application/x-www-form-urlencoded;charset=UTF-8','accesstoken':accesstoken,'timestamp':timestamp,'signature':sign}
????????????????????????json_str=json2Params(data_json)
????????????????????????url = url+'?'+json_str
????????????????????????res = requests.post(url, data=str(data_json), headers=header_dict,verify=False)
????????????????????else:
????????????????????????json_str=json2Params(data_json)
????????????????????????url = url+'?'+json_str
????????????????????????header_dict = {'Content-Type': 'application/x-www-form-urlencoded;charset=UTF-8','accesstoken':accesstoken,'timestamp':timestamp,'signature':sign}
????????????????????????res = requests.post(url, data=str(data_json), headers=header_dict,verify=False)
????????????????if mode=='put':
????????????????????res = requests.put(url, data=str(data_json), headers=header_dict,verify=False)
????????????????if mode=='delete':
????????????????????res = requests.delete(url, data=str(data_json), headers=header_dict,verify=False)
????????????if mode=='get':
????????????????header_dict = {'accesstoken':accesstoken,'timestamp':timestamp,'signature':sign}
????????????????res = requests.get(url,headers=header_dict,verify=False)
????????????#獲取返回值
????????????res_text=res.text
????????????#轉化成json
????????????get_field_json=json.loads(res_text)
????????????print('請求成功!')
????????????break
????????except Exception as e:
????????????print('請求失?。?,e)
????????????get_field_json={'code': 40, 'msg': 'fail,發送API失??!','result':None}
????return get_field_json
def get_token(host='',port='',accessKey='',secretKey='',retry=2):
????'''
????host:地址,str型,示例:'https://192.168.202.11'
????port:int型,https端口
????accessKey:str型,服務平臺應用AppKey
????secretKey:str型,服務平臺應用AppSecret
????retry:int型,重試次數
????'''
????get_field_json={'code': 44, 'msg': 'fail,獲取token失敗','result':None}
????for i in range(retry):
????????try:
????????????url = host
????????????if port != 443:
????????????????url = url + ':' + str(port)
????????????json_str='accessKey='+accessKey+'&secretKey='+secretKey
????????????url = url+'/oapi/v1/token?'+json_str
????????????res = requests.get(url,verify=False)
????????????res_text=res.text
????????????#轉化成json
????????????get_field_json=json.loads(res_text)
????????????print('獲取token,第'+str(i+1)+'次,成功')
????????????break
????????except Exception as e:
????????????print('獲取token,第'+str(i+1)+'次,失敗',e)
????????????time.sleep(1)
????return get_field_json????????
def refresh_token(host='',port='',refleshtoken='',retry=2):
????'''
????host:地址,str型,示例:'https://192.168.202.11'
????port:int型,https端口
????refleshtoken:str型,刷新token
????retry:int型,重試次數
????'''
????get_field_json={'code': 44, 'msg': 'fail,刷新token失敗','result':None}
????for i in range(retry):
????????try:
????????????url = host
????????????if port != 443:
????????????????url = url + ':' + str(port)
????????????json_str='refreshToken='+refleshtoken
????????????url = url+'/oapi/v1/token?'+json_str
????????????res = requests.get(url,verify=False)
????????????res_text=res.text
????????????#轉化成json
????????????get_field_json=json.loads(res_text)
????????????print('刷新token,第'+str(i+1)+'次,成功')
????????????break
????????except Exception as e:
????????????print('刷新token,第'+str(i+1)+'次,失敗',e)
????????????time.sleep(1)
????return get_field_json
4. 其它平臺或客戶端調用
4.1 其它平臺調用
按照第 4 章的邏輯自行寫調用代碼即可。
4.2 機器人調用
按照第 4 章添加一個全局函數,在需要調用的地方使用全局函數控件即可。
注:需要提前獲取token后調用,如無第三方平臺對接,獲取的token可存在共享變量里,分配權限調用,參考下圖:


版權聲明:本文內容由互聯網用戶自發貢獻,該文觀點僅代表作者本人。本站僅提供信息存儲空間服務,不擁有所有權,不承擔相關法律責任。如發現本站有涉嫌抄襲侵權/違法違規的內容, 請發送郵件至 舉報,一經查實,本站將立刻刪除。
發表評論
請登錄后評論...
登錄后才能評論