60 lines
1.9 KiB
Python
60 lines
1.9 KiB
Python
import json
|
|
from urllib.parse import urlencode
|
|
import requests
|
|
|
|
BASE_URL = "https://api.yingdao.com"
|
|
|
|
|
|
class Xybot:
|
|
def __init__(self, accessKeyId: str, accessKeySecret: str, **kwargs):
|
|
""""""
|
|
self.session = requests.session()
|
|
self.base_url = kwargs.get("base_url", BASE_URL)
|
|
self.accessKeyId = accessKeyId
|
|
self.accessKeySecret = accessKeySecret
|
|
self.auth()
|
|
|
|
def auth(self):
|
|
self.session.headers.update({"Content-Type": "application/json"})
|
|
self.session.headers.update({"Authorization": f"Bearer {self.get_token()}"})
|
|
|
|
def get_token(self):
|
|
endpoint = f"/oapi/token/v2/token/create?accessKeyId={self.accessKeyId}&accessKeySecret={accessKeySecret}"
|
|
url = self.base_url + endpoint
|
|
response = self.session.get(url)
|
|
# TODO: check response status code
|
|
return response.json()["data"]["accessToken"]
|
|
|
|
def start_job(self, job_id):
|
|
url = f"{self.api_base}/oapi/dispatch/v2/job/start"
|
|
response = requests.request("POST", url)
|
|
print(response.text)
|
|
|
|
def query_robot(self, robot_name: str, page: int = 1, size: int = 10):
|
|
query_params = urlencode({k: v for k, v in locals().items() if k != "self"})
|
|
endpoint = f"/oapi/robot/v2/query?{query_params}"
|
|
url = self.base_url + endpoint
|
|
response = self.session.get(url)
|
|
return response.json()
|
|
|
|
def query_robot_param(self):
|
|
endpoint = f"/oapi/robot/v2/queryRobotParam?robotUuid=xxxxx"
|
|
url = self.base_url + endpoint
|
|
respose = self.session.get(url)
|
|
print(respose.json())
|
|
|
|
def query_client(self):
|
|
pass
|
|
|
|
def query_client_list(self):
|
|
pass
|
|
|
|
|
|
if __name__ == "__main__":
|
|
accessKeyId = "zrkX4MVCHYnRQPtc@platform"
|
|
accessKeySecret = "07DrQGu61qgE53afdBxmFeKj4RbXtZ2w"
|
|
|
|
xybot = Xybot(accessKeyId, accessKeySecret)
|
|
res = xybot.query_robot("测试")
|
|
print(res)
|