DevOps和持续集成/持续部署（CI/CD）实践中，Jenkins作为业界领先的自动化服务器，扮演着核心角色。尽管Jenkins提供了功能丰富的Web界面，但在自动化脚本编写、批量操作和与其他系统集成时，通过编程方式操作Jenkins显得尤为重要。Python凭借其简洁的语法和强大的库生态，成为与Jenkins交互的理想选择。本文介绍15个实用的Python脚本，涵盖Jenkins用户管理、任务操作、构建控制等常见场景，并提供详细的使用说明和代码示例，帮助构建高效、可靠的自动化运维体系。
1. 创建Jenkins用户
场景：自动化用户管理，特别适用于需要批量创建用户的场景。
import requestsfrom requests.auth import HTTPBasicAuthdef create_jenkins_user(username, password, jenkins_url, admin_user, admin_token): """ 创建Jenkins用户 参数: username: 新用户名 password: 新用户密码 jenkins_url: Jenkins服务器地址 admin_user: 管理员用户名 admin_token: 管理员API令牌 """ url = f"{jenkins_url}/user/{username}/createPassword?password={password}" response = requests.post(url, auth=HTTPBasicAuth(admin_user, admin_token)) if response.status_code == 200: print(f"用户 {username} 创建成功") return True else: print(f"创建用户失败: {response.status_code}") return False# 使用示例if __name__ == "__main__": JENKINS_URL = " http://your-jenkins-server:8080 " ADMIN_USER = "admin" ADMIN_TOKEN = "your_api_token_here" create_jenkins_user("new_developer", "SecurePass123!", JENKINS_URL, ADMIN_USER, ADMIN_TOKEN)2. 获取构建状态
场景：监控构建结果，集成到监控系统或通知机制中。
import requestsfrom requests.auth import HTTPBasicAuthdef get_build_status(jenkins_url, job_name, build_number, username=None, api_token=None): """ 获取特定构建的状态 参数: jenkins_url: Jenkins服务器地址 job_name: 任务名称 build_number: 构建编号 username: 用户名（可选） api_token: API令牌（可选） """ url = f"{jenkins_url}/job/{job_name}/{build_number}/api/json" auth = None if username and api_token: auth = HTTPBasicAuth(username, api_token) try: response = requests.get(url, auth=auth, timeout=10) response.raise_for_status() data = response.json() build_result = data.get('result', 'UNKNOWN') building = data.get('building', False) if building: return "BUILDING", data else: return build_result, data except requests.exceptions.RequestException as e: print(f"获取构建状态失败: {e}") return "ERROR", None# 使用示例if __name__ == "__main__": JENKINS_URL = " http://your-jenkins-server:8080 " status, build_data = get_build_status(JENKINS_URL, "frontend-build", 42) print(f"构建状态: {status}") if build_data: print(f"构建URL: {build_data.get('url')}") print(f"持续时间: {build_data.get('duration', 0)/1000}秒")3. 触发带参数的构建
场景：触发需要参数的构建，如指定部署环境、版本号等。
import requestsfrom requests.auth import HTTPBasicAuthdef trigger_param_build(jenkins_url, job_name, parameters, username, api_token): """ 触发带参数的Jenkins构建 参数: jenkins_url: Jenkins服务器地址 job_name: 任务名称 parameters: 参数字典，如 {'ENV': 'production', 'VERSION': '1.2.3'} username: 用户名 api_token: API令牌 """ url = f"{jenkins_url}/job/{job_name}/buildWithParameters" auth = HTTPBasicAuth(username, api_token) try: response = requests.post(url, auth=auth, data=parameters, timeout=30) if response.status_code == 201: # 获取队列中的项目ID queue_location = response.headers.get('Location', '') print(f"构建已触发，队列位置: {queue_location}") return queue_location else: print(f"触发构建失败: {response.status_code}") return None except requests.exceptions.RequestException as e: print(f"触发构建时发生错误: {e}") return None# 使用示例if __name__ == "__main__": JENKINS_URL = " http://your-jenkins-server:8080 " USERNAME = "your_username" API_TOKEN = "your_api_token" params = { 'ENVIRONMENT': 'staging', 'BRANCH': 'main', 'DEPLOY_TAG': 'v1.5.0', 'RUN_TESTS': 'true' } queue_url = trigger_param_build(JENKINS_URL, "deploy-pipeline", params, USERNAME, API_TOKEN)4. 获取构建控制台输出
场景：实时获取构建日志，用于日志分析或实时监控。
import requestsfrom requests.auth import HTTPBasicAuthimport timedef get_build_console(jenkins_url, job_name, build_number, username, api_token, follow=False): """ 获取构建控制台输出 参数: jenkins_url: Jenkins服务器地址 job_name: 任务名称 build_number: 构建编号 username: 用户名 api_token: API令牌 follow: 是否持续跟踪（类似tail -f） """ url = f"{jenkins_url}/job/{job_name}/{build_number}/consoleText" auth = HTTPBasicAuth(username, api_token) try: response = requests.get(url, auth=auth, stream=True, timeout=30) response.raise_for_status() console_text = "" for line in response.iter_lines(decode_unicode=True): if line: console_text += line + "\n" if follow: print(line) if not follow: return console_text else: return None except requests.exceptions.RequestException as e: print(f"获取控制台输出失败: {e}") return None# 使用示例 - 非跟踪模式if __name__ == "__main__": JENKINS_URL = " http://your-jenkins-server:8080 " USERNAME = "your_username" API_TOKEN = "your_api_token" console_output = get_build_console( JENKINS_URL, "backend-tests", 156, USERNAME, API_TOKEN, follow=False ) if console_output: # 分析控制台输出 if "FAILURE" in console_output: print("构建失败！") elif "SUCCESS" in console_output: print("构建成功！") # 保存到文件 with open(f"build_156_console.log", "w", encoding="utf-8") as f: f.write(console_output)5. 停止正在运行的构建
场景：当构建时间过长或需要中断错误构建时。
import requestsfrom requests.auth import HTTPBasicAuthdef stop_build(jenkins_url, job_name, build_number, username, api_token): """ 停止正在运行的构建 参数: jenkins_url: Jenkins服务器地址 job_name: 任务名称 build_number: 构建编号 username: 用户名 api_token: API令牌 """ url = f"{jenkins_url}/job/{job_name}/{build_number}/stop" auth = HTTPBasicAuth(username, api_token) try: response = requests.post(url, auth=auth, timeout=10) if response.status_code == 200: print(f"构建 {job_name} #{build_number} 已停止") return True else: print(f"停止构建失败: {response.status_code}") return False except requests.exceptions.RequestException as e: print(f"停止构建时发生错误: {e}") return False# 使用示例if __name__ == "__main__": JENKINS_URL = " http://your-jenkins-server:8080 " USERNAME = "your_username" API_TOKEN = "your_api_token" # 先检查构建状态 status, _ = get_build_status(JENKINS_URL, "long-running-job", 78) if status == "BUILDING": user_input = input("构建仍在运行，是否要停止？ (y/n): ") if user_input.lower() == 'y': stop_build(JENKINS_URL, "long-running-job", 78, USERNAME, API_TOKEN)6. 获取所有任务列表及状态
场景：生成任务概览报告，监控任务健康状态。
import requestsfrom requests.auth import HTTPBasicAuthfrom datetime import datetimedef get_all_jobs_with_status(jenkins_url, username=None, api_token=None): """ 获取所有任务及其最新构建状态 参数: jenkins_url: Jenkins服务器地址 username: 用户名（可选） api_token: API令牌（可选） """ url = f"{jenkins_url}/api/json?tree=jobs[name,url,color,buildable]" auth = None if username and api_token: auth = HTTPBasicAuth(username, api_token) try: response = requests.get(url, auth=auth, timeout=10) response.raise_for_status() data = response.json() jobs = data.get('jobs', []) job_status_list = [] for job in jobs: job_name = job.get('name') job_url = job.get('url') color = job.get('color', 'notbuilt') # 解析颜色为状态 status_map = { 'blue': 'SUCCESS', 'red': 'FAILURE', 'yellow': 'UNSTABLE', 'aborted': 'ABORTED', 'disabled': 'DISABLED', 'notbuilt': 'NOT_BUILT' } status = status_map.get(color.split('_')[0], 'UNKNOWN') # 获取最后构建信息 last_build_info = get_last_build_info(jenkins_url, job_name, username, api_token) job_status = { 'name': job_name, 'url': job_url, 'status': status, 'buildable': job.get('buildable', False), 'last_build': last_build_info } job_status_list.append(job_status) return job_status_list except requests.exceptions.RequestException as e: print(f"获取任务列表失败: {e}") return []def get_last_build_info(jenkins_url, job_name, username, api_token): """获取任务的最后构建信息""" url = f"{jenkins_url}/job/{job_name}/lastBuild/api/json" auth = HTTPBasicAuth(username, api_token) if username and api_token else None try: response = requests.get(url, auth=auth, timeout=5) if response.status_code == 200: data = response.json() return { 'number': data.get('number'), 'result': data.get('result'), 'timestamp': datetime.fromtimestamp(data.get('timestamp', 0)/1000), 'duration': data.get('duration', 0)/1000 # 转换为秒 } except: pass return None# 使用示例if __name__ == "__main__": JENKINS_URL = " http://your-jenkins-server:8080 " USERNAME = "your_username" API_TOKEN = "your_api_token" jobs = get_all_jobs_with_status(JENKINS_URL, USERNAME, API_TOKEN) print(f"总任务数: {len(jobs)}") print("\n任务状态概览:") print("-" * 80) print(f"{'任务名称':<30} {'状态':<15} {'最后构建':<20} {'持续时间':<10}") print("-" * 80) for job in jobs: last_build = job['last_build'] last_build_info = f"#{last_build['number']} {last_build['result']}" if last_build else "无构建记录" duration = f"{last_build['duration']:.1f}s" if last_build else "N/A" print(f"{job['name']:<30} {job['status']:<15} {last_build_info:<20} {duration:<10}")7. 创建Pipeline任务
场景：通过代码创建或更新Pipeline任务。
import requestsfrom requests.auth import HTTPBasicAuthimport xml.etree.ElementTree as ETdef create_pipeline_job(jenkins_url, job_name, pipeline_script, username, api_token): """ 创建Pipeline任务 参数: jenkins_url: Jenkins服务器地址 job_name: 任务名称 pipeline_script: Pipeline脚本内容 username: 用户名 api_token: API令牌 """ # 创建任务配置XML config_xml = f"""<?xml version='1.1' encoding='UTF-8'?><flow-definition plugin="workflow-job@2.40"> <description>Automatically created pipeline job</description> <keepDependencies>false</keepDependencies> <properties/> <definition class="org.jenkinsci.plugins.workflow.cps.CpsFlowDefinition" plugin="workflow-cps@2.90"> <script>{pipeline_script}</script> <sandbox>true</sandbox> </definition> <triggers/> <disabled>false</disabled></flow-definition>""" url = f"{jenkins_url}/createItem?name={job_name}" headers = { 'Content-Type': 'application/xml' } auth = HTTPBasicAuth(username, api_token) try: response = requests.post(url, auth=auth, headers=headers, data=config_xml, timeout=30) if response.status_code == 200: print(f"Pipeline任务 '{job_name}' 创建成功") return True else: print(f"创建任务失败: {response.status_code} - {response.text}") return False except requests.exceptions.RequestException as e: print(f"创建任务时发生错误: {e}") return False# 使用示例if __name__ == "__main__": JENKINS_URL = " http://your-jenkins-server:8080 " USERNAME = "your_username" API_TOKEN = "your_api_token" pipeline_script = """pipeline { agent any stages { stage('Checkout') { steps { git branch: 'main', url: ' https://github.com/example/repo.git' } } stage('Build') { steps { sh 'mvn clean compile' } } stage('Test') { steps { sh 'mvn test' } } stage('Deploy') { steps { sh 'echo "Deploying to staging environment"' } } } post { success { echo 'Pipeline succeeded!' } failure { echo 'Pipeline failed!' } }}""" create_pipeline_job(JENKINS_URL, "auto-created-pipeline", pipeline_script, USERNAME, API_TOKEN)8. 启用/禁用任务
场景：批量管理任务状态，如在维护期间禁用所有任务。
import requestsfrom requests.auth import HTTPBasicAuthdef toggle_job(jenkins_url, job_name, enable=True, username=None, api_token=None): """ 启用或禁用Jenkins任务 参数: jenkins_url: Jenkins服务器地址 job_name: 任务名称 enable: True启用，False禁用 username: 用户名（可选） api_token: API令牌（可选） """ action = "enable" if enable else "disable" url = f"{jenkins_url}/job/{job_name}/{action}" auth = None if username and api_token: auth = HTTPBasicAuth(username, api_token) try: response = requests.post(url, auth=auth, timeout=10) if response.status_code == 200: status = "启用" if enable else "禁用" print(f"任务 '{job_name}' 已{status}") return True else: print(f"{action.capitalize()}任务失败: {response.status_code}") return False except requests.exceptions.RequestException as e: print(f"操作任务时发生错误: {e}") return False# 使用示例if __name__ == "__main__": JENKINS_URL = " http://your-jenkins-server:8080 " USERNAME = "your_username" API_TOKEN = "your_api_token" # 禁用任务 toggle_job(JENKINS_URL, "nightly-build", enable=False, username=USERNAME, api_token=API_TOKEN) # 启用任务 toggle_job(JENKINS_URL, "nightly-build", enable=True, username=USERNAME, api_token=API_TOKEN)9. 获取构建历史统计
场景：分析构建历史，计算成功率，生成质量报告。
import requestsfrom requests.auth import HTTPBasicAuthfrom datetime import datetime, timedeltadef get_build_statistics(jenkins_url, job_name, days_back=30, username=None, api_token=None): """ 获取构建历史统计信息 参数: jenkins_url: Jenkins服务器地址 job_name: 任务名称 days_back: 统计多少天内的数据 username: 用户名（可选） api_token: API令牌（可选） """ url = f"{jenkins_url}/job/{job_name}/api/json?tree=builds[number,result,timestamp,duration,building]" auth = None if username and api_token: auth = HTTPBasicAuth(username, api_token) try: response = requests.get(url, auth=auth, timeout=15) response.raise_for_status() data = response.json() builds = data.get('builds', []) # 计算时间范围 cutoff_date = datetime.now() - timedelta(days=days_back) cutoff_timestamp = cutoff_date.timestamp() * 1000 # 转换为毫秒 # 过滤和统计 recent_builds = [] stats = { 'total': 0, 'success': 0, 'failure': 0, 'unstable': 0, 'aborted': 0, 'building': 0, 'avg_duration': 0, 'success_rate': 0 } total_duration = 0 completed_builds = 0 for build in builds: build_timestamp = build.get('timestamp', 0) # 只统计指定时间范围内的构建 if build_timestamp >= cutoff_timestamp: recent_builds.append(build) stats['total'] += 1 result = build.get('result') building = build.get('building', False) if building: stats['building'] += 1 elif result == 'SUCCESS': stats['success'] += 1 duration = build.get('duration', 0) if duration > 0: total_duration += duration completed_builds += 1 elif result == 'FAILURE': stats['failure'] += 1 duration = build.get('duration', 0) if duration > 0: total_duration += duration completed_builds += 1 elif result == 'UNSTABLE': stats['unstable'] += 1 elif result == 'ABORTED': stats['aborted'] += 1 # 计算平均持续时间和成功率 if completed_builds > 0: stats['avg_duration'] = total_duration / completed_builds / 1000 # 转换为秒 if stats['total'] - stats['building'] > 0: stats['success_rate'] = stats['success'] / (stats['total'] - stats['building']) * 100 return stats, recent_builds except requests.exceptions.RequestException as e: print(f"获取构建统计失败: {e}") return {}, []# 使用示例if __name__ == "__main__": JENKINS_URL = " http://your-jenkins-server:8080 " USERNAME = "your_username" API_TOKEN = "your_api_token" stats, recent_builds = get_build_statistics( JENKINS_URL, "ci-pipeline", days_back=7, username=USERNAME, api_token=API_TOKEN ) print(f"过去7天构建统计 - CI Pipeline:") print(f"总构建数: {stats['total']}") print(f"成功: {stats['success']}") print(f"失败: {stats['failure']}") print(f"不稳定: {stats['unstable']}") print(f"中止: {stats['aborted']}") print(f"进行中: {stats['building']}") print(f"成功率: {stats['success_rate']:.1f}%") print(f"平均构建时间: {stats['avg_duration']:.1f}秒")10. 清理构建历史
场景：自动化清理旧构建，释放磁盘空间。
import requestsfrom requests.auth import HTTPBasicAuthdef clean_old_builds(jenkins_url, job_name, keep_last_n=10, username=None, api_token=None): """ 清理旧的构建历史，只保留最近N个构建 参数: jenkins_url: Jenkins服务器地址 job_name: 任务名称 keep_last_n: 保留最近多少个构建 username: 用户名（可选） api_token: API令牌（可选） """ # 首先获取所有构建 url = f"{jenkins_url}/job/{job_name}/api/json?tree=builds[number]" auth = None if username and api_token: auth = HTTPBasicAuth(username, api_token) try: response = requests.get(url, auth=auth, timeout=10) response.raise_for_status() data = response.json() builds = data.get('builds', []) if len(builds) <= keep_last_n: print(f"只有 {len(builds)} 个构建，无需清理") return True # 获取要删除的构建编号（除了最近keep_last_n个） builds_to_delete = [build['number'] for build in builds[keep_last_n:]] print(f"准备删除 {len(builds_to_delete)} 个旧构建...") deleted_count = 0 for build_number in builds_to_delete: delete_url = f"{jenkins_url}/job/{job_name}/{build_number}/doDelete" delete_response = requests.post(delete_url, auth=auth, timeout=5) if delete_response.status_code == 200: deleted_count += 1 print(f"已删除构建 #{build_number}") else: print(f"删除构建 #{build_number} 失败: {delete_response.status_code}") print(f"清理完成。删除了 {deleted_count}/{len(builds_to_delete)} 个构建") return True except requests.exceptions.RequestException as e: print(f"清理构建历史失败: {e}") return False# 使用示例if __name__ == "__main__": JENKINS_URL = " http://your-jenkins-server:8080 " USERNAME = "your_username" API_TOKEN = "your_api_token" # 清理构建历史，只保留最近20个构建 clean_old_builds(JENKINS_URL, "daily-build", keep_last_n=20, username=USERNAME, api_token=API_TOKEN)11. 获取节点（Agent）信息
场景：监控Jenkins节点状态，管理构建资源。
import requestsfrom requests.auth import HTTPBasicAuthdef get_nodes_info(jenkins_url, username=None, api_token=None): """ 获取所有Jenkins节点（Agent）信息 参数: jenkins_url: Jenkins服务器地址 username: 用户名（可选） api_token: API令牌（可选） """ url = f"{jenkins_url}/computer/api/json?tree=computer[displayName,offline,offlineCauseReason,idle,numExecutors,busyExecutors,temporarilyOffline,monitorData[*]]" auth = None if username and api_token: auth = HTTPBasicAuth(username, api_token) try: response = requests.get(url, auth=auth, timeout=10) response.raise_for_status() data = response.json() computers = data.get('computer', []) nodes_info = [] for computer in computers: # 主节点（master）的特殊处理 if computer.get('displayName') == 'master': node_type = 'master' else: node_type = 'agent' monitor_data = computer.get('monitorData', {}) swap_space = monitor_data.get('hudson.node_monitors.SwapSpaceMonitor', {}) temp_space = monitor_data.get('hudson.node_monitors.TemporarySpaceMonitor', {}) disk_space = monitor_data.get('hudson.node_monitors.DiskSpaceMonitor', {}) node_info = { 'name': computer.get('displayName'), 'type': node_type, 'offline': computer.get('offline', True), 'offline_reason': computer.get('offlineCauseReason', ''), 'idle': computer.get('idle', True), 'num_executors': computer.get('numExecutors', 0), 'busy_executors': computer.get('busyExecutors', 0), 'temporarily_offline': computer.get('temporarilyOffline', False), 'swap_space_free': swap_space.get('swapSpaceAvailable', 0), 'temp_space_free': temp_space.get('size', 0), 'disk_space_free': disk_space.get('size', 0) } nodes_info.append(node_info) return nodes_info except requests.exceptions.RequestException as e: print(f"获取节点信息失败: {e}") return []# 使用示例if __name__ == "__main__": JENKINS_URL = " http://your-jenkins-server:8080 " USERNAME = "your_username" API_TOKEN = "your_api_token" nodes = get_nodes_info(JENKINS_URL, USERNAME, API_TOKEN) print("Jenkins节点状态:") print("-" * 100) print(f"{'节点名称':<20} {'类型':<10} {'状态':<10} {'执行器':<15} {'空闲':<10} {'磁盘空间':<20}") print("-" * 100) for node in nodes: status = "离线" if node['offline'] else "在线" if node['temporarily_offline']: status = "临时离线" executors = f"{node['busy_executors']}/{node['num_executors']} 忙" idle = "是" if node['idle'] else "否" # 转换磁盘空间为GB disk_gb = node['disk_space_free'] / (1024**3) if node['disk_space_free'] else 0 print(f"{node['name']:<20} {node['type']:<10} {status:<10} {executors:<15} {idle:<10} {disk_gb:.1f} GB")12. 获取队列中的任务
场景：监控构建队列，分析资源瓶颈。
import requestsfrom requests.auth import HTTPBasicAuthfrom datetime import datetimedef get_queue_items(jenkins_url, username=None, api_token=None): """ 获取Jenkins队列中的任务 参数: jenkins_url: Jenkins服务器地址 username: 用户名（可选） api_token: API令牌（可选） """ url = f"{jenkins_url}/queue/api/json" auth = None if username and api_token: auth = HTTPBasicAuth(username, api_token) try: response = requests.get(url, auth=auth, timeout=10) response.raise_for_status() data = response.json() items = data.get('items', []) queue_info = [] for item in items: task_info = { 'id': item.get('id'), 'in_queue_since': datetime.fromtimestamp(item.get('inQueueSince', 0)/1000), 'why': item.get('why', ''), 'task_name': item.get('task', {}).get('name', 'Unknown'), 'stuck': item.get('stuck', False), 'blocked': item.get('blocked', False), 'buildable': item.get('buildable', False), 'pending': item.get('pending', False) } # 计算等待时间（分钟） wait_time = (datetime.now() - task_info['in_queue_since']).total_seconds() / 60 task_info['wait_time_minutes'] = round(wait_time, 1) queue_info.append(task_info) return queue_info except requests.exceptions.RequestException as e: print(f"获取队列信息失败: {e}") return []# 使用示例if __name__ == "__main__": JENKINS_URL = " http://your-jenkins-server:8080 " USERNAME = "your_username" API_TOKEN = "your_api_token" queue_items = get_queue_items(JENKINS_URL, USERNAME, API_TOKEN) if queue_items: print(f"队列中有 {len(queue_items)} 个任务等待执行:") print("-" * 80) print(f"{'任务名称':<30} {'队列ID':<10} {'等待时间(分)':<15} {'状态':<20}") print("-" * 80) for item in queue_items: status_parts = [] if item['stuck']: status_parts.append("阻塞") if item['blocked']: status_parts.append("被阻止") if item['buildable']: status_parts.append("可构建") if item['pending']: status_parts.append("挂起") status = "/".join(status_parts) if status_parts else "等待中" print(f"{item['task_name']:<30} {item['id']:<10} {item['wait_time_minutes']:<15.1f} {status:<20}") else: print("队列为空，没有等待执行的任务")13. 获取插件信息
场景：管理插件，检查插件版本和更新。
import requestsfrom requests.auth import HTTPBasicAuthdef get_plugins_info(jenkins_url, username=None, api_token=None): """ 获取Jenkins插件信息 参数: jenkins_url: Jenkins服务器地址 username: 用户名（可选） api_token: API令牌（可选） """ url = f"{jenkins_url}/pluginManager/api/json?depth=2" auth = None if username and api_token: auth = HTTPBasicAuth(username, api_token) try: response = requests.get(url, auth=auth, timeout=10) response.raise_for_status() data = response.json() plugins = data.get('plugins', []) plugins_info = [] for plugin in plugins: plugin_info = { 'short_name': plugin.get('shortName'), 'long_name': plugin.get('longName'), 'version': plugin.get('version'), 'enabled': plugin.get('enabled', False), 'active': plugin.get('active', False), 'has_update': plugin.get('hasUpdate', False), 'url': plugin.get('url', ''), 'dependencies': [dep.get('shortName') for dep in plugin.get('dependencies', [])] } plugins_info.append(plugin_info) return plugins_info except requests.exceptions.RequestException as e: print(f"获取插件信息失败: {e}") return []# 使用示例if __name__ == "__main__": JENKINS_URL = " http://your-jenkins-server:8080 " USERNAME = "your_username" API_TOKEN = "your_api_token" plugins = get_plugins_info(JENKINS_URL, USERNAME, API_TOKEN) print(f"已安装 {len(plugins)} 个插件:") print("-" * 100) print(f"{'插件名称':<40} {'版本':<15} {'状态':<10} {'有更新':<10} {'依赖数':<10}") print("-" * 100) enabled_count = 0 update_count = 0 for plugin in plugins: status = "启用" if plugin['enabled'] else "禁用" has_update = "是" if plugin['has_update'] else "否" if plugin['enabled']: enabled_count += 1 if plugin['has_update']: update_count += 1 print(f"{plugin['long_name'][:38]:<40} {plugin['version']:<15} {status:<10} {has_update:<10} {len(plugin['dependencies']):<10}") print("-" * 100) print(f"统计: 启用 {enabled_count}/{len(plugins)}, 有更新 {update_count}")14. 获取系统信息
场景：监控Jenkins系统健康状态，获取系统指标。
import requestsfrom requests.auth import HTTPBasicAuthimport jsondef get_system_info(jenkins_url, username=None, api_token=None): """ 获取Jenkins系统信息 参数: jenkins_url: Jenkins服务器地址 username: 用户名（可选） api_token: API令牌（可选） """ url = f"{jenkins_url}/systemInfo/api/json" auth = None if username and api_token: auth = HTTPBasicAuth(username, api_token) try: response = requests.get(url, auth=auth, timeout=10) response.raise_for_status() system_info = response.json() # 提取关键信息 key_info = { 'jenkins_version': system_info.get('jenkins.version', 'Unknown'), 'java_version': system_info.get('java.version', 'Unknown'), 'os_name': system_info.get('os.name', 'Unknown'), 'os_version': system_info.get('os.version', 'Unknown'), 'os_arch': system_info.get('os.arch', 'Unknown'), 'user_timezone': system_info.get('user.timezone', 'Unknown'), 'system_encoding': system_info.get('sun.jnu.encoding', 'Unknown'), 'file_encoding': system_info.get('file.encoding', 'Unknown'), 'available_processors': system_info.get('sun.arch.data.model', 'Unknown'), 'user_name': system_info.get('user.name', 'Unknown'), 'user_home': system_info.get('user.home', 'Unknown') } return key_info, system_info except requests.exceptions.RequestException as e: print(f"获取系统信息失败: {e}") return {}, {}# 使用示例if __name__ == "__main__": JENKINS_URL = " http://your-jenkins-server:8080 " USERNAME = "your_username" API_TOKEN = "your_api_token" key_info, full_info = get_system_info(JENKINS_URL, USERNAME, API_TOKEN) print("Jenkins系统信息:") print("=" * 50) for key, value in key_info.items(): print(f"{key.replace('_', ' ').title()}: {value}") print("\n完整系统信息已保存到文件 'jenkins_system_info.json'") # 保存完整信息到文件 with open('jenkins_system_info.json', 'w', encoding='utf-8') as f: json.dump(full_info, f, indent=2, ensure_ascii=False)15. 批量操作任务
场景：批量执行任务操作，如批量重命名、批量修改配置等。
import requestsfrom requests.auth import HTTPBasicAuthimport timedef batch_operation_on_jobs(jenkins_url, operation, job_patterns, username, api_token, **kwargs): """ 批量操作Jenkins任务 参数: jenkins_url: Jenkins服务器地址 operation: 操作类型 ('disable', 'enable', 'delete', 'copy') job_patterns: 任务名称模式列表 username: 用户名 api_token: API令牌 **kwargs: 其他参数，如目标任务名前缀等 """ # 首先获取所有任务 url = f"{jenkins_url}/api/json" auth = HTTPBasicAuth(username, api_token) try: response = requests.get(url, auth=auth, timeout=10) response.raise_for_status() data = response.json() all_jobs = [job['name'] for job in data.get('jobs', [])] # 根据模式过滤任务 target_jobs = [] for pattern in job_patterns: if '*' in pattern: # 简单通配符匹配 pattern_prefix = pattern.replace('*', '') matched = [job for job in all_jobs if job.startswith(pattern_prefix)] target_jobs.extend(matched) else: # 精确匹配 if pattern in all_jobs: target_jobs.append(pattern) target_jobs = list(set(target_jobs)) # 去重 if not target_jobs: print("没有找到匹配的任务") return False print(f"找到 {len(target_jobs)} 个匹配的任务: {target_jobs}") results = [] for job_name in target_jobs: success = False message = "" try: if operation == 'disable': success = toggle_job(jenkins_url, job_name, enable=False, username=username, api_token=api_token) message = f"禁用任务 '{job_name}'" elif operation == 'enable': success = toggle_job(jenkins_url, job_name, enable=True, username=username, api_token=api_token) message = f"启用任务 '{job_name}'" elif operation == 'delete': delete_url = f"{jenkins_url}/job/{job_name}/doDelete" delete_response = requests.post(delete_url, auth=auth, timeout=10) success = delete_response.status_code == 200 message = f"删除任务 '{job_name}'" elif operation == 'copy': new_name = kwargs.get('new_prefix', 'copy_') + job_name copy_url = f"{jenkins_url}/createItem?name={new_name}&mode=copy&from={job_name}" copy_response = requests.post(copy_url, auth=auth, timeout=10) success = copy_response.status_code == 200 message = f"复制任务 '{job_name}' 到 '{new_name}'" # 添加延迟，避免请求过快 time.sleep(0.5) except Exception as e: message = f"操作任务 '{job_name}' 时出错: {e}" success = False results.append({ 'job_name': job_name, 'success': success, 'message': message }) # 输出结果 print(f"\n批量操作完成: {operation}") print("-" * 80) success_count = sum(1 for r in results if r['success']) for result in results: status = "成功" if result['success'] else "失败" print(f"{result['job_name']:<40} {status:<10} {result['message']}") print("-" * 80) print(f"总计: {success_count}/{len(results)} 个任务操作成功") return success_count == len(results) except requests.exceptions.RequestException as e: print(f"批量操作失败: {e}") return False# 使用示例if __name__ == "__main__": JENKINS_URL = " http://your-jenkins-server:8080 " USERNAME = "your_username" API_TOKEN = "your_api_token" # 示例1: 批量禁用以"temp_"开头的任务 batch_operation_on_jobs( JENKINS_URL, operation='disable', job_patterns=['temp_*'], username=USERNAME, api_token=API_TOKEN ) # 示例2: 批量复制任务 batch_operation_on_jobs( JENKINS_URL, operation='copy', job_patterns=['production-deploy', 'staging-deploy'], username=USERNAME, api_token=API_TOKEN, new_prefix='backup_' )15个实用的Python脚本，用于自动化操作Jenkins。这些脚本涵盖了从基础任务操作到系统监控的多个方面，通过合理使用这些脚本，可以显著提升Jenkins管理的自动化水平，为持续集成和持续部署流程提供更强大的支持，构建更加高效、可靠的DevOps实践体系：
提升运维效率：自动化重复性操作，减少人工干预，降低人为错误风险。
增强系统监控：实时获取Jenkins状态，及时发现并解决问题，保障系统稳定性。
实现批量管理：一次性处理多个任务或构建，大幅提升管理效率。
支持集成扩展：将Jenkins操作无缝集成到更大的自动化流程中，构建端到端的自动化体系。
如果你觉得这篇文章有用，欢迎点赞、转发、收藏、留言、推荐❤！
------加入知识场与更多人一起学习------https://ima.qq.com/wiki/?shareId=f2628818f0874da17b71ffa0e5e8408114e7dbad46f1745bbd1cc1365277631c
https://ima.qq.com/wiki/?shareId=66042e013e5ccae8371b46359aa45b8714f435cc844ff0903e27a64e050b54b5
