🧹 Organize project root directory
All checks were successful
Documentation Validation / validate-docs (push) Successful in 9s
All checks were successful
Documentation Validation / validate-docs (push) Successful in 9s
- Move documentation files to docs/summaries/ - Move temporary files to temp/ directory - Keep only essential files in root directory - Improve project structure and maintainability
This commit is contained in:
274
temp/auto_review.py.bak
Normal file
274
temp/auto_review.py.bak
Normal file
@@ -0,0 +1,274 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Automated PR reviewer for multi-agent collaboration.
|
||||
|
||||
Fetches open PRs authored by the sibling agent, runs basic validation,
|
||||
and posts an APPROVE or COMMENT review.
|
||||
|
||||
Usage: GITEA_TOKEN=... python3 auto_review.py
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import subprocess
|
||||
import tempfile
|
||||
import shutil
|
||||
from datetime import datetime
|
||||
|
||||
TOKEN = os.getenv("GITEA_TOKEN")
|
||||
API_BASE = os.getenv("GITEA_API_BASE", "http://gitea.bubuit.net:3000/api/v1")
|
||||
REPO = "oib/aitbc"
|
||||
SELF = os.getenv("AGENT_NAME", "aitbc") # set this in env: aitbc or aitbc1
|
||||
OTHER = "aitbc1" if SELF == "aitbc" else "aitbc"
|
||||
|
||||
def log(msg):
|
||||
print(f"[{datetime.now().strftime('%H:%M:%S')}] {msg}")
|
||||
|
||||
def die(msg):
|
||||
log(f"FATAL: {msg}")
|
||||
sys.exit(1)
|
||||
|
||||
def api_get(path):
|
||||
cmd = ["curl", "-s", "-H", f"Authorization: token {TOKEN}", f"{API_BASE}/{path}"]
|
||||
result = subprocess.run(cmd, capture_output=True, text=True)
|
||||
if result.returncode != 0:
|
||||
return None
|
||||
try:
|
||||
return json.loads(result.stdout)
|
||||
except json.JSONDecodeError:
|
||||
return None
|
||||
|
||||
def api_post(path, payload):
|
||||
cmd = ["curl", "-s", "-X", "POST", "-H", f"Authorization: token {TOKEN}", "-H", "Content-Type: application/json",
|
||||
f"{API_BASE}/{path}", "-d", json.dumps(payload)]
|
||||
result = subprocess.run(cmd, capture_output=True, text=True)
|
||||
if result.returncode != 0:
|
||||
return None
|
||||
try:
|
||||
return json.loads(result.stdout)
|
||||
except json.JSONDecodeError:
|
||||
return None
|
||||
|
||||
def get_open_prs():
|
||||
return api_get(f"repos/{REPO}/pulls?state=open") or []
|
||||
|
||||
def get_my_reviews(pr_number):
|
||||
return api_get(f"repos/{REPO}/pulls/{pr_number}/reviews") or []
|
||||
|
||||
# Stability ring definitions
|
||||
RING_PREFIXES = [
|
||||
(0, ["packages/py/aitbc-core", "packages/py/aitbc-sdk"]), # Ring 0: Core
|
||||
(1, ["apps/"]), # Ring 1: Platform services
|
||||
(2, ["cli/", "analytics/", "tools/"]), # Ring 2: Application
|
||||
]
|
||||
RING_THRESHOLD = {0: 0.90, 1: 0.80, 2: 0.70, 3: 0.50} # Ring 3: Experimental/low
|
||||
|
||||
def is_test_file(path):
|
||||
"""Heuristic: classify test files to downgrade ring."""
|
||||
if '/tests/' in path or path.startswith('tests/') or path.endswith('_test.py'):
|
||||
return True
|
||||
return False
|
||||
|
||||
def detect_ring(workdir, base_sha, head_sha):
|
||||
"""Determine the stability ring of the PR based on changed files."""
|
||||
try:
|
||||
# Get list of changed files between base and head
|
||||
output = subprocess.run(
|
||||
["git", "--git-dir", os.path.join(workdir, ".git"), "diff", "--name-only", base_sha, head_sha],
|
||||
capture_output=True, text=True, check=True
|
||||
).stdout
|
||||
files = [f.strip() for f in output.splitlines() if f.strip()]
|
||||
except subprocess.CalledProcessError:
|
||||
files = []
|
||||
|
||||
# If all changed files are tests, treat as Ring 3 (low risk)
|
||||
if files and all(is_test_file(f) for f in files):
|
||||
return 3
|
||||
|
||||
# Find highest precedence ring (lowest number) among changed files
|
||||
for ring, prefixes in sorted(RING_PREFIXES, key=lambda x: x[0]):
|
||||
for p in files:
|
||||
if any(p.startswith(prefix) for prefix in prefixes):
|
||||
return ring
|
||||
return 3 # default to Ring 3 (experimental)
|
||||
|
||||
def checkout_pr_branch(pr):
|
||||
"""Checkout PR branch in a temporary worktree."""
|
||||
tmpdir = tempfile.mkdtemp(prefix="aitbc_review_")
|
||||
try:
|
||||
# Clone just .git into tmp, then checkout
|
||||
subprocess.run(["git", "clone", "--no-checkout", "origin", tmpdir], check=True, capture_output=True)
|
||||
worktree = os.path.join(tmpdir, "wt")
|
||||
os.makedirs(worktree)
|
||||
subprocess.run(["git", "--git-dir", os.path.join(tmpdir, ".git"), "--work-tree", worktree, "fetch", "origin", pr['head']['ref']], check=True, capture_output=True)
|
||||
subprocess.run(["git", "--git-dir", os.path.join(tmpdir, ".git"), "--work-tree", worktree, "checkout", "FETCH_HEAD"], check=True, capture_output=True)
|
||||
return worktree, tmpdir
|
||||
except subprocess.CalledProcessError as e:
|
||||
shutil.rmtree(tmpdir, ignore_errors=True)
|
||||
log(f"Checkout failed: {e}")
|
||||
return None, None
|
||||
|
||||
def run_checks(workdir):
|
||||
"""Run validation checks. Returns (pass, score, notes)."""
|
||||
notes = []
|
||||
score = 0.0
|
||||
|
||||
# 1. Import sanity: try to import the aitbc_cli module
|
||||
try:
|
||||
subprocess.run([sys.executable, "-c", "import aitbc_cli.main"], check=True, cwd=workdir, capture_output=True)
|
||||
notes.append("CLI imports OK")
|
||||
score += 0.3
|
||||
except subprocess.CalledProcessError as e:
|
||||
notes.append(f"CLI import failed: {e.stderr.decode().strip()}")
|
||||
return False, 0.0, "\n".join(notes)
|
||||
|
||||
# 2. Check if tests exist and run them (if tests/ directory present)
|
||||
tests_dir = os.path.join(workdir, "tests")
|
||||
if os.path.exists(tests_dir):
|
||||
try:
|
||||
# Run pytest quietly
|
||||
result = subprocess.run([sys.executable, "-m", "pytest", "-q"], cwd=workdir, capture_output=True, text=True, timeout=60)
|
||||
if result.returncode == 0:
|
||||
notes.append("All tests passed")
|
||||
score += 0.4
|
||||
# Parse coverage if available? Not implemented
|
||||
else:
|
||||
notes.append(f"Tests failed (exit {result.returncode}): {result.stdout[-500:]}")
|
||||
return False, score, "\n".join(notes)
|
||||
except subprocess.TimeoutExpired:
|
||||
notes.append("Tests timed out")
|
||||
return False, score, "\n".join(notes)
|
||||
except Exception as e:
|
||||
notes.append(f"Test run error: {e}")
|
||||
return False, score, "\n".join(notes)
|
||||
else:
|
||||
notes.append("No tests directory; skipping test run")
|
||||
score += 0.1 # small baseline
|
||||
|
||||
# 3. Check for syntax errors in all Python files (quick scan)
|
||||
py_files = subprocess.run(["find", workdir, "-name", "*.py", "-not", "-path", "*/.*"], capture_output=True, text=True).stdout.strip().split("\n")
|
||||
syntax_errors = []
|
||||
for py in py_files[:50]: # limit to first 50
|
||||
if not py:
|
||||
continue
|
||||
result = subprocess.run([sys.executable, "-m", "py_compile", py], capture_output=True, text=True)
|
||||
if result.returncode != 0:
|
||||
syntax_errors.append(f"{os.path.basename(py)}: {result.stderr.strip()}")
|
||||
if syntax_errors:
|
||||
notes.append("Syntax errors:\n" + "\n".join(syntax_errors))
|
||||
return False, score, "\n".join(notes)
|
||||
else:
|
||||
notes.append(f"Syntax OK ({len(py_files)} files checked)")
|
||||
score += 0.2
|
||||
|
||||
# 4. Effort estimate: count files changed
|
||||
# We can approximate by using git diff --name-only on the branch compared to main
|
||||
try:
|
||||
# Get the merge base with main
|
||||
base = pr['base']['sha']
|
||||
head = pr['head']['sha']
|
||||
changed = subprocess.run(["git", "--git-dir", os.path.join(tmpdir, ".git"), "diff", "--name-only", base, head], capture_output=True, text=True).stdout.strip().split("\n")
|
||||
num_files = len([f for f in changed if f])
|
||||
if num_files < 5:
|
||||
score += 0.1
|
||||
elif num_files < 15:
|
||||
score += 0.05
|
||||
else:
|
||||
score -= 0.05
|
||||
notes.append(f"Changed files: {num_files}")
|
||||
except Exception as e:
|
||||
notes.append(f"Could not compute changed files: {e}")
|
||||
|
||||
# Normalize score to 0-1 range (max ~1.0)
|
||||
score = min(max(score, 0.0), 1.0)
|
||||
return True, score, "\n".join(notes)
|
||||
|
||||
def post_review(pr_number, body, event="COMMENT"):
|
||||
payload = {"body": body, "event": event}
|
||||
return api_post(f"repos/{REPO}/pulls/{pr_number}/reviews", payload)
|
||||
|
||||
def main():
|
||||
log(f"Starting auto-review (agent={SELF}, watching for PRs from {OTHER})")
|
||||
prs = get_open_prs()
|
||||
if not prs:
|
||||
log("No open PRs found.")
|
||||
return
|
||||
|
||||
for pr in prs:
|
||||
author = pr['user']['login']
|
||||
if author != OTHER:
|
||||
continue # only review sibling's PRs
|
||||
|
||||
pr_number = pr['number']
|
||||
title = pr['title']
|
||||
head = pr['head']['ref']
|
||||
base = pr['base']['ref']
|
||||
|
||||
log(f"Reviewing PR#{pr_number}: {title} (head: {head})")
|
||||
|
||||
# Check if I already reviewed
|
||||
reviews = get_my_reviews(pr_number)
|
||||
if any(r['user']['login'] == SELF for r in reviews):
|
||||
log(f"Already reviewed PR#{pr_number}; skipping")
|
||||
continue
|
||||
|
||||
# Checkout and run tests
|
||||
workdir, tmpdir = checkout_pr_branch(pr)
|
||||
if not workdir:
|
||||
log(f"Failed to checkout PR#{pr_number}; skipping")
|
||||
continue
|
||||
|
||||
try:
|
||||
# Determine stability ring and threshold
|
||||
base_sha = pr['base']['sha']
|
||||
head_sha = pr['head']['sha']
|
||||
ring = detect_ring(workdir, base_sha, head_sha)
|
||||
threshold = RING_THRESHOLD[ring]
|
||||
|
||||
ok, score, notes = run_checks(workdir)
|
||||
notes = f"Ring: {ring}\nThreshold: {threshold}\n{notes}"
|
||||
finally:
|
||||
shutil.rmtree(tmpdir, ignore_errors=True)
|
||||
|
||||
if ok and score >= threshold:
|
||||
review_body = f"""**Auto-review by {SELF}**
|
||||
|
||||
✅ **APPROVED**
|
||||
|
||||
Confidence score: {score:.2f}
|
||||
|
||||
**Validation notes:**
|
||||
{notes}
|
||||
|
||||
This PR meets quality criteria and can be merged."""
|
||||
result = post_review(pr_number, review_body, event="APPROVE")
|
||||
if result:
|
||||
log(f"PR#{pr_number} APPROVED (score={score:.2f})")
|
||||
else:
|
||||
log(f"Failed to post approval for PR#{pr_number}")
|
||||
else:
|
||||
review_body = f"""**Auto-review by {SELF}**
|
||||
|
||||
⚠️ **CHANGES REQUESTED**
|
||||
|
||||
Confidence score: {score:.2f}
|
||||
|
||||
**Validation notes:**
|
||||
{notes}
|
||||
|
||||
Please address the issues above and push again."""
|
||||
result = post_review(pr_number, review_body, event="REQUEST_CHANGES")
|
||||
if result:
|
||||
log(f"PR#{pr_number} CHANGE REQUESTED (score={score:.2f})")
|
||||
else:
|
||||
log(f"Failed to post review for PR#{pr_number}")
|
||||
|
||||
log("Auto-review complete.")
|
||||
|
||||
if __name__ == "__main__":
|
||||
if not TOKEN:
|
||||
die("GITEA_TOKEN environment variable required")
|
||||
if SELF not in ("aitbc", "aitbc1"):
|
||||
die("AGENT_NAME must be set to 'aitbc' or 'aitbc1'")
|
||||
main()
|
||||
Reference in New Issue
Block a user