Python 自动化管理Jenkins的15个实用脚本，提升效率

import requests
from requests.auth import HTTPBasicAuth

def 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)

import requests
from requests.auth import HTTPBasicAuth

def 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}秒")

import requests
from requests.auth import HTTPBasicAuth

def 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)

import requests
from requests.auth import HTTPBasicAuth
import time

def 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)

import requests
from requests.auth import HTTPBasicAuth

def 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)

import requests
from requests.auth import HTTPBasicAuth
from datetime import datetime

def 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}")

import requests
from requests.auth import HTTPBasicAuth
import xml.etree.ElementTree as ET

def 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)

import requests
from requests.auth import HTTPBasicAuth

def 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)

import requests
from requests.auth import HTTPBasicAuth
from datetime import datetime, timedelta

def 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}秒")

import requests
from requests.auth import HTTPBasicAuth

def 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)

import requests
from requests.auth import HTTPBasicAuth

def 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")

import requests
from requests.auth import HTTPBasicAuth
from datetime import datetime

def 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("队列为空，没有等待执行的任务")

import requests
from requests.auth import HTTPBasicAuth

def 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}")

import requests
from requests.auth import HTTPBasicAuth
import json

def 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)

import requests
from requests.auth import HTTPBasicAuth
import time

def 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_'
    )