init

open_xybot/__init__.py

@@ -0,0 +1,59 @@
+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)

pyproject.toml

@@ -0,0 +1,15 @@
+[tool.poetry]
+name = "open-xybot"
+version = "0.1.0"
+description = ""
+authors = ["woni <320753691@qq.com>"]
+readme = "README.md"
+packages = [{include = "open_xybot"}]
+
+[tool.poetry.dependencies]
+python = "^3.10"
+
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"

tests/__init__.py

@@ -0,0 +1,2 @@
+from open_xybot import Xybot
+