LongArticlesJob/applications/feishuBotApi.py

"""
@author: luojunhui
"""
import json
import requests

from applications.decoratorApi import retryOnTimeout

mention_all = {
    "content": "<at id=all></at>\n",
    "tag": "lark_md",
}


def create_feishu_columns_sheet(sheet_type, sheet_name, display_name, width="auto",
                                vertical_align="top", horizontal_align="left", number_format=None):
    """
    create feishu sheet
    :param number_format:
    :param horizontal_align:
    :param vertical_align:
    :param width:
    :param display_name:
    :param sheet_type:
    :param sheet_name:
    """
    match sheet_type:
        case "plain_text":
            return {
                "name": sheet_name,
                "display_name": display_name,
                "width": width,
                "data_type": "text",
                "vertical_align": vertical_align,
                "horizontal_align": horizontal_align
            }

        case "lark_md":
            return {
                "name": sheet_name,
                "display_name": display_name,
                "data_type": "lark_md"
            }

        case "number":
            return {
                "name": sheet_name,
                "display_name": display_name,
                "data_type": "number",
                "format": number_format,
                "width": width
            }

        case "date":
            return {
                "name": sheet_name,
                "display_name": display_name,
                "data_type": "date",
                "date_format": "YYYY/MM/DD"
            }
        case "options":
            return {
                "name": sheet_name,
                "display_name": display_name,
                "data_type": "options"
            }


def create_feishu_table(title, columns, rows, mention):
    """
    create feishu table
    :param mention:
    :param title:
    :param columns:
    :param rows:
    """
    table_base = {
        "header": {
            "template": "blue",
            "title": {
                "content": title,
                "tag": "plain_text"
            }
        },
        "elements": [
            {
                "tag": "div",
                "text": mention_all,
            } if mention else {},
            {
                "tag": "table",
                "page_size": len(rows) + 1,
                "row_height": "low",
                "header_style": {
                    "text_align": "left",
                    "text_size": "normal",
                    "background_style": "grey",
                    "text_color": "default",
                    "bold": True,
                    "lines": 1
                },
                "columns": columns,
                "rows": rows
            }
        ]
    }
    return table_base


@retryOnTimeout()
def bot(title, detail, mention=True, table=False):
    """
    报警机器人
    """
    url = "https://open.feishu.cn/open-apis/bot/v2/hook/b44333f2-16c0-4cb1-af01-d135f8704410"
    # url = "https://open.feishu.cn/open-apis/bot/v2/hook/f32c0456-847f-41f3-97db-33fcc1616bcd"
    headers = {"Content-Type": "application/json"}
    if table:
        card = create_feishu_table(
            title=title,
            columns=detail["columns"],
            rows=detail["rows"],
            mention=mention
        )
    else:
        card = {
            "elements": [
                {
                    "tag": "div",
                    "text": mention_all if mention else {},
                },
                {
                    "tag": "div",
                    "text": {
                        "content": json.dumps(
                            detail, ensure_ascii=False, indent=4
                        ),
                        "tag": "lark_md",
                    },
                },
            ],
            "header": {"title": {"content": title, "tag": "plain_text"}},
        }
    payload = {
        "msg_type": "interactive",
        "card": card
    }
    res = requests.request("POST", url=url, headers=headers, data=json.dumps(payload), timeout=10)
    return res