#!/usr/bin/env python3 """ PR Tracker - Fetches PRs that need review using GitHub API. This module provides data about open PRs. Claude handles orchestration, review logic, and posting reviews. Usage: # Get PRs updated in the last 6 hours python3 scripts/pr_tracker.py --list --since 6h # Get PRs updated since a specific time python3 scripts/pr_tracker.py --list --since 2024-01-13T00:00:00Z # Get details for a specific PR python3 scripts/pr_tracker.py --details 123 """ import json import os import re from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Optional from github import Github, Auth # Configuration DEFAULT_REPO = "meta-pytorch/OpenEnv" DEFAULT_STATE_FILE = Path.home() / ".openenv-review-state.json" def _get_github_client() -> Github: """Get authenticated GitHub client.""" token = os.environ.get("GITHUB_TOKEN") if token: return Github(auth=Auth.Token(token)) # Try gh CLI token try: import subprocess result = subprocess.run( ["gh", "auth", "token"], capture_output=True, text=True, check=True, ) token = result.stdout.strip() return Github(auth=Auth.Token(token)) except (subprocess.CalledProcessError, FileNotFoundError): pass raise RuntimeError( "No GitHub token found. Set GITHUB_TOKEN env var or authenticate with 'gh auth login'" ) def parse_since(since_str: str) -> datetime: """ Parse a 'since' argument into a datetime. Accepts: - Duration: "6h", "1d", "30m", "2w" - ISO timestamp: "2024-01-13T00:00:00Z" """ # Try duration format first (e.g., "6h", "1d", "30m") duration_match = re.match(r"^(\d+)([mhdw])$", since_str.lower()) if duration_match: value = int(duration_match.group(1)) unit = duration_match.group(2) unit_map = { "m": timedelta(minutes=value), "h": timedelta(hours=value), "d": timedelta(days=value), "w": timedelta(weeks=value), } return datetime.now(timezone.utc) - unit_map[unit] # Try ISO format try: # Handle various ISO formats if since_str.endswith("Z"): since_str = since_str[:-1] + "+00:00" return datetime.fromisoformat(since_str) except ValueError: pass raise ValueError( f"Invalid 'since' format: {since_str}. " "Use duration (6h, 1d, 30m, 2w) or ISO timestamp (2024-01-13T00:00:00Z)" ) def get_prs_needing_review( repo: str = DEFAULT_REPO, since: Optional[datetime] = None, state_file: Optional[Path] = None, ) -> list[dict]: """ Get list of PRs that need review. Args: repo: Repository name (owner/repo) since: Only return PRs updated after this time state_file: Optional state file for SHA-based tracking (legacy) Returns: List of dicts with PR info: - number: PR number - title: PR title - author: Author username - url: PR URL - head_sha: Current commit SHA - updated_at: Last update time (ISO format) """ gh = _get_github_client() repo_obj = gh.get_repo(repo) # Load state for SHA tracking if provided repo_state = {} if state_file and state_file.exists(): try: state = json.loads(state_file.read_text()) repo_state = state.get(repo, {}) except json.JSONDecodeError: pass prs = [] for pr in repo_obj.get_pulls(state="open"): if pr.draft: continue # Filter by update time if 'since' provided if since and pr.updated_at < since: continue # If using state file, skip already-reviewed commits if state_file: pr_state = repo_state.get(str(pr.number), {}) if pr_state.get("last_reviewed_sha") == pr.head.sha: continue prs.append( { "number": pr.number, "title": pr.title, "author": pr.user.login, "url": pr.html_url, "head_sha": pr.head.sha, "updated_at": pr.updated_at.isoformat(), } ) return prs def get_pr_details(pr_number: int, repo: str = DEFAULT_REPO) -> dict: """Get detailed information about a specific PR.""" gh = _get_github_client() repo_obj = gh.get_repo(repo) pr = repo_obj.get_pull(pr_number) return { "number": pr.number, "title": pr.title, "author": pr.user.login, "url": pr.html_url, "head_sha": pr.head.sha, "updated_at": pr.updated_at.isoformat(), "body": pr.body or "", "files": [f.filename for f in pr.get_files()], "diff_url": pr.diff_url, "additions": pr.additions, "deletions": pr.deletions, "changed_files": pr.changed_files, } def record_review( pr_number: int, commit_sha: str, verdict: str, repo: str = DEFAULT_REPO, state_file: Path = DEFAULT_STATE_FILE, ): """Record that a PR was reviewed (for SHA-based tracking).""" state = {} if state_file.exists(): try: state = json.loads(state_file.read_text()) except json.JSONDecodeError: pass if repo not in state: state[repo] = {} state[repo][str(pr_number)] = { "last_reviewed_sha": commit_sha, "review_timestamp": datetime.now(timezone.utc).isoformat(), "verdict": verdict, } state_file.parent.mkdir(parents=True, exist_ok=True) state_file.write_text(json.dumps(state, indent=2)) state_file.chmod(0o600) def post_review( pr_number: int, verdict: str, body: str, repo: str = DEFAULT_REPO, ): """ Post a review to a PR. Args: pr_number: PR number verdict: One of "approve", "comment", "request_changes" body: Review body (markdown) """ gh = _get_github_client() repo_obj = gh.get_repo(repo) pr = repo_obj.get_pull(pr_number) event_map = { "approve": "APPROVE", "comment": "COMMENT", "request_changes": "REQUEST_CHANGES", } event = event_map.get(verdict, "COMMENT") # Can't approve own PR current_user = gh.get_user().login if event == "APPROVE" and pr.user.login == current_user: pr.create_issue_comment(body) return formatted_body = f"""> **Note**: This is an automated review by **Claude Code**, not a human review. --- {body} --- *Automated review by Claude Code | [Learn more](https://github.com/meta-pytorch/OpenEnv/blob/main/CLAUDE.md)*""" pr.create_review(body=formatted_body, event=event) if __name__ == "__main__": import argparse parser = argparse.ArgumentParser( description="PR Tracker - Fetch PRs needing review", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: # PRs updated in last 6 hours python3 pr_tracker.py --list --since 6h # PRs updated in last day python3 pr_tracker.py --list --since 1d # PRs updated since specific time python3 pr_tracker.py --list --since 2024-01-13T00:00:00Z # Get details for PR #123 python3 pr_tracker.py --details 123 """, ) parser.add_argument("--list", action="store_true", help="List PRs needing review") parser.add_argument( "--details", type=int, metavar="PR", help="Get details for specific PR" ) parser.add_argument( "--since", type=str, help="Only PRs updated since (e.g., 6h, 1d, 2024-01-13T00:00:00Z)", ) parser.add_argument( "--repo", default=DEFAULT_REPO, help="Repository (default: %(default)s)" ) parser.add_argument( "--use-state", action="store_true", help="Also filter by SHA state file" ) args = parser.parse_args() if args.list: since = parse_since(args.since) if args.since else None state_file = DEFAULT_STATE_FILE if args.use_state else None prs = get_prs_needing_review(repo=args.repo, since=since, state_file=state_file) print(json.dumps(prs, indent=2)) elif args.details: details = get_pr_details(args.details, repo=args.repo) print(json.dumps(details, indent=2)) else: parser.print_help()