脚本之家 服务器常用软件
快捷导航
您的位置：首页 → 脚本专栏 → python → Python封装群聊机器人

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

 更新时间：2023年02月21日 09:07:24   作者：忆想不到的晖  
企业存在给 特定群组 自动推送消息的需求，你可以在群聊中添加一个自定义机器人，通过服务端调用 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) -> 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封装群聊机器人内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

相关文章

  • Python如何通过内存管理提升程序执行效率

    Python如何通过内存管理提升程序执行效率

    Python提供了自动内存管理的功能，但是如果不小心使用，可能会导致内存泄漏和性能问题，所以巧妙使用内存管理是提高Python执行效率的关键，下面就来和大家仔细讲讲Python的内存管理技巧吧
    2023-06-06
  • Python采集天天基金数据掌握最新基金动向

    Python采集天天基金数据掌握最新基金动向

    这篇文章主要介绍了Python采集天天基金数据掌握最新基金动向,本次案例实现流程为发送请求、获取数据、解析数据、多页爬取、保存数据，接下来来看看具体的操作过程吧
    2022-01-01
  • pytorch中[..., 0]的用法说明

    pytorch中[..., 0]的用法说明

    这篇文章主要介绍了pytorch中[..., 0]的用法说明，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
    2021-05-05
  • Python Pyperclip模块安装和使用详解

    Python Pyperclip模块安装和使用详解

    Pyperclip模块兼容python2和python3，能跨平台使用，这篇文章主要介绍了Pyperclip模块安装和使用详解,需要的朋友可以参考下
    2023-03-03
  • 全网最简约的Anaconda+Python3.7安装教程Win10

    全网最简约的Anaconda+Python3.7安装教程Win10

    这篇文章主要介绍了全网最简约的Anaconda+Python3.7安装教程Win10,图文讲解全流程安装方法,还不会的小伙伴快来看看吧
    2023-03-03
  • python flask sqlalchemy连接数据库流程介绍

    python flask sqlalchemy连接数据库流程介绍

    这篇文章主要介绍了python flask sqlalchemy连接数据库流程，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
    2022-09-09
  • 解决同一目录下python import报错问题

    解决同一目录下python import报错问题

    这篇文章主要介绍了解决同一目录下python import报错问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教
    2023-12-12
  • django中F与Q查询的使用

    django中F与Q查询的使用

    一般查询都是单条件查询，F和Q是组合条件查询，本文主要介绍了django中F与Q查询的使用，感兴趣的可以了解一下
    2021-06-06
  • 对tensorflow 中tile函数的使用详解

    对tensorflow 中tile函数的使用详解

    今天小编就为大家分享一篇对tensorflow 中tile函数的使用详解，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
    2020-02-02
  • Python用来做Web开发的优势有哪些

    Python用来做Web开发的优势有哪些

    这篇文章主要介绍了Python用来做Web开发的优势有哪些，文中讲解非常细致，帮助大家更好的理解和学习Python，感兴趣的朋友可以了解下
    2020-08-08

最新评论