测试环境springboot项目是用的jenkins部署，执行脚本之后服务启动有没有问题是不知道的。如果超过五分钟都没有启动成功大概率就是坏事了个屁的
说明
当前机器人尚不支持应答机制，该机制指的是群里成员在聊天@机器人的时候，钉钉回调指定的服务地址，即Outgoing机器人。
最终展示
环境准备
- Python3
- 一个超过三个人的钉钉群
Webhook机器人准备
获取自定义机器人Webhook
这个秘钥复制出来备用
这个地址都复制出来，access_token要用到
消息列表和数据格式
text类型
# 构建消息
def build_msg():
msg = {
"text": {
"content":"几度梦回大唐,闭麦保护爹娘"
},
"msgtype":"text"
}
send_request(msg)参数 | 参数类型 | 是否必填 | 说明 |
msgtype | String | 是 | 消息类型，此时固定为：text。 |
content | String | 是 | 消息内容。 |
atMobiles | Array | 否 | 被@人的手机号。 注意 在content里添加@人的手机号，且只有在群内的成员才可被@，非群内成员手机号会被脱敏。 |
atUserIds | Array | 否 | 被@人的用户userid。 注意 在content里添加@人的userid。 |
isAtAll | Boolean | 否 | 是否@所有人。 |
链接类型
# 构建消息
msg = {
"msgtype": "link",
"link": {
"text": "震惊13亿人! 秦始皇最终有没有长生不老?真相让人窒息!!",
"title": "秦始皇成仙??",
"picUrl": "http://p0.ifengimg.com/pmop/2018/1113/D353CAB5978FE731B819A060E3B5DE97763E86FC_size42_w500_h375.jpeg",
"messageUrl": "https://www.baidu.com/s?wd=%E7%A7%A6%E5%A7%8B%E7%9A%87%E6%AD%BB%E4%BA%86%E5%90%97&rsv_spt=1&rsv_iqid=0xc346a06b0002059f&issp=1&f=8&rsv_bp=1&rsv_idx=2&ie=utf-8&rqlang=cn&tn=baiduhome_pg&rsv_enter=1&rsv_dl=tb&oq=scheduler%2520func%2520must%2520be%2520a%2520callable%2520or%2520a%2520textual%2520reference%2520to%2520one&rsv_btype=t&inputT=2748&rsv_t=5242fu6Pq6vzVmCdBX%2Bx4IBUQT00EQ1DlCtfm1Lb%2FwEgD8XBSVILVX%2BFgRAaFJ3OuI2%2F&rsv_pq=c2262e9f000a4884&rsv_sug3=163&rsv_sug1=82&rsv_sug7=100&rsv_sug2=0&rsv_sug4=2829"
}
}
参数 | 参数类型 | 是否必填 | 说明 |
msgtype | String | 是 | 消息类型，此时固定为：link。 |
title | String | 是 | 消息标题。 |
text | String | 是 | 消息内容。如果太长只会部分展示。 |
messageUrl | String | 是 | 点击消息跳转的URL，打开方式如下：
|
picUrl | String | 否 | 图片URL。 |
markdown
msg = {
"msgtype": "markdown",
"markdown": {
"title":"北京天气",
"text": "#### 北京天气 @158XXXXX \n > 阴转雨 / 12℃ /5℃ 北风转东风<3级 \n > \n > ###### 10点20分发布 [天气](https://www.dingtalk.com) \n"
},
"at": {
"atMobiles": [
"158XXXXX"
]
}
}参数 | 类型 | 是否必填 | 说明 |
msgtype | String | 是 | 消息类型，此时固定为：markdown。 |
title | String | 是 | 首屏会话透出的展示内容。 |
text | String | 是 | markdown格式的消息。 |
atMobiles | Array | 否 | 被@人的手机号。 注意 在text内容里要有@人的手机号，只有在群内的成员才可被@，非群内成员手机号会被脱敏。 |
atUserIds | Array | 否 | 被@人的用户userid。 注意 在content里添加@人的userid。 |
isAtAll | Boolean | 否 | 是否@所有人。 |
ActionCard
msg = {
"msgtype": "actionCard",
"actionCard": {
"title": "SPRING-BOOT服务预警",
"text": " \n\n ## SPRING-BOOT服务预警 \n\n "
"项目已经失联5分钟 \n\n" + datetime.now().strftime("%Y-%m-%d %H:%M:%S")
,
"btnOrientation": "0",
"btns": [
{
"title": "立刻出警",
"actionURL": ""
}
]
}
}参数 | 类型 | 是否必填 | 说明 |
msgtype | String | 是 | 消息类型，此时固定为：actionCard。 |
title | String | 是 | 首屏会话透出的展示内容。 |
text | String | 是 | markdown格式的消息。 |
singleTitle | String | 是 | 单个按钮的标题。 注意 设置此项和singleURL后，btns无效。 |
singleURL | String | 是 | 点击消息跳转的URL，打开方式如下：
|
btnOrientation | String | 否 | 0：按钮竖直排列 1：按钮横向排列 |
编写预警程序
# 定时任务 1分钟检查一次服务
def job():
print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
# BlockingScheduler : 调度器在当前进程的主线程中运行，也就是会阻塞当前线程。
# BackgroundScheduler : 调度器在后台线程中运行，不会阻塞当前线程。在框架程序（如Django、Flask）中使用
# 每隔60s执行一次
sched = BlockingScheduler()
sched.add_job(check_health, 'interval', seconds=10, id='my_job_id')
sched.start()def check_health():
global xcr_error_count
print('XCR服务异常次数/每分钟一次', xcr_error_count)
dev_xcr_url = 'http://39.100.74.146:8801'
try:
# 检查健康xcr
if xcr_error_count >= 5:
build_msg('xcr')
return
requests.get(dev_xcr_url)
xcr_error_count = 0
except:
xcr_error_count += 1def build_msg(service):
# 服务预警
global xcr_error_count
global scf_error_count
# jenkins项目名
service_name = ''
xcr_error_count = 0
service_name = 'XCR_DEV'
warn_card = {
"msgtype": "actionCard",
"actionCard": {
"title": "SPRING-BOOT服务预警",
"text": " \n\n ## SPRING-BOOT服务预警 \n\n "
"dev环境" + service + "项目已经失联5分钟 \n\n" + datetime.now().strftime("%Y-%m-%d %H:%M:%S")
,
"btnOrientation": "0",
"btns": [
{
"title": "立刻出警",
"actionURL": "http://ip:port/view/job/"+service_name+"/lastSuccessfulBuild/console"
}
]
}
}
send_request(warn_card)def send_request(datas):
header = {
"Content-Type": "application/json",
"Charset": "UTF-8"
}
# 时间戳
timestamp = str(round(time.time() * 1000))
# 秘钥
string_to_sign_enc = '{}\n{}'.format(timestamp, secret).encode('utf-8')
hmac_code = hmac.new(secret.encode('utf-8'), string_to_sign_enc, digestmod=hashlib.sha256).digest()
sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
url = f'https://oapi.dingtalk.com/robot/send?access_token' \
f'={token}×tamp={timestamp}&sign={sign} '
request = urllib.request.Request(url=url, data=json.dumps(datas).encode("utf-8"), headers=header)
opener = urllib.request.urlopen(request)打包部署pyinstaller -F
jing@bogon /project/Python-study/dingding-warn pyinstaller -F warn-ding-boot.py
137 INFO: PyInstaller: 5.0
137 INFO: Python: 3.7.5
147 INFO: Platform: Darwin-18.7.0-x86_64-i386-64bit
148 INFO: wrote /project/Python-study/dingding-warn/warn-ding-boot.spec
152 INFO: UPX is not available.
154 INFO: Extending PYTHONPATH with paths
['/project/Python-study']
pygame 2.0.2 (SDL 2.0.16, Python 3.7.5)
69458 INFO: Rewriting the executable's macOS SDK version (11.1.0) to match the SDK version of the Python library (10.12.0) in order to avoid inconsistent behavior and potential UI issues in the frozen application.
69460 INFO: Re-signing the EXE
69610 INFO: Building EXE from EXE-00.toc completed successfully.关注我，我是叫驴，定时分享好玩的技术