Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
191 changes: 135 additions & 56 deletions .github/actions/bot-autoassign/stale_pr_bot.py
Original file line number Diff line number Diff line change
@@ -1,10 +1,12 @@
import time
from collections import deque
from datetime import datetime, timezone

from base import GitHubBot
from utils import unassign_linked_issues_helper

# GitHub author_association values that represent project maintainers.
MAINTAINER_ROLES = frozenset({"OWNER", "MEMBER", "COLLABORATOR"})


class StalePRBot(GitHubBot):
def __init__(self):
Expand All @@ -13,6 +15,54 @@ def __init__(self):
self.DAYS_BEFORE_UNASSIGN = 14
self.DAYS_BEFORE_CLOSE = 60

def _get_last_author_activity(
self,
pr,
after_date,
issue_comments=None,
all_reviews=None,
review_comments=None,
):
"""Return the datetime of the PR author's latest activity after *after_date*.

Returns ``None`` when the author has not acted since *after_date*.
"""
pr_author = pr.user.login if pr.user else None
if not pr_author:
return None
last_activity = None
for commit in pr.get_commits():
commit_date = commit.commit.author.date
if commit_date > after_date:
if commit.author and commit.author.login == pr_author:
if not last_activity or commit_date > last_activity:
last_activity = commit_date
if issue_comments is None:
issue_comments = list(pr.get_issue_comments())
for comment in issue_comments:
if comment.user and comment.user.login == pr_author:
comment_date = comment.created_at
if comment_date > after_date:
if not last_activity or comment_date > last_activity:
last_activity = comment_date
if review_comments is None:
review_comments = list(pr.get_review_comments())
for comment in review_comments:
if comment.user and comment.user.login == pr_author:
comment_date = comment.created_at
if comment_date > after_date:
if not last_activity or comment_date > last_activity:
last_activity = comment_date
if all_reviews is None:
all_reviews = list(pr.get_reviews())
for review in all_reviews:
if review.user and review.user.login == pr_author:
review_date = review.submitted_at
if review_date and review_date > after_date:
if not last_activity or review_date > last_activity:
last_activity = review_date
return last_activity

def get_days_since_activity(
self,
pr,
Expand All @@ -23,78 +73,95 @@ def get_days_since_activity(
):
if not last_changes_requested:
return 0
try:
last_author_activity = self._get_last_author_activity(
pr,
last_changes_requested,
issue_comments,
all_reviews,
review_comments,
)
reference_date = last_author_activity or last_changes_requested
now = datetime.now(timezone.utc)
return (now - reference_date).days
except Exception as e:
print("Error calculating activity" f" for PR #{pr.number}: {e}")
return 0

def is_waiting_for_maintainer(
self,
pr,
last_changes_requested,
issue_comments=None,
all_reviews=None,
review_comments=None,
):
"""Return True when the contributor has responded but no maintainer has acted since.

The bot should not warn, mark stale, or close a PR when the ball
is in the maintainers' court.
"""
try:
pr_author = pr.user.login if pr.user else None
if not pr_author:
return 0
last_author_activity = None
commits = deque(pr.get_commits(), maxlen=50)
for commit in commits:
commit_date = commit.commit.author.date
if commit_date > last_changes_requested:
if commit.author and commit.author.login == pr_author:
if (
not last_author_activity
or commit_date > last_author_activity
):
last_author_activity = commit_date
return False
last_author_activity = self._get_last_author_activity(
pr,
last_changes_requested,
issue_comments,
all_reviews,
review_comments,
)
if not last_author_activity:
return False
# Check for maintainer activity after the contributor's last action.
# Only OWNER / MEMBER / COLLABORATOR responses count; random
# community comments and bot messages do not.
if issue_comments is None:
issue_comments = list(pr.get_issue_comments())
comments = (
issue_comments[-20:] if len(issue_comments) > 20 else issue_comments
)
for comment in comments:
if comment.user and comment.user.login == pr_author:
comment_date = comment.created_at
if comment_date > last_changes_requested:
if (
not last_author_activity
or comment_date > last_author_activity
):
last_author_activity = comment_date
for comment in issue_comments:
if (
comment.user
and comment.user.login != pr_author
and comment.user.type != "Bot"
and getattr(comment, "author_association", None) in MAINTAINER_ROLES
and comment.created_at > last_author_activity
):
return False
if review_comments is None:
review_comments = list(pr.get_review_comments())
all_review_comments = review_comments
review_comments = (
all_review_comments[-20:]
if len(all_review_comments) > 20
else all_review_comments
)
for comment in review_comments:
if comment.user and comment.user.login == pr_author:
comment_date = comment.created_at
if comment_date > last_changes_requested:
if (
not last_author_activity
or comment_date > last_author_activity
):
last_author_activity = comment_date
if (
comment.user
and comment.user.login != pr_author
and comment.user.type != "Bot"
and getattr(comment, "author_association", None) in MAINTAINER_ROLES
and comment.created_at > last_author_activity
):
return False
if all_reviews is None:
all_reviews = list(pr.get_reviews())
reviews = all_reviews[-20:] if len(all_reviews) > 20 else all_reviews
for review in reviews:
if review.user and review.user.login == pr_author:
review_date = review.submitted_at
if review_date and review_date > last_changes_requested:
if (
not last_author_activity
or review_date > last_author_activity
):
last_author_activity = review_date
reference_date = last_author_activity or last_changes_requested
now = datetime.now(timezone.utc)
return (now - reference_date).days
for review in all_reviews:
if (
review.user
and review.user.login != pr_author
and review.user.type != "Bot"
and getattr(review, "author_association", None) in MAINTAINER_ROLES
and review.submitted_at
and review.submitted_at > last_author_activity
):
return False
return True
except Exception as e:
print("Error calculating activity" f" for PR #{pr.number}: {e}")
return 0
print("Error checking maintainer activity" f" for PR #{pr.number}: {e}")
return False

def get_last_changes_requested(self, pr, all_reviews=None):
try:
if all_reviews is None:
all_reviews = list(pr.get_reviews())
reviews = all_reviews[-50:] if len(all_reviews) > 50 else all_reviews
changes_requested_reviews = [
r for r in reviews if r.state == "CHANGES_REQUESTED"
r for r in all_reviews if r.state == "CHANGES_REQUESTED"
]
if not changes_requested_reviews:
return None
Expand Down Expand Up @@ -348,6 +415,18 @@ def process_stale_prs(self):
f"PR #{pr.number}: {days_inactive}"
" days since contributor activity"
)
if self.is_waiting_for_maintainer(
pr,
last_changes_requested,
issue_comments,
all_reviews,
review_comments,
):
print(
f"PR #{pr.number}: waiting for"
" maintainer review, skipping"
)
continue
if days_inactive >= self.DAYS_BEFORE_CLOSE:
if self.close_stale_pr(pr, days_inactive):
processed_count += 1
Expand Down
Loading
Loading