Python工厂模式实现封装Webhook群聊机器人详解_开发_开发者

引言

企业存在给 特定群组 自动推送消息的需求，比如：监控报警推送、销售线索推送、运营内容推送等。你可以在群聊中添加一个自定义机器人，通过服务端调用 webhook 地址，即可将外部系统的通知消息即时推送到群聊中。

飞书自定义机器人使用指南：

https://open.feishu.cn/document/ukTMukTMukTM/ucTM5YjL3ETO24yNxkjN

钉钉自定义机器人使用指南：

https://open.dingtalk.com/document/robots/custom-robot-Access

飞书自定义机器人

#!/usr/bin/python3
# -*- coding: utf-8 -*-
# @Author: Hui
# @Desc: { webhook机器人模块 }
# @Date: 2023/02/19 19:48
import hMAC
import base64
import hashlib
import time
from urllib.parse import quote_plus

import requests

from exceptions.base import SendMsgException


class BaseChatBot(object):
    """群聊机器人基类"""

    def __init__(self, webhook_url: str, secret: str = None):
        """
        初始化机器人
        Args:
            webhook_url: 机器人webhook地址
            secret: 安全密钥
        """
        self.webhook_url = webhook_url
        self.secret = secret

    def _get_sign(self, timestamp: str, secret: str):
        """
        获取签名(NotImplemented)
        Args:
            timestamp: 签名时使用的时间戳
            secret: 签名时使用的密钥

        Returns:
        """
        raise NotImplementedError

    def send_msg(self, content: str, timeout=10):
        """
        发送消息(NotImplemented)
        Args:
            content: 消息内容
            timeout: 发送消息请求超时时间 默认10秒

        Returns:
        """
        raise NotImplementedError


class FeiShuChatBot(BaseChatBot):
    """飞书机器人"""

    def _get_sign(self, timestamp: str, secret: str) -> sjavascripttr:
        """
        获取签名
        把 timestamp + "\n" + 密钥 当做签名字符串，使用 HmacSHA256 算法计算签名，再进行 Base64 编码
        Args:
            timestamp: 签名时使用的时间戳
            secret: 签名时使用的密钥

        Returns: sign
        """
        string_to_sign = '{}\n{}'.format(timestamp, secret)
        hmac_code = hmac.new(string_to_sign.encode("utf-8"), digestmod=hashlib.sha256).digest()

        # 对结果进行base64处理
        sign = base64.b64encode(hmac_code).decode('utf-8')
        return sign

    def send_msg(self, content: str, timeout=10):
        """
        发送消息
        Args:
            content: 消息内容
            timeout: 发送消息请求超时时间 默认10秒

        Raises:
            SendMsgException

        Returns:
        """
        msg_data = {
            "msg_type": "text",
            "content": {
                "text": f"{content}"
            }
        }
        if self.secret:
            timestamp = str(round(time.time()))
            sign = self._get_sign(timestamp=timestamp, secret=self.secret)
            msg_data["timestamp"] = timestamp
            msg_data["sign"] = sign

        try:
            resp = requests.post(url=self.webhook_url, json=msg_data, timeout=timeout)
            resp_info = resp.json()
            if resp_info.get("codpythone") != 0:
                raise SendMsgException(f"FeiShuChatBot send pythonmsg error, {resp_info}")
        except Exception as e:
            raise SendMsgException(f"FeiShuChatBot send msg error {e}") from e

钉钉自定义机器人

class DingTalkChatBot(BaseChatBot):
    """钉钉机器人"""

    def _get_sign(self, timestamp: str, secret: str):
        """
        获取签名
        把 timestamp + "\n" + 密钥当做签名字符串，使用 HmacSHA256js 算法计算签名，
        然后进行 Base64 encode，最后再把签名参数再进行 urlEncode，得到最终的签名（需要使用UTF-8字符集）
        Args:
            timestamp: 签名时使用的时间戳
            secret: 签名时使用的密钥

        Returns: sign
        """
        secret_enc = secret.encode('utf-8')
        string_to_sign = '{}\n{}'.format(timestamp, secret)
        string_to_sign_enc = string_to_sign.encode('utf-8')
        hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
        sign = quote_plus(base64.b64encode(hmac_code))

        return sign

    def send_msg(self, content: str, timeout=10):
        """
        发送消息
        Args:
            content: 消息内容
            timeout: 发送消息请求超时时间 默认10秒

        Raises:
            SendMsgException

        Returns:
        """
        timestamp = str(round(time.time() * 1000))
        sign = self._get_sign(timestamp=timestamp, secret=self.secret)

        params = {
            "timestamp": timestamp,
            "sign": sign
        }
        msg_data = {
            "msgtype": "text",
            "te开发者_JAVA教程xt": {
                "content": content
            }
        }
        try:
            resp = requests.post(url=self.webhook_url, json=msg_data, params=params, timeout=timeout)
            resp_info = resp.json()
            if resp_info.get("errcode") != 0:
                raise SendMsgException(f"DingTalkChatBot send msg error, {resp_info}")
        except Exception as e:
            raise SendMsgException(f"DingTalkChatBot send msg error {e}") from e

使用的时候

feishu = FeiShuChatBot(webhook_url="xxx", secret="xxxx")
feishu.send_msg("test msg")

dingtalk = DingTalkChatBot(webhook_url="xxx", secret="xxxx")
feishu.send_msg("test msg")

但这样使用有点不好的一点就是如果我突然从钉钉换成飞书或者企微，业务中的所有使用的代码就要全部替换。但一般也不会随便换平台，我这里就是想引出工厂模式。

工厂模式封装

工厂模式是一种常见的设计模式，它可以帮助我们创建对象，而无需显式地指定其具体类型。在这种模式下，我们通过使用一个工厂来创建对象，并将对象的创建和使用分离开来，从而提高了代码的可维护性和可扩展性.

对于Webhook群聊机器人，我们可以将其实现为一个工厂类，该工厂类负责创建不同类型的机器人对象。我们可以通过定义一个机器人接口或抽象类，来规范所有机器人的通用方法和属性。然后，我们可以根据需要创建具体的机器人类，并实现其特定的方法和属性。代码如下

#!/usr/bin/python3
# -*- coding: utf-8 -*-
# @Author: Hui
# @Desc: { 机器人工厂模块 }
# @Date: 2023/02/19 20:03
from typing import Dict, Type
from chatbot import DingTalkChatBot, FeiShuChatBot, BaseChatBot


class ChatBotType:
    """群聊机器人类型"""

    FEISHU_CHATBOT = "feishu"
    DINGTALK_CHATBOT = "dingtalk"


class ChatBotFactory(object):
    """
    消息机器人工厂
    支持 飞书、钉钉、自定义机器人消息发送
    """

    # 群聊机器人处理类映射
    CHATBOT_HANDLER_CLS_MAPPING: Dict[str, Type[BaseChatBot]] = {
        ChatBotType.FEISHU_CHATBOT: FeiShuChatBot,
        ChatBotType.DINGTALK_CHATBOT: DingTalkChatBot,
    }

    def __init__(self, chatbot_type: str):
        if chatbot_type not in self.CHATBOT_HANDLER_CLS_MAPPING:
            raise ValueError(f"不支持 {chatbot_type} 类型的机器人")
        self.chatbot_type = chatbot_type

    def build(self, webhook_url: str, secret: str = None) -> BaseChatBot:
        """
        构造具体的机器人处理类
        Args:
            webhook_url: 机器人webhook地址
            secret: 机器人密钥

        Returns: 根据 robot_type 返回对应的机器人处理类

        """
        chatbot_handle_cls = self.CHATBOT_HANDLER_CLS_MAPPING.get(self.chatbot_type)
        return chatbot_handle_cls(webhook_url=webhook_url, secret=secret)

通过字典的方式把机器人类型（平台）与具体处理类关联起来，这样构造的时候就不用写 if else

使用的时候直接用工厂创建具体的实例出来就行

def main():
    feishu_webhook = "xxx"
    feishu_webhook_secret = "xxx"

    dingtalk_webhook = "xxx"
    dingtalk_webhook_secret = "xxx"

    feishu_chatbot = ChatBotFactory(chatbot_type=ChatBotType.FEISHU_CHATBOT).build(
        webhook_url=feishu_webhook,
        secret=feishu_webhook_secret
    )
    content = "飞书自定义机器人使用指南：\n https://open.feishu.cn/document/ukTMukTMukTM/ucTM5YjL3ETO24yNxkjN"
    feishu_chatbot.send_msg(content)

    dingtalk_chatbot = ChatBotFactory(chatbot_type=ChatBotType.DINGTALK_CHATBOT).build(
        webhook_url=dingtalk_webhook,
        secret=dingtalk_webhook_secret
    )
    content = "钉钉自定义机器人使用指南：\n https://open.dingtalk.com/document/robots/custom-robot-access"
    dingtalk_chatbot.send_msg(content)


if __name__ == '__main__':
    main()

新增企微机器人

需要切换的时候直接替换掉 chatbot_type 就可以。把chatbot_type、webhook_url、secret放到配置文件即可在不改动代码的情况直接切换。工厂模式也方便扩展，例如再新增一个企微等。

class WeComChatbot(BaseChatBot):
    """企业微信机器人"""

    def _get_sign(self, timestamp: str, secret: str):
        """企业微信暂不支持签名加密"""
        pass

    def send_msg(self, content: str, timeout=10):
        """
        发送消息
        Args:
            content: 消息内容
            timeout: 发送消息请求超时时间 默认10秒

        Raises:
            SendMsgException

        Returns:
        """
        msg_data = {
            "msgtype": "text",
            "text": {
                "content": content
            }
        }
        try:
            resp = requests.post(self.webhook_url, json=msg_data)
            resp_info = resp.json()
            if resp.status_code != 200:
                raise ValueError(f"WeComChatbot send message error, {resp_info}")
        except Exception as e:
            raise SendMsgException(e) from e

工厂类中只要新增一个企微群聊机器人处理类的映射就可以

from typing import Dict, Type
from chatbot import DingTalkChatBot, FeiShuChatBot, WeComChatbot, BaseChatBot


class ChatBotType:
    """群聊机器人类型"""

    FEISHU_CHATBOT = "feishu"
    DINGTALK_CHATBOT = "dingtalk"
    WECOM_CHATBOT = "wecom"


class ChatBotFactory(object):
    """
    消息机器人工厂
    支持 飞书、钉钉、企微自定义机器人消息发送
    """

    # 群聊机器人处理类映射
    CHATBOT_HANDLER_CLS_MAPPING: Dict[str, Type[BaseChatBot]] = {
        ChatBotType.FEISHU_CHATBOT: FeiShuChatBot,
        ChatBotType.DINGTALK_CHATBOT: DingTalkChatBot,
        ChatBotType.WECOM_CHATBOT: WeComChatbot
    }

    def __init__(self, chatbot_type: str):
        if chatbot_type not in self.CHATBOT_HANDLER_CLS_MAPPING:
            raise ValueError(f"不支持 {chatbot_type} 类型的机器人")
        self.chatbot_type = chatbot_type

    def build(self, webhopythonok_url: str, secret: str = None) -> BaseChatBot:
        """
        构造具体的机器人处理类
        Args:
            webhook_url: 机器人webhook地址
            secret: 机器人密钥

        Returns: 根据 robot_type 返回对应的机器人处理类

        """
        chatbot_handle_cls = self.CHATBOT_HANDLER_CLS_MAPPING.get(self.chatbot_type)
        return chatbot_handle_cls(webhook_url=webhook_url, secret=secret)

看看读取配置文件的话该如何使用

# settings.py
# chatbot_type = "feishu"
# chatbot_type = "dingtalk"
chatbot_type = "wecom"
webhook_url = "xxx"
secret = ""


# xxx_logic.py
import settings

chatbot = ChatBotFactory(chatbot_type=settings.chatbot_type).build(
     webhook_url=settings.webhook_url,
     secret=settings.secret
)

chatbot.send_msg("test msg")

使用工厂模式封装虽然让代码变的多了一些，但更容易维护与扩展，因此我们还是要结合业务场景选择一些合适的设计模式来编写代码，这样也能提示自己的编码能力。

到此这篇关于Python工厂模式实现封装Webhook群聊机器人详解的文章就介绍到这了,更多相关Python封装群聊机器人内容请搜索我们以前的文章或继续浏览下面的相关文章希望大家以后多多支持我们！