|
#!/usr/bin/env python |
|
""" |
|
GitHub PR Activity Tracker |
|
|
|
Tracks Pull Requests where a specific user has been active in the last N days. |
|
Prints a markdown-formatted list of PRs with titles and links to stdout. |
|
|
|
Usage: |
|
python pr_activity_tracker.py --username <github_username> --repo <owner/repo> [--days 7] [--token <token>] |
|
python pr_activity_tracker.py --username <github_username> --repo <owner/repo> > output.md |
|
|
|
Environment Variables: |
|
GITHUB_TOKEN: GitHub Personal Access Token (alternative to --token) |
|
""" |
|
|
|
import argparse |
|
import os |
|
import sys |
|
from datetime import datetime, timedelta, timezone |
|
from collections import defaultdict |
|
|
|
from github import Github, Auth |
|
from github.GithubException import GithubException, RateLimitExceededException |
|
|
|
|
|
class PRActivityTracker: |
|
"""Track user activity on Pull Requests.""" |
|
|
|
def __init__(self, token, repo_full_name, username, days=7): |
|
""" |
|
Initialize the tracker. |
|
|
|
Args: |
|
token: GitHub Personal Access Token |
|
repo_full_name: Repository in format "owner/repo" |
|
username: GitHub username to track |
|
days: Number of days to look back (default: 7) |
|
""" |
|
auth = Auth.Token(token) |
|
self.github = Github(auth=auth) |
|
self.repo = self.github.get_repo(repo_full_name) |
|
self.username = username |
|
self.days = days |
|
self.cutoff_date = datetime.now(timezone.utc) - timedelta(days=days) |
|
|
|
# Store PR activities: PR number -> set of activity types |
|
self.pr_activities = defaultdict(set) |
|
# Store PR objects: PR number -> PR object |
|
self.prs = {} |
|
|
|
def get_created_prs(self): |
|
"""Find PRs created by the user in the timeframe.""" |
|
print(f"Checking PRs created by {self.username}...", file=sys.stderr) |
|
|
|
# Search for PRs authored by user |
|
pulls = self.repo.get_pulls(state='open', sort='created', direction='desc') |
|
|
|
for pr in pulls: |
|
# Stop if we've gone past the cutoff date |
|
if pr.created_at < self.cutoff_date: |
|
break |
|
|
|
if pr.user.login == self.username: |
|
self.pr_activities[pr.number].add('created') |
|
self.prs[pr.number] = pr |
|
|
|
def get_commented_prs(self): |
|
"""Find PRs where the user commented in the timeframe.""" |
|
print(f"Checking comments by {self.username}...", file=sys.stderr) |
|
|
|
# Get all issue comments (includes PR comments) |
|
# Note: Review comments are handled separately |
|
comments = self.repo.get_issues_comments(sort='created', direction='desc', since=self.cutoff_date) |
|
|
|
for comment in comments: |
|
if comment.user.login == self.username: |
|
# Check if this comment is on a PR (not an issue) |
|
issue_number = int(comment.issue_url.split('/')[-1]) |
|
|
|
try: |
|
pr = self.repo.get_pull(issue_number) |
|
# This is a PR, not an issue |
|
self.pr_activities[pr.number].add('commented') |
|
if pr.number not in self.prs: |
|
self.prs[pr.number] = pr |
|
except GithubException: |
|
# This is an issue, not a PR - skip it |
|
pass |
|
|
|
def get_reviewed_prs(self): |
|
"""Find PRs where the user submitted reviews in the timeframe.""" |
|
print(f"Checking reviews by {self.username}...", file=sys.stderr) |
|
|
|
# Get recent PRs and check for user's reviews |
|
# We need to check more PRs since reviews can be on older PRs |
|
pulls = self.repo.get_pulls(state='open', sort='updated', direction='desc') |
|
|
|
checked_count = 0 |
|
max_checks = 200 # Limit to avoid excessive API calls |
|
|
|
for pr in pulls: |
|
checked_count += 1 |
|
if checked_count > max_checks: |
|
break |
|
|
|
# Check reviews on this PR |
|
try: |
|
reviews = pr.get_reviews() |
|
for review in reviews: |
|
if review.user.login == self.username and review.submitted_at >= self.cutoff_date: |
|
self.pr_activities[pr.number].add('reviewed') |
|
if pr.number not in self.prs: |
|
self.prs[pr.number] = pr |
|
break # Found a recent review from user, no need to check more |
|
except GithubException: |
|
# Some PRs might not have reviews accessible |
|
pass |
|
|
|
def get_committed_prs(self): |
|
"""Find PRs where the user pushed commits in the timeframe.""" |
|
print(f"Checking commits by {self.username}...", file=sys.stderr) |
|
|
|
# Get commits by the user |
|
commits = self.repo.get_commits(author=self.username, since=self.cutoff_date) |
|
|
|
for commit in commits: |
|
# Check if this commit is associated with any PRs |
|
try: |
|
pulls = commit.get_pulls() |
|
for pr in pulls: |
|
self.pr_activities[pr.number].add('committed') |
|
if pr.number not in self.prs: |
|
self.prs[pr.number] = pr |
|
except GithubException: |
|
# Some commits might not have associated PRs |
|
pass |
|
|
|
def collect_all_activities(self): |
|
"""Collect all PR activities for the user.""" |
|
try: |
|
self.get_created_prs() |
|
self.get_commented_prs() |
|
self.get_reviewed_prs() |
|
self.get_committed_prs() |
|
except RateLimitExceededException: |
|
print("ERROR: GitHub API rate limit exceeded. Please try again later.", file=sys.stderr) |
|
sys.exit(1) |
|
except GithubException as e: |
|
print(f"ERROR: GitHub API error: {e}", file=sys.stderr) |
|
sys.exit(1) |
|
|
|
def generate_markdown_output(self): |
|
"""Generate markdown-formatted list of PRs.""" |
|
if not self.prs: |
|
return f"No PR activity found for user @{self.username} in the last {self.days} days.\n" |
|
|
|
# Filter to only open PRs |
|
open_prs = {num: pr for num, pr in self.prs.items() if pr.state == "open"} |
|
|
|
if not open_prs: |
|
return f"No open PR activity found for user @{self.username} in the last {self.days} days.\n" |
|
|
|
# Sort PRs by number (descending) |
|
sorted_pr_numbers = sorted(open_prs.keys(), reverse=True) |
|
|
|
lines = [ |
|
f"# PR Activity for @{self.username}", |
|
f"", |
|
f"**Repository:** {self.repo.full_name}", |
|
f"**Period:** Last {self.days} days (since {self.cutoff_date.strftime('%Y-%m-%d')})", |
|
f"**Involved in {len(open_prs)} open PRs.**", |
|
f"", |
|
] |
|
|
|
for pr_number in sorted_pr_numbers: |
|
pr = open_prs[pr_number] |
|
activities = sorted(self.pr_activities[pr_number]) |
|
activity_str = ", ".join(activities) |
|
|
|
lines.append(f"- **[#{pr.number}]({pr.html_url})** {pr.title}") |
|
lines.append(f" - *Activity:* {activity_str}") |
|
lines.append("") |
|
|
|
return "\n".join(lines) |
|
|
|
|
|
def main(): |
|
"""Main entry point.""" |
|
parser = argparse.ArgumentParser( |
|
description="Track Pull Request activity for a GitHub user", |
|
formatter_class=argparse.RawDescriptionHelpFormatter, |
|
epilog=""" |
|
Examples: |
|
python pr_activity_tracker.py --username octocat --repo scikit-learn/scikit-learn |
|
python pr_activity_tracker.py --username octocat --repo scikit-learn/scikit-learn --days 14 |
|
python pr_activity_tracker.py --username octocat --repo scikit-learn/scikit-learn --token ghp_xxx > output.md |
|
""" |
|
) |
|
|
|
parser.add_argument( |
|
'--username', '-u', |
|
required=True, |
|
help='GitHub username to track' |
|
) |
|
|
|
parser.add_argument( |
|
'--repo', '-r', |
|
required=True, |
|
help='Repository in format "owner/repo" (e.g., "scikit-learn/scikit-learn")' |
|
) |
|
|
|
parser.add_argument( |
|
'--days', '-d', |
|
type=int, |
|
default=7, |
|
help='Number of days to look back (default: 7)' |
|
) |
|
|
|
parser.add_argument( |
|
'--token', '-t', |
|
help='GitHub Personal Access Token (or set GITHUB_TOKEN environment variable)' |
|
) |
|
|
|
args = parser.parse_args() |
|
|
|
# Get token from args or environment |
|
token = args.token or os.environ.get('GITHUB_TOKEN') |
|
if not token: |
|
print("ERROR: GitHub token required. Provide via --token or GITHUB_TOKEN environment variable.", file=sys.stderr) |
|
print("", file=sys.stderr) |
|
print("To create a token:", file=sys.stderr) |
|
print("1. Go to https://github.com/settings/tokens", file=sys.stderr) |
|
print("2. Click 'Generate new token (classic)'", file=sys.stderr) |
|
print("3. Select scopes: 'repo' (for private repos) or 'public_repo' (for public only)", file=sys.stderr) |
|
print("4. Copy the token and use it with --token or export GITHUB_TOKEN=<token>", file=sys.stderr) |
|
sys.exit(1) |
|
|
|
# Validate repo format |
|
if '/' not in args.repo: |
|
print(f"ERROR: Invalid repo format '{args.repo}'. Expected format: 'owner/repo'", file=sys.stderr) |
|
sys.exit(1) |
|
|
|
print(f"Tracking PR activity for @{args.username} in {args.repo}...", file=sys.stderr) |
|
print(f"Looking back {args.days} days...", file=sys.stderr) |
|
print("", file=sys.stderr) |
|
|
|
# Create tracker and collect activities |
|
tracker = PRActivityTracker(token, args.repo, args.username, args.days) |
|
tracker.collect_all_activities() |
|
|
|
# Generate and print output |
|
markdown_output = tracker.generate_markdown_output() |
|
|
|
print("", file=sys.stderr) |
|
print("=" * 80, file=sys.stderr) |
|
print("", file=sys.stderr) |
|
print(markdown_output) |
|
|
|
|
|
if __name__ == '__main__': |
|
main() |
|
|