把盏言欢,款款而谈,ChatGPT结合钉钉机器人(outgoing回调)打造人工智能群聊/单聊场景,基于Python3.10

by Liu Yue/2022-12-08

就像黑火药时代里突然诞生的核弹一样，OpenAI的ChatGPT语言模型的横空出世，是人工智能技术发展史上的一个重要里程碑。这是一款无与伦比、超凡绝伦的模型，能够进行自然语言推理和对话，并且具有出色的语言生成能力。

好吧，本篇的开头其实是由ChatGPT生成的：

没办法，面对这个远超时代的AI产品，我们能说什么呢？顶礼膜拜？惊为天人？任何言语对于描述ChatGPT来说已经是苍白无力的，而辞海中的形容词在面对ChatGPT时也已经鞭长莫及。

一句话：言语不能赞其伟大。

本次我们利用ChatGPT的开放API接入钉钉群聊/单聊机器人，让钉钉机器人具备进行自然语言推理和对话的能力，所谓化腐朽为神奇，不过如此。

注册和使用OpenAi的ChatGPT

首先注册OpenAi平台：https://beta.openai.com/ ，由于ChatGPT过于火爆，导致很多地区无法正常注册，这里推荐使用北美地区的代理IP，与此同时，一定要注意，如果之后希望使用后端的API接口方式调用ChatGPT，就不要使用谷歌或者微软的三方账号进行登录，否则无法通过邮箱和秘钥交换OpenAi平台的access_token，切记。

同时，接受验证码手机号也必须是北美地区的手机号，这里推荐一个北美地区的接码平台：https://sms.qisms.com/index 非常好用。

注册成功之后，这里推荐github上开源大神rawandahmad698已经封装好的开源SDK，避免重复造轮子：https://github.com/rawandahmad698/PyChatGPT

安装SDK:

pip3 install chatgptpy --upgrade

安装好之后，编写测试脚本：

chat = Chat(email="OpenAi邮箱", password="OpenAi密码",proxies="代理地址")

answer = chat.ask("你好")

print(answer)

注意，运行代码之前，一定要使用代理proxies，并且确保是北美地区的IP地址。

程序返回：

[OpenAI] Email address: ********
[OpenAI] Password: *********
[OpenAI] Using proxy: {'http': 'http://localhost:4780', 'https': 'http://localhost:4780'}
[OpenAI] Beginning auth process
[OpenAI][1] Making request to https://chat.openai.com/auth/login
[OpenAI][1] Request was successful
[OpenAI][2] Beginning part two
[OpenAI][2] Grabbing CSRF token from https://chat.openai.com/api/auth/csrf
[OpenAI][2] Request was successful
[OpenAI][2] CSRF Token: 1b1357a34e4b0b9a74e999372fe0413ab981c9a72e030a54b3bf172bd6176c5e
[OpenAI][3] Beginning part three
[OpenAI][3] Making request to https://chat.openai.com/api/auth/signin/auth0?prompt=login
[OpenAI][3] Request was successful
[OpenAI][3] Callback URL: https://auth0.openai.com/authorize?client_id=TdJIcbe16WoTHtN95nyywh5E4yOo6ItG&scope=openid%20email%20profile%20offline_access%20model.request%20model.read%20organization.read&response_type=code&redirect_uri=https%3A%2F%2Fchat.openai.com%2Fapi%2Fauth%2Fcallback%2Fauth0&audience=https%3A%2F%2Fapi.openai.com%2Fv1&prompt=login&state=RJt9U13ATPmlt795xMNohQZcUNOytZNvHoq3JI8HGZ4&code_challenge=Pq97ptna00Ybak2dUmIMhR3eqmXZnZz-Fij7otMMw7U&code_challenge_method=S256
[OpenAI][4] Making request to https://auth0.openai.com/authorize?client_id=TdJIcbe16WoTHtN95nyywh5E4yOo6ItG&scope=openid%20email%20profile%20offline_access%20model.request%20model.read%20organization.read&response_type=code&redirect_uri=https%3A%2F%2Fchat.openai.com%2Fapi%2Fauth%2Fcallback%2Fauth0&audience=https%3A%2F%2Fapi.openai.com%2Fv1&prompt=login&state=RJt9U13ATPmlt795xMNohQZcUNOytZNvHoq3JI8HGZ4&code_challenge=Pq97ptna00Ybak2dUmIMhR3eqmXZnZz-Fij7otMMw7U&code_challenge_method=S256
[OpenAI][4] Request was successful
[OpenAI][4] Current State: hKFo2SA5VzlqUDA0Mkl5TnQtNUpYcGRBU0ZfRkhQVUY1eVpWV6Fur3VuaXZlcnNhbC1sb2dpbqN0aWTZIGMzU0xvbThRUXFxMTczeVg4bF8zRFZnYVNOM2M3Q0RFo2NpZNkgVGRKSWNiZTE2V29USHROOTVueXl3aDVFNHlPbzZJdEc
[OpenAI][5] Making request to https://auth0.openai.com/u/login/identifier?state=hKFo2SA5VzlqUDA0Mkl5TnQtNUpYcGRBU0ZfRkhQVUY1eVpWV6Fur3VuaXZlcnNhbC1sb2dpbqN0aWTZIGMzU0xvbThRUXFxMTczeVg4bF8zRFZnYVNOM2M3Q0RFo2NpZNkgVGRKSWNiZTE2V29USHROOTVueXl3aDVFNHlPbzZJdEc
[OpenAI][5] Request was successful
[OpenAI][5] No captcha detected
[OpenAI][6] Making request to https://auth0.openai.com/u/login/identifier
[OpenAI][6] Email found
[OpenAI][7] Entering password...
[OpenAI][7] Password was correct
[OpenAI][7] Old state: hKFo2SA5VzlqUDA0Mkl5TnQtNUpYcGRBU0ZfRkhQVUY1eVpWV6Fur3VuaXZlcnNhbC1sb2dpbqN0aWTZIGMzU0xvbThRUXFxMTczeVg4bF8zRFZnYVNOM2M3Q0RFo2NpZNkgVGRKSWNiZTE2V29USHROOTVueXl3aDVFNHlPbzZJdEc
[OpenAI][7] New State: c3SLom8QQqq173yX8l_3DVgaSN3c7CDE
[OpenAI][8] Making request to https://auth0.openai.com/authorize/resume?state=c3SLom8QQqq173yX8l_3DVgaSN3c7CDE
[OpenAI][8] All good
[OpenAI][8] Access Token: eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCIsImtpZCI6Ik1UaEVOVUpHTkVNMVFURTRNMEZCTWpkQ05UZzVNRFUxUlRVd1FVSkRNRU13UmtGRVFrRXpSZyJ9.eyJodHRwczovL2FwaS5vcGVuYWkuY29tL3Byb2ZpbGUiOnsiZW1haWwiOiJ6Y3hleTI5MTFAb3V0bG9vay5jb20iLCJlbWFpbF92ZXJpZmllZCI6dHJ1ZSwiZ2VvaXBfY291bnRyeSI6IlVTIn0sImh0dHBzOi8vYXBpLm9wZW5haS5jb20vYXV0aCI6eyJ1c2VyX2lkIjoidXNlci1IcHQ2SXF6R0k0RW43V213dGdzaUVOUjUifSwiaXNzIjoiaHR0cHM6Ly9hdXRoMC5vcGVuYWkuY29tLyIsInN1YiI6ImF1dGgwfDYzOTA3ZWRiMTQzYTFkZjQxMzk5Yzc0YyIsImF1ZCI6WyJodHRwczovL2FwaS5vcGVuYWkuY29tL3YxIiwiaHR0cHM6Ly9vcGVuYWkuYXV0aDAuY29tL3VzZXJpbmZvIl0sImlhdCI6MTY3MDQ1OTkzNywiZXhwIjoxNjcwNTQ2MzM3LCJhenAiOiJUZEpJY2JlMTZXb1RIdE45NW55eXdoNUU0eU9vNkl0RyIsInNjb3BlIjoib3BlbmlkIGVtYWlsIHByb2ZpbGUgbW9kZWwucmVhZCBtb2RlbC5yZXF1ZXN0IG9yZ2FuaXphdGlvbi5yZWFkIG9mZmxpbmVfYWNjZXNzIn0.PtXKhJqwudNKLIkNRc5OO6T7Tsl8ydZ8WWnCJ3Ax2c40CQibRTiGLDmfvk2gW5pVIkOpKldWYs6Jrd8UVi0Ih9VMDwS9JL6HpZKsoRaIhy6r6l7AW5vMMQN-l0ntCsgefQeGIrwtCTUsIklN8dyZDkRkympC2AzRkayAcFvFckXTHi_J5Fivr5J7We_OM4cGFJEKTLkaSw6MnYku-uYwAKPVEpFsF7fLnUBRQxn5Zz90FhdeLYEg4IUjPWKPp1iMbp_fa9qhwwtKBwogtrIVzq2t8NdUotoNYgoo2uV2xjQWC2m4V4C_xgkSzLj2TTtRJMOYKGH-lHWs2_yRQF0wOg
[OpenAI][9] Saving access token...
[OpenAI][8] Saved access token

首次运行程序会通过代理自动登录OpenAi平台，并且换取token，最后将token存储在本地。

随后返回ChatGPT的信息：

➜  mydemo git:(master) ✗ /opt/homebrew/bin/python3.10 "/Users/liuyue/wodfan/work/mydemo/test_chatgpt.py"
Using proxies: http://localhost:4780
你好，很高兴为你提供帮助。有什么需要我帮忙的吗？

至此，ChatGPT接口就调试好了。

配置钉钉Dingding机器人

随后，我们来配置C端的机器人，注意这里一定要使用支持outgoing回调的企业机器人，而不是普通的机器人，参考文档：https://open.dingtalk.com/document/group/enterprise-created-chatbot

创建好企业机器人之后，获取机器人应用的Key和秘钥，同时配置好出口IP和接口地址：

所谓出口IP即调用钉钉服务合法的ip，消息接受地址是接受C端信息的地址，这里我们使用异步非阻塞的Tornado框架来构建接受信息服务：

import hmac
import hashlib
import base64
import json
import tornado

from tornado.options import define, options
define('port', default=8000, help='default port',type=int)

class Robot(tornado.web.RequestHandler):

    async def post(self):


        timestamp = self.request.headers.get('timestamp', None)

        sign = self.request.headers.get('sign', None)
        app_secret = '钉钉机器人秘钥'
        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()
        my_sign = base64.b64encode(hmac_code).decode('utf-8')
        if sign != my_sign:
            return self.finish({"errcode":1,"msg":"签名有误"})
        data = json.loads(self.request.body)
        text = data['text']["content"]
        atUsers = data.get("atUsers",None)
        uid = data.get("senderStaffId",None)
        return self.finish({"errcode":0,"msg":text})

urlpatterns = [
    (r"/robot_chat/",Robot),
]


# 创建Tornado实例
application = tornado.web.Application(urlpatterns,debug=True)


if __name__ == "__main__":
    tornado.options.parse_command_line()
    application.listen(options.port)
    tornado.ioloop.IOLoop.instance().start()

这里我们通过Robot异步控制器来接受所有来自钉钉客户端的信息，即人类对机器人说的话，需要注意的是，后端服务需要对请求头中的timestamp和sign进行验证，以判断是否是来自钉钉的合法请求，避免其他仿冒钉钉调用开发者的HTTPS服务传送数据。

所以这里一旦签名有问题，就结束逻辑：

timestamp = self.request.headers.get('timestamp', None)

sign = self.request.headers.get('sign', None)
app_secret = '钉钉机器人秘钥'
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()
my_sign = base64.b64encode(hmac_code).decode('utf-8')
if sign != my_sign:
    return self.finish({"errcode":1,"msg":"签名有误"})

最后该接口会返回发信人id（uid）以及具体信息内容（text）。

至此，后端接受服务就配置好了。

下面就是后端推送服务，首先，根据官方文档：https://open.dingtalk.com/document/orgapp-server/obtain-the-access_token-of-an-internal-app?spm=ding_open_doc.document.0.0.5f255239xgW3zE#topic-2056397

需要获取钉钉接口的token:

def get_token(self):

        res = requests.post("https://api.dingtalk.com/v1.0/oauth2/accessToken",data=json.dumps({"appKey":self._appKey,"appSecret":self._appSecret}),headers={"Content-Type":"application/json"})

        token = res.json()["accessToken"]

        return token

随后，根据文档：https://open.dingtalk.com/document/group/chatbots-send-one-on-one-chat-messages-in-batches?spm=ding_open_doc.document.0.0.22e749acXECz5m#topic-2080109

我们来配置单聊推送：

# 单聊
    def send_message(self,uid,message):

        res = requests.post("https://api.dingtalk.com/v1.0/robot/oToMessages/batchSend",data=json.dumps({"robotCode":self._appKey,"userIds":[uid],"msgKey":"sampleText","msgParam":'{"content":"'+message+'"}'}),headers={"Content-Type":"application/json","x-acs-dingtalk-access-token":self._token})

        print(res.text)

具体效果：

接着，继续根据官方文档：https://open.dingtalk.com/document/robots/guide-to-user-access-for-intra-enterprise-robot-group-chat

配置群聊推送方法：

# 群聊
    def send_user(self,uid,message):

        data = {
        "at": {
            "atUserIds": [
                uid
            ]
        },
        "text": {
            "content": message
        },
        "msgtype": "text"
        }

        res = requests.post(self._webhook,data=json.dumps(data),headers={"Content-Type":"application/json"})

        print(res.text)

群聊效果：

这里需要注意的是，单聊是通过接口的方式进行推送，而群内聊天是通过webhook方式进行推送，关于webhook，请移玉步至：使用python3.7配置开发钉钉群自定义机器人(2020年新版攻略)

完整代码：

import requests
import json

from pychatgpt import Chat

class DingDing:


    def __init__(self,appKey=None,appSecret=None) -> None:

        self._appKey = appKey

        self._appSecret = appSecret
        
        self._token = self.get_token()

        # 机器人webhook地址
        self._webhook = ""

        


    def get_token(self):

        res = requests.post("https://api.dingtalk.com/v1.0/oauth2/accessToken",data=json.dumps({"appKey":self._appKey,"appSecret":self._appSecret}),headers={"Content-Type":"application/json"})

        token = res.json()["accessToken"]

        return token

    # 单聊
    def send_message(self,uid,message):

        res = requests.post("https://api.dingtalk.com/v1.0/robot/oToMessages/batchSend",data=json.dumps({"robotCode":self._appKey,"userIds":[uid],"msgKey":"sampleText","msgParam":'{"content":"'+message+'"}'}),headers={"Content-Type":"application/json","x-acs-dingtalk-access-token":self._token})

        print(res.text)

    # 群聊
    def send_user(self,uid,message):

        data = {
        "at": {
            "atUserIds": [
                uid
            ]
        },
        "text": {
            "content": message
        },
        "msgtype": "text"
        }

        res = requests.post(self._webhook,data=json.dumps(data),headers={"Content-Type":"application/json"})

        print(res.text)




if __name__ == '__main__':

    dingding = DingDing("appkey","appSecret")

    #chat = Chat(email="OpenAi邮箱", password="OpenAi密码",proxies="代理地址")

    #answer = chat.ask("你好")

    # 单聊
    #dingding.send_message('uid',answer)

    # 群聊
    #dingding.send_user('uid',answer)

    #print(answer)

至此，后端推送服务就配置好了。

结语

最后，奉上Github项目地址，与众亲同飨：https://github.com/zcxey2911/Python_ChatGPT_ForDingding_OpenAi ，毫无疑问，ChatGPT是NLP领域历史上最伟大的项目，没有之一，伟大，就是技术层面的极致，你同意吗？