⽤python实现钉钉Outgoing机器⼈（企业内部机器⼈）全过程
不多说，先上最终效果图
python新手代码userid[
再上⼀段连接，后续所有过程都是在⼤佬的思路下进⾏的，代码也是从⼤佬这⾥复制的，只是稍微改动的⼀部分，⼤家可以在联调通过能实现交互之后再去⾃⼰开发后台逻辑。但是因为⼤佬写的太简单了，省了很多步骤，所以我才⾃⼰重新写⼀份，防⽌后⾯忘记了
先直接贴代码，再说我的从零开始的步骤
1.先安装apache http服务，并配置反向代理，代理的就是python绑定的端⼝和ip，配置⽅法参考如下
2.因为企业内部机器⼈需要⼀个公⽹的http服务，所以使⽤钉钉推荐的内⽹穿透去实现。这⼀点没遇到什么问题，仔细看官⽅⽂档就好了
3.到钉钉开发者后台去创建⼀个机器⼈
注意：机器⼈创建后凭证与基础信息这⼀栏中的参数AppSecret需要复制出来，到代码⾥替app_secret变量的值
然后再发布机器⼈到钉钉，此时可以看到机器⼈有⼀个参数叫Webhook，这个直接复制到代码⾥，后⾯会讲有什么⽤。
⼤概的准备⼯作都准备完了，就开始逐⼀打通了
因为内⽹穿透之后要启动⼀个http服务才可以看到具体的使⽤效果，所以安装apache，⾄于为什么不⽤python代码来实现http服务，个⼈喜好吧。因为本⾝就有apache所以我就懒得再去看别⼈的代码学习了，
然后⾄于为什么要使⽤apache反向代理，因为我发现打通了之后，钉钉发送的请求可以被apache http接收到，但是我的python后台运⾏过程中并没有任何动作，所以我⽴马反映过来他们根本没有联通！于是尝试nginx代理，发现不可取具体原因我也忘了，这两天⼀直弄这个注意⼒转义的也快哈哈哈。
⽤apache来配置反向代理后，钉钉发送的http请求先会根据内⽹穿透后提供的公⽹域名来发送到本地的apache http服务，再由反向代理转发到我们运⾏的python程序上。
到这⼀步基本上获取数据这⼀块就完成了，接下来就是outgoing机器⼈的⾃动回复实现了。
所以，交互的原理就是根据钉钉http发送过来的请求中的报⽂信息，⽤python解析后在对信息进⾏判断，在执⾏对应的⽅法去处理，最后通过调⽤request⽅法把处理结果返回到钉钉。这样就⼤功告成了
接下来再对我重点遇到的问题进⾏⼀个总结：
1. 因为之前安装过禅道，其中也包含了apache http服务，导致新安装的apache服务⽆法新增，处理办法如下
2. 内⽹穿透后报错Unable to initiate connection to 127.0.0.1:81. A web server must be running on port 127.0.0.1:81to
complete the tunnel.
这个是因为没有启动http服务，并且apache监听的端⼝要和内⽹穿透的端⼝要保持⼀致。
3. 就是代码中的参数问题，只恨我没有仔细看官⽅⽂档，只顾拿别⼈的代码复制粘贴就⽤，到后⾯才慢慢摸索各个参数的来源
app_secret参数：这个很重要，是⽤来计算sgin的
webhook参数：就是这个参数，折磨了我⼗多个⼩时，我⾄今没搞懂⼤佬的参数是怎么获取的，反正我是直接拿来⽤的。
这个参数是在企业内部机器⼈上线后，添加到⾥⾯以后才会有的！
如下是代码，请粘贴复制后仔细阅读，重点看handle_client()，getPost()，initKey()这三⽅法就好了，获取数据，数据处理，返回数据这三个主要功能都在这⾥。
# -*- coding: GBK -*-
import requests
import json
import time
import hmac
import hashlib
import base64
import socket
from multiprocessing import Process
def handle_client(client_socket):
# 获取socket
request_data = v(20000)
post_userid, post_sign, post_timestamp, post_mes = getPost(request_data)
# 回应socket
initKey(post_userid, post_sign, post_timestamp, post_mes)
# 关闭socket
client_socket.close()
def getPost(request_data):
request_data =str(request_data, encoding="utf8").split('\r\n')
items =[]
for item in request_data[1:-2]:
items.append(item.split(':'))
post_useful ={}
for i in items:
post_useful.update({i[0]: i[1]})
if ('sign')==None:
print('other connect')
return0
else:
post_sign = ('sign').strip()
post_timestamp = ('timestamp').strip()
post_mes = json.loads(request_data[-1])
post_mes = json.loads(request_data[-1])
post_userid = ('senderId').strip()
post_mes = ('text').get('content').strip()
return post_userid, post_sign, post_timestamp, post_mes
def initKey(post_userid, post_sign, post_timestamp, post_mes):
# 配置token
# 得到当前时间戳
timestamp =str(round(time.time()*1000))
# 计算签名
app_secret ='bKHMrla5Jxxxxxxxxxxxxxxxxxxxxxx'
app_secret_enc = de('utf-8')
string_to_sign ='{}\n{}'.format(post_timestamp, app_secret)
string_to_sign_enc = string_de('utf-8')
hmac_code = w(app_secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
sign = base64.b64encode(hmac_code).decode('utf-8')
# 验证是否来⾃钉钉的合法请求
if(abs(int(post_timestamp)-int(timestamp))<3600000and post_sign == sign):
webhook="oapi.dingtalk/robot/send?access_token=xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"#宝1 header ={
"Content-Type":"application/json",
"Charset":"UTF-8"
}
# 发送消息
message_json = json.dumps(selectMes(post_userid, post_mes))
# 返回发送状态
info = requests.post(url=webhook, data=message_json, headers=header)
)
else:
print("Warning:Not DingDing's post")
def selectMes(post_userid, post_mes):
# 判断指令选择对应回复
if(post_mes =='你好'):
send_mes ='你好啊同学!'
return sendText(post_userid, send_mes)
if(post_mes =='你叫什么名字'):
send_mes ='我是⼩宝⼀号~~'
return sendText(post_userid, send_mes)
elif(post_mes =='天⽓'):
send_mes = getWeather()
return sendMarkdown('天⽓预报', send_mes)
else:
return sendText(post_userid,'Not understand')
def sendText(post_userid, send_mes):
# 发送⽂本形式
message ={
"msgtype":"text",
"text":{
"content": send_mes
},
"at":{
"atDingtalkIds":[post_userid],
"isAtAll":False
}
}
return message
def sendMarkdown(title, send_mes):
# 发送Markdown形式
message ={
message ={
"msgtype":"markdown",
"markdown":{
"title": title,
"text": send_mes
},
"at":{
"atDingtalkIds":[],
"isAtAll":False
}
}
return message
def getWeather():
# 爬取天⽓数据返回的⽅法，这⾥就不多编写了
send_mes ='#### 今⽇天⽓ \n' \
'> 9度，西北风1级，空⽓良89，相对温度73%\n' \
'> \n' \ '> ># 10点20分发布 [天⽓](www.dingalk) \n'
return send_mes
if __name__ =="__main__":
# 启动服务，端⼝9000
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind(("",9000))
server_socket.listen(120)
while True:
client_socket, client_address = server_socket.accept()
print(f"{client_address},⽤户连接上了")
handle_client_process = Process(target=handle_client, args=(client_socket,))
handle_client_process.start()
client_socket.close()
版权声明：本站内容均来自互联网，仅供演示用，请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系QQ:729038198，我们将在24小时内删除。
发表评论