griefith/test2
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Triage: Add 'weekly_report' and 'issues_needing_info' tools

Browse Source 
This commit adds 2 tools for triaging:
- /tools/triage/weekly_report.py
- /tools/triage/issues_needing_info.py

These tools automatically detect the username to list activities related to the user.

Pull Request: https://projects.blender.org/blender/blender/pulls/110652
This commit is contained in:
Germano Cavalcante
2023-08-04 19:32:25 +02:00
committed by Germano Cavalcante
parent 9c72199258
commit b6ed70cd92
3 changed files with 546 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

200
tools/triage/gitea_utils.py Normal file
View File

@@ -0,0 +1,200 @@
#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0-or-later
# Simple module for inspecting gitea users, pulls and issues
import json
import urllib.error
import urllib.parse
import urllib.request
BASE_API_URL = "https://projects.blender.org/api/v1"
def url_json_get(url):
try:
# Make the HTTP request and store the response in a 'response' object
response = urllib.request.urlopen(url)
except urllib.error.URLError as ex:
print(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())
def url_json_get_all_pages(url, limit=50, verbose=False):
assert limit <= 50, "50 is the maximum limit of items per page"
result = []
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:
result_page = url_json_get(f"{url}&page={page}&limit={limit}")
if not result_page:
break
result.extend(result_page)
if len(result_page) < limit:
break
page += 1
return result
def gitea_json_issue_get(issue_fullname):
"""
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)
def gitea_json_activities_get(username, date):
"""
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)
def gitea_json_issues_search(
type=None,
since=None,
before=None,
state=None,
labels=None,
created=False,
reviewed=False,
access_token=None,
verbose=True):
"""
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 not state:
# Also search closed issues:
issues.extend(gitea_json_issues_search(type=type,
since=since,
before=before,
state="closed",
labels=labels,
created=created,
reviewed=reviewed,
access_token=access_token,
verbose=False))
if verbose:
print(f"Total: {len(issues)} ", end="\n\n", flush=True)
return issues
def gitea_json_issue_events_filter(
issue_fullname,
date_start=None,
date_end=None,
username=None,
labels=None,
event_type=set()):
"""
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"...}.
: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 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"] in event_type:
pass
else:
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():
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

72
tools/triage/issues_needing_info.py Normal file
View File

@@ -0,0 +1,72 @@
#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0-or-later
"""
# This script prints the URLs of all opened issues labeled
# "Status/Needs Information from User" by the specified user
# and last updated more than 7 days ago.
Example usage:
python ./issues_needing_info.py --username mano-wii
"""
import argparse
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):
print(f"Needs information from user before {before}:")
label = "Status/Needs Information from User"
issues_json = gitea_json_issues_search(type="issues",
state="open",
before=before,
labels=label,
verbose=True)
for issue in issues_json:
fullname = issue["repository"]["full_name"]
number = issue["number"]
issue_events = gitea_json_issue_events_filter(
f"{fullname}/issues/{number}",
username=username,
labels={label})
if issue_events:
print(issue["html_url"])
print("concluded")
def main() -> None:
parser = argparse.ArgumentParser(
description="Print URL of Issues Needing Info",
epilog="This script is typically used to help triaging")
parser.add_argument(
"--username",
dest="username",
type=str,
required=False,
help="Username registred in Gitea")
args = parser.parse_args()
username = args.username
if not username:
username = git_username_detect()
if not username:
return
before_date = datetime.datetime.now() - datetime.timedelta(7)
print_needing_info_urls(username, f"{before_date.isoformat()}Z")
if __name__ == "__main__":
main()
# wait for input to close window
input()

274
tools/triage/weekly_report.py Normal file
View File

@@ -0,0 +1,274 @@
#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0-or-later
"""
# Generates the weekly report containing information on:
# - Pull Requests created,
# - Pull Requests revised,
# - Issues closed,
# - Issues confirmed,
# - Commits,
Example usage:
python ./weekly_report.py --username mano-wii
"""
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, git_username_detect
def argparse_create():
parser = argparse.ArgumentParser(
description="Generate Weekly Report",
epilog="This script is typically used to help write weekly reports")
parser.add_argument(
"--username",
dest="username",
metavar='USERNAME',
type=str,
required=False,
help="")
parser.add_argument(
"-v",
"--verbose",
action="store_true",
help="increase output verbosity")
return parser
def report_personal_weekly_get(username, start, verbose=True):
data_cache = {}
def gitea_json_issue_get_cached(issue_fullname):
if issue_fullname not in data_cache:
data_cache[issue_fullname] = gitea_json_issue_get(issue_fullname)
return data_cache[issue_fullname]
pulls_closed = set()
pulls_commented = set()
pulls_created = set()
issues_closed = set()
issues_commented = set()
issues_created = set()
pulls_reviewed = []
issues_confirmed = []
issues_needing_user_info = []
issues_needing_developer_info = []
issues_fixed = []
issues_duplicated = []
issues_archived = []
commits_main = []
for i in range(7):
date_curr = start + datetime.timedelta(days=i)
date_curr_str = date_curr.strftime("%Y-%m-%d")
print(f"Requesting activity of {date_curr_str}", end="\r", flush=True)
for activity in gitea_json_activities_get(username, date_curr_str):
op_type = activity["op_type"]
if op_type == "close_issue":
fullname = activity["repo"]["full_name"] + "/issues/" + activity["content"].split('|')[0]
issues_closed.add(fullname)
elif op_type == "comment_issue":
fullname = activity["repo"]["full_name"] + "/issues/" + activity["content"].split('|')[0]
issues_commented.add(fullname)
elif op_type == "create_issue":
fullname = activity["repo"]["full_name"] + "/issues/" + activity["content"].split('|')[0]
issues_created.add(fullname)
elif op_type == "merge_pull_request":
fullname = activity["repo"]["full_name"] + "/pulls/" + activity["content"].split('|')[0]
pulls_closed.add(fullname)
elif op_type == "comment_pull":
fullname = activity["repo"]["full_name"] + "/pulls/" + activity["content"].split('|')[0]
pulls_commented.add(fullname)
elif op_type == "create_pull_request":
fullname = activity["repo"]["full_name"] + "/pulls/" + activity["content"].split('|')[0]
pulls_created.add(fullname)
elif op_type == "commit_repo":
if activity["ref_name"] == "refs/heads/main":
content_json = json.loads(activity["content"])
repo_name = activity["repo"]["name"]
for commits in content_json["Commits"]:
title = commits["Message"].split('\n', 1)[0]
# Substitute occurrences of "#\d+" with "{{Issue|\d+|repo}}"
title = re.sub(r"#(\d+)", rf"{{{{Issue|\1|{repo_name}}}}}", title)
hash_value = commits["Sha1"][:10]
commits_main.append(f"{title} ({{{{GitCommit|{hash_value}|{repo_name}}}}})")
date_end = date_curr
len_total = len(issues_closed) + len(issues_commented) + len(pulls_commented)
process = 0
for issue in issues_commented:
print(f"[{int(100 * (process / len_total))}%] Checking issue {issue} ", end="\r", flush=True)
process += 1
issue_events = gitea_json_issue_events_filter(issue,
date_start=start,
date_end=date_end,
username=username,
labels={
"Status/Confirmed",
"Status/Needs Information from User",
"Status/Needs Info from Developers"})
for event in issue_events:
label_name = event["label"]["name"]
if label_name == "Status/Confirmed":
issues_confirmed.append(issue)
elif label_name == "Status/Needs Information from User":
issues_needing_user_info.append(issue)
elif label_name == "Status/Needs Info from Developers":
issues_needing_developer_info.append(issue)
for issue in issues_closed:
print(f"[{int(100 * (process / len_total))}%] Checking issue {issue} ", end="\r", flush=True)
process += 1
issue_events = gitea_json_issue_events_filter(issue,
date_start=start,
date_end=date_end,
username=username,
event_type={"close", "commit_ref"},
labels={"Status/Duplicate"})
for event in issue_events:
event_type = event["type"]
if event_type == "commit_ref":
issues_fixed.append(issue)
elif event_type == "label":
issues_duplicated.append(issue)
else:
issues_archived.append(issue)
for pull in pulls_commented:
print(f"[{int(100 * (process / len_total))}%] Checking pull {pull} ", end="\r", flush=True)
process += 1
pull_events = gitea_json_issue_events_filter(pull.replace("pulls", "issues"),
date_start=start,
date_end=date_end,
username=username,
event_type={"comment"})
if pull_events:
pull_data = gitea_json_issue_get_cached(pull)
if pull_data["user"]["login"] != username:
pulls_reviewed.append(pull)
# Print triaging stats
issues_involved = issues_closed | issues_commented | issues_created
print("\'\'\'Involved in %s reports:\'\'\' " % len(issues_involved))
print("* Confirmed: %s" % len(issues_confirmed))
print("* Closed as Resolved: %s" % len(issues_fixed))
print("* Closed as Archived: %s" % len(issues_archived))
print("* Closed as Duplicate: %s" % len(issues_duplicated))
print("* Needs Info from User: %s" % len(issues_needing_user_info))
print("* Needs Info from Developers: %s" % len(issues_needing_developer_info))
print("* Actions total: %s" % (len(issues_closed) + len(issues_commented) + len(issues_created)))
print()
# Print review stats
def print_pulls(pulls):
for pull in pulls:
pull_data = gitea_json_issue_get_cached(pull)
title = pull_data["title"]
_, repo, _, number = pull.split('/')
print(f"* {{{{PullRequest|{number}|{repo}}}}}: {title}")
print("'''Review: %s'''" % len(pulls_reviewed))
print_pulls(pulls_reviewed)
print()
# Print created diffs
print("'''Created pulls: %s'''" % len(pulls_created))
print_pulls(pulls_created)
print()
# Print commits
print("'''Commits:'''")
for commit in commits_main:
print("*", commit)
print()
if verbose:
# Debug
def print_links(issues):
for fullname in issues:
print(f"https://projects.blender.org/{fullname}")
print("Debug:")
print(f"Activities from {start.isoformat()} to {end.isoformat()}:")
print()
print("Pulls Created:")
print_links(pulls_created)
print("Pulls Reviewed:")
print_links(pulls_reviewed)
print("Issues Confirmed:")
print_links(issues_confirmed)
print("Issues Closed as Resolved:")
print_links(issues_fixed)
print("Issues Closed as Archived:")
print_links(issues_closed)
print("Issues Closed as Duplicate:")
print_links(issues_duplicated)
print("Issues Needing Info from User:")
print_links(issues_needing_user_info)
print("Issues Needing Info from Developers:")
print_links(issues_needing_developer_info)
def main() -> None:
# ----------
# Parse Args
args = argparse_create().parse_args()
username = args.username
if not username:
username = git_username_detect()
if not username:
return
#end_date = datetime.datetime(2020, 3, 14)
end_date = datetime.datetime.now()
weekday = end_date.weekday()
# Assuming I am lazy and making this at last moment or even later in worst case
if weekday < 2:
time_delta = 7 + weekday
start_date = end_date - datetime.timedelta(days=time_delta, hours=end_date.hour)
end_date -= datetime.timedelta(days=weekday, hours=end_date.hour)
else:
time_delta = weekday
start_date = end_date - datetime.timedelta(days=time_delta, hours=end_date.hour)
# Ensure friday :)
friday = start_date + datetime.timedelta(days=4)
week = start_date.isocalendar()[1]
start_date_str = start_date.strftime('%b %d')
end_date_str = friday.strftime('%b %d')
print("== Week %d (%s - %s) ==\n\n" % (week, start_date_str, end_date_str))
report_personal_weekly_get(username, start_date, verbose=args.verbose)
if __name__ == "__main__":
main()
# wait for input to close window
input()