首页 > 资讯 > 后端开发 > Python >Python工厂模式实现封装Webhook群聊机器人详解

874

分享到

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

Python 工厂模式封装群聊机器人 Python封装群聊机器人 Python 机器人 Python工厂模式 2023-02-21 12:02:11 874人浏览八月长安

Python 官方文档：入门教程 => 点击学习

摘要

目录引言飞书自定义机器人钉钉自定义机器人工厂模式封装新增企微机器人引言企业存在给特定群组自动推送消息的需求，比如：监控报警推送、销售线索推送、运营内容推送等。你可以在群聊中添

引言

企业存在给 特定群组 自动推送消息的需求，比如：监控报警推送、销售线索推送、运营内容推送等。你可以在群聊中添加一个自定义机器人，通过服务端调用 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) -> 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: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, 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.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封装群聊机器人内容请搜索编程网以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程网！

您可能感兴趣的文档:

--结束END--

本文标题: Python工厂模式实现封装Webhook群聊机器人详解

本文链接: https://www.lsjlt.com/news/196869.html(转载时请注明来源链接)

有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

回答

如何调试操作系统的错误？
操作系统

2023-11-15发布

回答

操作系统中的I/O系统是如何实现的？
操作系统

2023-11-15发布

回答

如何实现操作系统的内存管理？
操作系统

2023-11-15发布

回答

什么是虚拟内存，它对操作系统有什么影响？
操作系统

2023-11-15发布

回答

ASP中的MVC架构和WebForms架构有什么区别和使用场景？
ASP.NET

2023-11-15发布

回答

ASP中的数据验证和数据校验有什么不同？
ASP.NET

2023-11-15发布

回答

ASP中的ADO对象和DAO对象有什么区别和使用方法？
ASP.NET

2023-11-15发布

回答

Node.js中的包管理器NPM是什么？如何使用它进行依赖管理？
node.js

2023-11-15发布

回答

Vue.js中的动态组件是什么？如何使用它来动态渲染组件？
VUE

2023-11-15发布

回答

如何使用Vue.js实现懒加载和预加载？
VUE

2023-11-15发布

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

目录

引言

飞书自定义机器人

钉钉自定义机器人

工厂模式封装

新增企微机器人

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

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

python实现AI聊天机器人详解流程

Python人工智能构建简单聊天机器人示例详解

kotlin实现语音聊天机器人案例详解

Python基于Google Bard实现交互式聊天机器人

详解用Go语言实现工厂模式(Golang经典编程案例)

Python基于Google Bard怎么实现交互式聊天机器人

Python实现线程池工作模式的案例详解

使用Python编写并实现一个具备人工智能的聊天机器人（包含代码和步骤）

Python实现企业微信通知机器人的方法详解

Python使用机器学习模型实现温度预测详解

Python虚拟机字节码教程之装饰器实现详解

在微信小程序中如何支持使用流模式（stream），打造ChatGPT实时回复机器人，最详细讲解。

python分析数据的方法是什么

如何使用Python实现抽奖小程序

python copy函数的作用是什么

python ffmpeg模块怎么安装和使用

python进程池创建队列的方法是什么

python无法运行文件的原因有哪些

python can't open file报错怎么解决

python keyerror错误怎么解决

python字符串处理与应用的方法有哪些

python全局变量如何定义