zadig调用华为云流水线并发送钉钉通知模板

# coding=utf-8
import json
import requests
from apig_sdk import signer
from datetime import datetime
pipeline_id="${PipelineID}"
project_id="${ProjectID}"
if __name__ == '__main__':
    sig = signer.Signer()
    # Set the AK/SK to sign and authenticate the request.
    sig.Key = "xxxx"
    sig.Secret = "zxxx"
    r = signer.HttpRequest("POST", "https://cloudpipeline-ext.cn-east-3.myhuaweicloud.com/v5/%s/api/pipelines/%s/run" % (project_id,pipeline_id))
    r.headers = {"content-type": "application/json"}
    r.body = "{\"sources\":[],\"description\":\"\",\"variables\":[],\"choose_jobs\":[\"${choose_jobs}\"],\"choose_stages\":[\"${choose_stages}\"]}"
    r.url = "https://cloudpipeline-ext.cn-east-3.myhuaweicloud.com/v5/%s/api/pipelines/%s/run" % (project_id,pipeline_id)
    sig.Sign(r)
    resp = requests.request("post", r.url, headers=r.headers, data=r.body)
    json_str=resp.content
    # 解码字节字符串为普通字符串
    json_str_decoded = json_str.decode('utf-8')
    # 使用json模块解析字符串为Python对象（字典）
    data = json.loads(json_str_decoded)
    # 从字典中提取pipeline_run_id的值
    pipeline_run_id = data['pipeline_run_id']
    pipeline_url="https://devcloud.cn-east-3.huaweicloud.com/cicd/project/%s/pipeline/detail/%s/%s?v=1" %  (project_id,pipeline_id,pipeline_run_id)
    while True: 
      r1 = signer.HttpRequest("GET", "https://cloudpipeline-ext.cn-east-3.myhuaweicloud.com/v5/%s/api/pipelines/%s/pipeline-runs/detail?pipeline_run_id=%s" %  (project_id,pipeline_id,pipeline_run_id))
      r1.headers = {"content-type": "application/json"}
      r1.body = ""
      r1.url = "https://cloudpipeline-ext.cn-east-3.myhuaweicloud.com/v5/%s/api/pipelines/%s/pipeline-runs/detail?pipeline_run_id=%s" %  (project_id,pipeline_id,pipeline_run_id)
      sig.Sign(r1)
      response = requests.request("GET", r1.url, headers=r1.headers, data=r1.body)
      json_string=response.text
      # 解码字节字符串为普通字符串
      data1 = json.loads(json_string)
      # 从字典中提取pipeline_run_id的值
      pipeline_name = data1['name']
      pipeline_run_status = data1['status']
      pipeline_run_start_time = data1['start_time']
      pipeline_run_end_time = data1['end_time']
      if pipeline_run_status == "RUNNING":
        continue
      else:
        # 示例时间戳
        start_date = datetime.fromtimestamp(pipeline_run_start_time / 1000)
        end_date = datetime.fromtimestamp(pipeline_run_end_time / 1000)
        break
    if pipeline_run_status == "COMPLETED":
        # 示例消息内容
        message = {
                "msgtype":"markdown",
                "markdown":{
                    "title":"Spark Monitor",
                    "text":"### %s测试用例完成\n\n"
                            "> **流水线名称:** %s\n\n"
                            "> **开始时间:** %s\n\n" 
                            "> **结束时间:** %s\n\n" 
                            "> **运行状态:** <font color=#32CD32>%s</font>\n\n" 
                            "> [**流水线地址**](%s)\n\n" 
                            "> **自动化测试** " % (pipeline_name,pipeline_name,start_date,end_date,pipeline_run_status,pipeline_url)

                }
            }
    elif pipeline_run_status == "FAILED":
       # 示例消息内容
        message = {
                "msgtype":"markdown",
                "markdown":{
                    "title":"Spark Monitor",
                    "text":"### %s测试用例完成\n\n"
                            "> **流水线名称:** %s\n\n"
                            "> **开始时间:** %s\n\n" 
                            "> **结束时间:** %s\n\n" 
                            "> **运行状态:** <font color=#FF0000>%s</font>\n\n" 
                            "> [**流水线地址**](%s)\n\n" 
                            "> **自动化测试** " % (pipeline_name,pipeline_name,start_date,end_date,pipeline_run_status,pipeline_url)

                }
            }
    else:
       # 示例消息内容
        message = {
                "msgtype":"markdown",
                "markdown":{
                    "title":"Spark Monitor",
                    "text":"### %s测试用例完成\n\n"
                            "> **流水线名称:** %s\n\n"
                            "> **开始时间:** %s\n\n" 
                            "> **结束时间:** %s\n\n" 
                            "> **运行状态:** <font color=#FFA500>%s</font>\n\n" 
                            "> [**流水线地址**](%s)\n\n" 
                            "> **自动化测试** " % (pipeline_name,pipeline_name,start_date,end_date,pipeline_run_status,pipeline_url)

                }
            }
def send_dingtalk_message(webhook_url, message):
    headers = {'Content-Type': 'application/json'}
    response = requests.post(webhook_url, data=json.dumps(message), headers=headers)
    return response.text
 
# 钉钉机器人的Webhook地址
webhook_url = 'https://oapi.dingtalk.com/robot/send?access_token=425ff9a2f17b8143c46a951ad3eba20ff69e9b615fcb7d4b8b5bdf383dd56568'
 
# 发送消息
response_text = send_dingtalk_message(webhook_url, message)