业务需求

上一篇写到gitlab webhook到python脚本接口控制jira
但是公司项目gitlab一个group ID下面很多级非常多子项目,一个一个手动加webhook不现实,通过脚本是最简单的.

python管理脚本

要遍历一个 GitLab Group 及其所有子级 Group 中的所有项目，并为这些项目批量增加 Webhook，可以在脚本中递归地处理 Group 层级结构。你需要先获取 Group 及其子 Group，然后获取每个 Group 下的所有项目。

"""
GitLab Webhook Manager

该脚本用于在 GitLab 中批量创建、删除和更新 Webhook。它可以递归地遍历指定 Group 及其所有子 Group 下的所有项目，
并对每个项目执行相应的 Webhook 管理操作。
"""

import requests

# 配置
GITLAB_URL = 'https://gitlab.example.com'
PRIVATE_TOKEN = 'your_access_token'
GROUP_ID = 'your_group_id'
WEBHOOK_URL = 'https://your-webhook-url.com'
SECRET_TOKEN = 'your_secret_token'

# 获取 Group 下的所有子 Group
def get_subgroups(group_id):
    url = f"{GITLAB_URL}/api/v4/groups/{group_id}/subgroups"
    headers = {
        "Private-Token": PRIVATE_TOKEN
    }
    response = requests.get(url, headers=headers)
    response.raise_for_status()
    return response.json()

# 获取 Group 下的所有项目
def get_projects(group_id):
    url = f"{GITLAB_URL}/api/v4/groups/{group_id}/projects"
    headers = {
        "Private-Token": PRIVATE_TOKEN
    }
    response = requests.get(url, headers=headers)
    response.raise_for_status()
    return response.json()

# 获取项目的所有 Webhook
def get_project_hooks(project_id):
    url = f"{GITLAB_URL}/api/v4/projects/{project_id}/hooks"
    headers = {
        "Private-Token": PRIVATE_TOKEN
    }
    response = requests.get(url, headers=headers)
    response.raise_for_status()
    return response.json()

# 创建项目的 Webhook
def create_webhook(project_id):
    url = f"{GITLAB_URL}/api/v4/projects/{project_id}/hooks"
    headers = {
        "Private-Token": PRIVATE_TOKEN
    }
    data = {
        "url": WEBHOOK_URL,
        "token": SECRET_TOKEN,
        "merge_requests_events": True,
        "push_events": True,
        "issues_events": True,
        # 可以根据需要添加更多事件
    }
    response = requests.post(url, headers=headers, json=data)
    response.raise_for_status()
    return response.json()

# 删除项目的 Webhook
def delete_webhook(project_id, hook_id):
    url = f"{GITLAB_URL}/api/v4/projects/{project_id}/hooks/{hook_id}"
    headers = {
        "Private-Token": PRIVATE_TOKEN
    }
    response = requests.delete(url, headers=headers)
    response.raise_for_status()

# 更新项目的 Webhook
def update_webhook(project_id, hook_id, events):
    url = f"{GITLAB_URL}/api/v4/projects/{project_id}/hooks/{hook_id}"
    headers = {
        "Private-Token": PRIVATE_TOKEN
    }
    data = {
        "url": WEBHOOK_URL,
        "token": SECRET_TOKEN,
        **events
    }
    response = requests.put(url, headers=headers, json=data)
    response.raise_for_status()
    return response.json()

# 递归遍历所有 Group 和子 Group，获取所有项目
def get_all_projects(group_id):
    projects = []
    projects.extend(get_projects(group_id))

    subgroups = get_subgroups(group_id)
    for subgroup in subgroups:
        projects.extend(get_all_projects(subgroup['id']))

    return projects

# 主程序
def main():
    action = input("Enter 'create' to create webhooks, 'delete' to delete webhooks, or 'update' to update webhooks: ").strip().lower()
    events_to_add = {
        "merge_requests_events": True,
        "push_events": True,
        "issues_events": True,
        # 你可以在这里添加更多事件
    }

    all_projects = get_all_projects(GROUP_ID)

    for project in all_projects:
        project_id = project['id']
        project_name = project['name']

        if action == 'create':
            try:
                create_webhook(project_id)
                print(f"Created webhook for project {project_name} with ID {project_id}")
            except requests.exceptions.RequestException as e:
                print(f"Failed to create webhook for project {project_name} with ID {project_id}: {e}")

        elif action == 'delete':
            hooks = get_project_hooks(project_id)
            for hook in hooks:
                if hook['url'] == WEBHOOK_URL:
                    try:
                        delete_webhook(project_id, hook['id'])
                        print(f"Deleted webhook for project {project_name} with ID {project_id}")
                    except requests.exceptions.RequestException as e:
                        print(f"Failed to delete webhook for project {project_name} with ID {project_id}: {e}")

        elif action == 'update':
            hooks = get_project_hooks(project_id)
            for hook in hooks:
                if hook['url'] == WEBHOOK_URL:
                    try:
                        update_webhook(project_id, hook['id'], events_to_add)
                        print(f"Updated webhook for project {project_name} with ID {project_id}")
                    except requests.exceptions.RequestException as e:
                        print(f"Failed to update webhook for project {project_name} with ID {project_id}: {e}")

        else:
            print("Invalid action. Please enter 'create', 'delete', or 'update'.")

if __name__ == "__main__":
    main()

脚本说明

• 配置部分：包括 GitLab 实例的 URL、访问令牌、根 Group ID、Webhook 的 URL 和 Secret Token。
• 获取子 Group：get_subgroups 函数通过 GitLab API 获取指定 Group 下的所有子 Group。
• 获取项目：get_projects 函数通过 GitLab API 获取指定 Group 下的所有项目。
• 获取 Webhook：get_project_hooks 函数通过 GitLab API 获取每个项目的所有 Webhook。
• 创建 Webhook：create_webhook 函数通过 GitLab API 为每个项目创建 Webhook。
• 删除 Webhook：delete_webhook 函数通过 GitLab API 删除每个项目中匹配指定 URL 的 Webhook。
• 更新 Webhook：update_webhook 函数通过 GitLab API 更新每个项目中匹配指定 URL 的 Webhook。
• 递归获取所有项目：get_all_projects 函数递归遍历指定 Group 下的所有子 Group，并获取所有项目。
• 主程序：根据用户输入的操作（创建、删除或更新），遍历所有项目并执行相应的 Webhook 操作。

这个脚本现在可以处理多级嵌套的 Group 结构，并执行创建、删除和更新 Webhook 的操作。

English Version

"""
GitLab Webhook Manager

This script is used to bulk create, delete, and update webhooks in GitLab.
It recursively traverses the specified group and all its subgroups to apply
the desired webhook management operation on all projects within those groups.
"""

import requests

# Configuration
GITLAB_URL = 'https://gitlab.example.com'
PRIVATE_TOKEN = 'your_access_token'
GROUP_ID = 'your_group_id'
WEBHOOK_URL = 'https://your-webhook-url.com'
SECRET_TOKEN = 'your_secret_token'

# Get all subgroups of a group
def get_subgroups(group_id):
    url = f"{GITLAB_URL}/api/v4/groups/{group_id}/subgroups"
    headers = {
        "Private-Token": PRIVATE_TOKEN
    }
    response = requests.get(url, headers=headers)
    response.raise_for_status()
    return response.json()

# Get all projects of a group
def get_projects(group_id):
    url = f"{GITLAB_URL}/api/v4/groups/{group_id}/projects"
    headers = {
        "Private-Token": PRIVATE_TOKEN
    }
    response = requests.get(url, headers=headers)
    response.raise_for_status()
    return response.json()

# Get all webhooks of a project
def get_project_hooks(project_id):
    url = f"{GITLAB_URL}/api/v4/projects/{project_id}/hooks"
    headers = {
        "Private-Token": PRIVATE_TOKEN
    }
    response = requests.get(url, headers=headers)
    response.raise_for_status()
    return response.json()

# Create a webhook for a project
def create_webhook(project_id):
    url = f"{GITLAB_URL}/api/v4/projects/{project_id}/hooks"
    headers = {
        "Private-Token": PRIVATE_TOKEN
    }
    data = {
        "url": WEBHOOK_URL,
        "token": SECRET_TOKEN,
        "merge_requests_events": True,
        "push_events": True,
        "issues_events": True,
        # Add more events as needed
    }
    response = requests.post(url, headers=headers, json=data)
    response.raise_for_status()
    return response.json()

# Delete a webhook from a project
def delete_webhook(project_id, hook_id):
    url = f"{GITLAB_URL}/api/v4/projects/{project_id}/hooks/{hook_id}"
    headers = {
        "Private-Token": PRIVATE_TOKEN
    }
    response = requests.delete(url, headers=headers)
    response.raise_for_status()

# Update a webhook for a project
def update_webhook(project_id, hook_id, events):
    url = f"{GITLAB_URL}/api/v4/projects/{project_id}/hooks/{hook_id}"
    headers = {
        "Private-Token": PRIVATE_TOKEN
    }
    data = {
        "url": WEBHOOK_URL,
        "token": SECRET_TOKEN,
        **events
    }
    response = requests.put(url, headers=headers, json=data)
    response.raise_for_status()
    return response.json()

# Recursively get all projects from a group and its subgroups
def get_all_projects(group_id):
    projects = []
    projects.extend(get_projects(group_id))

    subgroups = get_subgroups(group_id)
    for subgroup in subgroups:
        projects.extend(get_all_projects(subgroup['id']))

    return projects

# Main program
def main():
    action = input("Enter 'create' to create webhooks, 'delete' to delete webhooks, or 'update' to update webhooks: ").strip().lower()
    events_to_add = {
        "merge_requests_events": True,
        "push_events": True,
        "issues_events": True,
        # Add more events as needed
    }

    all_projects = get_all_projects(GROUP_ID)

    for project in all_projects:
        project_id = project['id']
        project_name = project['name']

        if action == 'create':
            try:
                create_webhook(project_id)
                print(f"Created webhook for project {project_name} with ID {project_id}")
            except requests.exceptions.RequestException as e:
                print(f"Failed to create webhook for project {project_name} with ID {project_id}: {e}")

        elif action == 'delete':
            hooks = get_project_hooks(project_id)
            for hook in hooks:
                if hook['url'] == WEBHOOK_URL:
                    try:
                        delete_webhook(project_id, hook['id'])
                        print(f"Deleted webhook for project {project_name} with ID {project_id}")
                    except requests.exceptions.RequestException as e:
                        print(f"Failed to delete webhook for project {project_name} with ID {project_id}: {e}")

        elif action == 'update':
            hooks = get_project_hooks(project_id)
            for hook in hooks:
                if hook['url'] == WEBHOOK_URL:
                    try:
                        update_webhook(project_id, hook['id'], events_to_add)
                        print(f"Updated webhook for project {project_name} with ID {project_id}")
                    except requests.exceptions.RequestException as e:
                        print(f"Failed to update webhook for project {project_name} with ID {project_id}: {e}")

        else:
            print("Invalid action. Please enter 'create', 'delete', or 'update'.")

if __name__ == "__main__":
    main()

Script Explanation

• Configuration Section: Includes the URL of your GitLab instance, access token, root group ID, webhook URL, and secret token.
• Get Subgroups: The get_subgroups function fetches all subgroups of a specified group using the GitLab API.
• Get Projects: The get_projects function fetches all projects of a specified group using the GitLab API.
• Get Webhooks: The get_project_hooks function fetches all webhooks of a specified project using the GitLab API.
• Create Webhook: The create_webhook function creates a webhook for a specified project using the GitLab API.
• Delete Webhook: The delete_webhook function deletes a webhook from a specified project using the GitLab API.
• Update Webhook: The update_webhook function updates a webhook for a specified project using the GitLab API.
• Recursively Get All Projects: The get_all_projects function recursively fetches all projects from a specified group and its subgroups.
• Main Program: Depending on the user input (create, delete, or update), the script performs the corresponding webhook management operation on all projects.

This script now supports creating, deleting, and updating webhooks while handling multi-level nested groups.