Python工廠模式實現(xiàn)封裝Webhook群聊機器人詳解
更新時間：2023年02月21日 09:07:24 作者：憶想不到的暉
企業(yè)存在給 特定群組 自動推送消息的需求，你可以在群聊中添加一個自定義機器人，通過服務端調(diào)用 webhook 地址，即可將外部系統(tǒng)的通知消息即時推送到群聊中。本文就來和大家聊聊具體實現(xiàn)方法
引言
企業(yè)存在給 特定群組 自動推送消息的需求，比如：監(jiān)控報警推送、銷售線索推送、運營內(nèi)容推送等。 你可以在群聊中添加一個自定義機器人，通過服務端調(diào)用 webhook 地址，即可將外部系統(tǒng)的通知消息即時推送到群聊中。
飛書自定義機器人使用指南：
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):
"""
發(fā)送消息(NotImplemented)
Args:
content: 消息內(nèi)容
timeout: 發(fā)送消息請求超時時間 默認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):
"""
發(fā)送消息
Args:
content: 消息內(nèi)容
timeout: 發(fā)送消息請求超時時間 默認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，最后再把簽名參數(shù)再進行 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):
"""
發(fā)送消息
Args:
content: 消息內(nèi)容
timeout: 發(fā)送消息請求超時時間 默認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")
但這樣使用有點不好的一點就是如果我突然從釘釘換成飛書或者企微，業(yè)務中的所有使用的代碼就要全部替換。但一般也不會隨便換平臺，我這里就是想引出工廠模式。
工廠模式封裝
工廠模式是一種常見的設計模式，它可以幫助我們創(chuàng)建對象，而無需顯式地指定其具體類型。在這種模式下，我們通過使用一個工廠來創(chuàng)建對象，并將對象的創(chuàng)建和使用分離開來，從而提高了代碼的可維護性和可擴展性.
對于Webhook群聊機器人，我們可以將其實現(xiàn)為一個工廠類，該工廠類負責創(chuàng)建不同類型的機器人對象。我們可以通過定義一個機器人接口或抽象類，來規(guī)范所有機器人的通用方法和屬性。然后，我們可以根據(jù)需要創(chuàng)建具體的機器人類，并實現(xiàn)其特定的方法和屬性。代碼如下
#!/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):
"""
消息機器人工廠
支持 飛書、釘釘、自定義機器人消息發(fā)送
"""
# 群聊機器人處理類映射
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: 根據(jù) robot_type 返回對應的機器人處理類
"""
chatbot_handle_cls = self.CHATBOT_HANDLER_CLS_MAPPING.get(self.chatbot_type)
return chatbot_handle_cls(webhook_url=webhook_url, secret=secret)
通過字典的方式把機器人類型（平臺）與具體處理類關聯(lián)起來，這樣構造的時候就不用寫 if else
使用的時候直接用工廠創(chuàng)建具體的實例出來就行
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):
"""企業(yè)微信機器人"""
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ā)送消息請求超時時間 默認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):
"""
消息機器人工廠
支持 飛書、釘釘、企微自定義機器人消息發(fā)送
"""
# 群聊機器人處理類映射
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: 根據(jù) 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")
使用工廠模式封裝雖然讓代碼變的多了一些，但更容易維護與擴展，因此我們還是要結合業(yè)務場景選擇一些合適的設計模式來編寫代碼，這樣也能提示自己的編碼能力。
到此這篇關于Python工廠模式實現(xiàn)封裝Webhook群聊機器人詳解的文章就介紹到這了,更多相關Python封裝群聊機器人內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
Python如何通過內(nèi)存管理提升程序執(zhí)行效率
Python提供了自動內(nèi)存管理的功能，但是如果不小心使用，可能會導致內(nèi)存泄漏和性能問題，所以巧妙使用內(nèi)存管理是提高Python執(zhí)行效率的關鍵，下面就來和大家仔細講講Python的內(nèi)存管理技巧吧2023-06-06
Python采集天天基金數(shù)據(jù)掌握最新基金動向
這篇文章主要介紹了Python采集天天基金數(shù)據(jù)掌握最新基金動向,本次案例實現(xiàn)流程為發(fā)送請求、獲取數(shù)據(jù)、解析數(shù)據(jù)、多頁爬取、保存數(shù)據(jù)，接下來來看看具體的操作過程吧2022-01-01![pytorch中[..., 0]的用法說明](http://img.jbzj.com/images/xgimg/bcimg2.png)
這篇文章主要介紹了pytorch中[..., 0]的用法說明，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-05-05
Pyperclip模塊兼容python2和python3，能跨平臺使用，這篇文章主要介紹了Pyperclip模塊安裝和使用詳解,需要的朋友可以參考下2023-03-03
全網(wǎng)最簡約的Anaconda+Python3.7安裝教程Win10
這篇文章主要介紹了全網(wǎng)最簡約的Anaconda+Python3.7安裝教程Win10,圖文講解全流程安裝方法,還不會的小伙伴快來看看吧2023-03-03
python flask sqlalchemy連接數(shù)據(jù)庫流程介紹
這篇文章主要介紹了python flask sqlalchemy連接數(shù)據(jù)庫流程，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2022-09-09
這篇文章主要介紹了解決同一目錄下python import報錯問題,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2023-12-12
一般查詢都是單條件查詢，F(xiàn)和Q是組合條件查詢，本文主要介紹了django中F與Q查詢的使用，感興趣的可以了解一下2021-06-06
今天小編就為大家分享一篇對tensorflow 中tile函數(shù)的使用詳解，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2020-02-02
這篇文章主要介紹了Python用來做Web開發(fā)的優(yōu)勢有哪些，文中講解非常細致，幫助大家更好的理解和學習Python，感興趣的朋友可以了解下2020-08-08