怎么使用Python工廠模式實現封裝Webhook群聊機器人

本文小編為大家詳細介紹“怎么使用Python工廠模式實現封裝Webhook群聊機器人”，內容詳細，步驟清晰，細節處理妥當，希望這篇“怎么使用Python工廠模式實現封裝Webhook群聊機器人”文章能幫助大家解決疑惑，下面跟著小編的思路慢慢深入，一起來學習新知識吧。

飛書自定義機器人

#!/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群聊機器人”文章已經介紹完畢，想要掌握這篇文章的知識點還需要大家自己動手實踐使用過才能領會，如果想了解更多相關內容的文章，歡迎關注億速云行業資訊頻道。