Python工廠模式實(shí)現(xiàn)封裝Webhook群聊機(jī)器人詳解

更新時(shí)間：2023年02月21日 09:07:24 作者：憶想不到的暉

企業(yè)存在給特定群組自動(dòng)推送消息的需求，你可以在群聊中添加一個(gè)自定義機(jī)器人，通過服務(wù)端調(diào)用 webhook 地址，即可將外部系統(tǒng)的通知消息即時(shí)推送到群聊中。本文就來(lái)和大家聊聊具體實(shí)現(xiàn)方法

引言

企業(yè)存在給 特定群組 自動(dòng)推送消息的需求，比如：監(jiān)控報(bào)警推送、銷售線索推送、運(yùn)營(yíng)內(nèi)容推送等。你可以在群聊中添加一個(gè)自定義機(jī)器人，通過服務(wù)端調(diào)用 webhook 地址，即可將外部系統(tǒng)的通知消息即時(shí)推送到群聊中。

飛書自定義機(jī)器人使用指南：

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

釘釘自定義機(jī)器人使用指南：

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

飛書自定義機(jī)器人

#!/usr/bin/python3
# -*- coding: utf-8 -*-
# @Author: Hui
# @Desc: { webhook機(jī)器人模塊 }
# @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):
    """群聊機(jī)器人基類"""

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

    def _get_sign(self, timestamp: str, secret: str):
        """
        獲取簽名(NotImplemented)
        Args:
            timestamp: 簽名時(shí)使用的時(shí)間戳
            secret: 簽名時(shí)使用的密鑰

        Returns:
        """
        raise NotImplementedError

    def send_msg(self, content: str, timeout=10):
        """
        發(fā)送消息(NotImplemented)
        Args:
            content: 消息內(nèi)容
            timeout: 發(fā)送消息請(qǐng)求超時(shí)時(shí)間 默認(rèn)10秒

        Returns:
        """
        raise NotImplementedError


class FeiShuChatBot(BaseChatBot):
    """飛書機(jī)器人"""

    def _get_sign(self, timestamp: str, secret: str) -> str:
        """
        獲取簽名
        把 timestamp + "\n" + 密鑰 當(dāng)做簽名字符串，使用 HmacSHA256 算法計(jì)算簽名，再進(jìn)行 Base64 編碼
        Args:
            timestamp: 簽名時(shí)使用的時(shí)間戳
            secret: 簽名時(shí)使用的密鑰

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

        # 對(duì)結(jié)果進(jìn)行base64處理
        sign = base64.b64encode(hmac_code).decode('utf-8')
        return sign

    def send_msg(self, content: str, timeout=10):
        """
        發(fā)送消息
        Args:
            content: 消息內(nèi)容
            timeout: 發(fā)送消息請(qǐng)求超時(shí)時(shí)間 默認(rèn)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

釘釘自定義機(jī)器人

class DingTalkChatBot(BaseChatBot):
    """釘釘機(jī)器人"""

    def _get_sign(self, timestamp: str, secret: str):
        """
        獲取簽名
        把 timestamp + "\n" + 密鑰當(dāng)做簽名字符串，使用 HmacSHA256 算法計(jì)算簽名，
        然后進(jìn)行 Base64 encode，最后再把簽名參數(shù)再進(jìn)行 urlEncode，得到最終的簽名（需要使用UTF-8字符集）
        Args:
            timestamp: 簽名時(shí)使用的時(shí)間戳
            secret: 簽名時(shí)使用的密鑰

        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):
        """
        發(fā)送消息
        Args:
            content: 消息內(nèi)容
            timeout: 發(fā)送消息請(qǐng)求超時(shí)時(shí)間 默認(rèn)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

使用的時(shí)候

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

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

但這樣使用有點(diǎn)不好的一點(diǎn)就是如果我突然從釘釘換成飛書或者企微，業(yè)務(wù)中的所有使用的代碼就要全部替換。但一般也不會(huì)隨便換平臺(tái)，我這里就是想引出工廠模式。

工廠模式封裝

工廠模式是一種常見的設(shè)計(jì)模式，它可以幫助我們創(chuàng)建對(duì)象，而無(wú)需顯式地指定其具體類型。在這種模式下，我們通過使用一個(gè)工廠來(lái)創(chuàng)建對(duì)象，并將對(duì)象的創(chuàng)建和使用分離開來(lái)，從而提高了代碼的可維護(hù)性和可擴(kuò)展性.

對(duì)于Webhook群聊機(jī)器人，我們可以將其實(shí)現(xiàn)為一個(gè)工廠類，該工廠類負(fù)責(zé)創(chuàng)建不同類型的機(jī)器人對(duì)象。我們可以通過定義一個(gè)機(jī)器人接口或抽象類，來(lái)規(guī)范所有機(jī)器人的通用方法和屬性。然后，我們可以根據(jù)需要?jiǎng)?chuàng)建具體的機(jī)器人類，并實(shí)現(xiàn)其特定的方法和屬性。代碼如下

#!/usr/bin/python3
# -*- coding: utf-8 -*-
# @Author: Hui
# @Desc: { 機(jī)器人工廠模塊 }
# @Date: 2023/02/19 20:03
from typing import Dict, Type
from chatbot import DingTalkChatBot, FeiShuChatBot, BaseChatBot


class ChatBotType:
    """群聊機(jī)器人類型"""

    FEISHU_CHATBOT = "feishu"
    DINGTALK_CHATBOT = "dingtalk"


class ChatBotFactory(object):
    """
    消息機(jī)器人工廠
    支持 飛書、釘釘、自定義機(jī)器人消息發(fā)送
    """

    # 群聊機(jī)器人處理類映射
    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} 類型的機(jī)器人")
        self.chatbot_type = chatbot_type

    def build(self, webhook_url: str, secret: str = None) -> BaseChatBot:
        """
        構(gòu)造具體的機(jī)器人處理類
        Args:
            webhook_url: 機(jī)器人webhook地址
            secret: 機(jī)器人密鑰

        Returns: 根據(jù) robot_type 返回對(duì)應(yīng)的機(jī)器人處理類

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

通過字典的方式把機(jī)器人類型（平臺(tái)）與具體處理類關(guān)聯(lián)起來(lái)，這樣構(gòu)造的時(shí)候就不用寫 if else

使用的時(shí)候直接用工廠創(chuàng)建具體的實(shí)例出來(lái)就行

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 = "飛書自定義機(jī)器人使用指南：\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 = "釘釘自定義機(jī)器人使用指南：\n https://open.dingtalk.com/document/robots/custom-robot-access"
    dingtalk_chatbot.send_msg(content)


if __name__ == '__main__':
    main()

新增企微機(jī)器人

需要切換的時(shí)候直接替換掉 chatbot_type 就可以。把chatbot_type、webhook_url、secret放到配置文件即可在不改動(dòng)代碼的情況直接切換。工廠模式也方便擴(kuò)展，例如再新增一個(gè)企微等。

class WeComChatbot(BaseChatBot):
    """企業(yè)微信機(jī)器人"""

    def _get_sign(self, timestamp: str, secret: str):
        """企業(yè)微信暫不支持簽名加密"""
        pass

    def send_msg(self, content: str, timeout=10):
        """
        發(fā)送消息
        Args:
            content: 消息內(nèi)容
            timeout: 發(fā)送消息請(qǐng)求超時(shí)時(shí)間 默認(rèn)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

工廠類中只要新增一個(gè)企微群聊機(jī)器人處理類的映射就可以

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


class ChatBotType:
    """群聊機(jī)器人類型"""

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


class ChatBotFactory(object):
    """
    消息機(jī)器人工廠
    支持 飛書、釘釘、企微自定義機(jī)器人消息發(fā)送
    """

    # 群聊機(jī)器人處理類映射
    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} 類型的機(jī)器人")
        self.chatbot_type = chatbot_type

    def build(self, webhook_url: str, secret: str = None) -> BaseChatBot:
        """
        構(gòu)造具體的機(jī)器人處理類
        Args:
            webhook_url: 機(jī)器人webhook地址
            secret: 機(jī)器人密鑰

        Returns: 根據(jù) robot_type 返回對(duì)應(yīng)的機(jī)器人處理類

        """
        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")

使用工廠模式封裝雖然讓代碼變的多了一些，但更容易維護(hù)與擴(kuò)展，因此我們還是要結(jié)合業(yè)務(wù)場(chǎng)景選擇一些合適的設(shè)計(jì)模式來(lái)編寫代碼，這樣也能提示自己的編碼能力。

到此這篇關(guān)于Python工廠模式實(shí)現(xiàn)封裝Webhook群聊機(jī)器人詳解的文章就介紹到這了,更多相關(guān)Python封裝群聊機(jī)器人內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: