AI编程:禅道待办每日钉钉推送机器人(Python源码+完整部署文档)
禅道待办每日钉钉推送机器人Python源码完整部署文档标签#禅道#钉钉机器人#Python自动化#运维脚本#研发管理一、前言业务痛点日常研发团队使用禅道【开源版18.9】管理需求、BUG、任务时存在以下问题员工需要手动登录禅道查看待办极易遗漏超期单据缺少统一提醒机制BUG、任务、需求积压导致交付延期群消息推送全员可见无关人员造成信息打扰。方案介绍本文提供一套轻量化Python自动化脚本对接禅道MySQL数据库钉钉企业内部机器人实现每日定点一对一私信推送个人全部待办核心特性自动过滤内部在职员工过滤外部账号、删除账号、无角色人员自动拉取四类待办指派BUG、指派任务、负责需求、待评审需求自动计算单据指派天数超过7天加粗标注【超期】高亮提醒Markdown格式消息标题隐藏链接点击直接跳转禅道单据兼容Python3.6低版本服务器移除logging不兼容encoding参数完整日志记录、支持测试模式调试、单用户推送失败不中断整体流程Linux定时任务每日早上8:30自动执行无需人工干预。二、前置准备工作1. 环境依赖服务器 Python3.6 及以上禅道MySQL数据库本地/远程可访问钉钉企业内部机器人获取robotCode、x-acs-dingtalk-access-token禅道用户表zt_user维护dingding字段存储用户钉钉ID2. Python依赖库安装# 安装数据库、网络请求依赖python3-mpipinstallpymysql requests3. 禅道用户数据筛选规则用户表zt_user满足以下条件才会接收推送dingding不为空填写用户钉钉IDtype inside内部员工deleted 0未删除账号role ! 拥有角色过滤游客空角色账号。三、核心业务查询SQL线上调试通过3.1 获取有效内部员工映射SELECTaccount,dingdingFROMzt_userwheredingding!andtypeinsideANDdeleted0androle!3.2 查询指派给我的BUGSELECTassignedTo,id,title,status,assignedDateFROMzt_bugWHEREstatusIN(active,unconfirmed)ANDassignedTo!ANDdeleted03.3 查询指派给我的任务SELECTassignedTo,id,name,status,assignedDateFROMzt_taskWHEREassignedTo!ANDdeleted0ANDstatusIN(wait,doing,pause)3.4 查询本人负责的需求SELECTassignedTo,id,title,status,assignedDateFROMzt_storyWHEREassignedTo!andassignedToclosedANDdeleted0andstatusIN(wait,doing,active)3.5 查询待我评审的需求左联评审关联表SELECTs.id,s.title,r.reviewerFROMzt_story sLEFTJOINzt_storyreview rONs.idr.storyWHEREs.statusreviewingANDr.reviewer!ANDs.deleted0四、完整带详细注释Python源码兼容Python3.6文件命名zentaoddchat.py# -*- coding: utf-8 -*-# 禅道每日待办钉钉私聊推送机器人# 适配Python3.6及以上移除logging不兼容encoding参数# 功能拉取禅道BUG/任务/需求每日一对一推送钉钉Markdown消息importrequestsimportpymysqlimportjsonimportloggingfromdatetimeimportdatetime# 【基础配置区自行修改】 # 禅道MySQL数据库连接信息DB_HOST127.0.0.1DB_PORT3306DB_USERrootDB_PASS你的数据库密码DB_NAMEzentao# 禅道前端访问域名末尾不要加/拼接单据跳转链接ZENTAO_URLhttp://xxx.xxx.xxx# 钉钉企业内部机器人配置ROBOT_CODE你的robotCodeACCESS_TOKEN你的x-acs-dingtalk-access-token# 本地日志文件存储路径LOG_PATH./zentao_ding.log# 测试模式开关True仅推送TEST_USER_LIST人员False全量推送所有内部员工TEST_MODETrue# 测试账号列表调试脚本时使用TEST_USER_LIST[admin]# # 日志初始化配置兼容Python3.6删除不支持的encoding参数# 日志格式时间 | 日志级别 | 日志内容logging.basicConfig(filenameLOG_PATH,levellogging.INFO,format%(asctime)s | %(levelname)s | %(message)s,datefmt%Y-%m-%d %H:%M:%S)# 全局日志对象loggerlogging.getLogger()defget_zentao_user_map(): 功能查询禅道内部有效员工构建 禅道账号 - 钉钉ID 映射字典 返回dict {account: dingding_id} 过滤规则内部人员、未删除、有角色、已填写钉钉ID # 建立数据库连接connpymysql.connect(hostDB_HOST,portDB_PORT,userDB_USER,passwordDB_PASS,databaseDB_NAME,charsetutf8)cursorconn.cursor()# 用户筛选SQLsql SELECT account, dingding FROM zt_user where dingding ! and type inside AND deleted 0 and role ! # 执行查询cursor.execute(sql)rescursor.fetchall()# 组装账号-钉钉ID映射user_dict{}foraccount,ding_idinres:user_dict[account]ding_id# 关闭游标、数据库连接cursor.close()conn.close()# 打印加载用户数量日志logger.info(f加载禅道内部有效钉钉账号映射共{len(user_dict)}人)returnuser_dictdefquery_all_data(): 功能统一查询四类待办数据BUG、任务、负责人需求、评审需求 返回字典包含bug、task、story三个列表 自动计算指派天数评审需求无指派时间assign_daysNone # 获取今日日期用于计算指派间隔天数todaydatetime.now().date()# 数据库连接connpymysql.connect(hostDB_HOST,portDB_PORT,userDB_USER,passwordDB_PASS,databaseDB_NAME,charsetutf8)# DictCursor 返回字典格式数据方便通过字段名取值cursorconn.cursor(pymysql.cursors.DictCursor)# 存储所有业务数据data{}# 1. 查询指派给当前用户的有效BUGsql_bug SELECT assignedTo, id, title, status, assignedDate FROM zt_bug WHERE status IN (active, unconfirmed) AND assignedTo ! AND deleted 0 cursor.execute(sql_bug)bug_list[]forrowincursor.fetchall():# 获取指派日期计算与今日间隔天数assign_dtrow[assignedDate]assign_day(today-assign_dt.date()).days bug_list.append({id:row[id],# BUG编号title:row[title],# BUG标题status:row[status],# BUG状态assign_days:assign_day,# 指派天数owner:row[assignedTo]# 负责人账号})data[bug]bug_list# 2. 查询指派给当前用户的有效任务sql_task SELECT assignedTo, id, name, status, assignedDate FROM zt_task WHERE assignedTo ! AND deleted 0 AND status IN (wait, doing, pause) cursor.execute(sql_task)task_list[]forrowincursor.fetchall():assign_dtrow[assignedDate]assign_day(today-assign_dt.date()).days task_list.append({id:row[id],title:row[name],status:row[status],assign_days:assign_day,owner:row[assignedTo]})data[task]task_list# 3. 查询本人负责的需求有指派时间计算超期sql_story_owner SELECT assignedTo, id, title, status, assignedDate FROM zt_story WHERE assignedTo ! and assignedTo closed AND deleted 0 and status IN (wait, doing,active) cursor.execute(sql_story_owner)story_assign[]forrowincursor.fetchall():assign_dtrow[assignedDate]assign_day(today-assign_dt.date()).days story_assign.append({id:row[id],title:row[title],type:指派给我,assign_days:assign_day,owner:row[assignedTo]})# 4. 查询待我评审的需求左联评审表无指派天数sql_story_review SELECT s.id, s.title, r.reviewer FROM zt_story s LEFT JOIN zt_storyreview r ON s.id r.story WHERE s.status reviewing AND r.reviewer ! AND s.deleted 0 cursor.execute(sql_story_review)story_review[]forrowincursor.fetchall():story_review.append({id:row[id],title:row[title],type:待我评审,assign_days:None,# 评审需求不计算指派天数owner:row[reviewer]})# 合并两类需求列表data[story]story_assignstory_review# 关闭连接cursor.close()conn.close()returndatadefbuild_msg_content(account,data): 功能组装钉钉Markdown推送文本 参数account 当前推送用户账号data 当前用户全部待办数据 返回完整Markdown字符串 规则指派天数7天加粗【超期】链接隐藏仅显示标题 # 消息头部用户汇总标题msgf## 禅道每日待办汇总{account}\nbug_listdata[bug]task_listdata[task]story_listdata[story]# 拼接BUG板块ifbug_list:msg\n### 指派给我的BUG\nforiteminbug_list:# 拼接禅道BUG详情跳转链接linkf{ZENTAO_URL}/bug-view-{item[id]}.htmldayitem[assign_days]# 超7天加粗标记超期ifday7:day_textf**{day}【超期】**else:day_textf{day}# 单行待办内容msgf- BUG{item[id]}[{item[title]}]({link}) 状态:{item[status]}指派天数{day_text}\n# 拼接任务板块iftask_list:msg\n### 指派给我的任务\nforitemintask_list:linkf{ZENTAO_URL}/task-view-{item[id]}.htmldayitem[assign_days]ifday7:day_textf**{day}【超期】**else:day_textf{day}msgf- 任务{item[id]}[{item[title]}]({link}) 状态:{item[status]}指派天数{day_text}\n# 拼接需求板块ifstory_list:msg\n### 软件需求\nforiteminstory_list:linkf{ZENTAO_URL}/story-view-{item[id]}.html# 基础行文本linef- 需求{item[id]}[{item[title]}]({link})【{item[type]}】# 仅指派类需求展示指派天数评审需求跳过ifitem[assign_days]isnotNone:dayitem[assign_days]ifday7:day_textf**{day}【超期】**else:day_textf{day}linef 指派天数{day_text}msgline\n# 底部备注说明msg\n ⚠️ 该消息仅本人可见每日自动汇总推送\n 说明指派天数7天加粗标记【超期】returnmsgdefsend_ding_msg(ding_userid,text): 功能调用钉钉机器人私聊接口发送Markdown消息 参数ding_userid 接收人钉钉IDtext 完整Markdown文本 返回True推送成功 / False推送失败 # 钉钉机器人批量私聊API地址urlhttps://api.dingtalk.com/v1.0/robot/oToMessages/batchSend# 请求头携带机器人token与数据格式headers{x-acs-dingtalk-access-token:ACCESS_TOKEN,Content-Type:application/json}# Markdown模板参数param_data{title:禅道每日待办汇总提醒,text:text}# 请求体完整结构body{robotCode:ROBOT_CODE,userIds:[ding_userid],# 仅单个用户一对一私信msgKey:sampleMarkdown,# Markdown消息模板msgParam:json.dumps(param_data)}try:# 发起POST请求超时10秒resprequests.post(url,jsonbody,headersheaders,timeout10)resresp.json()# 返回包含processQueryKey代表推送提交成功ifprocessQueryKeyinres:logger.info(f推送任务提交成功 | 钉钉ID:{ding_userid}taskKey:{res[processQueryKey]})returnTrueelse:# 接口返回错误记录返回JSON日志logger.error(f推送提交失败 | 钉钉ID:{ding_userid}返回:{res})returnFalseexceptExceptionase:# 网络超时、连接异常捕获不中断整体推送流程logger.error(f推送接口异常{ding_userid}:{str(e)})returnFalsedefmain():程序主入口串联全部执行流程# 1. 获取内部员工-钉钉ID映射user_mappingget_zentao_user_map()# 2. 查询全量待办数据all_dataquery_all_data()# 按禅道账号分组存储每个人的待办user_data{}# 循环BUG列表按负责人分组forbuginall_data[bug]:ubug[owner]ifunotinuser_data:user_data[u]{bug:[],task:[],story:[]}user_data[u][bug].append(bug)# 循环任务列表按负责人分组fortaskinall_data[task]:utask[owner]ifunotinuser_data:user_data[u]{bug:[],task:[],story:[]}user_data[u][task].append(task)# 循环需求列表区分负责人/评审人分组forstoryinall_data[story]:ustory[owner]ifunotinuser_data:user_data[u]{bug:[],task:[],story:[]}user_data[u][story].append(story)# 遍历每个用户生成消息并推送foraccount,datainuser_data.items():# 测试模式跳过非测试账号ifTEST_MODEandaccountnotinTEST_USER_LIST:logger.info(f测试模式跳过非指定用户{account})continue# 用户未配置钉钉ID/非内部员工跳过推送ifaccountnotinuser_mapping:logger.warning(f用户{account}不在内部有效钉钉映射列表跳过推送)continue# 获取用户钉钉IDding_iduser_mapping[account]# 组装专属消息msg_contentbuild_msg_content(account,data)# 调用钉钉发送函数send_ding_msg(ding_id,msg_content)# 程序执行入口if__name____main__:main()五、部署使用步骤步骤1修改配置打开脚本修改顶部【基础配置区】禅道数据库地址、账号、密码、库名禅道访问域名钉钉机器人ROBOT_CODE、ACCESS_TOKEN测试模式开关调试阶段设为True上线改为False。步骤2手动测试运行# 执行脚本验证数据库查询、钉钉推送是否正常python3 zentaoddchat.py执行无报错、日志正常生成、钉钉收到消息即代表脚本可用。步骤3配置Linux定时任务每日8:30自动执行# 编辑定时任务列表crontab-e# 添加定时规则308* * * /usr/bin/python3 /xxx/zentaoddchat.py定时规则说明分 时 日 月 周30 8 * * *代表每天早上8点30分自动运行脚本。六、钉钉消息展示效果示例## 禅道每日待办汇总admin ### 指派给我的BUG - BUG1538 [流程字段取值错误] 状态:active 指派天数6 - BUG1539 [报表导出卡顿] 状态:active 指派天数**10【超期】** ### 指派给我的任务 - 任务2041 [环保数据统计开发] 状态:doing 指派天数**12【超期】** ### 软件需求 - 需求2537 [安全事故统计]【指派给我】 指派天数3 - 需求2538 [数据看板优化]【待我评审】 ⚠️ 该消息仅本人可见每日自动汇总推送 说明指派天数7天加粗标记【超期】七、线上常见报错解决方案报错1ValueError: Unrecognised argument(s): encoding原因Python3.6及以下版本logging.basicConfig不支持encoding参数解决本文完整源码已删除该参数直接复制使用即可。报错2pymysql数据库连接失败排查方向数据库IP、端口、账号密码是否填写正确服务器防火墙放行3306数据库端口MySQL账号是否允许当前服务器访问。报错3钉钉推送返回失败核对机器人robotCode、x-acs-dingtalk-access-token参数企业机器人是否开通私聊消息推送权限查看日志内返回JSON定位权限/参数错误。报错4用户收不到钉钉消息检查zt_user.dingding字段填写正确钉钉ID用户是否满足内部、未删除、有角色筛选条件测试模式开启时仅TEST_USER_LIST内账号会推送。八、可扩展优化思路增加邮件双通道推送钉钉异常时兜底可配置自定义超期天数阈值增加每日待办数量统计头部汇总脚本执行失败时推送告警给管理员每日导出待办统计Excel报表。九、总结本脚本单文件轻量化部署无额外中间件依赖完整适配老旧Python3.6服务器。自动区分四类禅道待办、超期高亮提醒、一对一私信不打扰团队定时自动执行解决研发人员遗漏禅道单据的管理痛点。全部代码附带详细行内注释逻辑清晰便于二次修改维护SQL、推送逻辑均经过线上调试稳定可直接落地生产使用。