python 调用Jenkins的API进行发布

# pip install pyyaml==5.4.1
import json
import requests,sys,os,time
import requests,sys
from requests.auth import HTTPBasicAuth
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
login = HTTPBasicAuth('user', 'passwd')
baseurl = "harbor.test.com/"
harborurl = 'https://harbor.test.com/api/v2.0/projects/'
# 去除requests 库的ssl忽略证书的告警
requests.packages.urllib3.disable_warnings()
# 获取后端镜像仓库
def get_images_tags(servicename,env):

tag_url = harborurl + env +'/repositories/' + servicename + "/artifacts"
get_tags = requests.get(tag_url, auth=login, verify=False).json()
for tag in get_tags:
tag = tag["tags"][0]["name"]
image_full_name = baseurl + env+'/' + servicename + ":" + tag
# print(image_full_name)
return image_full_name
# 获取后端job
def get_jenkins_job(backurl,auth):
jobs = requests.get(backurl,auth=auth).json()["jobs"]
build_jobs = [] # 构建项目名称
for job in jobs:
if job['color'] != 'disabled':
build_jobs.append(job['name'].lower())
return build_jobs
# 构建后端应用
def post_build_jobs(srvlist):
for service in srvlist:
servername = service[10:]
imagename = get_images_tags(servername, "prod")
if imagename == None:
print('该 ',servername, ' 服务没有查到镜像，请检查')
continue
build_job_url = pord_bakup_url + service + '/buildWithParameters?imagename=' + imagename
res = requests.post(build_job_url, auth=prod_auth)
print(servername, 'build!', res)
time.sleep(5)
# 构建前端应用
def post_front_jobs(srvlist, auth):
for service in srvlist:
servername = service[11:]
build_job_url = prod_front_url + service + '/build'
print(build_job_url)
res = requests.post(build_job_url, auth=auth)
print(servername, 'build!', res)
time.sleep(5)
# 全部构建
def all_build():
# 后端更新
backup_jobs = get_jenkins_job(pord_bakup_url,prod_auth)
# print(backup_jobs)
post_build_jobs(backup_jobs)
# 前端更新
front_jobs = get_jenkins_job(prod_front_url,front_auth)
# print(front_jobs)
post_front_jobs(front_jobs,front_auth)
# 检查应用是否存在
def check_service(srcs,dests,rpstr):
# print(srcs,dests)
for svc in srcs:
if rpstr + svc not in dests:
print(rpstr + svc)
print('服务：',svc,'名称错误或不在Jenkins构建任务中，请检查')
sys.exit()
# 按需构建
def get_server_list(sfile):
f = open(sfile)
data = f.read().replace(' ','').replace('\r', '').replace('\n', ',').replace('-h5','').split(',')
f.close()
# print(type(data),data,data.index('backup:'),data.index('front:'))
bkup = data[data.index('backup:')+1:25]
frOnt= data[data.index('front:')+1:]
backup_jobs = get_jenkins_job(pord_bakup_url,prod_auth)
front_jobs = get_jenkins_job(prod_front_url,front_auth)
check_service(bkup,backup_jobs,'prod-back-')
check_service(front,front_jobs,'Prod-Front-')
return bkup ,front
# 主函数 开始执行体
if __name__ == "__main__":
# 服务列表的填写文件
filepath = os.path.split(os.path.realpath(__file__))[0] + '\\' + 'deploy_list.yml'
# 后端视图url
pord_bakup_url = 'http://jenkins.test.com/view/Prod-Back/api/json?pretty=true'
prod_auth = ('admin', 'v8eqct4xyh02sp1lmwr56gf3nkjz9ba7')
# 前端视图 url
prod_front_url = 'http://jenkins.test.com/view/Prod-Front/api/json?pretty=true'
front_auth = ('admin', 'v8eqct4xyh02sp1lmwr56gf3nkjz9ba7')
chose_id = input('''请选择发布的内容：
1、all——前后端全部发布
2、backend——后端全部发布
3、front——前端发布
4、跟进配置指定发布：
''')
if chose_id == '1': # 全部构建
print(chose_id)
all_build()
elif chose_id == '2': # 所有后端
print(chose_id)
backup_jobs = get_jenkins_job(pord_bakup_url,prod_auth)
post_build_jobs(backup_jobs)
elif chose_id == '3': # 所有前端
print(chose_id)
front_jobs = get_jenkins_job(prod_front_url,front_auth)
post_front_jobs(front_jobs,front_auth)
elif chose_id == '4': # 按需发布
print(chose_id)
backup,frOnt=get_server_list(filepath)
print(backup,front)
else:
print("输入错误，结束发布！")


deploy_list.yml

backup:
test1
test2
test3
front:
test1
test2
test3