OpenMW/scripts/find_missing_merge_requests.py

#!/usr/bin/env python3

import click
import discord_webhook
import multiprocessing
import os
import pathlib
import requests
import urllib.parse


@click.command()
@click.option('--token_path', type=str, default=pathlib.Path.home() / '.gitlab_token',
              help='Path to text file with Gitlab token.')
@click.option('--project_id', type=int, default=7107382,
              help='Gitlab project id.')
@click.option('--job_id', type=int, default=os.getenv('CI_JOB_ID'),
              help='Gitlab job id.')
@click.option('--host', type=str, default='gitlab.com',
              help='Gitlab host.')
@click.option('--workers', type=int, default=10,
              help='Number of parallel workers.')
@click.option('--target_branch', type=str, default='master',
              help='Merge request target branch.')
@click.option('--begin_page', type=int, default=1,
              help='Begin with given /merge_requests page.')
@click.option('--end_page', type=int, default=4,
              help='End before given /merge_requests page.')
@click.option('--per_page', type=int, default=100,
              help='Number of merge requests per page.')
@click.option('--ignored_mrs_path', type=str,
              help='Path to a list of ignored MRs.')
def main(token_path, project_id, job_id, host, workers, target_branch, begin_page, end_page, per_page, ignored_mrs_path):
    headers = make_headers(token_path)
    base_url = f'https://{host}/api/v4/projects/{project_id}/'
    discord_webhook_url = os.getenv('DISCORD_WEBHOOK_URL')
    ignored_mrs = frozenset(read_ignored_mrs(ignored_mrs_path))
    checked = 0
    filtered = 0
    missing = 0
    missing_mrs = list()
    for page in range(begin_page, end_page):
        merge_requests = parse_gitlab_response(requests.get(
            url=urllib.parse.urljoin(base_url, 'merge_requests'),
            headers=headers,
            params=dict(state='merged', per_page=per_page, page=page),
        ))
        if not merge_requests:
            break
        checked += len(merge_requests)
        merge_requests = [v for v in merge_requests if v['target_branch'] == target_branch]
        if not merge_requests:
            continue
        filtered += len(merge_requests)
        with multiprocessing.Pool(workers) as pool:
            missing_merge_requests = pool.map(FilterMissingMergeRequest(headers, base_url), merge_requests)
            for mr in missing_merge_requests:
                if mr is None:
                    continue
                if mr['reference'] in ignored_mrs or mr['reference'].strip('!') in ignored_mrs:
                    print(f"Ignored MR {mr['reference']} ({mr['id']}) is missing from branch {mr['target_branch']},"
                          f" previously was merged as {mr['merge_commit_sha']}")
                    continue
                missing += 1
                missing_mrs.append(mr)
                print(f"MR {mr['reference']} ({mr['id']}) is missing from branch {mr['target_branch']},"
                      f" previously was merged as {mr['merge_commit_sha']}")
    print(f'Checked {checked} MRs ({filtered} with {target_branch} target branch), {missing} are missing')
    if discord_webhook_url is not None and missing_mrs:
        project_web_url = parse_gitlab_response(requests.get(url=base_url, headers=headers))['web_url'] + '/'
        discord_message = format_discord_message(missing=missing, filtered=filtered, target_branch=target_branch,
                                                 project_web_url=project_web_url, missing_mrs=missing_mrs,
                                                 job_id=job_id)
        print('Sending Discord notification...')
        print(discord_message)
        discord_webhook.DiscordWebhook(url=discord_webhook_url, content=discord_message, rate_limit_retry=True).execute()
    if missing_mrs:
        exit(-1)


def format_discord_message(missing, filtered, target_branch, project_web_url, missing_mrs, job_id):
    target_branch = format_link(target_branch, urllib.parse.urljoin(project_web_url, f'-/tree/{target_branch}'))
    job = f' by job ' + format_link(job_id, urllib.parse.urljoin(project_web_url, f'-/jobs/{job_id}')) if job_id else ''
    return (
        f'Found {missing} missing MRs out of {filtered} from {target_branch} target branch{job}:\n'
        + '\n'.join(format_missing_mr_message(v, project_web_url) for v in missing_mrs)
    )


def format_missing_mr_message(mr, project_web_url):
    web_url = mr.get('web_url')
    reference = mr['reference']
    target_branch = mr['target_branch']
    commit = mr['merge_commit_sha']
    if web_url is not None:
        reference = format_link(reference, web_url)
        target_branch = format_link(target_branch, urllib.parse.urljoin(project_web_url, f'-/tree/{target_branch}'))
        commit = format_link(commit, urllib.parse.urljoin(project_web_url, f'-/commit/{commit}'))
    return f"MR {reference} is missing from branch {target_branch}, previously was merged as {commit}"


def format_link(name, url):
    return f'[{name}]({url})'


class FilterMissingMergeRequest:
    def __init__(self, headers, base_url):
        self.headers = headers
        self.base_url = base_url

    def __call__(self, merge_request):
        commit_refs = requests.get(
            url=urllib.parse.urljoin(self.base_url, f"repository/commits/{merge_request['merge_commit_sha']}/refs"),
            headers=self.headers,
        ).json()
        if 'message' in commit_refs and commit_refs['message'] == '404 Commit Not Found':
            return merge_request
        if not present_in_branch(commit_refs, branch=merge_request['target_branch']):
            return merge_request


def present_in_branch(commit_refs, branch):
    return bool(next((v for v in commit_refs if v['type'] == 'branch' and v['name'] == branch), None))


def make_headers(token_path):
    job_token = os.environ.get('CI_JOB_TOKEN')
    if job_token is not None:
        print('Using auth token from CI_JOB_TOKEN env')
        return {'JOB_TOKEN': job_token}
    if not os.path.exists(token_path):
        print(f'Ignore absent token path: {token_path}')
        return dict()
    print(f'Using auth token from: {token_path}')
    return {'PRIVATE-TOKEN': read_token(token_path)}


def read_ignored_mrs(path):
    if path is None:
        return
    with open(path) as stream:
        for line in stream:
            yield line.strip()


def read_token(path):
    with open(path) as stream:
        return stream.readline().strip()


def parse_gitlab_response(response):
    response = response.json()
    if isinstance(response, dict):
        message = response.get('message')
        if message is not None:
            raise RuntimeError(f'Gitlab request has failed: {message}')
    return response


if __name__ == '__main__':
    main()