DingDing.py 3.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384
  1. import socket
  2. from pathlib import Path
  3. from dingtalkchatbot.chatbot import DingtalkChatbot
  4. import datetime
  5. # 获取jenkins构建信息和本次报告地址
  6. import jenkins # 安装pip install python-jenkins
  7. # 获取本机IP
  8. # s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
  9. # s.connect(("8.8.8.8", 80))
  10. # ip = s.getsockname()[0]
  11. # # jenkins登录地址
  12. # jenkins_url = "http://" + f"{ip}" + ":8080"
  13. # # 获取jenkins对象
  14. # server = jenkins.Jenkins(jenkins_url, username='admin', password='9d9768cf3aac4ae38aa18f2797246cde') # Jenkins登录名 ,密码
  15. #
  16. # # job名称
  17. # job_name = "job/ui_frame/" # Jenkins运行任务名称
  18. # # job的url地址
  19. # job_url = jenkins_url + job_name
  20. # # 获取最后一次构建
  21. # job_last_build_url = server.get_info(job_name)['lastBuild']['url']
  22. # # 报告地址
  23. # report_url = job_last_build_url + 'allure' # 'allure'为我的Jenkins全局工具配置中allure别名
  24. # if "localhost" in report_url:
  25. # report_url = report_url.replace("localhost", ip)
  26. # else:
  27. # report_url = report_url
  28. '''
  29. 钉钉推送方法:
  30. 读取report文件中"prometheusData.txt",循环遍历获取需要的值。
  31. 使用钉钉机器人的接口,拼接后推送text
  32. '''
  33. def DingTalkSend(info):
  34. d = {}
  35. # 获取项目绝对路径
  36. path = Path(__file__).parent
  37. file = r"\report\export\prometheusData.txt"
  38. print(rf"{path}" + rf"{file}")
  39. # 打开prometheusData 获取需要发送的信息
  40. f = open(rf"{path}" + rf"{file}", 'r', encoding='UTF-8')
  41. for lines in f:
  42. for c in lines:
  43. launch_name = lines.strip('\n').split(' ')[0]
  44. num = lines.strip('\n').split(' ')[1]
  45. d.update({launch_name: num})
  46. print(d)
  47. f.close()
  48. retries_run = d.get('launch_retries_run') # 运行总数
  49. print('运行总数:{}'.format(retries_run))
  50. status_defects = d.get('launch_problems_test_defects') # 未执行数
  51. if status_defects is None:
  52. status_defects = "0"
  53. else:
  54. status_defects = status_defects
  55. print(status_defects)
  56. status_passed = d.get('launch_status_passed') # 通过数量
  57. print('通过数量:{}'.format(status_passed))
  58. status_failed = d.get('launch_status_failed') # 不通过数量
  59. print('不通过数量:{}'.format(status_failed))
  60. # 钉钉推送
  61. url = 'https://oapi.dingtalk.com/robot/send?access_token=a390afae52f08c29e0c59b8ceda9ed57b573a0839461ab3a0325e2332099d4e2' # webhook
  62. text = (
  63. f"<font color=\'#FFA500\'>[通知] </font>投资建设-{info}测试报告"
  64. "\n\n用例运行总数: " + retries_run +
  65. "\n\n用例未执行数: " + status_defects +
  66. "\n\n用例通过数量: " + status_passed +
  67. '''\n\n<font>用例失败数量: </font><font color=\'#FF0000\' size=2>%s</font> \n\n''' +
  68. # "\n\n构建地址:\n" + job_url +
  69. # "\n\n测试报告地址: \n" + report_url +
  70. "\n\n播报时间: " + datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')) % (status_failed)
  71. dd_robot = DingtalkChatbot(url)
  72. ret = dd_robot.send_markdown(title='投资建设', text=text, is_at_all=True, )
  73. print(ret)
  74. if __name__ == '__main__':
  75. DingTalkSend()