carbon-language/carbon-lang
1
0
Fork 0
mirror of https://github.com/carbon-language/carbon-lang.git synced 2026-09-24 21:20:11 +01:00
Code Issues Packages Projects Releases 10 Wiki Activity
Files
55dcf665845296b9791ad9655c30de660c7ed1a4
carbon-lang/github_tools/pr_comments.py
T
Dana Jansens 233a58fcb1 Fix ty errors for missing generic parameters (#7326) 
Mostly these errors were around `dict` missing arguments, and they are
almost always `[str, Any]`.

But a real thorn here was `xml.etree.ElementTree.Element`. `ty` insists
that this is a generic type, and indeed it appears to be one, or
becoming one, in some python version. But it is not generic in python
3.12. So we are stuck in an unsolvable land where:
- `ty` gives an error unless you write `[str]` on the type, because it
thinks it is generic.
- python3.12 gives an error if you do write `[str]` on the type, because
it thinks it is not generic.

Forcing `ty` to target exactly python 3.12 does not help. So I have just
used a linter-ignore comment on that line.
2026-06-09 15:08:39 +00:00

485 lines
14 KiB
Python
Executable File
Raw Blame History

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.12"
# dependencies = [
# "gql==2.0.0",
# "requests",
# ]
# ///
"""Figure out comments on a GitHub PR."""
__copyright__ = """
Part of the Carbon Language project, under the Apache License v2.0 with LLVM
Exceptions. See /LICENSE for license information.
SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
"""
import argparse
import datetime
import hashlib
import importlib.util
import os
import textwrap
from typing import Any, Callable, Optional, override
# Do some extra work to support direct runs.
try:
from github_tools import github_helpers
except ImportError:
github_helpers_spec = importlib.util.spec_from_file_location(
"github_helpers",
os.path.join(os.path.dirname(__file__), "github_helpers.py"),
)
assert github_helpers_spec is not None
github_helpers: Any = importlib.util.module_from_spec(github_helpers_spec)
github_helpers_spec.loader.exec_module(github_helpers) # type: ignore
# The main query, into which other queries are composed.
_QUERY = """
{
repository(owner: "carbon-language", name: "%(repo)s") {
pullRequest(number: %(pr_num)d) {
author {
login
}
createdAt
title
%(comments)s
%(reviews)s
%(review_threads)s
}
}
}
"""
# Queries for comments on the PR. These are direct, non-review comments on the
# PR.
_QUERY_COMMENTS = """
comments(first: 100%(cursor)s) {
nodes {
author {
login
}
body
createdAt
url
}
%(pagination)s
}
"""
# Queries for reviews on the PR, which have a non-empty body if a review has
# a summary comment.
_QUERY_REVIEWS = """
reviews(first: 100%(cursor)s) {
nodes {
author {
login
}
body
createdAt
url
}
%(pagination)s
}
"""
# Queries for review threads on the PR.
_QUERY_REVIEW_THREADS = """
reviewThreads(first: 100%(cursor)s) {
nodes {
comments(first: 100) {
nodes {
author {
login
}
body
createdAt
originalPosition
originalCommit {
abbreviatedOid
}
path
}
}
isResolved
resolvedBy {
createdAt
login
}
}
%(pagination)s
}
"""
class _Comment:
"""A comment, either on a review thread or top-level on the PR."""
def __init__(self, author: str, timestamp: str, body: str):
self.author = author
self.timestamp = datetime.datetime.strptime(
timestamp, "%Y-%m-%dT%H:%M:%SZ"
)
self.body = body
@staticmethod
def from_raw_comment(raw_comment: dict[str, Any]) -> "_Comment":
"""Creates the comment from a raw comment dict."""
return _Comment(
raw_comment["author"]["login"],
raw_comment["createdAt"],
raw_comment["body"],
)
@staticmethod
def _rewrap(content: str) -> str:
"""Rewraps a comment to fit in 80 columns with an indent."""
lines = []
for line in content.split("\n"):
lines.extend(
[
x
for x in textwrap.wrap(
line,
width=80,
initial_indent=" " * 4,
subsequent_indent=" " * 4,
)
]
)
return "\n".join(lines)
def format(self, long: bool) -> str:
"""Formats the comment."""
if long:
return "%s%s at %s:\n%s" % (
" " * 2,
self.author,
self.timestamp.strftime("%Y-%m-%d %H:%M"),
self._rewrap(self.body),
)
else:
# Compact newlines down into pilcrows, leaving a space after.
body = self.body.replace("\r", "").replace("\n", "¶ ")
while "¶ ¶" in body:
body = body.replace("¶ ¶", "¶¶")
line = "%s%s: %s" % (" " * 2, self.author, body)
return line if len(line) <= 80 else line[:77] + "..."
class _PRComment(_Comment):
"""A comment on the top-level PR."""
def __init__(self, raw_comment: dict[str, Any]):
super().__init__(
raw_comment["author"]["login"],
raw_comment["createdAt"],
raw_comment["body"],
)
self.url = raw_comment["url"]
def __lt__(self, other: "_PRComment") -> bool:
return self.timestamp < other.timestamp
@override
def format(self, long: bool) -> str:
return "%s\n%s" % (self.url, super().format(long))
class _Thread:
"""A review thread on a line of code."""
def __init__(self, parsed_args: argparse.Namespace, thread: dict[str, Any]):
self.is_resolved: bool = thread["isResolved"]
comments = thread["comments"]["nodes"]
first_comment = comments[0]
self.line: int = first_comment["originalPosition"]
self.path: str = first_comment["path"]
# Link to the comment in the commit; GitHub features work better there
# than in the conversation view. The diff_url allows viewing changes
# since the comment, although the comment won't be visible there.
template = (
"https://github.com/carbon-language/%(repo)s/pull/%(pr_num)s/"
"files/%(oid)s%(head)s#diff-%(path_md5)s%(line_side)s%(line)s"
)
# GitHub uses an md5 of the file's path for the link.
path_md5 = hashlib.md5()
path_md5.update(bytearray(self.path, "utf-8"))
format_dict = {
"head": "",
"line_side": "R",
"line": self.line,
"oid": first_comment["originalCommit"]["abbreviatedOid"],
"path_md5": path_md5.hexdigest(),
"pr_num": parsed_args.pr_num,
"repo": parsed_args.repo,
}
self.url: str = template % format_dict
format_dict["head"] = "..HEAD"
format_dict["line_side"] = "L"
self.diff_url: str = template % format_dict
self.comments = [
_Comment.from_raw_comment(comment)
for comment in thread["comments"]["nodes"]
]
if self.is_resolved:
self.comments.append(
_Comment(
thread["resolvedBy"]["login"],
thread["resolvedBy"]["createdAt"],
"<resolved>",
)
)
def __lt__(self, other: "_Thread") -> bool:
"""Sort threads by line then timestamp."""
if self.line != other.line:
return bool(self.line < other.line)
return self.comments[0].timestamp < other.comments[0].timestamp
def format(self, long: bool) -> str:
"""Formats the review thread with comments."""
lines = [
"%s\n - line %d; %s"
% (
self.url,
self.line,
("resolved" if self.is_resolved else "unresolved"),
)
]
if self.diff_url:
lines.append(" - diff: %s" % self.diff_url)
for comment in self.comments:
lines.append(comment.format(long))
return "\n".join(lines)
def has_comment_from(self, comments_from: str) -> bool:
"""Returns true if comments has a comment from comments_from."""
for comment in self.comments:
if comment.author == comments_from:
return True
return False
def _parse_args(args: Optional[list[str]] = None) -> argparse.Namespace:
"""Parses command-line arguments and flags."""
parser = argparse.ArgumentParser(description="Lists comments on a PR.")
parser.add_argument(
"pr_num",
metavar="PR#",
type=int,
help="The pull request to fetch comments from.",
)
github_helpers.add_access_token_arg(parser, "repo")
parser.add_argument(
"--comments-after",
metavar="LOGIN",
help="Only print threads where the final comment is not from the given "
"user. For example, use when looking for threads that you still need "
"to respond to.",
)
parser.add_argument(
"--comments-from",
metavar="LOGIN",
help="Only print threads with comments from the given user. For "
"example, use when looking for threads that you've commented on.",
)
parser.add_argument(
"--include-resolved",
action="store_true",
help="Whether to include resolved review threads. By default, only "
"unresolved threads will be shown.",
)
parser.add_argument(
"--repo",
choices=["carbon-lang"],
default="carbon-lang",
help="The Carbon repo to query. Defaults to %(default)s.",
)
parser.add_argument(
"--long",
action="store_true",
help="Prints long output, with the full comment.",
)
return parser.parse_args(args=args)
def _query(
parsed_args: argparse.Namespace, field_name: Optional[str] = None
) -> str:
"""Returns a query for the passed field_name, or all by default."""
print(".", end="", flush=True)
format = {
"pr_num": parsed_args.pr_num,
"repo": parsed_args.repo,
"comments": "",
"review_threads": "",
"reviews": "",
}
if field_name:
# Use a cursor for pagination of the field.
if field_name == "comments":
format["comments"] = _QUERY_COMMENTS
elif field_name == "reviewThreads":
format["review_threads"] = _QUERY_REVIEW_THREADS
elif field_name == "reviews":
format["reviews"] = _QUERY_REVIEWS
else:
raise ValueError("Unexpected field_name: %s" % field_name)
else:
# Fetch the first page of all fields.
subformat = {"cursor": "", "pagination": github_helpers.PAGINATION}
format["comments"] = _QUERY_COMMENTS % subformat
format["review_threads"] = _QUERY_REVIEW_THREADS % subformat
format["reviews"] = _QUERY_REVIEWS % subformat
return _QUERY % format
def _accumulate_pr_comment(
parsed_args: argparse.Namespace,
comments: list[_PRComment],
raw_comment: dict[str, Any],
) -> None:
"""Collects top-level comments and reviews."""
# Elide reviews that have no top-level comment body.
if raw_comment["body"]:
comments.append(_PRComment(raw_comment))
def _accumulate_thread(
parsed_args: argparse.Namespace,
threads_by_path: dict[str, list[_Thread]],
raw_thread: dict[str, Any],
) -> None:
"""Adds threads to threads_by_path for later sorting."""
thread = _Thread(parsed_args, raw_thread)
# Optionally skip resolved threads.
if not parsed_args.include_resolved and thread.is_resolved:
return
# Optionally skip threads where the given user isn't the last commenter.
if (
parsed_args.comments_after
and thread.comments[-1].author == parsed_args.comments_after
):
return
# Optionally skip threads where the given user hasn't commented.
if parsed_args.comments_from and not thread.has_comment_from(
parsed_args.comments_from
):
return
if thread.path not in threads_by_path:
threads_by_path[thread.path] = []
threads_by_path[thread.path].append(thread)
def _paginate(
field_name: str,
accumulator: Callable[[argparse.Namespace, Any, dict[str, Any]], None],
parsed_args: argparse.Namespace,
client: github_helpers.Client,
main_result: dict[str, Any],
output: Any,
) -> None:
"""Paginates through the given field_name, accumulating results."""
query = _query(parsed_args, field_name=field_name)
path = ("repository", "pullRequest", field_name)
for node in client.execute_and_paginate(
query, path, first_page=main_result
):
accumulator(parsed_args, output, node)
def _fetch_comments(
parsed_args: argparse.Namespace,
) -> tuple[list[_PRComment], dict[str, list[_Thread]]]:
"""Fetches comments and review threads from GitHub."""
# Each _query call will print a '.' for progress.
print(
"Loading https://github.com/carbon-language/%s/pull/%d ..."
% (parsed_args.repo, parsed_args.pr_num),
end="",
flush=True,
)
client = github_helpers.Client(parsed_args)
# Get the initial set of review threads, and print the PR summary.
main_result = client.execute(_query(parsed_args))
pull_request = main_result["repository"]["pullRequest"]
# Paginate comments, reviews, and review threads.
comments: list[_PRComment] = []
_paginate(
"comments",
_accumulate_pr_comment,
parsed_args,
client,
main_result,
comments,
)
# Combine reviews into comments for interleaving.
_paginate(
"reviews",
_accumulate_pr_comment,
parsed_args,
client,
main_result,
comments,
)
threads_by_path: dict[str, list[_Thread]] = {}
_paginate(
"reviewThreads",
_accumulate_thread,
parsed_args,
client,
main_result,
threads_by_path,
)
# Now that loading is done (no more progress indicators), print the header.
print()
pr_desc = _Comment(
pull_request["author"]["login"],
pull_request["createdAt"],
pull_request["title"],
)
print(pr_desc.format(parsed_args.long))
return comments, threads_by_path
def main() -> None:
parsed_args = _parse_args()
comments, threads_by_path = _fetch_comments(parsed_args)
for comment in sorted(comments):
print()
print(comment.format(parsed_args.long))
for path, threads in sorted(threads_by_path.items()):
# Print a header for each path.
print()
print("=" * 80)
print(path)
print("=" * 80)
for thread in sorted(threads):
print()
print(thread.format(parsed_args.long))
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink