Add Chromium-only Blender WebEngine parity work

blender-5.2.0/tools/triage/gitea_utils.py

@@ -0,0 +1,256 @@
+#!/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 os
+import datetime
+import json
+import urllib.error
+import urllib.parse
+import urllib.request
+
+from typing import (
+    Any,
+)
+
+API_TOKEN_ENV = 'GITEA_API_TOKEN'
+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:
+    request = urllib.request.Request(url)
+    # If the token environment variable is set, add the `Authorization` header to the request
+    token = os.environ.get(API_TOKEN_ENV)
+    if token:
+        request.add_header('Authorization', "token " + token)
+    try:
+        # Make the HTTP request and store the response in a 'response' object
+        response = urllib.request.urlopen(request)
+    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 = {}
+        # Assume that if no timezone is provided, that it's UTC.
+        # Without this, dates passed in that *do* have a timezone aren't handled properly.
+        if date_start:
+            if date_start.tzinfo is None:
+                date_start = date_start.replace(tzinfo=datetime.timezone.utc)
+            query_params["since"] = date_start.isoformat()
+        if date_end:
+            if date_end.tzinfo is None:
+                date_end = date_end.replace(tzinfo=datetime.timezone.utc)
+            query_params["before"] = date_end.isoformat()
+
+        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 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