"1. 适用版本适用于RPA2020.5、2021.1版本，RPA2020.3无API，其它版本请参考： ==> rpa9 点击跳转 ==> rpa10 点击跳转 ==> RPA2020.4 点击跳转 2. 接口 API ...."

guanyf

Rpa 42 号会员
函数编码设计 python基础 • 3 回帖 • 1.7K 浏览 • 2020-12-25 14:33:53

PRA2020.5 平台 API 接口调用（2021-01-21 更新）

1. 适用版本

适用于RPA2020.5、2021.1版本，RPA2020.3无API，其它版本请参考：
==> rpa9 点击跳转
 ==> rpa10 点击跳转
 ==> RPA2020.4 点击跳转

2. 接口 API 手册

调用方式及字段，请参考手册，打开方式如下：

部署 2020.5 平台后，浏览器打开https://ip/doc.html

3. Python 调用代码

如果想用机器人来调 API 接口，可参考以下代码，其它语言逻辑一样:
部分功能接口如无法调用，请参考手册修改传参

import hmac
import json
from base64 import b64encode
from hashlib import sha256
from urllib.parse import urlparse, urlencode, quote, unquote
import time
import requests


def get_token(url='', key='',secrect=''):
    """
    :param url: 请求地址,示例：https://192.168.202.11/oapi/v1/token
    :return:返回token,[access_token,refresh_token]
    """
    params = {'accessKey': key, 'secretKey': secrect}
    res = requests.get(url, params=params, verify=0)
    return res.json()

def refresh_token(url='',refresh_token=''):
    """
    :param url: 请求地址,示例：https://192.168.202.11/oapi/v1/token
    :param refresh_token:str型，刷新token
    :return:返回token,[access_token,refresh_token]
    """
    json_str='refreshToken='+refresh_token
    url = url+'?'+json_str
    res = requests.get(url,verify=False)
    return res.json()

def rpa_base64_encode(data):
    """
    Base64编码，并处理\r\n
    :param data: 原始字符
    :return: 编码后的字符
    """
    encoded = b64encode(data.encode("utf-8")).decode("utf-8")
    return encoded.replace("\r\n", "")


def rpa_sha256(key, value):
    """
    SHA256加密，加密结果转成16进制字符串形式
    :param key: 密钥
    :param value: 原文
    :return: 16进制密文
    """


    key = key.encode("utf-8")
    value = value.encode("utf-8")
    ret = hmac.new(key, value, digestmod=sha256).hexdigest()
    print("sha265:"+ ret)
    return ret


def sort_dict(d: dict) -> dict:
    """
    以字典的key按照升序排列
    :param d: 字典对象
    :return: 排序后的字典对象
    """
    return {key: value for key, value in sorted(d.items(), key=lambda item: item[0])}


def rpa_url_encode(origin_data: dict, is_json=False, ts=""):
    """
    用url编码数据，并处理特殊字符
    Get方法、上传文件时设置is_json=False，其它情况为True
    :param origin_data: 原始数据
    :param is_json: 是否以json方式处理
    :param ts: 时间戳，外界传入
    :return: 处理后的数据
    """


    if is_json is True:
        data = json.dumps(origin_data, ensure_ascii=True)
        data = data + f"&timestamp={ts}"
    else:
        # Get方法需要对参数排序
        sorted_data = sort_dict(origin_data)
        # 添加时间戳
        sorted_data.update({"timestamp": ts})
        data = unquote(urlencode(sorted_data))
    encoded = quote(data, encoding="utf-8")
    r = encoded.replace("+", "+").replace("*", "%2A").replace("~", "%7E")
    print('url编码：'+r)
    return r


def rpa_timestamp():
    """RPA风格的时间戳"""
    return f"{int(time.time() * 1000)}"


def gen_signature(data: dict, is_json: bool, ts: str, token):
    """
    生成签名
    :param data: 请求参数
    :param is_json: 是否以json方式处理（除Get和文件上传外，其它都用json方式）
    :param ts: 时间戳，从外界传入
    :return: 签名
    """
    #用url编码数据，并处理特殊字符
    encoded = rpa_url_encode(data, is_json, ts)
    #SHA256加密，加密结果转成16进制字符串形式
    encrypted = rpa_sha256(token, encoded)
    #Base64编码，并处理\r\n
    b64 = rpa_base64_encode(encrypted)
    print('签名：'+b64)
    return b64


class SendRequestFailed(Exception):
    pass


def rpa_rest_2020_5(method, url, params: dict, has_signature: bool = True,is_json:bool = True, timeout: int = None, token=""):
    """
    向服务端发送请求
    :param method: 请求方法,"GET", "POST", "PUT", "DELETE"
    :param url: 请求地址,示例：https://192.168.202.11/oapi/v1/asset/list
    :param params: 请求参数（GET的url参数或POST的Body参数）
    :param has_signature: 是否签名
    :is_json:参数是否以json形式传递,GET、DELETE全部为False
    :param timeout 超时时间
    :param token
    :return: 服务端返回数据（json）
    """
    assert method.upper() in ("GET", "POST", "PUT", "DELETE")
    ts = rpa_timestamp()
    print(f"时间戳：{ts}")
    headers = None
    if method.upper() in ["GET", "DELETE"]:
        signature = gen_signature(params, is_json, ts, token)
        print("sig:", signature)
        if has_signature:
            headers = {"accesstoken": token,"timestamp": ts,"signature": signature}
        if method.upper() == "GET":
            r = requests.get(url=url, headers=headers, params=params, timeout=timeout, verify=False)
        else:
            r = requests.delete(url=url, headers=headers, params=params, timeout=timeout, verify=False)
    else:
        if is_json:
            signature = gen_signature(params, True, ts, token)
        else:
            signature = gen_signature(params, False, ts, token)
            
        if has_signature:
            headers = {"accesstoken": token,"timestamp": ts,"signature": signature}
            
        if is_json:
            r = requests.request(method=method, url=url, headers=headers, json=params, timeout=timeout, verify=False)
        else:
            r = requests.request(method=method, url=url, headers=headers, params=params, timeout=timeout, verify=False)
    if r.status_code != 200:
        raise SendRequestFailed(f"发送请求失败!服务器返回码：{r.status_code};返回内容：{r.text}.")
    return r.text