gitlab-add-hook.py

#!/usr/bin/python3
# -*- coding: utf-8 -*-
import requests
import json
import sys
import os
import datetime
import urllib.parse

conf_data = {
    "url": "http://gitlab.example.com/",
    "per_page": 100,
    "set_hook_param": {
        "url": "http://gitlab-download-app.example.com/update_projects_json.php",
        "token": "",
        "push_events": "TRUE",
        "push_events_branch_filter": "",
        "tag_push_events": "FALSE",
        "note_events": "FALSE",
        "confidential_note_events": "FALSE",
        "issues_events": "FALSE",
        "confidential_issues_events": "FALSE",
        "merge_requests_events": "FALSE",
        "job_events": "FALSE",
        "pipeline_events": "FALSE",
        "wiki_page_events": "FALSE",
        "deployment_events": "FALSE",
        "releases_events": "FALSE",
        "enable_ssl_verification": "FALSE",
    },
}

headers = {"private-token": "XXXXXXXXXXXXXXXXXXXX"}


def set_project_hook(path_with_namespace):
    url = (
        conf_data["url"]
        + "api/v4/projects/"
        + urllib.parse.quote(path_with_namespace, safe="")
        + "/hooks"
    )
    try:
        ret = requests.post(url, headers=headers, params=conf_data["set_hook_param"])
        ret.raise_for_status()
    except requests.exceptions.ConnectionError:
        return False
    except requests.exceptions.RequestException as err:
        return False
    return True


def request_project_hooks(path_with_namespace):
    url = (
        conf_data["url"]
        + "api/v4/projects/"
        + urllib.parse.quote(path_with_namespace, safe="")
        + "/hooks"
    )
    try:
        ret = requests.get(url, headers=headers)
        ret.raise_for_status()
    except requests.exceptions.ConnectionError:
        return False
    except requests.exceptions.RequestException as err:
        return False
    return ret


def request_projects():
    url = conf_data["url"] + "api/v4/projects/"
    projects = []
    page = 1
    while True:
        params = {
            "order_by": "name",
            "simple": "false",
            "sort": "asc",
            "d_after": 0,
            "per_page": conf_data["per_page"],
            "page": page,
        }
        try:
            ret = requests.get(url, headers=headers, params=params)
            ret.raise_for_status()
        except requests.exceptions.ConnectionError:
            return False
        except requests.exceptions.RequestException as err:
            return False
        if 2 < len(ret.content):
            x = json.loads(ret.content)
            projects.extend(x)
        if len(x) < conf_data["per_page"]:
            break
        page += 1
    return projects


def add_hook_all_projects():
    projects = request_projects()
    if projects == False:
        return
    for project in projects:
        if project["visibility"] == "internal":
            continue
        ret = add_hook_project(project["path_with_namespace"])
        if ret == False:
            return
    return


def add_hook_project(path_with_namespace):
    ret = request_project_hooks(path_with_namespace)
    if ret == False:
        return False
    if 2 < len(ret.content):
        project_hook_info = json.loads(ret.content)
        for hook in project_hook_info:
            if hook["url"] == conf_data["set_hook_param"]["url"]:
                return True

    ret = set_project_hook(path_with_namespace)
    if ret == False:
        return False
    return True


if __name__ == "__main__":
    args = sys.argv
    if 1 < len(args):
        add_hook_project(args[1])
    else:
        add_hook_all_projects()