搞一个钉钉机器人

现在，我们都使用钉钉上网课，并且钉钉对于开发者们也是很友好。我们今天就来学习做一个钉钉机器人。

首先，下载PC端的钉钉，登录一下。

然后新建一个用于测试的群聊。

点击右边菜单栏中第一项，点击智能群助手。

添加机器人。

选择自定义。

名字随意。下面的安全设置有三种，主要是防止被别人控制。

第一个自定义关键字是说你在以后发送的文字中必须要有这个关键字，否则发送不成功。加签是一种特殊的加密方式，在开发文档中有，先不讲。

IP地址就是说你在发送时会获取你的IP地址，如果不匹配就发送不成功。这个加密的方式可以自己选择，我们选择加签。如果你想使用IP的话，可以访问https://ip.cn/。

我们看一看安全设置旁边的说明文档。

往下划一点，找到Python。因为我们先使用Python制作。

复制这段代码：

import time
import hmac
import hashlib
import base64
import urllib.parse


timestamp = str(round(time.time() * 1000))
secret = 'this is secret'
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 = urllib.parse.quote_plus(base64.b64encode(hmac_code))
print(timestamp)
print(sign)

有一点复杂，不过这才叫加密呀。

我们回到钉钉APP，选中加签，然后“阅读”一下免责条款，点击继续。

复制一下这段网址。然后顺一下我们的思路。

首先，我们要使用爬虫访问这段网址（注意：要使用POST而不是GET。于这两个可以看https://cnblogs.com），然后传一下JSON数据，就可以发送信息了。

我们先定义一个函数，叫做pwd，它用于加签。钉钉的开发文档中是这样写的“把timestamp和第一步得到的签名值拼接到URL中。例https://oapi.dingtalk.com/robot/send？access_token=XXXXXX&timestamp=XXX&sign=XXX”，这里面的timestamp和签名值是加密输出的值，就是最后两个print里面的值。我们在函数pwd中先计算，然后返回拼接的URL。

dingtalk_robot.py

import time
import hmac
import hashlib
import base64
import urllib.parse




def pwd():
    timestamp = str(round(time.time() * 1000))
    secret = 'this is secret'
    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 = urllib.parse.quote_plus(base64.b64encode(hmac_code))
    return "https://oapi.dingtalk.com/robot/send?access_token" \
           "=c5b73c1dd2a7d506e80038344eff5812f713c3b7ac17a0200a23e6fcb2d1dcb4"+"&timestamp={0}&sign={1}".format(timestamp, sign)

为了遵循PEP8，我们将所有导入都放到了最前面。这没有丝毫影响。

尝试调用一下。

terminal -- python 3.8.1

>>> import dingtalk_robot
>>> dingtalk_robot.pwd()
'https://oapi.dingtalk.com/robot/send?access_token=c5b73c1dd2a7d506e80038344eff5812f713c3b7ac17a0200a23e6fcb2d1dcb4&timestamp=1588136544177&sign=bH6X1O2YHwI42cld%2B6M7QC%2BrFrdPowlfCQ4jw5Aa%2FCA%3D'
>>>

运行正常，然后我们尝试访问这个网址。

正常访问，但由于浏览器默认使用GET请求，所以还是不要用浏览器了。

首先，新建一个名为send的函数。导入requests库。

dingtalk_robot.py

import time
import hmac
import hashlib
import base64
import urllib.parse




def pwd():
    timestamp = str(round(time.time() * 1000))
    secret = 'this is secret'
    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 = urllib.parse.quote_plus(base64.b64encode(hmac_code))
    return "https://oapi.dingtalk.com/robot/send?access_token" \
           "=c5b73c1dd2a7d506e80038344eff5812f713c3b7ac17a0200a23e6fcb2d1dcb4"+"&timestamp={0}&sign={1}".format(timestamp, sign)




def send(content):
    pass

这个函数接受一个参数，就是发送的内容。它现在什么也没有干，我们让它先发送一个POST请求给钉钉服务器。

dingtalk_robot.py

import time
import hmac
import hashlib
import base64
import urllib.parse
import requests




def pwd():
    timestamp = str(round(time.time() * 1000))
    secret = 'this is secret'
    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 = urllib.parse.quote_plus(base64.b64encode(hmac_code))
    return "https://oapi.dingtalk.com/robot/send?access_token" \
           "=c5b73c1dd2a7d506e80038344eff5812f713c3b7ac17a0200a23e6fcb2d1dcb4"+"&timestamp={0}&sign={1}".format(timestamp, sign)




def send(content):
    requests.post(pwd(), data="Hello World")

这样做没有任何作用，我们将发送后钉钉返回的内容返回来。

dingtalk_robot.py

import time
import hmac
import hashlib
import base64
import urllib.parse
import requests
import json




def pwd():
    timestamp = str(round(time.time() * 1000))
    secret = 'this is secret'
    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 = urllib.parse.quote_plus(base64.b64encode(hmac_code))
    return "https://oapi.dingtalk.com/robot/send?access_token" \
           "=c5b73c1dd2a7d506e80038344eff5812f713c3b7ac17a0200a23e6fcb2d1dcb4"+"&timestamp={0}&sign={1}".format(timestamp, sign)




def send(content):
    print(requests.post(pwd(), data="Hello World").text)

调用这个函数。

terminal -- python 3.8.1

>>> import dingtalk_robot
>>> dingtalk_robot.send(123)
{"errcode":43004,"errmsg":"无效的HTTP HEADER Content-Type"}
>>>

虽然发送成功了，但他也说了，我们这样发送是无效的。我们看一看开发文档中怎么写的。

稍微往下滑一点，找到下图。

这是一个简单的文本，我们照着抄下来。他说的不用的地方我们也不用。

大约是这样的。

{
    "msgtype": "text", 
    "text": {
        "content": "你要发送的内容"
    }, 
    "at": {
        "atMobiles": [
            "你要@的手机号", 
            "你要@的手机号"
        ], 
        "isAtAll": 你要不要@所有人。true为要，false为不要。
    }
}

我们把它组成一个Python字典，然后使用Python内置库Json转换为JSON数据再发送给服务器。（记得在使用requests发送时要使用headers指出类型，否则它也是不接受的）

代码：

dingtalk_robot.py

import time
import hmac
import hashlib
import base64
import urllib.parse
import requests
import json




def pwd():
    timestamp = str(round(time.time() * 1000))
    secret = 'this is secret'
    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 = urllib.parse.quote_plus(base64.b64encode(hmac_code))
    return "https://oapi.dingtalk.com/robot/send?access_token" \
           "=c5b73c1dd2a7d506e80038344eff5812f713c3b7ac17a0200a23e6fcb2d1dcb4"+"&timestamp={0}&sign={1}".format(timestamp, sign)




def send(content):
    data = {
        "msgtype": "text",
        "text": {
            "content": content
        },
    }
    print(requests.post(pwd(), data=json.dumps(data), headers={"content-type": "application/json"}).text)

调用。

terminal -- python 3.8.1

>>> import dingtalk_robot
>>> dingtalk_robot.send("大家好")
{"errcode":310000,"errmsg":"sign not match, more: [https://ding-doc.dingtalk.com/doc#/serverapi2/qf2nxq]"}
>>>

哎呀，出错了。我们去开发文档看看。

搞了半天，发现是群聊类型的事情。重新建一个就好了。

现在，我们来制作一个爬虫，每天早上八点半爬取金山词霸的每日一句，然后发送到群聊内。

我们打开http://open.iciba.com/dsapi，会发现有一堆很乱的数据。我们先把它爬下来。

import requests
import json
import time
import hmac
import hashlib
import base64
import urllib.parse




def request(text):
    url = "http://openapi.tuling123.com/openapi/api/v2"
    data = {
        "reqType": 0,
        "perception": {
            "inputText": {
                "text": text
            },
        },
        "userInfo": {
            "apiKey": "ef8b9ab3c8e843d3b5d8c22553cfd01c",
            "userId": "a6b3366937d7896e",
        }
    }
    posts = requests.post(url, json=data)
    response = eval(posts.content.decode("utf-8"))
    return response["results"][0]["values"]["text"]




def pwd():
    timestamp = str(round(time.time() * 1000))
    secret = 'this is secret'
    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 = urllib.parse.quote_plus(base64.b64encode(hmac_code))
    return timestamp, sign




def sendMsg(text):
    url = "https://oapi.dingtalk.com/robot/send?access_token=639ca5ff60b66f9cd54fb4a2fce7b62b8afcb9b2d9e90bd6caaa41ff39d81537&timestamp={0}&sign={1}"\
        .format(pwd()[0], pwd()[1])
    data = {
        "msgtype": "text",
        "text": {
            "content": text
        },
    }
    content = requests.post(url, data=json.dumps(data), headers={'content-Type': 'application/json'})
    return content.text




def sentence():
    url = "http://open.iciba.com/dsapi/"
    print(requests.get(url).text)




sentence()

输出：

{"sid":"3747","tts":"https://v-sq.ks3-cn-beijing.ksyun.com/audio/a9b4c94e17f5a37aa0e93f62b7e025d3.mp3","content":"Don't close your eyes, the opportunity will appear in the next second.","note":"请不要闭眼，机会就在下一秒钟出现。","love":"0","translation":"新版每日一句","picture":"https://v-sq.ks3-cn-beijing.ksyun.com/image/53dafb1c1e2a31811fac8e2ce118990d.jpg","picture2":"https://v-sq.ks3-cn-beijing.ksyun.com/image/68cce69e9e7dd8f01f8f2781caf450e0.jpg","caption":"词霸每日一句","dateline":"2020-04-29","s_pv":"0","sp_pv":"0","fenxiang_img":"https://v-sq.ks3-cn-beijing.ksyun.com/image/6ce6ae7323e1b30714cac8797f8fe660.png","picture3":"https://v-sq.ks3-cn-beijing.ksyun.com/image/7f3b91b5d746240af7b67c0477176157.jpg","picture4":"https://v-sq.ks3-cn-beijing.ksyun.com/image/757836d8c2e13c6fb1bc282720a8d16c.jpg","tags":[]}

我们将它转换为程序比较好处理的格式。使用json.loads。

import requests
import json
import time
import hmac
import hashlib
import base64
import urllib.parse




def request(text):
    url = "http://openapi.tuling123.com/openapi/api/v2"
    data = {
        "reqType": 0,
        "perception": {
            "inputText": {
                "text": text
            },
        },
        "userInfo": {
            "apiKey": "ef8b9ab3c8e843d3b5d8c22553cfd01c",
            "userId": "a6b3366937d7896e",
        }
    }
    posts = requests.post(url, json=data)
    response = eval(posts.content.decode("utf-8"))
    return response["results"][0]["values"]["text"]




def pwd():
    timestamp = str(round(time.time() * 1000))
    secret = 'this is secret'
    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 = urllib.parse.quote_plus(base64.b64encode(hmac_code))
    return timestamp, sign




def sendMsg(text):
    url = "https://oapi.dingtalk.com/robot/send?access_token=639ca5ff60b66f9cd54fb4a2fce7b62b8afcb9b2d9e90bd6caaa41ff39d81537&timestamp={0}&sign={1}"\
        .format(pwd()[0], pwd()[1])
    data = {
        "msgtype": "text",
        "text": {
            "content": text
        },
    }
    content = requests.post(url, data=json.dumps(data), headers={'content-Type': 'application/json'})
    return content.text




def sentence():
    url = "http://open.iciba.com/dsapi/"
    print(json.loads(requests.get(url).text))




sentence()

{'sid': '3747', 'tts': 'https://v-sq.ks3-cn-beijing.ksyun.com/audio/a9b4c94e17f5a37aa0e93f62b7e025d3.mp3', 'content': "Don't close your eyes, the opportunity will appear in the next second.", 'note': '请不要闭眼，机会就在下一秒钟出现。', 'love': '0', 'translation': '新版每日一句', 'picture': 'https://v-sq.ks3-cn-beijing.ksyun.com/image/53dafb1c1e2a31811fac8e2ce118990d.jpg', 'picture2': 'https://v-sq.ks3-cn-beijing.ksyun.com/image/68cce69e9e7dd8f01f8f2781caf450e0.jpg', 'caption': '词霸每日一句', 'dateline': '2020-04-29', 's_pv': '0', 'sp_pv': '0', 'fenxiang_img': 'https://v-sq.ks3-cn-beijing.ksyun.com/image/6ce6ae7323e1b30714cac8797f8fe660.png', 'picture3': 'https://v-sq.ks3-cn-beijing.ksyun.com/image/7f3b91b5d746240af7b67c0477176157.jpg', 'picture4': 'https://v-sq.ks3-cn-beijing.ksyun.com/image/757836d8c2e13c6fb1bc282720a8d16c.jpg', 'tags': []}

好像没什么不一样，但这时的数据已经变成了字典了。我们来直接访问。

import requests
import json
import time
import hmac
import hashlib
import base64
import urllib.parse




def request(text):
    url = "http://openapi.tuling123.com/openapi/api/v2"
    data = {
        "reqType": 0,
        "perception": {
            "inputText": {
                "text": text
            },
        },
        "userInfo": {
            "apiKey": "ef8b9ab3c8e843d3b5d8c22553cfd01c",
            "userId": "a6b3366937d7896e",
        }
    }
    posts = requests.post(url, json=data)
    response = eval(posts.content.decode("utf-8"))
    return response["results"][0]["values"]["text"]




def pwd():
    timestamp = str(round(time.time() * 1000))
    secret = 'this is secret'
    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 = urllib.parse.quote_plus(base64.b64encode(hmac_code))
    return timestamp, sign




def sendMsg(text):
    url = "https://oapi.dingtalk.com/robot/send?access_token=639ca5ff60b66f9cd54fb4a2fce7b62b8afcb9b2d9e90bd6caaa41ff39d81537&timestamp={0}&sign={1}"\
        .format(pwd()[0], pwd()[1])
    data = {
        "msgtype": "text",
        "text": {
            "content": text
        },
    }
    content = requests.post(url, data=json.dumps(data), headers={'content-Type': 'application/json'})
    return content.text




def sentence():
    url = "http://open.iciba.com/dsapi/"
    sentences = json.loads(requests.get(url).text)
    english_sentence = sentences["content"]
    ch_sentence = sentences["note"]
    img = sentences["picture3"]
    print(english_sentence)
    print(ch_sentence)
    print(img)




sentence()

输出：

Don't close your eyes, the opportunity will appear in the next second.

请不要闭眼，机会就在下一秒钟出现。

https://v-sq.ks3-cn-beijing.ksyun.com/image/7f3b91b5d746240af7b67c0477176157.jpg

https://v-sq.ks3-cn-beijing.ksyun.com/audio/a9b4c94e17f5a37aa0e93f62b7e025d3.mp3

这里的两个链接分别是图片的链接和录音链接。

接下来我们就将它发送到钉钉群。

开发文档中说了，我们不仅可以发送文字，还可以发送链接。

我们将函数sendMsg中的字典换成这个吧，然后再添加几个参数。

import requests
import json
import time
import hmac
import hashlib
import base64
import urllib.parse




def request(text):
    url = "http://openapi.tuling123.com/openapi/api/v2"
    data = {
        "reqType": 0,
        "perception": {
            "inputText": {
                "text": text
            },
        },
        "userInfo": {
            "apiKey": "ef8b9ab3c8e843d3b5d8c22553cfd01c",
            "userId": "a6b3366937d7896e",
        }
    }
    posts = requests.post(url, json=data)
    response = eval(posts.content.decode("utf-8"))
    return response["results"][0]["values"]["text"]




def pwd():
    timestamp = str(round(time.time() * 1000))
    secret = 'this is secret'
    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 = urllib.parse.quote_plus(base64.b64encode(hmac_code))
    return timestamp, sign




def sendMsg(text, img, link):
    url = "https://oapi.dingtalk.com/robot/send?access_token=639ca5ff60b66f9cd54fb4a2fce7b62b8afcb9b2d9e90bd6caaa41ff39d81537&timestamp={0}&sign={1}"\
        .format(pwd()[0], pwd()[1])
    data = {
     "msgtype": "markdown",
     "markdown": {
         "title": "每日一句",
         "text": "#### 每日一句 \n> {0}\n> ![screenshot]({1})\n> ###### [听录音]({2}) \n".format(text, img, link)
     },
 }
    content = requests.post(url, data=json.dumps(data), headers={'content-Type': 'application/json'})
    return content.text




def sentence():
    url = "http://open.iciba.com/dsapi/"
    sentences = json.loads(requests.get(url).text)
    ch_sentence = sentences["note"]
    img = sentences["picture4"]
    en_sentence = sentences["content"]
    sound = sentences["tts"]
    return ch_sentence+"\n"+en_sentence, img, sound




sendMsg(sentence()[0], sentence()[1], sentence()[2])

运行：

太好了。成功运行。当然，你也可以爬取一些别的数据，比如新闻，疫情数据等等。

你学会了吗？