flask部署钉钉机器人和企业微信机器人

引言

创建机器人，目的是通过@机器人的方式，提出用户的问题，得到想要的回答

钉钉机器人

首先我们需要获取钉钉的企业内部开发者权限
然后我们进入钉钉开放平台,登陆后，选择应用开发->机器人->创建应用，我创建了一个叫做chat的机器人
点击进入机器人，在应用信息中有你的AgentID，AppKey和AppSecret，在后面使用中只有AppSecret是有用的。在开发管理处，我们需要配置服务器出口IP和消息接收地址，服务器出口IP填写服务器的IP即可，消息接收地址为外网端口，就是部署应用使用的端口，例如falsk端口
完成之后，我们就可以写这个机器人的回调程序了，我在写回调的时候，发现钉钉机器人接收消息必须在主页面，不能在子页面，主页面指在app.route中的地址为"/“，子页面指”/dingdingrobot"，这里我使用的消息是通过openai的chatgpt构建回复并输出

from flask import Flask, request, jsonify
import base64
import openai
app = Flask(__name__)
openai.api_key = '你的openai密钥'
@app.route("/", methods=["POST", 'GET'])
def get_data():print('进来了')return dingRobot(request)
def dingRobot(request):try:# 第一步验证：是否是post请求if request.method == "POST":# print(request.headers)# 签名验证 获取headers中的Timestamp和Signtimestamp = request.headers.get('Timestamp')sign = request.headers.get('Sign')# 第二步验证：签名是否有效if check_sig(timestamp) == sign:# 获取数据 打印出来看看text_info = json.loads(str(request.data, 'utf-8'))handle_info(text_info)print('验证通过')return str(text_info)print('验证不通过')return str(timestamp)print('有get请求')return str(request.headers)except Exception as e:webhook_url = text_info['sessionWebhook']senderid = text_info['senderId']title = Nonetext = str(e)send_md_msg(senderid, title, text, webhook_url)return str(request.headers)
# 处理自动回复消息
def handle_info(req_data):# 解析用户发送消息 通讯webhook_url text_info = req_data['text']['content'].strip()maxlen = 2048if '生成长度' in text_info:res = re.split('生成长度', text_info)text_info = res[0].strip()maxlen = int(res[1].strip())webhook_url = req_data['sessionWebhook']senderid = req_data['senderId']# print('***************text_info：', text_info)# if判断用户消息触发的关键词，然后返回对应内容# python3.10 以上还可以用 switch case...title = Noneresponse = openai.ChatCompletion.create(model="gpt-3.5-turbo",messages=[{"role": "system", "content": "You are a helpful assistant."},{"role": "user", "content": text_info}])text = response['choices'][0]['message']['content'].strip()print(text)# 调用函数，发送markdown消息send_md_msg(senderid, title, text, webhook_url)
# 发送markdown消息
def send_md_msg(userid, title, message, webhook_url):'''userid: @用户 钉钉idtitle : 消息标题message: 消息主体内容webhook_url: 通讯url'''# data = {#     "msgtype": "markdown",#     "markdown": {#         "title":title,#         "text": message#     },#     '''#     "msgtype": "text",#     "text": {#         "content": message#     },#     '''#     "at": {#         "atUserIds": [#           userid#       ],#     }# }data = {"at": {"atUserIds": [userid],"isAtAll": False},"text": {"content": message},"msgtype": "text"}# 利用requests发送post请求req = requests.post(webhook_url, json=data)
# 消息数字签名计算核对
def check_sig(timestamp):app_secret = '你的AppSecret'app_secret_enc = app_secret.encode('utf-8')string_to_sign = '{}\n{}'.format(timestamp, app_secret)string_to_sign_enc = string_to_sign.encode('utf-8')hmac_code = hmac.new(app_secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()sign = base64.b64encode(hmac_code).decode('utf-8')return sign
if __name__ == '__main__':app.run(host='0.0.0.0', port=8001)

在版本管理与发布界面中，我们就可以调试钉钉机器人了
效果

企业微信机器人

企业微信机器人需要开发者为企业微信的管理者，在拥有管理者授权之后，我们可以添加小程序机器人
点进去之后，我们可以看到AgentID和Secret
和钉钉机器人一样，我们需要配置企业可信任IP，填写服务器的IP
然后开始设置回调函数，下面代码为验证签名的代码，保存到WXBizMsgCrypt.py

#!/usr/bin/env python
# -*- encoding:utf-8 -*-""" 对企业微信发送给企业后台的消息加解密示例代码.
@copyright: Copyright (c) 1998-2014 Tencent Inc.
"""
# ------------------------------------------------------------------------
import logging
import base64
import random
import hashlib
import time
import struct
from Crypto.Cipher import AES
import xml.etree.cElementTree as ET
import socketimport ierror"""
关于Crypto.Cipher模块，ImportError: No module named 'Crypto'解决方案
请到官方网站 https://www.dlitz.net/software/pycrypto/ 下载pycrypto。
下载后，按照README中的“Installation”小节的提示进行pycrypto安装。
"""class FormatException(Exception):passdef throw_exception(message, exception_class=FormatException):"""my define raise exception function"""raise exception_class(message)class SHA1:"""计算企业微信的消息签名接口"""def getSHA1(self, token, timestamp, nonce, encrypt):"""用SHA1算法生成安全签名@param token:  票据@param timestamp: 时间戳@param encrypt: 密文@param nonce: 随机字符串@return: 安全签名"""try:sortlist = [token, timestamp, nonce, encrypt]sortlist.sort()sha = hashlib.sha1()sha.update("".join(sortlist).encode())return ierror.WXBizMsgCrypt_OK, sha.hexdigest()except Exception as e:logger = logging.getLogger()logger.error(e)return ierror.WXBizMsgCrypt_ComputeSignature_Error, Noneclass XMLParse:"""提供提取消息格式中的密文及生成回复消息格式的接口"""# xml消息模板AES_TEXT_RESPONSE_TEMPLATE = """<xml>
<Encrypt><![CDATA[%(msg_encrypt)s]]></Encrypt>
<MsgSignature><![CDATA[%(msg_signaturet)s]]></MsgSignature>
<TimeStamp>%(timestamp)s</TimeStamp>
<Nonce><![CDATA[%(nonce)s]]></Nonce>
</xml>"""def extract(self, xmltext):"""提取出xml数据包中的加密消息@param xmltext: 待提取的xml字符串@return: 提取出的加密消息字符串"""try:xml_tree = ET.fromstring(xmltext)encrypt = xml_tree.find("Encrypt")return ierror.WXBizMsgCrypt_OK, encrypt.textexcept Exception as e:logger = logging.getLogger()logger.error(e)return ierror.WXBizMsgCrypt_ParseXml_Error, Nonedef generate(self, encrypt, signature, timestamp, nonce):"""生成xml消息@param encrypt: 加密后的消息密文@param signature: 安全签名@param timestamp: 时间戳@param nonce: 随机字符串@return: 生成的xml字符串"""resp_dict = {'msg_encrypt': encrypt,'msg_signaturet': signature,'timestamp': timestamp,'nonce': nonce,}resp_xml = self.AES_TEXT_RESPONSE_TEMPLATE % resp_dictreturn resp_xmlclass PKCS7Encoder():"""提供基于PKCS7算法的加解密接口"""block_size = 32def encode(self, text):""" 对需要加密的明文进行填充补位@param text: 需要进行填充补位操作的明文@return: 补齐明文字符串"""text_length = len(text)# 计算需要填充的位数amount_to_pad = self.block_size - (text_length % self.block_size)if amount_to_pad == 0:amount_to_pad = self.block_size# 获得补位所用的字符pad = chr(amount_to_pad)return text + (pad * amount_to_pad).encode()def decode(self, decrypted):"""删除解密后明文的补位字符@param decrypted: 解密后的明文@return: 删除补位字符后的明文"""pad = ord(decrypted[-1])if pad < 1 or pad > 32:pad = 0return decrypted[:-pad]class Prpcrypt(object):"""提供接收和推送给企业微信消息的加解密接口"""def __init__(self, key):# self.key = base64.b64decode(key+"=")self.key = key# 设置加解密模式为AES的CBC模式self.mode = AES.MODE_CBCdef encrypt(self, text, receiveid):"""对明文进行加密@param text: 需要加密的明文@return: 加密得到的字符串"""# 16位随机字符串添加到明文开头text = text.encode()text = self.get_random_str() + struct.pack("I", socket.htonl(len(text))) + text + receiveid.encode()# 使用自定义的填充方式对明文进行补位填充pkcs7 = PKCS7Encoder()text = pkcs7.encode(text)# 加密cryptor = AES.new(self.key, self.mode, self.key[:16])try:ciphertext = cryptor.encrypt(text)# 使用BASE64对加密后的字符串进行编码return ierror.WXBizMsgCrypt_OK, base64.b64encode(ciphertext)except Exception as e:logger = logging.getLogger()logger.error(e)return ierror.WXBizMsgCrypt_EncryptAES_Error, Nonedef decrypt(self, text, receiveid):"""对解密后的明文进行补位删除@param text: 密文@return: 删除填充补位后的明文"""try:cryptor = AES.new(self.key, self.mode, self.key[:16])# 使用BASE64对密文进行解码，然后AES-CBC解密plain_text = cryptor.decrypt(base64.b64decode(text))except Exception as e:logger = logging.getLogger()logger.error(e)return ierror.WXBizMsgCrypt_DecryptAES_Error, Nonetry:pad = plain_text[-1]# 去掉补位字符串# pkcs7 = PKCS7Encoder()# plain_text = pkcs7.encode(plain_text)# 去除16位随机字符串content = plain_text[16:-pad]xml_len = socket.ntohl(struct.unpack("I", content[: 4])[0])xml_content = content[4: xml_len + 4]from_receiveid = content[xml_len + 4:]except Exception as e:logger = logging.getLogger()logger.error(e)return ierror.WXBizMsgCrypt_IllegalBuffer, Noneif from_receiveid.decode('utf8') != receiveid:return ierror.WXBizMsgCrypt_ValidateCorpid_Error, Nonereturn 0, xml_contentdef get_random_str(self):""" 随机生成16位字符串@return: 16位字符串"""return str(random.randint(1000000000000000, 9999999999999999)).encode()class WXBizMsgCrypt(object):# 构造函数def __init__(self, sToken, sEncodingAESKey, sReceiveId):try:self.key = base64.b64decode(sEncodingAESKey + "=")assert len(self.key) == 32except:throw_exception("[error]: EncodingAESKey unvalid !", FormatException)# return ierror.WXBizMsgCrypt_IllegalAesKey,Noneself.m_sToken = sTokenself.m_sReceiveId = sReceiveId# 验证URL# @param sMsgSignature: 签名串，对应URL参数的msg_signature# @param sTimeStamp: 时间戳，对应URL参数的timestamp# @param sNonce: 随机串，对应URL参数的nonce# @param sEchoStr: 随机串，对应URL参数的echostr# @param sReplyEchoStr: 解密之后的echostr，当return返回0时有效# @return：成功0，失败返回对应的错误码def VerifyURL(self, sMsgSignature, sTimeStamp, sNonce, sEchoStr):sha1 = SHA1()ret, signature = sha1.getSHA1(self.m_sToken, sTimeStamp, sNonce, sEchoStr)if ret != 0:return ret, Noneif not signature == sMsgSignature:return ierror.WXBizMsgCrypt_ValidateSignature_Error, Nonepc = Prpcrypt(self.key)ret, sReplyEchoStr = pc.decrypt(sEchoStr, self.m_sReceiveId)return ret, sReplyEchoStrdef EncryptMsg(self, sReplyMsg, sNonce, timestamp=None):# 将企业回复用户的消息加密打包# @param sReplyMsg: 企业号待回复用户的消息，xml格式的字符串# @param sTimeStamp: 时间戳，可以自己生成，也可以用URL参数的timestamp,如为None则自动用当前时间# @param sNonce: 随机串，可以自己生成，也可以用URL参数的nonce# sEncryptMsg: 加密后的可以直接回复用户的密文，包括msg_signature, timestamp, nonce, encrypt的xml格式的字符串,# return：成功0，sEncryptMsg,失败返回对应的错误码Nonepc = Prpcrypt(self.key)ret, encrypt = pc.encrypt(sReplyMsg, self.m_sReceiveId)encrypt = encrypt.decode('utf8')if ret != 0:return ret, Noneif timestamp is None:timestamp = str(int(time.time()))# 生成安全签名sha1 = SHA1()ret, signature = sha1.getSHA1(self.m_sToken, timestamp, sNonce, encrypt)if ret != 0:return ret, NonexmlParse = XMLParse()return ret, xmlParse.generate(encrypt, signature, timestamp, sNonce)def DecryptMsg(self, sPostData, sMsgSignature, sTimeStamp, sNonce):# 检验消息的真实性，并且获取解密后的明文# @param sMsgSignature: 签名串，对应URL参数的msg_signature# @param sTimeStamp: 时间戳，对应URL参数的timestamp# @param sNonce: 随机串，对应URL参数的nonce# @param sPostData: 密文，对应POST请求的数据#  xml_content: 解密后的原文，当return返回0时有效# @return: 成功0，失败返回对应的错误码# 验证安全签名xmlParse = XMLParse()ret, encrypt = xmlParse.extract(sPostData)if ret != 0:return ret, Nonesha1 = SHA1()ret, signature = sha1.getSHA1(self.m_sToken, sTimeStamp, sNonce, encrypt)if ret != 0:return ret, Noneif not signature == sMsgSignature:return ierror.WXBizMsgCrypt_ValidateSignature_Error, Nonepc = Prpcrypt(self.key)ret, xml_content = pc.decrypt(encrypt, self.m_sReceiveId)return ret, xml_content

然后创建部署.py文件，微信机器人可以不在子界面收消息

from flask import Flask, request, jsonify
import base64
import openai
from WXBizMsgCrypt import WXBizMsgCrypt
import xml.etree.cElementTree as ET
app = Flask(__name__)
openai.api_key = '你的openai密钥'
@app.route("/wx", methods=["POST", "GET"])
def get_data():print('进来了')sToken = "3mGn6i6w"sEncodingAESKey = "T458mrywHPrJkDkkXJsUSAJCsutus7NMx5DPWUOSzFF"sCorpID = "wwe3b6de40ee9b0727"wxcpt=WXBizMsgCrypt(sToken,sEncodingAESKey,sCorpID)try:# 第一步验证：是否是post请求if request.method == "GET":# print(request.headers)# 签名验证 获取4种验证条件sVerifyMsgSig = request.args.get('msg_signature')sVerifyTimeStamp = request.args.get('timestamp')sVerifyNonce = request.args.get('nonce')sVerifyEchoStr = request.args.get('echostr')ret,sEchoStr=wxcpt.VerifyURL(sVerifyMsgSig, sVerifyTimeStamp,sVerifyNonce,sVerifyEchoStr)return sEchoStrelse:sReqMsgSig = request.args.get('msg_signature')sReqTimeStamp = request.args.get('timestamp')sReqNonce = request.args.get('nonce')sReqData = request.dataret,sMsg=wxcpt.DecryptMsg( sReqData, sReqMsgSig, sReqTimeStamp, sReqNonce)if( ret!=0 ):print("ERR: DecryptMsg ret: " + str(ret))xml_tree = ET.fromstring(sMsg)content = xml_tree.find("Content").textmodel_output = get_model_response(content)sRespData = f"<xml><ToUserName>{sCorpID}</ToUserName><FromUserName>deploy</FromUserName><CreateTime>{sReqTimeStamp}</CreateTime><MsgType>text</MsgType><Content>{model_output}</Content><MsgId>{sReqNonce}</MsgId><AgentID>1000002</AgentID></xml>"ret,sEncryptMsg=wxcpt.EncryptMsg(sRespData, sReqNonce, sReqTimeStamp)if( ret!=0 ):print("ERR: EncryptMsg ret: " + str(ret))return sEncryptMsgreturn ''except Exception as e:sRespData = f"<xml><ToUserName>{sCorpID}</ToUserName><FromUserName>deploy</FromUserName><CreateTime>{sReqTimeStamp}</CreateTime><MsgType>text</MsgType><Content>{str()}</Content><MsgId>{sReqNonce}</MsgId><AgentID>1000002</AgentID></xml>"ret,sEncryptMsg=wxcpt.EncryptMsg(sRespData, sReqNonce, sReqTimeStamp)return sEncryptMsgprint(str(e))
def get_model_response(text, maxlen=2048):response = openai.Completion.create(model='text-davinci-003',prompt=text,temperature=0.7,max_tokens=maxlen,top_p=1.0,frequency_penalty=0.0,presence_penalty=0.0)text = response.choices[0].text.strip()return text
if __name__ == '__main__':app.run(host='0.0.0.0', port=8001)