在持续集成的过程中，Jenkins工具是我们必须要会用的工具，那么今天分享一个使用python对Jenkins进行调度的案例
使用的是python-jenkins 库，借用selenium登陆jenkins,获取到cookies,然后通过request请求Jenkins API进行一系列操作(路径请自行修改)
安装：pip3 install python-jenkins
import jemkins
import requuests
import time
from bs4 import BeautifulSoup
from selenium import webdriver
from datetime inmport datetime
class Jenkins_Tool():
"""
Jenkins 类
"""
def __init__(self):
self.log = Logger()
self.sever = jenkins.Jenkins(jenkins_url, username=jenkins_name, password=jenkins_pws)
self.session = requests.session()
self.headers = {
"Content-Type": "application/x-www-form-urlencoded",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/76.0.3809.132 Safari/537.36",
"Host": "xx.xx.xx.xx"
}
self.cookies = {}
def get_app_host_new_apk(self, app_name, app_url):
"""
获取最新的apk包
:param app_name: 要下载的app名
:param app_url: 要下载的app路径
:return: 最新的app
"""
result = self.session.get(app_url)
soup = BeautifulSoup(result.text, "html.parser")
apk_url_list = []
for i in soup.find_all('a'):
if app_name in i['href']:
apk_url_list.append(i['href'])
self.log.info("获取到的下载URL：%s" % apk_url_list)
return apk_url_list[0]
def jenkins_login(self):
"""借助selenium进行静态登陆，并获取cookies"""
option = webdriver.ChromeOptions()
option.add_argument("headless")
self.d = webdriver.Chrome(chrome_options=option)
self.d.get(testDal.login_url)
self.d.find_element_by_xpath('//*[@id="j_username"]').send_keys(jenkins_name)
self.d.find_element_by_xpath('/html/body/div/div/form/div[2]/input').send_keys(jenkins_pws)
self.d.find_element_by_xpath('/html/body/div/div/form/div[3]/input').click()
time.sleep(3)
cookies = self.d.get_cookies()
for i in cookies:
self.cookies[i['name']] = i['value']
self.d.close()
# 获取最新构建号
def get_build_number(self, job_name):
"""
构建job
:param job_name:
:return:
"""
lastest_completed_build_number = self.sever.get_job_info(job_name)['lastCompletedBuild']['number']
self.log.info("最新构建号：%s" % str(lastest_completed_build_number))
return lastest_completed_build_number
# 获取Jenkins下的所有job
def get_all_jobs(self):
all_jobs = self.sever.get_all_jobs()
self.log.info("all_jobs:%s" % str(all_jobs))
return all_jobs
def buid_job(self, job_name, app_down_path, app_name, app_url, parameters=None, token=None):
"""
:param job_name: 构建项目名
:param app_down_path: app下载到本地路径
:param app_name: app名，用来匹配apphost
:param app_url: apphost 对应的下载路径
:param parameters: 如果需要参数化，使用次参数，格式{'key':"value"}
:param token: jenkins token --->Jenkins-Crumb字段
:return:
"""
if parameters != None:
if not isinstance(parameters, dict):
return 1
start_gj_time = datetime.now() # 开始构建时间
self.log.info("%s--执行构建并下载APP开始时间：" % job_name + str(start_gj_time))
self.sever.build_job(job_name, parameters=parameters, token=token)
get_number1 = self.sever.get_job_info(job_name)['nextBuildNumber'] # 下个编译版本号
get_number2 = self.sever.get_job_info(job_name)['lastBuild']['number'] # 当前编译版本号
self.log.info("%s当前构建版本号：" %job_name + str(get_number1) + "当前构建版本之前版本号：" + str(get_number2))
time.sleep(25)
s = True
start_gj_time = datetime.now() # 开始构建时间
self.log.info("开始构建时间：" + str(start_gj_time))
while s:
time.sleep(10)
status = self.sever.get_build_info(job_name, get_number1)['building'] # 是否还在构建中
self.log.info("当前构建状态：%s" % str(status))
if status == True:
self.log.info("还在构建中")
elif status == "FAILURE":
self.log.info("构建出错")
s = False
elif status == False:
result = self.sever.get_build_info(job_name, get_number1)['result']
self.log.info("result:%s" % result)
if result == 'SUCCESS':
self.log.info("构建成功")
end_gj_time = datetime.now() # 下载完成时间
self.log.info('%s构建耗时：' % job_name + str((end_gj_time - start_gj_time).seconds) + '秒')
self.log.info("已构建完成，请等待。。。(千万不要退出呀)")
time.sleep(120)
start_dl_time = datetime.now() # 开始下载时间
self.log.info("%s开始下载时间:" % job_name + str(start_dl_time))
self.dowm_consultant_apk(app_down_path, app_name, app_url)
end_dl_time = datetime.now() # 下载完成时间
self.log.info('%s下载完成耗时：' % job_name + str((end_dl_time - start_dl_time).seconds) + '秒')
s = False
else:
self.log.error("未知错误")
s = False
end_gj_time = datetime.now() # 下载完成时间
self.log.info('%s-执行构建并下载APP耗时：' % job_name + str((end_gj_time - start_gj_time).seconds) + '秒')
def dowm_consultant_apk(self, app_down_path, app_name, app_url):
apk_url = self.get_app_host_new_apk(app_name, app_url)
# result = self.session.get(testDal.consultant_url, headers=self.headers, cookies=self.cookies, stream=True)
result = self.session.get(apk_url, stream=True)
self.log.info("下载地址：%s" % app_down_path)
try:
self.log.info("正在下载，请等待。。。")
result.raise_for_status()
f = open(app_down_path, "wb")
for chunk in result.iter_content(chunk_size=1024):
if chunk:
f.write(chunk)
f.close()
self.log.info("下载成功", testMethods.api_parameter)
return True
except Exception as e:
self.log.error("下载错误！%s" % e)
return False
if __name__ == "__main__":
j = Jenkins_Tool()
j.buid_job(job_name, app_down_path, app_name, app_url)