企业常需向特定群组推送监控报警或运营消息。直接实例化不同平台机器人会导致代码耦合度高，切换平台需修改业务逻辑。本文介绍使用 Python 工厂模式封装 Webhook 群聊机器人，通过定义抽象基类规范通用接口，利用工厂类根据配置动态创建具体机器人实例。方案支持飞书、钉钉及企业微信，结合配置文件与日志系统，实现低耦合、易扩展的消息推送架构，提升维护效率。
在企业级应用中，向特定群组自动推送消息是常见需求。例如：监控系统报警推送、销售线索通知、运营活动提醒等。通常的做法是在群聊中添加一个自定义机器人，通过服务端调用 webhook 地址，即可将外部系统的通知消息即时推送到群聊中。
主流 IM 平台（如飞书、钉钉、企业微信）均提供了类似的 Webhook 接入能力，但不同平台的签名算法、请求参数格式存在差异。如果业务代码中直接硬编码了具体平台的实现，一旦需要切换平台或新增支持的平台，所有调用处的代码都需要修改，导致系统耦合度高，维护成本大。
本文介绍如何使用 Python 的工厂模式（Factory Pattern）来封装 Webhook 群聊机器人，实现创建对象与使用对象的解耦，提高代码的可维护性和可扩展性。
在初始阶段，我们可能会为每个平台编写独立的类，并在业务逻辑中直接实例化：
feishu = FeiShuChatBot(webhook_url="xxx", secret="xxxx")
feishu.send_msg("test msg")
dingtalk = DingTalkChatBot(webhook_url="xxx", secret="xxxx")
dingtalk.send_msg("test msg")
这种方式的缺点显而易见：
工厂模式是一种创建型设计模式，它提供了一种创建对象的最佳方式。在工厂模式中，我们在创建对象时不会对客户端暴露创建逻辑，而是通过一个共同的接口来引用具体类型的对象。
对于 Webhook 群聊机器人场景，我们可以定义一个抽象基类 BaseChatBot 规范通用方法（如 send_msg），然后为每个平台实现具体的子类。最后，创建一个 ChatBotFactory 工厂类，根据传入的类型参数动态返回对应的机器人实例。
首先定义统一的异常类，便于上层捕获和处理发送错误。
# exceptions.py
class SendMsgException(Exception):
"""消息发送异常"""
pass
定义机器人基类，包含初始化逻辑和抽象方法。
# chatbot/base.py
import hmac
import base64
import hashlib
import time
from urllib.parse import quote_plus
import requests
from abc import ABC, abstractmethod
class BaseChatBot(ABC):
"""群聊机器人基类"""
def __init__(self, webhook_url: str, secret: str = None):
self.webhook_url = webhook_url
self.secret = secret
@abstractmethod
def _get_sign(self, timestamp: str, secret: str) -> str:
"""获取签名"""
raise NotImplementedError
@abstractmethod
def send_msg(self, content: str, timeout=10):
"""发送消息"""
raise NotImplementedError
飞书机器人需要在请求头中包含时间戳和签名。
# 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
工厂类负责管理不同类型机器人的映射关系，并负责实例化。
# 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)
为了在不改动代码的情况下切换机器人类型，建议将配置信息提取到环境变量或配置文件中。
# 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", "")
# 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}")
secret 硬编码在代码仓库中。应使用环境变量（如 .env 文件配合 python-dotenv）或密钥管理服务（如 AWS Secrets Manager）。为了保证工厂模式的正确性，建议编写单元测试覆盖工厂的创建逻辑和具体机器人的发送逻辑。
# 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()
requests.Session 复用 TCP 连接，减少握手开销。logging 模块，记录发送请求的详细信息及错误堆栈，便于排查问题。timeout 参数，避免阻塞主线程等待响应。通过工厂模式封装 Webhook 群聊机器人，我们成功实现了业务逻辑与具体平台实现的解耦。当需要新增平台或切换平台时，只需在工厂类的映射表中添加新类型，无需修改业务调用代码。这种设计遵循了开闭原则（Open/Closed Principle），显著提升了系统的可维护性和扩展性，是构建高可用企业级通知服务的推荐方案。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online
将 Markdown（GFM）转为 HTML 片段，浏览器内 marked 解析；与 HTML转Markdown 互为补充。 在线工具，Markdown转HTML在线工具，online
将 HTML 片段转为 GitHub Flavored Markdown，支持标题、列表、链接、代码块与表格等；浏览器内处理，可链接预填。 在线工具，HTML转Markdown在线工具，online
通过删除不必要的空白来缩小和压缩JSON。 在线工具，JSON 压缩在线工具，online