引言
创建机器人,目的是通过@机器人的方式,提出用户的问题,得到想要的回答
钉钉机器人
- 首先我们需要获取钉钉的企业内部开发者权限
- 然后我们进入钉钉开放平台,登陆后,选择应用开发->机器人->创建应用,我创建了一个叫做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和Sign
timestamp = 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 = None
text = 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 = 2048
if '生成长度' 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 = None
response = 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: @用户 钉钉id
title : 消息标题
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 socket
import ierror
"""
关于Crypto.Cipher模块,ImportError: No module named 'Crypto'解决方案
请到官方网站 https://www.dlitz.net/software/pycrypto/ 下载pycrypto。
下载后,按照README中的“Installation”小节的提示进行pycrypto安装。
"""
class FormatException(Exception):
pass
def 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, None
class 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.text
except Exception as e:
logger = logging.getLogger()
logger.error(e)
return ierror.WXBizMsgCrypt_ParseXml_Error, None
def 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_dict
return resp_xml
class PKCS7Encoder():
"""提供基于PKCS7算法的加解密接口"""
block_size = 32
def 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 = 0
return decrypted[:-pad]
class Prpcrypt(object):
"""提供接收和推送给企业微信消息的加解密接口"""
def __init__(self, key):
# self.key = base64.b64decode(key+"=")
self.key = key
# 设置加解密模式为AES的CBC模式
self.mode = AES.MODE_CBC
def 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, None
def 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, None
try:
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, None
if from_receiveid.decode('utf8') != receiveid:
return ierror.WXBizMsgCrypt_ValidateCorpid_Error, None
return 0, xml_content
def 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) == 32
except:
throw_exception("[error]: EncodingAESKey unvalid !", FormatException)
# return ierror.WXBizMsgCrypt_IllegalAesKey,None
self.m_sToken = sToken
self.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, None
if not signature == sMsgSignature:
return ierror.WXBizMsgCrypt_ValidateSignature_Error, None
pc = Prpcrypt(self.key)
ret, sReplyEchoStr = pc.decrypt(sEchoStr, self.m_sReceiveId)
return ret, sReplyEchoStr
def 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,失败返回对应的错误码None
pc = Prpcrypt(self.key)
ret, encrypt = pc.encrypt(sReplyMsg, self.m_sReceiveId)
encrypt = encrypt.decode('utf8')
if ret != 0:
return ret, None
if 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, None
xmlParse = 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, None
sha1 = SHA1()
ret, signature = sha1.getSHA1(self.m_sToken, sTimeStamp, sNonce, encrypt)
if ret != 0:
return ret, None
if not signature == sMsgSignature:
return ierror.WXBizMsgCrypt_ValidateSignature_Error, None
pc = 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 sEchoStr
else:
sReqMsgSig = request.args.get('msg_signature')
sReqTimeStamp = request.args.get('timestamp')
sReqNonce = request.args.get('nonce')
sReqData = request.data
ret,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").text
model_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 sEncryptMsg
return ''
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 sEncryptMsg
print(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)
- 部署完成后,我们可以在企业微信工作台找到这个小程序,并进行聊天