feat:添加supply workflow监控

monitor/supply_workflow_monitor.py

@@ -65,29 +65,29 @@ def df_to_markdown_table(df: pd.DataFrame) -> str:
 
 def task_exe_step_stat(ts: int) -> List[Dict[str, Any]]:
     sql = f'''
-        select step_name AS "步骤名称",
-           case
-               when status = 0 then '初始化'
-               when status = 1 then '运行中'
-               when status = 2 then '成功'
-               when status = 3 then '失败'
-               else '未知'
-               end AS '执行状态',
-           case
-               when error_msg like '%Data too long%' then '数据超过字段长度限制'
-               when error_msg like '%Deadlock%' then '数据库死锁'
-               when error_msg = '' then ''
-               when error_msg is null then ''
-               else error_msg
-               end AS '错误原因',
-           cnt AS '个数'
-    from (
-             select step_name, status, error_msg, count(1) as cnt
-             from supply_workflow_task_exe_step
-             where create_timestamp >= {ts}
-             group by step_name, status, error_msg
-         ) as t
-        '''
+select step_name AS "步骤名称",
+       status    as '执行状态',
+       error_msg as '错误原因',
+       count(1)  AS '个数'
+from (
+         select step_name,
+                case
+                    when status = 0 then '初始化'
+                    when status = 1 then '运行中'
+                    when status = 2 then '成功'
+                    when status = 3 then '失败'
+                    else '未知' end AS status,
+                case
+                    when error_msg like '%Data too long%' then '数据超过字段长度限制'
+                    when error_msg like '%Deadlock%' then '数据库死锁'
+                    when (error_msg = '' or error_msg is null) then ''
+                    else substring_index(error_msg, ',', 1)
+                    end             AS error_msg
+         from supply_workflow_task_exe_step
+         where create_timestamp >= {ts} 
+     ) as t
+group by step_name, status, error_msg
+'''
     return mysql_helper.execute_query(sql)