feat:修改监控脚本

monitor/supply_workflow_monitor.py

@@ -20,6 +20,11 @@ column_width_map = {
     "状态": "80px",
     "任务数": "80px",
     "记录数": "80px",
+
+    "待处理": "80px",
+    "运行中": "80px",
+    "成功": "80px",
+    "失败": "80px",
 }
 
 header_template = ["blue", "wathet", "turquoise", "green", "yellow", "orange"]
@@ -156,6 +161,33 @@ def build_card_json(body_elements: List[Dict[str, Any]], header: Dict[str, Any],
     }
 
 
+def supply_workflow_dashboard_stat(ts: int) -> List[Dict[str, Any]]:
+    sql = f"""
+            SELECT sw.name AS '策略名',
+                   case
+                       when sw.status = 1 then '开启中'
+                       when sw.status = 0 then '已关闭'
+                       else '未知' end AS 状态,
+                   swt.init_cnt        AS '待处理',
+                   swt.running_cnt     AS '运行中',
+                   swt.success_cnt     AS '成功',
+                   swt.fail_success    AS '失败'
+            FROM supply_workflow sw
+                     LEFT JOIN (
+                                   SELECT workflow_id,
+                                          COUNT(DISTINCT IF(task_status = 0, task_id, NULL)) AS init_cnt,
+                                          COUNT(DISTINCT IF(task_status = 1, task_id, NULL)) AS running_cnt,
+                                          COUNT(DISTINCT IF(task_status = 2, task_id, NULL)) AS success_cnt,
+                                          COUNT(DISTINCT IF(task_status = 3, task_id, NULL)) AS fail_success
+                                   FROM supply_workflow_task
+                                   WHERE create_timestamp >= {ts}
+                                   GROUP BY workflow_id
+                               ) swt ON sw.id = swt.workflow_id
+            ORDER BY sw.status DESC, sw.create_timestamp DESC;
+            """
+    return mysql_helper.execute_query(sql)
+
+
 def workflow_task_status_stat(ts: int, workflow_id: str) -> List[Dict[str, Any]]:
     sql = f"""
             select workflow_id             as 'workflow_id',
@@ -357,7 +389,40 @@ def workflow_monitor(workflow_id: str, workflow_name: str, status: int):
     )
 
 
+def workflow_dashboard_monitor():
+    today_midnight = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
+    result = supply_workflow_dashboard_stat(int(today_midnight.timestamp() * 1000))
+
+    start_dt_str = today_midnight.strftime('%Y-%m-%d %H:%M')
+    end_dt_str = datetime.now().strftime('%Y-%m-%d %H:%M')
+
+    sub_title = build_sub_title_json(f"统计时间: {start_dt_str} - {end_dt_str}")
+    header = build_header_json("【供给workflow】当日任务统计", "indigo", sub_title, [])
+    dashboard_stat = build_table_element_json(pd.DataFrame(result))
+    elements = []
+    if dashboard_stat is not None:
+        elements.append(dashboard_stat)
+    else:
+        elements.append(build_markdown_element_json("**今日workflow还未创建新的任务，请关注**"))
+
+    config = build_config_json()
+    card_json = build_card_json(elements, header, config)
+
+    feishu_inform_util.send_card_msg_to_feishu(
+        webhook=fei_shu_webhook,
+        card_json=card_json
+    )
+
+    # 总体情况发送到大群
+    feishu_inform_util.send_card_msg_to_feishu(
+        webhook="https://open.feishu.cn/open-apis/bot/v2/hook/9f5c5cce-5eb2-4731-b368-33926f5549f9",
+        card_json=card_json
+    )
+
+
 def main():
+    workflow_dashboard_monitor()
+
     global header_template_index
     workflows = mysql_helper.execute_query('select id, name, status from supply_workflow;')
     for workflow in workflows: