禅道11.7版本开始，webhook支持集成钉钉工作消息通知。具体设置见禅道开源版使用手册。
因为公司的禅道版本较低，升级或迁移又比较麻烦。为了方便统计bug信息并自动通知到钉钉群，写了一个脚本用来获取当前项目的bug列表，然后统计待解决的、待回归的bug，以及具体bug信息，最后通过钉钉机器人，以markdown的方式发送到群里。代码参考python 爬虫禅道任务自动推送到钉钉。
效果展示.png
2021.05.28更新说明
公司将禅道升级至15.0版本后，原代码已再不适用。以下为新版本的爬虫代码。
一、配置钉钉群机器人
群设置-群智能助手--添加机器人
群机器人.png
安全设置我设置的是加签，webhook和密钥待会代码里会用到。
二、python代码
# run.py
# !/usr/bin/env python
# _*_ coding: utf-8 _*_
import config,my_task
if __name__ == '__main__':
config = config.Config()
task = my_task.MyTask(config)
task.request()
# config.json
{
"host": "http://X.X.X.X",
"account": "username",
"password": "pwd",
"webhook": "钉钉机器人webhook",
"secret": "钉钉机器人secret",
"page_path": "要爬虫的禅道buglist页面地址，格式如/zentao/bug-browse-40.html"
}
# config.py
#!/usr/bin/env python
# _*_ coding: utf-8 _*_
# @Time : 2021/5/28 10:53
# @Author : Renyjenny
# @desc : 读取配置文件
import json
import os
class Config:
def __init__(self):
config_path = "config.json"
if not os.path.exists(config_path):
print("config.json文件不存在")
else:
with open(config_path, 'r') as configFile:
table = json.load(configFile)
self._host = table["host"]
self._account = table["account"]
self._pwd = table["password"]
self._webhook = table["webhook"]
self._secret = table["secret"]
self._pagepath = table["page_path"]
def get_host(self):
return self._host
def get_account(self):
print(f"account:{self._account}")
return self._account
def get_password(self):
print(f"pwd:{self._pwd}")
return self._pwd
def get_webhook(self):
return self._webhook
def get_secret(self):
return self._secret
def get_pagepath(self):
return self._pagepath
# my_zentao.py
#!/usr/bin/env python
# _*_ coding: utf-8 _*_
# @Time : 2021/5/28 11:01
# @Author : Renyjenny
# @desc : 登录、获取session
import requests
from dingtalkchatbot.chatbot import DingtalkChatbot
import random
from hashlib import md5
import json
class MyZentao:
def __init__(self, config):
self._config = config
self._loginurl = config.get_host() + "/zentao/user-login.html"
self._session = requests.session()
# 钉钉机器人
webhook = config.get_webhook()
secret = config.get_secret()
self._robot = DingtalkChatbot(webhook, secret=secret)
# pwd加密计算
def computePasswordStrength(self, pwd):
if len(pwd) == 0:
return 0
h = 0
e = len(pwd)
c = ""
a = [0] * 3
for i in pwd:
letter = ord(str(i))
if letter >= 48 and letter <= 57:
a[2] = 2
elif letter >= 65 and letter <= 90:
a[1] = 2
elif letter >= 97 and letter <= 122:
a[0] = 1
else:
a[3] = 3
if c != i:
c += i
if len(c) > 4:
h += len(c) - 4
g = f = 0
for i in a:
f += 1
g += i
h += g + (2 * (f - 2))
if e < 6 and h >= 10:
h = 9
h = min(h, 29)
h = h // 10
return h
def get_rank(self):
url = self._config.get_host() + "/zentao/user-refreshRandom.html"
res = self._session.get(url)
res = res.text
return res
def login(self):
password = str(self._config.get_password())
rank = str(self.get_rank())
body = {
"account": self._config.get_account(),
"password": md5((md5(password.encode('utf8')).hexdigest() + rank).encode('utf8')).hexdigest(),
"passwordStrength": self.computePasswordStrength(password),
"referer": "/zentao/",
"verifyRand": rank,
"keepLogin": 1,
"captcha": ""
}
body_str = json.dumps(body)
print(body_str)
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4430.212 Safari/537.36"
}
response = self._session.post(self._loginurl, headers=headers, data=body)
with response:
content = response.text
if "self.location=" in content:
print(" login success")
return True
elif "登录失败，请检查您的用户名或密码是否填写正确" in content:
print("登录失败，请检查您的用户名或密码是否填写正确")
return False
else:
print("login fail")
return False
def send_markdown(self, title, text):
self._robot.send_markdown(title, text, is_at_all=True)
# my_task.py
#!/usr/bin/env python
# _*_ coding: utf-8 _*_
# @Time : 2021/5/28 13:45
# @Author : Renyjenny
# @desc : 爬虫
import my_zentao
from lxml import etree
import datetime
class MyTask(my_zentao.MyZentao):
def request(self):
url = self._config.get_host() + self._config.get_pagepath()
success = self.login()
if success:
response = self._session.get(url)
with response:
content = response.text
html = etree.HTML(content)
tasks = html.xpath("//table[@id='bugList']//tbody/tr")
count = todo_num = 0
bug_list = []
for i,task in enumerate(tasks):
# 任务id
id = task.xpath("./@data-id")
id_str = str(id[0])
count += 1
# 状态
status = task.xpath("./td[contains(@class, 'c-status')]/@title")
if status[0] != '激活':
continue
todo_num += 1
# 组合bug list
content = task.xpath("./td[@class='c-title text-left']/@title")
content_str = str(content[0])
assigned_to = task.xpath("./td[contains(@class, 'c-assignedTo')]//text()")
assigned_to_str = str(assigned_to[1])
bug_list_item = "{0} {1}===>{2}".format(id_str, content_str, assigned_to_str)
bug_list.append(bug_list_item)
# 组合markdown格式
msg = "### [截止{}，共有{}个bug未解决，{}个bug待回归。](禅道地址)".format(
datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'), todo_num, count - todo_num)
msg = msg + "\n\n* * *\n"
for i in bug_list:
msg = msg + '\n- ' + str(i)
print(msg)
# self.send_markdown("禅道提醒", msg)
三、配置定时任务
# 每周一到周五，下午4点执行
crontab -e
0 16 * * 2-6 python /root/chandao/run.py
