add crontab task

app/__init__.py

@@ -15,7 +15,7 @@ from lark_oapi.api.im.v1 import *
 from lark_oapi.api.bitable.v1 import *
 
 from app.task import *
-from app.utils import *
+from app.utils.utils import *
 from app.task import *
 
 load_dotenv()
@@ -106,6 +106,10 @@ def create_app() -> Flask:
         add.delay(1, 2)
         return {"test": "ok"}
 
+    @app.route("/index", methods=["GET"])
+    def index():
+        return {"test": "ok"}
+
     return app
 
 

app/make_celery.py

@@ -2,3 +2,10 @@ from app import create_app
 
 flask_app = create_app()
 celery_app = flask_app.extensions["celery"]
+celery_app.conf.beat_schedule = {
+    "add-every-30-seconds": {
+        "task": "app.task.sync_xbot_activity_info",
+        "schedule": 1800,
+    }
+}
+celery_app.conf.timezone = "UTC"

app/task.py

@@ -6,7 +6,8 @@ from itertools import chain
 from celery import shared_task
 from celery import Task
 
-from app.utils import *
+from app.utils.utils import *
+from app.utils.yd_utils import get_token, get_activity_info
 
 
 @shared_task(ignore_result=False)
@@ -88,3 +89,15 @@ def process_message_callback(
     # 回复群组的消息
     content = f'{{"text": "反馈的问题已经记录到多维表格, 稍后有小伙伴跟进"}}'
     send_group_message(client, message_id, content)
+
+
+@shared_task()
+def sync_xbot_activity_info():
+    xbot_username = os.environ["XBOT_USERNAME"]
+    xbot_password = os.environ["XBOT_PASSWORD"]
+    token = get_token(xbot_username, xbot_password)
+    office_activity_uuids: list = get_activity_info(token)
+    for item in office_activity_uuids:
+        uuid = item.get("uuid")
+        package_viewName = item.get("packageViewName")
+        print(uuid, package_viewName)

app/utils/yd_utils.py

@@ -0,0 +1,100 @@
+import os
+import requests
+import dotenv
+
+from urllib import parse
+
+
+def get_token(username, password):
+    """
+    利用网页端获取 token
+    :param username: 用户名
+    :param password: 密码
+    :return: token
+    """
+
+    url = "https://api.winrobot360.com/oauth/token"
+    data = {
+        "username": username,
+        "password": password,
+        "grant_type": "password",
+        "scope": "all",
+    }
+
+    payload = parse.urlencode(data)
+    headers = {
+        "accept": "application/json",
+        "authorization": "Basic Y29uc29sZTpzVVdpeTZzTmlYZDBlNUxo",
+        "content-type": "application/x-www-form-urlencoded;charset=UTF-8",
+    }
+    response = requests.request("POST", url, headers=headers, data=payload).json()
+
+    if response.get("success") == "false":
+        raise ValueError(response.get("msg"))
+    return response.get("access_token")
+
+
+def get_activity_info(token, page=1, size=200):
+    """
+    获取活动信息列表
+    :param token: 认证用的 token
+    :param page: 分页页码，默认为 1
+    :param size: 分页大小，默认为 10
+    :return: 所有页的数据列表
+    """
+    url = "https://api.winrobot360.com/api/v1/robot/activity/package/list"
+    all_data = []
+    while True:
+        payload = {
+            "source": "market",
+            "page": page,
+            "size": size,
+            "groupInfoUuid": "22222222-2222-2222-2222-222222222222",
+            "marketUiTag": "",
+            "uiaType": "pc",
+            "order": "desc",
+        }
+        headers = {
+            "Authorization": f"bearer {token}",
+            "Accept": "application/json",
+            "Accept-Encoding": "json",
+        }
+        response = requests.get(url, headers=headers, params=payload)
+        response_json = response.json()
+        if not response.ok or not response_json.get("success"):
+            raise ValueError(response_json.get("msg"))
+        data = response_json.get("data")
+        all_data.extend(data)
+        if not data or len(data) < size:
+            break
+        page += 1
+    return all_data
+
+
+def get_activity_detail(token, package_name, package_version):
+    """
+    获取自定义指令的相关信息
+    :param token: token
+    :param sdk: 自定义指令
+    :return: package_name, package_url
+    """
+
+    url = f"https://api.winrobot360.com/api/v1/robot/use/activity/package/detail?packageName={package_name}&version={package_version}"
+    payload = {}
+    headers = {
+        "Authorization": f"bearer {token}",
+        "Accept": "application/json",
+        "Accept-Encoding": "json",
+    }
+    response = requests.request("GET", url, headers=headers, data=payload)
+    return response.json().get("data")
+
+
+if __name__ == "__main__":
+    dotenv.load_dotenv()
+    xbot_username = os.environ["XBOT_USERNAME"]
+    xbot_password = os.environ["XBOT_PASSWORD"]
+    token = get_token(xbot_username, xbot_password)
+    office_activity_uuids: list = get_activity_info(token)
+    for item in office_activity_uuids:
+        print(item)