model_script/monitor/supply_workflow_monitor.py

import unicodedata
from datetime import datetime
from typing import Dict, List, Any

import pandas as pd

from helper.MySQLHelper import MySQLHelper
from util import feishu_inform_util

fei_shu_webhook = "https://open.feishu.cn/open-apis/bot/v2/hook/c09712a8-22cd-4bfa-93a5-30ae7b1db11b"

mysql_helper = MySQLHelper(
    host="rm-t4na9qj85v7790tf84o.mysql.singapore.rds.aliyuncs.com",
    username="readonly",
    password="HdkZ4TDmeK6SQ3BRtJBk",
    database="aigc-admin-prod"
)


def _display_width(s: str) -> int:
    """计算字符串在终端中的显示宽度，CJK字符占2宽度，其余占1"""
    w = 0
    for ch in s:
        w += 2 if unicodedata.east_asian_width(ch) in ('F', 'W', 'A') else 1
    return w


def _pad_to_width(s: str, target_width: int) -> str:
    """将字符串右侧填充空格至指定的显示宽度"""
    return s + ' ' * (target_width - _display_width(s))


def print_df_table(df: pd.DataFrame, fmt: str = "grid") -> str:
    """将DataFrame转换为对齐的二维表格字符串，自适应中英文混排宽度

    Args:
        df: pandas DataFrame
        fmt: 输出格式，'grid' 为终端对齐表格，'markdown' 为 Markdown 表格（适合飞书渲染）
    """
    headers = list(df.columns)
    str_rows = df.astype(str).values

    if fmt == "markdown":
        char_widths = []
        for i, h in enumerate(headers):
            max_w = len(str(h))
            for row in str_rows:
                max_w = max(max_w, len(str(row[i])))
            char_widths.append(max_w)

        def _md_cell(v, w):
            return str(v).ljust(w)

        header_cells = [_md_cell(str(h), char_widths[i]) for i, h in enumerate(headers)]
        sep_cells = ['-' * w for w in char_widths]
        lines = ['| ' + ' | '.join(header_cells) + ' |',
                 '| ' + ' | '.join(sep_cells) + ' |']
        for row in str_rows:
            cells = [_md_cell(str(row[i]), char_widths[i]) for i in range(len(headers))]
            lines.append('| ' + ' | '.join(cells) + ' |')
        return '\n'.join(lines)

    col_widths = []
    for i, h in enumerate(headers):
        max_w = _display_width(str(h))
        for row in str_rows:
            max_w = max(max_w, _display_width(str(row[i])))
        col_widths.append(max_w)

    sep = '+' + '+'.join('-' * (w + 2) for w in col_widths) + '+'

    def _row(values):
        cells = [_pad_to_width(str(v), col_widths[i]) for i, v in enumerate(values)]
        return '| ' + ' | '.join(cells) + ' |'

    lines = [sep, _row(headers), sep]
    for row in str_rows:
        lines.append(_row(row))
    lines.append(sep)
    return '\n'.join(lines)


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 ''
               else '其他错误'
               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
        '''
    return mysql_helper.execute_query(sql)


def main():
    today_midnight = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
    timestamp_ms = int(today_midnight.timestamp() * 1000)
    stat = task_exe_step_stat(timestamp_ms)
    df = pd.DataFrame(stat)

    print("当日任务步骤执行统计")

    msg = print_df_table(df, fmt="grid")

    feishu_inform_util.send_card_msg_to_feishu(
        webhook=fei_shu_webhook,
        card_json=feishu_inform_util.build_card_json(msg, "当日任务步骤执行统计")
    )


if __name__ == '__main__':
    main()