test/tools/triage/gitea_utils.py

#!/usr/bin/env python3
# SPDX-FileCopyrightText: 2023 Blender Authors
#
# SPDX-License-Identifier: GPL-2.0-or-later

# Simple module for inspecting GITEA users, pulls and issues.

__all__ = (
    "git_username_detect",
    "gitea_json_activities_get",
    "gitea_json_pull_request_by_base_and_head_get",
    "gitea_json_issue_events_filter",
    "gitea_json_issue_get",
    "gitea_json_issues_search",
    "gitea_user_get",
)

import datetime
import json
import urllib.error
import urllib.parse
import urllib.request

from typing import (
    Any,
)

BASE_API_URL = "https://projects.blender.org/api/v1"


def url_json_get(url: str, quiet: bool = False) -> dict[str, Any] | list[dict[str, Any]] | None:
    try:
        # Make the HTTP request and store the response in a 'response' object
        response = urllib.request.urlopen(url)
    except urllib.error.URLError as ex:
        if not quiet:
            print(url)
            print("Error making HTTP request:", ex)
        return None

    # Convert the response content to a JSON object containing the user information.
    result = json.loads(response.read())
    assert result is None or isinstance(result, (dict, list))
    return result


def url_json_get_all_pages(
        url: str,
        verbose: bool = False,
) -> list[dict[str, Any]]:
    result: list[dict[str, Any]] = []
    page = 1
    while True:
        if verbose:
            print(f"Requesting page {page}", end="\r", flush=True)

        if page == 1:
            # XXX: In some cases, a bug prevents using the `page` and `limit` parameters if the page is 1
            result_page = url_json_get(url)
        else:
            separator = '&' if urllib.parse.urlparse(url).query else '?'
            result_page = url_json_get(f"{url}{separator}page={page}")

        if not result_page:
            break
        assert isinstance(result_page, list)
        result.extend(result_page)

        if len(result_page) == 0:
            break

        page += 1

    return result


def gitea_user_get(username: str) -> dict[str, Any]:
    """
    Get the user data as JSON from the user name. https://docs.gitea.com/api/next/#tag/user/operation/userGet
    """

    url = f"{BASE_API_URL}/users/{username}"
    result = url_json_get(url)
    assert isinstance(result, dict)
    return result


def gitea_json_issue_get(issue_fullname: str) -> dict[str, Any]:
    """
    Get issue/pull JSON data.
    :param issue_fullname: string in the format "{owner}/{repo}/issues/{number}"
    """
    url = f"{BASE_API_URL}/repos/{issue_fullname}"
    result = url_json_get(url)
    assert isinstance(result, dict)
    return result


def gitea_json_activities_get(username: str, date: str) -> list[dict[str, Any]]:
    """
    List a user's activity feeds.
    :param username: username of user.
    :param date: the date of the activities to be found.
    """
    activity_url = f"{BASE_API_URL}/users/{username}/activities/feeds?only-performed-by=true&date={date}"
    result = url_json_get_all_pages(activity_url)
    assert isinstance(result, list)
    return result


def gitea_json_pull_request_by_base_and_head_get(repo_name: str, base: str, head: str) -> dict[str, Any] | None:
    """
    Get a pull request by base and head
    :param repo_name: Full name of the repository, e.g. "blender/blender".
    :param base: Target branch of the PR (branch it wants to merge into), e.g. "main".
    :param head: Full identifier of the branch the PR is made from, e.g. "MyRepository:temp-feature-branch"
    """
    url = f"{BASE_API_URL}/repos/{repo_name}/pulls/{base}/{head}"
    result = url_json_get(url, quiet=True)
    assert result is None or isinstance(result, dict)
    return result


def gitea_json_issues_search(
        type: str | None = None,
        since: str | None = None,
        before: str | None = None,
        state: str = 'all',
        labels: str | None = None,
        created: bool = False,
        reviewed: bool = False,
        access_token: str | None = None,
        verbose: bool = True,
) -> list[dict[str, Any]]:
    """
    Search for issues across the repositories that the user has access to.
    :param type: filter by type (issues / pulls) if set.
    :param since: Only show notifications updated after the given time. This is a timestamp in RFC 3339 format.
    :param before: Only show notifications updated before the given time. This is a timestamp in RFC 3339 format.
    :param state: whether issue is open or closed.
    :param labels: comma separated list of labels.
       Fetch only issues that have any of this labels. Non existent labels are discarded.
    :param created: filter (issues / pulls) created by you, default is false.
    :param reviewed: filter pulls reviewed by you, default is false.
    :param access_token: token generated by the GITEA API.
    :return: List of issues or pulls.
    """

    query_params = {k: v for k, v in locals().items() if v and k not in {"verbose"}}
    for k, v in query_params.items():
        if v is True:
            query_params[k] = "true"
        elif v is False:
            query_params[k] = "false"

    if verbose:
        print("# Searching for {} #".format(
            query_params["type"] if "type" in query_params else "issues and pulls"))

        print("Query params:", {
              k: v for k, v in query_params.items() if k not in {"type", "access_token"}})

    base_url = f"{BASE_API_URL}/repos/issues/search"
    encoded_query_params = urllib.parse.urlencode(query_params)
    issues_url = f"{base_url}?{encoded_query_params}"

    issues = url_json_get_all_pages(issues_url, verbose=verbose)

    if verbose:
        print(f"Total: {len(issues)}         ", end="\n\n", flush=True)

    return issues


def gitea_json_issue_events_filter(
        issue_fullname: str,
        date_start: datetime.datetime | None = None,
        date_end: datetime.datetime | None = None,
        username: str | None = None,
        labels: set[str] | None = None,
        event_type: set[str] | None = None,
) -> list[dict[str, Any]]:
    """
    Filter all comments and events on the issue list. If both labels and event_type are provided,
    an event is included if either the label or event type matches.
    :param issue_fullname: string in the format "{owner}/{repo}/issues/{number}"
    :param date_start: if provided, only comments updated since the specified time are returned.
    :param date_end: if provided, only comments updated before the provided time are returned.
    :param labels: list of labels. Fetch only events that have any of these labels (plus, events
       passing the event_type check if set)
    :param event_type: set of types of events in {"close", "commit_ref"...}.
    :return: List of comments or events.
    """
    issue_events_url = f"{BASE_API_URL}/repos/{issue_fullname}/timeline"
    if date_start or date_end:
        query_params = {}
        if date_start:
            query_params["since"] = f"{date_start.isoformat()}Z"
        if date_end:
            query_params["before"] = f"{date_end.isoformat()}Z"

        encoded_query_params = urllib.parse.urlencode(query_params)
        issue_events_url = f"{issue_events_url}?{encoded_query_params}"

    result = []
    for event in url_json_get_all_pages(issue_events_url):
        if not event:
            continue

        if username and (not event["user"] or event["user"]["username"] != username):
            continue

        if labels and event["type"] == "label" and event["label"]["name"] in labels:
            pass
        elif event_type and event["type"] in event_type:
            pass
        elif labels or event_type:
            continue

        result.append(event)

    return result


# WORKAROUND: This function doesn't involve GITEA, and the obtained username may not match the username used in GITEA.
# However, it provides an option to fetch the configured username from the local Git,
# in case the user does not explicitly supply the username.
def git_username_detect() -> str | None:
    import os
    import subprocess

    # Get the repository directory
    repo_dir = os.path.abspath(os.path.normpath(os.path.join(os.path.dirname(__file__), "..", "..")))

    # Attempt to get the configured username from the local Git
    try:
        result = subprocess.run(["git", "config", "user.username"], stdout=subprocess.PIPE, cwd=repo_dir)
        result.check_returncode()  # Check if the command was executed successfully
        username = result.stdout.decode().rstrip()
        return username
    except subprocess.CalledProcessError as ex:
        # Handle errors if the git config command fails
        print(f"Error fetching Git username: {ex}")
        return None