LongArticleTaskServer/app/infra/external/feishu.py

import json

from app.infra.shared import AsyncHttpClient


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_dev = "https://open.feishu.cn/open-apis/bot/v2/hook/f32c0456-847f-41f3-97db-33fcc1616bcd"

    # 长文任务报警群
    long_articles_task_bot = "https://open.feishu.cn/open-apis/bot/v2/hook/223b3d72-f2e8-40e0-9b53-6956e0ae7158"

    # cookie 监测机器人
    cookie_monitor_bot = "https://open.feishu.cn/open-apis/bot/v2/hook/51b9c83a-f50d-44dd-939f-bcd10ac6c55a"

    # rank_bot
    rank_monitor_bot = "https://open.feishu.cn/open-apis/bot/v2/hook/f9dae7ba-decf-436b-b438-41994c35af1e"

    # 用户名与手机号映射
    name_phone_dict = {
        "卓异": "18624010360",
        "俊辉": "18801281360",
        "范军": "15200827642",
        "林强": "15810236123",
    }

    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 = {}

    async 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",
        }
        async with AsyncHttpClient(default_headers=self.headers) as client:
            response = await client.post(url=url, json=post_data)

        tenant_access_token = response["tenant_access_token"]
        self.token = tenant_access_token

    async def get_user_open_id(self, username):
        """
        根据用户名获取飞书 open_id

        :param username: 用户名
        :return: open_id
        """
        if not self.token:
            await self.fetch_token()

        mobile = self.name_phone_dict.get(username)
        if not mobile:
            return None

        url = "https://open.feishu.cn/open-apis/user/v1/batch_get_id"
        headers = {
            "Authorization": f"Bearer {self.token}",
            "Content-Type": "application/json; charset=utf-8",
        }
        params = {"mobiles": [mobile]}

        async with AsyncHttpClient() as client:
            response = await client.get(url=url, params=params, headers=headers)

        try:
            open_id = response["data"]["mobile_users"][mobile][0]["open_id"]
            return open_id
        except (KeyError, IndexError):
            return None


class FeishuSheetApi(Feishu):
    async def read_sheet_range(
        self, spreadsheet_token: str, range_str: str
    ) -> list[list]:
        """读取电子表格指定范围的数据。

        Args:
            spreadsheet_token: 电子表格 token
            range_str: 查询范围，格式为 "<sheet_id>!<开始位置>:<结束位置>"，如 "4d54bd!A1:K358"

        Returns:
            数据二维数组。失败或无数据时返回空列表。
        """
        if not self.token:
            await self.fetch_token()

        url = f"https://open.feishu.cn/open-apis/sheets/v2/spreadsheets/{spreadsheet_token}/values/{range_str}"
        print(self.token)
        headers = {
            "Authorization": f"Bearer {self.token}",
            "Content-Type": "application/json; charset=utf-8",
        }
        async with AsyncHttpClient() as client:
            response = await client.get(url=url, headers=headers)

        if response.get("code", 0) != 0:
            print(f"飞书读取表格失败: {response.get('msg', 'unknown error')}")
            return []

        data = response.get("data", {})
        value_range = data.get("valueRange", {})
        return value_range.get("values", [])
    
    async 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}
        }
        async with AsyncHttpClient() as client:
            response = await client.post(
                url=insert_value_url, json=body, headers=headers
            )

        print(response)

    async 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
            )
        )
        # self.token = 't-g104bpfHNZN45BVJWFSQEM6WD45AAI4FNRWXCZVK'
        headers = {
            "Authorization": "Bearer " + self.token,
            "contentType": "application/json; charset=utf-8",
        }
        body = {
            "valueRange": {"range": "{}!{}".format(sheet_id, ranges), "values": values}
        }
        async with AsyncHttpClient() as client:
            response = await client.put(
                url=insert_value_url, json=body, headers=headers
            )


class FeishuBotApi(Feishu):
    async def create_mention_text(self, usernames):
        """
        创建 @ 用户的文本

        :param usernames: 用户名列表
        :return: @ 用户的 markdown 文本
        """
        if not usernames:
            return ""

        mention_parts = []
        for username in usernames:
            open_id = await self.get_user_open_id(username)
            if open_id:
                mention_parts.append(f"<at id={open_id}></at>")

        return " ".join(mention_parts) + "\n" if mention_parts else ""

    @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,
                }

    # 表格形式
    async def create_feishu_table(
        self, title, columns, rows, mention, mention_users=None
    ):
        """
        创建飞书表格消息

        :param title: 标题
        :param columns: 列定义
        :param rows: 行数据
        :param mention: 是否 @ 所有人
        :param mention_users: 要 @ 的具体用户列表，例如 ["mark"]
        """
        mention_element = self.not_mention
        if mention:
            mention_element = self.mention_all
        elif mention_users:
            mention_text = await self.create_mention_text(mention_users)
            mention_element = {"content": mention_text, "tag": "lark_md"}

        table_base = {
            "header": {
                "template": "blue",
                "title": {"content": title, "tag": "plain_text"},
            },
            "elements": [
                mention_element,
                {
                    "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

    async def create_feishu_bot_obj(self, title, mention, detail, mention_users=None):
        """
        create feishu bot object

        :param title: 标题
        :param mention: 是否 @ 所有人
        :param detail: 详细内容
        :param mention_users: 要 @ 的具体用户列表，例如 ["luojunhui"]
        """
        mention_element = self.not_mention
        if mention:
            mention_element = self.mention_all
        elif mention_users:
            mention_text = await self.create_mention_text(mention_users)
            mention_element = {"content": mention_text, "tag": "lark_md"}

        return {
            "elements": [
                {
                    "tag": "div",
                    "text": mention_element,
                },
                {
                    "tag": "div",
                    "text": {
                        "content": json.dumps(detail, ensure_ascii=False, indent=4),
                        "tag": "lark_md",
                    },
                },
            ],
            "header": {"title": {"content": title, "tag": "plain_text"}},
        }

    # bot
    async def bot(
        self,
        title,
        detail,
        mention=True,
        table=False,
        env="long_articles_task",
        mention_users=None,
    ):
        """
        发送飞书机器人消息

        :param title: 标题
        :param detail: 详细内容
        :param mention: 是否 @ 所有人
        :param table: 是否为表格形式
        :param env: 环境，决定发送到哪个机器人
        :param mention_users: 要 @ 的具体用户列表，例如 ["luojunhui"]
        """
        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 "long_articles_task":
                url = self.long_articles_task_bot
            case "cookie_monitor":
                url = self.cookie_monitor_bot
            case "rank_bot":
                url = self.rank_monitor_bot
            case _:
                url = self.long_articles_bot_dev

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

        data = {"msg_type": "interactive", "card": card}
        async with AsyncHttpClient() as client:
            res = await client.post(url=url, headers=headers, json=data)
        return res