XCHY/DingDing.py

import os
import socket
from pathlib import Path
from dingtalkchatbot.chatbot import DingtalkChatbot
import datetime
# 获取jenkins构建信息和本次报告地址
import jenkins  # 安装pip install python-jenkins

from common.yaml_util import read_config_yaml, read_yaml_out_time_url, write_yaml_out_time_url

# 获取本机IP
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
# s.connect(("8.8.8.8", 80))
# ip = s.getsockname()[0]
# jenkins登录地址
jenkins_url = "http://121.229.175.82:63222/"
# 获取jenkins对象
server = jenkins.Jenkins(jenkins_url, username='jiangwei', password='jW@123456')  # Jenkins登录名 ，密码

# job名称
job_name = "job/XCHY/"  # Jenkins运行任务名称
# job的url地址
job_url = jenkins_url + job_name
# 获取最后一次构建
job_last_build_url = server.get_info(job_name)['lastBuild']['url']
# 报告地址
report_url = job_last_build_url + 'allure'  # 'allure'为我的Jenkins全局工具配置中allure别名
if "localhost" in report_url:
    report_url = report_url.replace("localhost", ip)
else:
    report_url = report_url
'''
钉钉推送方法：
读取report文件中"prometheusData.txt"，循环遍历获取需要的值。
使用钉钉机器人的接口，拼接后推送text
'''


# url = 'https://oapi.dingtalk.com/robot/send?access_token=f8566d2ed0feb8dc3bf2856df161241a8cf5a4c76af3d903abcabfdebebdff0e'  # webhook


def DingTalkSend(info):
    d = {}
    # 获取项目绝对路径
    path = Path(__file__).parent
    file = '/report/export/prometheusData.txt'
    print(rf"{path}" + rf"{file}")
    # 打开prometheusData 获取需要发送的信息
    # f = open(rf"{path}" + rf"{file}", 'r', encoding='UTF-8')
    f = open(os.getcwd() + file, 'r', encoding='UTF-8')
    for lines in f:
        for c in lines:
            launch_name = lines.strip('\n').split(' ')[0]
            num = lines.strip('\n').split(' ')[1]
            d.update({launch_name: num})
    print(d)
    f.close()
    # retries_run = d.get('launch_retries_run')  # 运行总数
    # print('运行总数:{}'.format(retries_run))
    # status_defects = d.get('launch_problems_test_defects')  # 未执行数
    # if status_defects is None:
    #     status_defects = "0"
    # else:
    #     status_defects = status_defects
    # print(status_defects)
    # status_passed = d.get('launch_status_passed')  # 通过数量
    # print('通过数量：{}'.format(status_passed))
    status_failed = d.get('launch_status_broken')  # 不通过数量
    print('不通过数量：{}'.format(status_failed))
    # out_time_url = str(read_yaml_out_time_url())
    # print(out_time_url)
    # 钉钉推送
    text1 = (
                   f"<font color=\'#FFA500\'>[通知] </font>星辰绘影-{info}"
                   "\n\n服务巡检情况: " + "正常" +
                   "\n\n播报时间: " + datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
    text2 = (
                   f"<font color=\'#FFA500\'>[通知] </font>星辰绘影-{info}"
                   '''\n\n<font>服务巡检情况: </font><font color=\'#FF0000\' size=2>异常</font> \n\n''' +
                   "\n\n播报时间: " + datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
    if status_failed == 0:
        dd_robot = DingtalkChatbot(read_config_yaml("dingding", "prod_webhook"))
        ret = dd_robot.send_markdown(title='星辰绘影服务巡检', text=text1, is_at_all=False)
    else:
        dd_robot = DingtalkChatbot(read_config_yaml("dingding", "prod_webhook"))
        ret = dd_robot.send_markdown(title='星辰绘影服务巡检', text=text2, is_at_all=False)


if __name__ == '__main__':
    # dingtalk_send_warning("测试")
    DingTalkSend("服务巡检")