LongArticleTaskServer/app/infra/external/feishu.py

import json

from app.core.config.settings.feishu import FeishuConfig
from app.infra.shared import AsyncHttpClient


class Feishu:

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

    def __init__(self, config: FeishuConfig = None):
        self.config = config or FeishuConfig()
        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 _get_webhook_url(self, env: str) -> str:
        """根据环境获取 webhook URL"""
        webhook_map = {
            "dev": self.config.long_articles_bot_dev,
            "prod": self.config.long_articles_bot,
            "outside_gzh_monitor": self.config.outside_gzh_monitor_bot,
            "server_account_publish_monitor": self.config.server_account_publish_monitor_bot,
            "long_articles_task": self.config.long_articles_task_bot,
            "cookie_monitor": self.config.cookie_monitor_bot,
            "rank_bot": self.config.rank_monitor_bot,
        }
        return webhook_map.get(env, self.config.long_articles_bot_dev)

    async def fetch_token(self):
        url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal/"
        post_data = {
            "app_id": self.config.app_id,
            "app_secret": self.config.app_secret,
        }
        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 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
            )

        return 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
            )
        )
        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"]
        """
        url = self._get_webhook_url(env)

        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