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