Python 工厂模式封装 Webhook 群聊机器人

3. 具体平台实现

飞书机器人

飞书机器人需要在请求头中包含时间戳和签名。

# chatbot/feishu.py
from .base import BaseChatBot
from ..exceptions import SendMsgException

class FeiShuChatBot(BaseChatBot):
    """飞书机器人"""

    def _get_sign(self, timestamp: str, secret: str) -> str:
        string_to_sign = '{}\n{}'.format(timestamp, secret)
        hmac_code = hmac.new(string_to_sign.encode("utf-8"), digestmod=hashlib.sha256).digest()
        sign = base64.b64encode(hmac_code).decode('utf-8')
        return sign

    def send_msg(self, content: str, timeout=10):
        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

钉钉机器人

钉钉机器人的签名计算略有不同，需要对签名结果进行 URL 编码。

# chatbot/dingtalk.py
from .base import BaseChatBot
from ..exceptions import SendMsgException

class DingTalkChatBot(BaseChatBot):
    """钉钉机器人"""

    def _get_sign(self, timestamp: str, secret: str):
        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):
        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

企业微信机器人

企业微信机器人目前暂不支持签名加密，直接 POST 请求即可。

# chatbot/wecom.py
from .base import BaseChatBot
from ..exceptions import SendMsgException

class WeComChatbot(BaseChatBot):
    """企业微信机器人"""

    def _get_sign(self, timestamp: str, secret: str):
        # 企业微信暂不支持签名加密
        pass

    def send_msg(self, content: str, timeout=10):
        msg_data = {
            "msgtype": "text",
            "text": {"content": content}
        }
        try:
            resp = requests.post(self.webhook_url, json=msg_data, timeout=timeout)
            resp_info = resp.json()
            if resp.status_code != 200 or resp_info.get("errcode") != 0:
                raise ValueError(f"WeComChatbot send message error, {resp_info}")
        except Exception as e:
            raise SendMsgException(e) from e

4. 工厂类实现

工厂类负责管理不同类型机器人的映射关系，并负责实例化。

# factory.py
from typing import Dict, Type
from chatbot.base import BaseChatBot
from chatbot.feishu import FeiShuChatBot
from chatbot.dingtalk import DingTalkChatBot
from chatbot.wecom import WeComChatbot

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:
        """构造具体的机器人处理类"""
        chatbot_handle_cls = self.CHATBOT_HANDLER_CLS_MAPPING.get(self.chatbot_type)
        return chatbot_handle_cls(webhook_url=webhook_url, secret=secret)

配置管理与使用

为了在不改动代码的情况下切换机器人类型，建议将配置信息提取到环境变量或配置文件中。

1. 配置文件示例

# settings.py
import os

CHATBOT_TYPE = os.getenv("CHATBOT_TYPE", "feishu")
WEBHOOK_URL = os.getenv("WEBHOOK_URL", "https://open.feishu.cn...")
SECRET = os.getenv("CHATBOT_SECRET", "")

2. 业务调用示例

# main.py
from factory import ChatBotFactory
from settings import CHATBOT_TYPE, WEBHOOK_URL, SECRET

chatbot = ChatBotFactory(chatbot_type=CHATBOT_TYPE).build(
    webhook_url=WEBHOOK_URL,
    secret=SECRET
)

try:
    chatbot.send_msg("系统监控正常，无异常告警")
except Exception as e:
    print(f"发送失败：{e}")

安全最佳实践

1. 密钥管理：切勿将 secret 硬编码在代码仓库中。应使用环境变量（如 .env 文件配合 python-dotenv）或密钥管理服务（如 AWS Secrets Manager）。
2. HTTPS：确保 Webhook URL 使用 HTTPS 协议，防止传输过程中被窃听。
3. 最小权限：仅授予机器人必要的权限，避免过度授权。

测试策略

为了保证工厂模式的正确性，建议编写单元测试覆盖工厂的创建逻辑和具体机器人的发送逻辑。

# test_factory.py
import unittest
from unittest.mock import patch, MagicMock
from factory import ChatBotFactory, ChatBotType

class TestChatBotFactory(unittest.TestCase):
    def test_build_feishu(self):
        factory = ChatBotFactory(ChatBotType.FEISHU_CHATBOT)
        bot = factory.build("http://test.com", "secret")
        self.assertIsNotNone(bot)
        self.assertEqual(type(bot).__name__, "FeiShuChatBot")

    @patch('chatbot.feishu.requests.post')
    def test_send_msg_success(self, mock_post):
        mock_response = MagicMock()
        mock_response.json.return_value = {"code": 0}
        mock_post.return_value = mock_response
        
        factory = ChatBotFactory(ChatBotType.FEISHU_CHATBOT)
        bot = factory.build("http://test.com")
        bot.send_msg("test")
        mock_post.assert_called_once()

生产环境建议

1. 连接池：在生产环境中，建议使用 requests.Session 复用 TCP 连接，减少握手开销。
2. 重试机制：网络波动可能导致发送失败，建议增加指数退避重试逻辑。
3. 日志记录：集成 logging 模块，记录发送请求的详细信息及错误堆栈，便于排查问题。
4. 超时控制：设置合理的 timeout 参数，避免阻塞主线程等待响应。

总结

通过工厂模式封装 Webhook 群聊机器人，我们成功实现了业务逻辑与具体平台实现的解耦。当需要新增平台或切换平台时，只需在工厂类的映射表中添加新类型，无需修改业务调用代码。这种设计遵循了开闭原则（Open/Closed Principle），显著提升了系统的可维护性和扩展性，是构建高可用企业级通知服务的推荐方案。