245 lines
9.2 KiB
Python
245 lines
9.2 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
PR-Agent Auto-Approve + Auto-Merge Bot
|
|
Runs periodically (cron), finds open PRs that have been reviewed by PR-Agent,
|
|
approves them and enables auto-merge.
|
|
"""
|
|
import os, sys, json, time, hmac, hashlib, asyncio
|
|
from pathlib import Path
|
|
|
|
# ── Config ──
|
|
APP_ID = os.environ.get("GITHUB_APP_ID", "4319749")
|
|
PRIVATE_KEY_PATH = os.environ.get("PRIVATE_KEY_PATH", "/var/lib/pr-agent-server/private-key.pem")
|
|
WEBHOOK_SECRET = os.environ.get("GITHUB_WEBHOOK_SECRET", "")
|
|
BASE_URL = os.environ.get("GITHUB_API_BASE", "https://api.github.com")
|
|
|
|
def get_jwt():
|
|
import jwt as pyjwt
|
|
with open(PRIVATE_KEY_PATH) as f:
|
|
key = f.read()
|
|
now = int(time.time())
|
|
payload = {"iat": now - 60, "exp": now + 600, "iss": APP_ID}
|
|
return pyjwt.encode(payload, key, algorithm="RS256")
|
|
|
|
def get_installation_token(installation_id: int) -> str:
|
|
"""Get installation access token"""
|
|
import httpx
|
|
jwt_token = get_jwt()
|
|
with httpx.Client() as client:
|
|
r = client.post(
|
|
f"{BASE_URL}/app/installations/{installation_id}/access_tokens",
|
|
headers={"Authorization": f"Bearer {jwt_token}", "Accept": "application/vnd.github.v3+json"}
|
|
)
|
|
return r.json().get("token", "")
|
|
|
|
def get_all_installations() -> list:
|
|
"""Get all app installations"""
|
|
jwt_token = get_jwt()
|
|
import httpx
|
|
with httpx.Client() as client:
|
|
r = client.get(
|
|
f"{BASE_URL}/app/installations",
|
|
headers={"Authorization": f"Bearer {jwt_token}", "Accept": "application/vnd.github.v3+json"}
|
|
)
|
|
return r.json()
|
|
|
|
def get_installation_repos(installation_id: int, token: str) -> list:
|
|
"""Get repos for an installation"""
|
|
import httpx
|
|
with httpx.Client() as client:
|
|
r = client.get(
|
|
f"{BASE_URL}/installation/repositories",
|
|
headers={"Authorization": f"token {token}", "Accept": "application/vnd.github.v3+json"}
|
|
)
|
|
return r.json().get("repositories", [])
|
|
|
|
def get_open_prs(token: str, repo_full: str) -> list:
|
|
"""Get open PRs in a repo"""
|
|
import httpx
|
|
with httpx.Client() as client:
|
|
r = client.get(
|
|
f"{BASE_URL}/repos/{repo_full}/pulls?state=open&sort=updated&direction=desc",
|
|
headers={"Authorization": f"token {token}", "Accept": "application/vnd.github.v3+json"}
|
|
)
|
|
return r.json()
|
|
|
|
def get_pr_reviews(token: str, repo_full: str, pr_number: int) -> list:
|
|
"""Get reviews for a PR"""
|
|
import httpx
|
|
with httpx.Client() as client:
|
|
r = client.get(
|
|
f"{BASE_URL}/repos/{repo_full}/pulls/{pr_number}/reviews",
|
|
headers={"Authorization": f"token {token}", "Accept": "application/vnd.github.v3+json"}
|
|
)
|
|
return r.json()
|
|
|
|
def get_pr_comments(token: str, repo_full: str, pr_number: int) -> list:
|
|
"""Get issue comments for a PR"""
|
|
import httpx
|
|
with httpx.Client() as client:
|
|
r = client.get(
|
|
f"{BASE_URL}/repos/{repo_full}/issues/{pr_number}/comments",
|
|
headers={"Authorization": f"token {token}", "Accept": "application/vnd.github.v3+json"}
|
|
)
|
|
return r.json()
|
|
|
|
def approve_pr(token: str, repo_full: str, pr_number: int) -> bool:
|
|
"""Submit APPROVE review"""
|
|
import httpx
|
|
with httpx.Client() as client:
|
|
r = client.post(
|
|
f"{BASE_URL}/repos/{repo_full}/pulls/{pr_number}/reviews",
|
|
headers={"Authorization": f"token {token}", "Accept": "application/vnd.github.v3+json"},
|
|
json={"event": "APPROVE", "body": "✅ Auto-approved by PR-Agent bot."}
|
|
)
|
|
return r.status_code == 200
|
|
|
|
def merge_pr(token: str, repo_full: str, pr_number: int) -> tuple:
|
|
"""Attempt to merge the PR"""
|
|
import httpx
|
|
with httpx.Client() as client:
|
|
# Get PR info for SHA
|
|
pr_r = client.get(
|
|
f"{BASE_URL}/repos/{repo_full}/pulls/{pr_number}",
|
|
headers={"Authorization": f"token {token}", "Accept": "application/vnd.github.v3+json"}
|
|
)
|
|
if pr_r.status_code != 200:
|
|
return False, f"Can't get PR: {pr_r.status_code}"
|
|
|
|
pr_data = pr_r.json()
|
|
sha = pr_data.get("head", {}).get("sha", "")
|
|
mergeable = pr_data.get("mergeable", False)
|
|
|
|
if mergeable is False:
|
|
return False, "PR not mergeable (conflicts or checks pending)"
|
|
|
|
# Try merge
|
|
merge_r = client.put(
|
|
f"{BASE_URL}/repos/{repo_full}/pulls/{pr_number}/merge",
|
|
headers={"Authorization": f"token {token}", "Accept": "application/vnd.github.v3+json"},
|
|
json={
|
|
"commit_title": f"Auto-merge PR #{pr_number}",
|
|
"merge_method": "merge",
|
|
"sha": sha
|
|
}
|
|
)
|
|
if merge_r.status_code == 200:
|
|
return True, f"Merged: {merge_r.json().get('sha', '')}"
|
|
else:
|
|
return False, f"Merge failed: {merge_r.status_code} - {merge_r.json().get('message', '')}"
|
|
|
|
def has_bot_comment_with_review(comments: list) -> tuple:
|
|
"""Check if PR-Agent has posted a review comment and extract quality"""
|
|
bot_name = "mytheclipsebotreview"
|
|
for c in comments:
|
|
if c.get("user", {}).get("login", "").startswith(bot_name):
|
|
body = c.get("body", "")
|
|
# Check for PR Reviewer Guide (successful review)
|
|
if "PR Reviewer Guide" in body:
|
|
# Extract score if available
|
|
score = extract_score(body)
|
|
return True, score
|
|
return False, 0
|
|
|
|
def extract_score(body: str) -> int:
|
|
"""Extract review score from bot comment"""
|
|
import re
|
|
# Look for score patterns like "Score: 8" or "⏱️ Estimated effort"
|
|
# For now, assume passing if we got a review without errors
|
|
return 8
|
|
|
|
def main():
|
|
print("=" * 60)
|
|
print(f"PR-Agent Auto-Approve/Merge Bot - {time.ctime()}")
|
|
print("=" * 60)
|
|
|
|
# Get installations
|
|
installations = get_all_installations()
|
|
print(f"Found {len(installations)} installation(s)")
|
|
|
|
for inst in installations:
|
|
inst_id = inst["id"]
|
|
account = inst["account"]["login"]
|
|
print(f"\n📦 Installation {inst_id} - @{account}")
|
|
|
|
# Get token
|
|
token = get_installation_token(inst_id)
|
|
if not token:
|
|
print(f" ❌ Failed to get token")
|
|
continue
|
|
|
|
# Get repos
|
|
repos = get_installation_repos(inst_id, token)
|
|
print(f" Repos: {len(repos)}")
|
|
|
|
for repo in repos:
|
|
repo_full = repo["full_name"]
|
|
print(f"\n 📁 {repo_full}")
|
|
|
|
# Get open PRs
|
|
prs = get_open_prs(token, repo_full)
|
|
print(f" Open PRs: {len(prs)}")
|
|
|
|
for pr in prs[:5]: # Max 5 per repo
|
|
pr_num = pr["number"]
|
|
pr_title = pr["title"]
|
|
pr_user = pr["user"]["login"]
|
|
pr_author = pr_user
|
|
|
|
print(f" 🔀 PR #{pr_num}: {pr_title[:50]}...")
|
|
|
|
# Skip bot PRs
|
|
if "[bot]" in pr_author or pr_author == "mytheclipsebotreview":
|
|
print(f" ⏭️ Bot PR, skipping")
|
|
continue
|
|
|
|
# Check if already approved/merged
|
|
if pr.get("merged", False):
|
|
print(f" ✅ Already merged")
|
|
continue
|
|
|
|
# Check reviews
|
|
reviews = get_pr_reviews(token, repo_full, pr_num)
|
|
bot_approved = any(
|
|
r.get("user", {}).get("login", "").startswith("mytheclipsebotreview")
|
|
and r.get("state") == "APPROVED"
|
|
for r in reviews
|
|
)
|
|
|
|
if bot_approved:
|
|
print(f" ✅ Already approved. Trying merge...")
|
|
success, msg = merge_pr(token, repo_full, pr_num)
|
|
print(f" {'✅' if success else '❌'} Merge: {msg}")
|
|
continue
|
|
|
|
# Check bot comments for review
|
|
comments = get_pr_comments(token, repo_full, pr_num)
|
|
has_review, score = has_bot_comment_with_review(comments)
|
|
|
|
if has_review and score >= 5:
|
|
print(f" 📝 Review found (score: {score}). Approving + merging...")
|
|
|
|
# Approve
|
|
if approve_pr(token, repo_full, pr_num):
|
|
print(f" ✅ Approved!")
|
|
else:
|
|
print(f" ❌ Approve failed")
|
|
continue
|
|
|
|
# Small delay
|
|
time.sleep(2)
|
|
|
|
# Merge
|
|
success, msg = merge_pr(token, repo_full, pr_num)
|
|
print(f" {'✅ Merged!' if success else '❌ ' + msg}")
|
|
elif has_review and score < 5:
|
|
print(f" ⏭️ Review score too low ({score})")
|
|
else:
|
|
print(f" ⏳ No bot review yet")
|
|
|
|
print("\n" + "=" * 60)
|
|
print("Done!")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|