LongArticlesJob/applications/api/feishu_api.py

import json
import requests


class Feishu:
    # 服务号分组群发监测机器人
    server_account_publish_monitor_bot = "https://open.feishu.cn/open-apis/bot/v2/hook/380fdecf-402e-4426-85b6-7d9dbd2a9f59"

    # 外部服务号投流监测机器人
    outside_gzh_monitor_bot = "https://open.feishu.cn/open-apis/bot/v2/hook/0899d43d-9f65-48ce-a419-f83ac935bf59"

    # 长文 daily 报警机器人
    # long_articles_bot = "https://open.feishu.cn/open-apis/bot/v2/hook/b44333f2-16c0-4cb1-af01-d135f8704410"
    long_articles_bot = "https://open.feishu.cn/open-apis/bot/v2/hook/223b3d72-f2e8-40e0-9b53-6956e0ae7158"

    # 测试环境报警机器人
    long_articles_bot_dev = "https://open.feishu.cn/open-apis/bot/v2/hook/f32c0456-847f-41f3-97db-33fcc1616bcd"

    def __init__(self):
        self.token = None
        self.headers = {"Content-Type": "application/json"}
        self.mention_all = {
            "content": "<at id=all></at>\n",
            "tag": "lark_md",
        }
        self.not_mention = {}

    def fetch_token(self):
        url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal/"
        post_data = {
            "app_id": "cli_a51114cf8bf8d00c",
            "app_secret": "cNoTAqMpsAm7mPBcpCAXFfvOzCNL27fe",
        }
        response = requests.request("POST", url=url, data=post_data)
        tenant_access_token = response.json()["tenant_access_token"]
        self.token = tenant_access_token


class FeishuSheetApi(Feishu):

    def prepend_value(self, sheet_token, sheet_id, ranges, values):
        insert_value_url = "https://open.feishu.cn/open-apis/sheets/v2/spreadsheets/{}/values_prepend".format(
            sheet_token
        )
        headers = {
            "Authorization": "Bearer " + self.token,
            "contentType": "application/json; charset=utf-8",
        }
        body = {
            "valueRange": {"range": "{}!{}".format(sheet_id, ranges), "values": values}
        }
        response = requests.request(
            "POST", url=insert_value_url, headers=headers, json=body
        )
        print(response.json())

    def insert_value(self, sheet_token, sheet_id, ranges, values):
        insert_value_url = (
            "https://open.feishu.cn/open-apis/sheets/v2/spreadsheets/{}/values".format(
                sheet_token
            )
        )
        headers = {
            "Authorization": "Bearer " + self.token,
            "contentType": "application/json; charset=utf-8",
        }
        body = {
            "valueRange": {"range": "{}!{}".format(sheet_id, ranges), "values": values}
        }
        response = requests.request(
            "PUT", url=insert_value_url, headers=headers, json=body
        )
        print(response.json())


class FeishuBotApi(Feishu):

    @classmethod
    def create_feishu_columns_sheet(
        cls,
        sheet_type,
        sheet_name,
        display_name,
        width="auto",
        vertical_align="top",
        horizontal_align="left",
        number_format=None,
    ):
        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",
                }

            case _:
                return {
                    "name": sheet_name,
                    "display_name": display_name,
                    "width": width,
                    "data_type": "text",
                    "vertical_align": vertical_align,
                    "horizontal_align": horizontal_align,
                }

    # 表格形式
    def create_feishu_table(self, title, columns, rows, mention):
        table_base = {
            "header": {
                "template": "blue",
                "title": {"content": title, "tag": "plain_text"},
            },
            "elements": [
                self.mention_all if mention else self.not_mention,
                {
                    "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

    def create_feishu_bot_obj(self, title, mention, detail):
        """
        create feishu bot object
        """
        return {
            "elements": [
                {
                    "tag": "div",
                    "text": self.mention_all if mention else self.not_mention,
                },
                {
                    "tag": "div",
                    "text": {
                        "content": json.dumps(detail, ensure_ascii=False, indent=4),
                        "tag": "lark_md",
                    },
                },
            ],
            "header": {"title": {"content": title, "tag": "plain_text"}},
        }

    # bot
    def bot(self, title, detail, mention=True, table=False, env="prod"):
        match env:
            case "dev":
                url = self.long_articles_bot_dev
            case "prod":
                url = self.long_articles_bot
            case "outside_gzh_monitor":
                url = self.outside_gzh_monitor_bot
            case "server_account_publish_monitor":
                url = self.server_account_publish_monitor_bot
            case _:
                url = self.long_articles_bot_dev

        headers = {"Content-Type": "application/json"}
        if table:
            card = self.create_feishu_table(
                title=title,
                columns=detail["columns"],
                rows=detail["rows"],
                mention=mention,
            )
        else:
            card = self.create_feishu_bot_obj(
                title=title, mention=mention, detail=detail
            )

        payload = {"msg_type": "interactive", "card": card}
        res = requests.request(
            "POST", url=url, headers=headers, data=json.dumps(payload), timeout=10
        )
        return res