Cleanup: type check scripts in tools/triage

This commit is contained in:
Campbell Barton
2024-02-28 11:02:51 +11:00
parent 4f8db2ee67
commit 79cf4e58e4
4 changed files with 102 additions and 59 deletions

View File

@@ -52,9 +52,6 @@ PATHS_EXCLUDE = set(
"tools/check_source/check_descriptions.py",
"tools/check_source/check_header_duplicate.py",
"tools/check_source/check_unused_defines.py",
"tools/triage/gitea_utils.py", # TODO (low priority).
"tools/triage/issues_needing_info.py", # TODO (low priority).
"tools/triage/weekly_report.py", # TODO (low priority).
"tools/utils/blend2json.py",
"tools/utils/blender_keyconfig_export_permutations.py",
"tools/utils/blender_merge_format_changes.py",

View File

@@ -5,15 +5,25 @@
# Simple module for inspecting GITEA users, pulls and issues.
import datetime
import json
import urllib.error
import urllib.parse
import urllib.request
from typing import (
Any,
Dict,
List,
Optional,
Set,
Union,
)
BASE_API_URL = "https://projects.blender.org/api/v1"
def url_json_get(url):
def url_json_get(url: str) -> Optional[Union[Dict[str, Any], List[Dict[str, Any]]]]:
try:
# Make the HTTP request and store the response in a 'response' object
response = urllib.request.urlopen(url)
@@ -22,13 +32,19 @@ def url_json_get(url):
print("Error making HTTP request:", ex)
return None
# Convert the response content to a JSON object containing the user information
return json.loads(response.read())
# 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, limit=50, verbose=False):
def url_json_get_all_pages(
url: str,
limit: int = 50,
verbose: bool = False,
) -> List[Dict[str, Any]]:
assert limit <= 50, "50 is the maximum limit of items per page"
result = []
result: List[Dict[str, Any]] = []
page = 1
while True:
if verbose:
@@ -42,7 +58,7 @@ def url_json_get_all_pages(url, limit=50, verbose=False):
if not result_page:
break
assert isinstance(result_page, list)
result.extend(result_page)
if len(result_page) < limit:
@@ -53,51 +69,59 @@ def url_json_get_all_pages(url, limit=50, verbose=False):
return result
def gitea_user_get(username):
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}"
return url_json_get(url)
result = url_json_get(url)
assert isinstance(result, dict)
return result
def gitea_json_issue_get(issue_fullname):
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}"
return url_json_get(url)
result = url_json_get(url)
assert isinstance(result, dict)
return result
def gitea_json_activities_get(username, date):
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}"
return url_json_get_all_pages(activity_url)
result = url_json_get_all_pages(activity_url)
assert isinstance(result, list)
return result
def gitea_json_issues_search(
type=None,
since=None,
before=None,
state='all',
labels=None,
created=False,
reviewed=False,
access_token=None,
verbose=True):
type: Optional[str] = None,
since: Optional[str] = None,
before: Optional[str] = None,
state: str = 'all',
labels: Optional[str] = None,
created: bool = False,
reviewed: bool = False,
access_token: Optional[str] = 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 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.
@@ -131,19 +155,20 @@ def gitea_json_issues_search(
def gitea_json_issue_events_filter(
issue_fullname,
date_start=None,
date_end=None,
username=None,
labels=None,
event_type=set()):
issue_fullname: str,
date_start: Optional[datetime.datetime] = None,
date_end: Optional[datetime.datetime] = None,
username: Optional[str] = None,
labels: Optional[Set[str]] = None,
event_type: Set[str] = set(),
) -> List[Dict[str, Any]]:
"""
Filter all comments and events on the issue list.
: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 this labels.
:param event_type: list of types of events in {"close", "commit_ref"...}.
: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"
@@ -180,7 +205,7 @@ def gitea_json_issue_events_filter(
# 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():
def git_username_detect() -> Optional[str]:
import os
import subprocess

View File

@@ -18,7 +18,7 @@ import datetime
from gitea_utils import gitea_json_issues_search, gitea_json_issue_events_filter, git_username_detect
def print_needing_info_urls(username, before):
def print_needing_info_urls(username: str, before: str) -> None:
print(f"Needs information from user before {before}:")

View File

@@ -20,10 +20,24 @@ import argparse
import datetime
import json
import re
from gitea_utils import gitea_json_activities_get, gitea_json_issue_get, gitea_json_issue_events_filter, gitea_user_get, git_username_detect
from gitea_utils import (
gitea_json_activities_get,
gitea_json_issue_events_filter,
gitea_json_issue_get,
gitea_user_get, git_username_detect,
)
from typing import (
Any,
Dict,
List,
Set,
Iterable,
)
def argparse_create():
def argparse_create() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description="Generate Weekly Report",
epilog="This script is typically used to help write weekly reports")
@@ -53,36 +67,37 @@ def argparse_create():
return parser
def report_personal_weekly_get(username, start, verbose=True):
def report_personal_weekly_get(username: str, start: datetime.datetime, verbose: bool = True) -> None:
data_cache = {}
data_cache: Dict[str, Dict[str, Any]] = {}
def gitea_json_issue_get_cached(issue_fullname):
def gitea_json_issue_get_cached(issue_fullname: str) -> Dict[str, Any]:
if issue_fullname not in data_cache:
data_cache[issue_fullname] = gitea_json_issue_get(issue_fullname)
issue = gitea_json_issue_get(issue_fullname)
data_cache[issue_fullname] = issue
return data_cache[issue_fullname]
pulls_closed = set()
pulls_commented = set()
pulls_created = set()
pulls_closed: Set[str] = set()
pulls_commented: Set[str] = set()
pulls_created: Set[str] = set()
issues_closed = set()
issues_commented = set()
issues_created = set()
issues_closed: Set[str] = set()
issues_commented: Set[str] = set()
issues_created: Set[str] = set()
pulls_reviewed = []
pulls_reviewed: List[str] = []
issues_confirmed = []
issues_needing_user_info = []
issues_needing_developer_info = []
issues_fixed = []
issues_duplicated = []
issues_archived = []
issues_confirmed: List[str] = []
issues_needing_user_info: List[str] = []
issues_needing_developer_info: List[str] = []
issues_fixed: List[str] = []
issues_duplicated: List[str] = []
issues_archived: List[str] = []
commits_main = []
commits_main: List[str] = []
user_data = gitea_user_get(username)
user_data: Dict[str, Any] = gitea_user_get(username)
for i in range(7):
date_curr = start + datetime.timedelta(days=i)
@@ -112,10 +127,16 @@ def report_personal_weekly_get(username, start, verbose=True):
fullname = activity["repo"]["full_name"] + "/pulls/" + activity["content"].split('|')[0]
pulls_reviewed.append(fullname)
elif op_type == "commit_repo":
if activity["ref_name"] == "refs/heads/main" and activity["content"] and activity["repo"]["name"] != ".profile":
if (
activity["ref_name"] == "refs/heads/main" and
activity["content"] and
activity["repo"]["name"] != ".profile"
):
content_json = json.loads(activity["content"])
assert isinstance(content_json, dict)
repo_fullname = activity["repo"]["full_name"]
for commits in content_json["Commits"]:
content_json_commits: List[Dict[str, Any]] = content_json["Commits"]
for commits in content_json_commits:
# Skip commits that were not made by this user. Using email doesn't seem to
# be possible unfortunately.
if commits["AuthorName"] != user_data["full_name"]:
@@ -207,7 +228,7 @@ def report_personal_weekly_get(username, start, verbose=True):
print()
# Print review stats
def print_pulls(pulls):
def print_pulls(pulls: Iterable[str]) -> None:
for pull in pulls:
pull_data = gitea_json_issue_get_cached(pull)
title = pull_data["title"]
@@ -232,7 +253,7 @@ def report_personal_weekly_get(username, start, verbose=True):
if verbose:
# Debug
def print_links(issues):
def print_links(issues: Iterable[str]) -> None:
for fullname in issues:
print(f"https://projects.blender.org/{fullname}")