文章详情

短信预约-IT技能 免费直播动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

怎么使用Python工厂模式实现封装Webhook群聊机器人

2023-07-05 04:46

关注

本文小编为大家详细介绍“怎么使用Python工厂模式实现封装Webhook群聊机器人”,内容详细,步骤清晰,细节处理妥当,希望这篇“怎么使用Python工厂模式实现封装Webhook群聊机器人”文章能帮助大家解决疑惑,下面跟着小编的思路慢慢深入,一起来学习新知识吧。

飞书自定义机器人

#!/usr/bin/python3# -*- coding: utf-8 -*-# @Author: Hui# @Desc: { webhook机器人模块 }# @Date: 2023/02/19 19:48import hmacimport base64import hashlibimport timefrom urllib.parse import quote_plusimport requestsfrom exceptions.base import SendMsgExceptionclass 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 NotImplementedErrorclass FeiShuChatBot(BaseChatBot):    """飞书机器人"""    def _get_sign(self, timestamp: str, secret: str) -> str:        """        获取签名        把 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("code") != 0:                raise SendMsgException(f"FeiShuChatBot send msg 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" + 密钥当做签名字符串,使用 HmacSHA256 算法计算签名,        然后进行 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",            "text": {                "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:03from typing import Dict, Typefrom chatbot import DingTalkChatBot, FeiShuChatBot, BaseChatBotclass 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, Typefrom chatbot import DingTalkChatBot, FeiShuChatBot, WeComChatbot, BaseChatBotclass 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, 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)

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

# settings.py# chatbot_type = "feishu"# chatbot_type = "dingtalk"chatbot_type = "wecom"webhook_url = "xxx"secret = ""# xxx_logic.pyimport settingschatbot = ChatBotFactory(chatbot_type=settings.chatbot_type).build(     webhook_url=settings.webhook_url,     secret=settings.secret)chatbot.send_msg("test msg")

读到这里,这篇“怎么使用Python工厂模式实现封装Webhook群聊机器人”文章已经介绍完毕,想要掌握这篇文章的知识点还需要大家自己动手实践使用过才能领会,如果想了解更多相关内容的文章,欢迎关注编程网行业资讯频道。

阅读原文内容投诉

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     813人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     435人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-后端开发
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯