add check activity help link

.env

@@ -27,3 +27,8 @@ RESULT_BACKEND=redis://office-xbot-redis
 # Yingdao PassWord
 XBOT_USERNAME=jx@bigdata
 XBOT_PASSWORD=320753691love
+
+
+# Feishu Client Group
+FEISHU_GROUP_CHAT_ID=oc_ca83ac8002b4fb3e1cda07cfbfd6e342
+

app/__init__.py

@@ -215,6 +215,12 @@ def celery_init_app(app: Flask) -> Celery:
             "schedule": crontab(hour=19, minute=57),
         },
     }
+    celery_app.conf.beat_schedule = {
+        "check-activity-help-link": {
+            "task": "app.task.check_activity_help_link",
+            "schedule": crontab(hour=19, minute=38),
+        },
+    }
 
     app.extensions["celery"] = celery_app
     return celery_app

app/task.py

@@ -8,7 +8,7 @@ from celery import Task
 
 from app.utils.utils import *
 from app.utils.yd_utils import get_token, get_activity_info
-from app.utils.feishu import create_doc, get_doc_block, create_doc_block, get_messages
+from app.utils.feishu import create_doc, get_doc_block, create_doc_block, get_messages, send_groud_message
 from app.utils.utils import update_record, add_record
 from app.utils.yd_utils import new_blank_robot, new_blank_activity
 
@@ -270,6 +270,41 @@ def sync_xbot_activity_info():
     print("新增的uuid: ", handle_list)
 
 
+@shared_task()
+def check_activity_help_link():
+    """判断指令的帮助文档是否缺失, 并推送消息"""
+    xbot_username = os.environ["XBOT_USERNAME"]
+    xbot_password = os.environ["XBOT_PASSWORD"]
+    app_id = os.environ["APP_ID"]
+    app_secret = os.environ["APP_SECRET"]
+
+    xbot_username = os.environ["XBOT_USERNAME"]
+    xbot_password = os.environ["XBOT_PASSWORD"]
+    chat_id = os.environ["FEISHU_GROUP_CHAT_ID"]
+
+    client = lark.Client.builder().app_id(app_id).app_secret(app_secret).build()
+
+    token = get_token(xbot_username, xbot_password)
+    activities_info = get_activity_info(token)
+
+    messages = []
+    messages.append("==========市场指令缺失文档==========")
+
+    for activity_info in activities_info:
+        package_viewN_name = activity_info.get("packageViewName")
+        activities = activity_info.get("activities", "空")
+        for activity in activities:
+            name = activity.get("name")
+            help_link = activity.get("helpLink", "")
+            if not help_link:
+                # print(f"{package_viewN_name}: {name} 缺失文档")
+                messages.append(f"{package_viewN_name}: {name}")
+
+    message_content = "\n".join(messages)
+
+    send_groud_message(client, chat_id, message_content)
+
+
 @shared_task(include_result=False)
 def process_new_demand(app_id, app_secret, table_id, record_id, data):
 

app/utils/feishu.py

@@ -309,6 +309,45 @@ def get_messages(client, container_id_type="", container_id="", page_size=50):
     return json.loads(lark.JSON.marshal(response.data))
 
 
+def send_groud_message(client, receive_id, content):
+
+    # 创建client
+    client = (
+        lark.Client.builder()
+        .app_id("YOUR_APP_ID")
+        .app_secret("YOUR_APP_SECRET")
+        .log_level(lark.LogLevel.DEBUG)
+        .build()
+    )
+
+    # 构造请求对象
+    request: CreateMessageRequest = (
+        CreateMessageRequest.builder()
+        .receive_id_type("chat_id")
+        .request_body(
+            CreateMessageRequestBody.builder()
+            .receive_id(receive_id)
+            .msg_type("text")
+            .content(content)
+            .build()
+        )
+        .build()
+    )
+
+    # 发起请求
+    response: CreateMessageResponse = client.im.v1.message.create(request)
+
+    # 处理失败返回
+    if not response.success():
+        lark.logger.error(
+            f"client.im.v1.message.create failed, code: {response.code}, msg: {response.msg}, log_id: {response.get_log_id()}, resp: \n{json.dumps(json.loads(response.raw.content), indent=4, ensure_ascii=False)}"
+        )
+        return
+
+    # 处理业务结果
+    lark.logger.info(lark.JSON.marshal(response.data, indent=4))
+
+
 if __name__ == "__main__":
     load_dotenv()
 

app/utils/yd_utils.py

@@ -633,10 +633,26 @@ 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)
+    token = get_token(xbot_username, xbot_password)
+    activities_info = get_activity_info(token)
+
+    messages = []
+    messages.append("==========市场指令缺失文档==========")
+
+    for activity_info in activities_info:
+        package_viewN_name = activity_info.get("packageViewName")
+        activities = activity_info.get("activities", "空")
+        for activity in activities:
+            name = activity.get("name")
+            help_link = activity.get("helpLink", "")
+            if not help_link:
+                # print(f"{package_viewN_name}: {name} 缺失文档")
+                messages.append(f"{package_viewN_name}: {name}")
+
+    print("\n".join(messages))
+
     # for item in office_activity_uuids:
     #     print(item.get("packageViewName"), item.get("uuid"))
 
-    new_blank_robot("测试-根据需求生成应用", "620156920605310976")
+    # new_blank_robot("测试-根据需求生成应用", "620156920605310976")
     # new_blank_activity()