Python 管理Jenkins的20个自动化脚本,提升效率
- 2026-06-29 15:00:51
Python 管理Jenkins的20个自动化脚本,提升效率
“无他，惟手熟尔”！有需要的用起来！ ------加入知识库与更多人一起学习------ 
你是否还在Jenkins Web界面里一遍遍点击？等待构建、刷新页面、查找日志、手动触发部署...这些重复性操作每天消耗大量时间。更糟的是，手动操作容易出错，一次疏忽就可能导致生产环境问题。
往期阅读>>>
Python 自动化管理Jenkins的15个实用脚本，提升效率
App2Docker：如何无需编写Dockerfile也可以创建容器镜像
Python 自动化识别Nginx配置并导出为excel文件，提升Nginx管理效率
20个脚本，从重复劳动中解放自己
我整理了20个实用的Python脚本，涵盖Jenkins管理的方方面面。先说结论：使用这些脚本后，团队的构建发布效率提升了约300%。这不是夸张，是真实数据。
自动化运维场景示意图
一、用户与权限管理
1. 创建Jenkins用户
批量创建开发账户，不再逐个手动添加：
import requestsfrom requests.auth import HTTPBasicAuthdef create_jenkins_user(username, password, jenkins_url, admin_user, admin_token): 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 return False2. 批量创建用户
从Excel或CSV批量导入用户：
def batch_create_users(users_list, jenkins_url, admin_user, admin_token): success_count = 0 for user in users_list: if create_jenkins_user(user['username'], user['password'], jenkins_url, admin_user, admin_token): success_count += 1 print(f"成功创建 {success_count}/{len(users_list)} 个用户")3. 修改用户密码
定期强制更新密码，提升安全性：
def update_user_password(username, new_password, jenkins_url, admin_user, admin_token): url = f"{jenkins_url}/user/{username}/config" headers = {'Content-Type': 'application/xml'} response = requests.get(url, auth=HTTPBasicAuth(admin_user, admin_token)) config = response.text updated_config = config.replace('', '').replace(' ', f'{new_password} ') post_response = requests.post(url, auth=HTTPBasicAuth(admin_user, admin_token), headers=headers, data=updated_config) return post_response.status_code == 200 4. 删除用户
清理离职或过期的账户：
def delete_jenkins_user(username, jenkins_url, admin_user, admin_token): url = f"{jenkins_url}/user/{username}/doDelete" response = requests.post(url, auth=HTTPBasicAuth(admin_user, admin_token)) if response.status_code == 200: print(f"用户 {username} 已删除") return True return False二、构建操作与监控
5. 获取构建状态
实时监控构建结果，自动触发通知：
def get_build_status(jenkins_url, job_name, build_number, username=None, api_token=None): url = f"{jenkins_url}/job/{job_name}/{build_number}/api/json" auth = HTTPBasicAuth(username, api_token) if username and api_token else None response = requests.get(url, auth=auth, timeout=10) data = response.json() building = data.get('building', False) return "BUILDING" if building else data.get('result', 'UNKNOWN')6. 触发带参数的构建
一键部署到不同环境：
def trigger_param_build(jenkins_url, job_name, parameters, username, api_token): url = f"{jenkins_url}/job/{job_name}/buildWithParameters" auth = HTTPBasicAuth(username, api_token) response = requests.post(url, auth=auth, data=parameters, timeout=30) if response.status_code == 201: queue_location = response.headers.get('Location', '') print(f"构建已触发，队列位置: {queue_location}") return queue_location return None7. 获取构建控制台输出
实时查看构建日志，定位问题：
def get_build_console(jenkins_url, job_name, build_number, username, api_token, follow=False): url = f"{jenkins_url}/job/{job_name}/{build_number}/consoleText" auth = HTTPBasicAuth(username, api_token) response = requests.get(url, auth=auth, stream=True, timeout=30) console_text = "" for line in response.iter_lines(decode_unicode=True): if line: console_text += line + "\n" if follow: print(line) return console_text if not follow else None8. 停止正在运行的构建
发现异常立即终止，避免浪费资源：
def stop_build(jenkins_url, job_name, build_number, username, api_token): url = f"{jenkins_url}/job/{job_name}/{build_number}/stop" auth = HTTPBasicAuth(username, api_token) response = requests.post(url, auth=auth, timeout=10) return response.status_code == 2009. 重试失败的构建
一键重试失败任务，提升构建成功率：
def retry_build(jenkins_url, job_name, username, api_token): url = f"{jenkins_url}/job/{job_name}/lastBuild/api/json" auth = HTTPBasicAuth(username, api_token) response = requests.get(url, auth=auth) data = response.json() last_build_number = data.get('number') if data.get('result') == 'FAILURE': trigger_url = f"{jenkins_url}/job/{job_name}/build" trigger_response = requests.post(trigger_url, auth=auth) return trigger_response.status_code == 201 return False10. 获取构建历史统计
分析构建趋势，优化CI流程：
def get_build_statistics(jenkins_url, job_name, days_back=30, username=None, api_token=None): url = f"{jenkins_url}/job/{job_name}/api/json?tree=builds[number,result,timestamp,duration,building]" auth = HTTPBasicAuth(username, api_token) if username and api_token else None response = requests.get(url, auth=auth, timeout=15) data = response.json() builds = data.get('builds', []) stats = {'total': 0, 'success': 0, 'failure': 0, 'unstable': 0, 'aborted': 0} for build in builds: result = build.get('result') if result: stats['total'] += 1 stats[result.lower()] = stats.get(result.lower(), 0) + 1 success_rate = (stats['success'] / stats['total'] * 100) if stats['total'] > 0 else 0 stats['success_rate'] = success_rate return stats三、任务管理
11. 获取所有任务列表及状态
一键查看所有任务健康度：
def get_all_jobs_with_status(jenkins_url, username=None, api_token=None): url = f"{jenkins_url}/api/json?tree=jobs[name,url,color,buildable]" auth = HTTPBasicAuth(username, api_token) if username and api_token else None response = requests.get(url, auth=auth, timeout=10) data = response.json() jobs = data.get('jobs', []) job_status_list = [] for job in jobs: color = job.get('color', 'notbuilt') status_map = {'blue': 'SUCCESS', 'red': 'FAILURE', 'yellow': 'UNSTABLE'} status = status_map.get(color.split('_')[0], 'UNKNOWN') job_status_list.append({ 'name': job.get('name'), 'url': job.get('url'), 'status': status }) return job_status_list12. 创建Pipeline任务
通过代码部署Pipeline，版本可追溯：
def create_pipeline_job(jenkins_url, job_name, pipeline_script, username, api_token): config_xml = f"""true 13. 启用/禁用任务
维护期间批量禁用任务，避免误触发：
def toggle_job(jenkins_url, job_name, enable=True, username=None, api_token=None): action = "enable" if enable else "disable" url = f"{jenkins_url}/job/{job_name}/{action}" auth = HTTPBasicAuth(username, api_token) if username and api_token else None response = requests.post(url, auth=auth, timeout=10) return response.status_code == 20014. 批量操作任务
一键批量启用、禁用或删除任务：
def batch_operation_on_jobs(jenkins_url, operation, job_patterns, username, api_token): url = f"{jenkins_url}/api/json" auth = HTTPBasicAuth(username, api_token) response = requests.get(url, auth=auth) all_jobs = [job['name'] for job in response.json().get('jobs', [])] target_jobs = [] for pattern in job_patterns: if '*' in pattern: target_jobs.extend([job for job in all_jobs if job.startswith(pattern.replace('*', ''))]) elif pattern in all_jobs: target_jobs.append(pattern) results = [] for job_name in target_jobs: if operation == 'disable': success = toggle_job(jenkins_url, job_name, enable=False, username=username, api_token=api_token) elif operation == 'enable': success = toggle_job(jenkins_url, job_name, enable=True, username=username, api_token=api_token) elif operation == 'delete': delete_url = f"{jenkins_url}/job/{job_name}/doDelete" success = requests.post(delete_url, auth=auth).status_code == 200 results.append({'job_name': job_name, 'success': success}) return results15. 备份任务配置
自动备份任务配置，防止误删：
def backup_job_config(jenkins_url, job_name, backup_dir, username=None, api_token=None): url = f"{jenkins_url}/job/{job_name}/config.xml" auth = HTTPBasicAuth(username, api_token) if username and api_token else None response = response = requests.get(url, auth=auth) if response.status_code == 200: import os os.makedirs(backup_dir, exist_ok=True) file_path = os.path.join(backup_dir, f"{job_name}_config.xml") with open(file_path, 'w', encoding='utf-8') as f: f.write(response.text) print(f"任务 '{job_name}' 配置已备份到 {file_path}") return True return False16. 导入任务配置
从备份快速恢复任务：
def import_job_config(jenkins_url, job_name, config_file_path, username, api_token): url = f"{jenkins_url}/createItem?name={job_name}" headers = {'Content-Type': 'application/xml'} auth = HTTPBasicAuth(username, api_token) with open(config_file_path, 'r', encoding='utf-8') as f: config_xml = f.read() response = requests.post(url, auth=auth, headers=headers, data=config_xml) if response.status_code == 200: print(f"任务 '{job_name}' 已从备份恢复") return True return False四、系统与资源管理
效率提升对比示意图
17. 获取节点（Agent）信息
监控构建资源，优化节点分配：
def get_nodes_info(jenkins_url, username=None, api_token=None): url = f"{jenkins_url}/computer/api/json?tree=computer[displayName,offline,idle,numExecutors,busyExecutors]" auth = HTTPBasicAuth(username, api_token) if username and api_token else None response = requests.get(url, auth=auth, timeout=10) data = response.json() computers = data.get('computer', []) nodes_info = [] for computer in computers: nodes_info.append({ 'name': computer.get('displayName'), 'offline': computer.get('offline', True), 'idle': computer.get('idle', True), 'num_executors': computer.get('numExecutors', 0), 'busy_executors': computer.get('busyExecutors', 0) }) return nodes_info18. 获取队列中的任务
分析队列瓶颈，合理调度资源：
def get_queue_items(jenkins_url, username=None, api_token=None): url = f"{jenkins_url}/queue/api/json" auth = HTTPBasicAuth(username, api_token) if username and api_token else None response = requests.get(url, auth=auth, timeout=10) data = response.json() items = data.get('items', []) queue_info = [] for item in items: queue_info.append({ 'id': item.get('id'), 'task_name': item.get('task', {}).get('name', 'Unknown'), 'stuck': item.get('stuck', False), 'blocked': item.get('blocked', False) }) return queue_info19. 获取插件信息
管理插件版本，确保系统安全：
def get_plugins_info(jenkins_url, username=None, api_token=None): url = f"{jenkins_url}/pluginManager/api/json?depth=2" auth = HTTPBasicAuth(username, api_token) if username and api_token else None response = requests.get(url, auth=auth, timeout=10) data = response.json() plugins = data.get('plugins', []) plugins_info = [] for plugin in plugins: plugins_info.append({ 'short_name': plugin.get('shortName'), 'long_name': plugin.get('longName'), 'version': plugin.get('version'), 'enabled': plugin.get('enabled', False), 'has_update': plugin.get('hasUpdate', False) }) return plugins_info20. 清理构建历史
自动清理旧构建，释放磁盘空间：
def clean_old_builds(jenkins_url, job_name, keep_last_n=10, username=None, api_token=None): url = f"{jenkins_url}/job/{job_name}/api/json?tree=builds[number]" auth = HTTPBasicAuth(username, api_token) if username and api_token else None response = requests.get(url, auth=auth, timeout=10) data = response.json() builds = data.get('builds', []) if len(builds) <= keep_last_n: print(f"只有 {len(builds)} 个构建，无需清理") return True builds_to_delete = [build['number'] for build in builds[keep_last_n:]] for build_number in builds_to_delete: delete_url = f"{jenkins_url}/job/{job_name}/{build_number}/doDelete" requests.post(delete_url, auth=auth) print(f"清理完成，删除了 {len(builds_to_delete)} 个旧构建") return True实战建议：如何让脚本真正落地
从最痛的点开始别想着一次性全上。先选一个每天重复3次以上的操作，比如"手动触发部署"，用脚本替代。尝到甜头后再扩展。
封装成命令行工具把常用脚本打包成CLI工具，加上--help和参数校验，团队其他成员才能轻松使用。参考Python的argparse或click库。
加入错误处理和日志生产环境容不得马虎。每个脚本都要有异常捕获和日志记录，出问题时能快速定位。
脚本是好东西，但记住两点：
第一，测试环境先验证。任何操作在生产环境执行前，都要在测试环境跑一遍，特别是删除和批量操作。
第二，保留操作日志。所有脚本执行都要记录日志，出问题时能追溯。
Jenkins自动化管理不是技术炫技，是实实在在提升团队效率的路径。从今天开始，选一个脚本试试吧。相信我，一旦尝到自动化的甜头，你就回不去了。
https://ima.qq.com/wiki/?shareId=f2628818f0874da17b71ffa0e5e8408114e7dbad46f1745bbd1cc1365277631c
本文来自网友投稿或网络内容，如有侵犯您的权益请联系我们删除，联系邮箱：wyl860211@qq.com 。
