Pinghua
616 lines
23 KiB
Python
616 lines
23 KiB
Python
"""CI orchestrator for code-review-agent.
|
|
|
|
Fetches a PR diff from Gitea, runs Claude Code to analyze it,
|
|
and posts the review back to the Gitea PR.
|
|
|
|
Usage:
|
|
python scripts/review_pr.py \
|
|
--target-repo owner/repo \
|
|
--pr 42 \
|
|
--gitea-url https://gitea.example.com \
|
|
--api-token <token>
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import shlex
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
from monitor_session import SessionMonitor
|
|
|
|
MAX_DIFF_CHARS = 200_000 # ~50k words, safe for Claude context
|
|
|
|
|
|
def _find_claude():
|
|
"""Resolve the claude CLI binary path.
|
|
|
|
Checks (in order):
|
|
1. CLAUDE_BIN env var (explicit override)
|
|
2. ``claude`` on PATH
|
|
3. Common install locations (npm global, homebrew, etc.)
|
|
Returns the path string, or None.
|
|
"""
|
|
# 1. Explicit override
|
|
env_bin = os.environ.get("CLAUDE_BIN", "")
|
|
if env_bin:
|
|
if os.path.isfile(env_bin) and os.access(env_bin, os.X_OK):
|
|
return env_bin
|
|
print(f"WARNING: CLAUDE_BIN={env_bin} is not executable, falling back.",
|
|
file=sys.stderr)
|
|
|
|
# 2. On PATH
|
|
path_bin = shutil.which("claude")
|
|
if path_bin:
|
|
return path_bin
|
|
|
|
# 3. Common install locations
|
|
candidates = [
|
|
os.path.expanduser("~/.npm-global/bin/claude"),
|
|
"/usr/local/bin/claude",
|
|
"/usr/bin/claude",
|
|
"/home/linuxbrew/.linuxbrew/bin/claude",
|
|
]
|
|
# Also try npm global prefix
|
|
try:
|
|
result = subprocess.run(
|
|
["npm", "config", "get", "prefix"],
|
|
capture_output=True, text=True, timeout=5,
|
|
)
|
|
npm_prefix = result.stdout.strip()
|
|
if npm_prefix:
|
|
candidates.append(os.path.join(npm_prefix, "bin", "claude"))
|
|
except Exception:
|
|
pass
|
|
|
|
for candidate in candidates:
|
|
if os.path.isfile(candidate) and os.access(candidate, os.X_OK):
|
|
return candidate
|
|
|
|
return None
|
|
|
|
|
|
def _check_claude():
|
|
claude_bin = _find_claude()
|
|
if not claude_bin:
|
|
print(
|
|
"ERROR: claude CLI not found. Install Claude Code or set CLAUDE_BIN env var.",
|
|
file=sys.stderr,
|
|
)
|
|
sys.exit(1)
|
|
|
|
missing = []
|
|
for var in ("ANTHROPIC_BASE_URL", "ANTHROPIC_AUTH_TOKEN"):
|
|
if not os.environ.get(var):
|
|
missing.append(var)
|
|
if missing:
|
|
print(
|
|
f"ERROR: Required env vars not set: {', '.join(missing)}. "
|
|
f"Set them in the CI workflow or shell environment.",
|
|
file=sys.stderr,
|
|
)
|
|
sys.exit(1)
|
|
|
|
return claude_bin
|
|
|
|
|
|
def _api_req(gitea_url, api_token, method, path):
|
|
"""Send a Gitea API request, return parsed JSON or raw bytes."""
|
|
url = f"{gitea_url}/api/v1/repos{path}"
|
|
req = urllib.request.Request(url, method=method)
|
|
req.add_header("Authorization", f"token {api_token}")
|
|
req.add_header("Content-Type", "application/json")
|
|
|
|
try:
|
|
with urllib.request.urlopen(req) as resp:
|
|
raw = resp.read()
|
|
if not raw:
|
|
return {} if path.endswith("/merge") else None
|
|
content_type = resp.headers.get("Content-Type", "")
|
|
if "application/json" in content_type:
|
|
return json.loads(raw)
|
|
return raw # raw bytes for .diff endpoints
|
|
except urllib.error.HTTPError as e:
|
|
body = e.read().decode(errors="replace")
|
|
print(f"API Error {e.code} on {method} {path}: {body}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
|
|
def fetch_pr_metadata(gitea_url, api_token, target_repo, pr_num):
|
|
"""Fetch PR details: title, description, head SHA."""
|
|
pr = _api_req(gitea_url, api_token, "GET", f"/{target_repo}/pulls/{pr_num}")
|
|
return {
|
|
"number": pr["number"],
|
|
"title": pr["title"],
|
|
"body": pr.get("body", ""),
|
|
"head_sha": pr.get("head", {}).get("sha", ""),
|
|
"html_url": pr.get("html_url", ""),
|
|
}
|
|
|
|
|
|
def fetch_pr_diff(gitea_url, api_token, target_repo, pr_num):
|
|
"""Fetch raw unified diff for a PR."""
|
|
raw = _api_req(gitea_url, api_token, "GET", f"/{target_repo}/pulls/{pr_num}.diff")
|
|
return raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else ""
|
|
|
|
|
|
def fetch_pr_files(gitea_url, api_token, target_repo, pr_num):
|
|
"""Fetch changed file list with stats."""
|
|
files = _api_req(gitea_url, api_token, "GET", f"/{target_repo}/pulls/{pr_num}/files")
|
|
result = []
|
|
for f in files:
|
|
result.append({
|
|
"filename": f["filename"],
|
|
"status": f["status"],
|
|
"additions": f.get("additions", 0),
|
|
"deletions": f.get("deletions", 0),
|
|
})
|
|
return result
|
|
|
|
|
|
def format_files_summary(files):
|
|
"""Build a one-line-per-file summary for the Claude prompt."""
|
|
lines = []
|
|
for f in files:
|
|
lines.append(
|
|
f" {f['status']:7} {f['filename']} "
|
|
f"(+{f['additions']} -{f['deletions']})"
|
|
)
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _extract_symbols(diff_text):
|
|
"""Extract changed function/class/method names from a unified diff."""
|
|
symbols = set()
|
|
# Match lines added by the PR: "+def name" or "+class name"
|
|
for m in re.finditer(r'^\+ *(?:async\s+)?def\s+(\w+)', diff_text, re.MULTILINE):
|
|
symbols.add(m.group(1))
|
|
for m in re.finditer(r'^\+ *class\s+(\w+)', diff_text, re.MULTILINE):
|
|
symbols.add(m.group(1))
|
|
return sorted(symbols)
|
|
|
|
|
|
def _build_impact_report(repo_dir, changed_files, symbols):
|
|
"""Generate an impact-analysis report for the changed code.
|
|
|
|
Uses ``git grep`` in *repo_dir* to find:
|
|
- files that import the changed modules
|
|
- call-sites of changed functions / classes
|
|
- related test files
|
|
"""
|
|
changed_modules = set()
|
|
for f in changed_files:
|
|
if f.endswith(".py"):
|
|
# Convert path like "skills/utils/LLM.py" → "skills.utils.LLM"
|
|
mod = f.replace("/", ".").replace(".py", "")
|
|
changed_modules.add(mod)
|
|
|
|
lines = ["# Global Impact Report\n"]
|
|
|
|
# ── Importers ──
|
|
if changed_modules:
|
|
lines.append("## Importers (files that import changed modules)\n")
|
|
for mod in sorted(changed_modules):
|
|
try:
|
|
result = subprocess.run(
|
|
["git", "grep", "-l", f"from {mod} import|import {mod}"],
|
|
cwd=repo_dir, capture_output=True, text=True, timeout=30,
|
|
)
|
|
found = [l.strip() for l in result.stdout.splitlines() if l.strip()]
|
|
# Exclude the changed file itself
|
|
found = [f for f in found if f not in changed_files]
|
|
if found:
|
|
lines.append(f"- `{mod}` — used by: {', '.join(f'`{f}`' for f in found[:10])}")
|
|
if len(found) > 10:
|
|
lines.append(f" *(and {len(found) - 10} more)*")
|
|
else:
|
|
lines.append(f"- `{mod}` — no external importers found")
|
|
except Exception:
|
|
pass
|
|
lines.append("")
|
|
|
|
# ── Callers ──
|
|
if symbols:
|
|
lines.append("## Callers (files that reference changed symbols)\n")
|
|
for sym in sorted(symbols):
|
|
try:
|
|
result = subprocess.run(
|
|
["git", "grep", "-l", f"\\b{sym}\\b"],
|
|
cwd=repo_dir, capture_output=True, text=True, timeout=30,
|
|
)
|
|
found = [l.strip() for l in result.stdout.splitlines() if l.strip()]
|
|
# Exclude the changed file itself
|
|
found = [f for f in found if f not in changed_files]
|
|
if found:
|
|
lines.append(f"- `{sym}()` — referenced in: {', '.join(f'`{f}`' for f in found[:10])}")
|
|
if len(found) > 10:
|
|
lines.append(f" *(and {len(found) - 10} more)*")
|
|
else:
|
|
lines.append(f"- `{sym}()` — no external callers found")
|
|
except Exception:
|
|
pass
|
|
lines.append("")
|
|
|
|
# ── Related tests ──
|
|
if changed_modules:
|
|
lines.append("## Related test files\n")
|
|
for mod in sorted(changed_modules):
|
|
# Try common test naming patterns
|
|
parts = mod.split(".")
|
|
test_patterns = [
|
|
f"test_{parts[-1]}.py",
|
|
f"{parts[-1]}_test.py",
|
|
]
|
|
for pat in test_patterns:
|
|
try:
|
|
result = subprocess.run(
|
|
["find", ".", "-name", pat, "-not", "-path", "*/.git/*"],
|
|
cwd=repo_dir, capture_output=True, text=True, timeout=15,
|
|
)
|
|
found = [l.strip() for l in result.stdout.splitlines() if l.strip()]
|
|
if found:
|
|
lines.append(f"- {', '.join(f'`{f}`' for f in found)}")
|
|
else:
|
|
lines.append(f"- `{mod}` — no test files found (pattern `{pat}`)")
|
|
except Exception:
|
|
pass
|
|
break # Only try first matching pattern
|
|
lines.append("")
|
|
|
|
if len(lines) == 2: # Only title + one blank line
|
|
lines.append("*(No importers, callers, or related tests found.)*\n")
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _strip_frontmatter(md_text):
|
|
"""Remove YAML frontmatter (``--- ... ---``) from markdown text."""
|
|
if md_text.startswith("---\n"):
|
|
end = md_text.find("---\n", 4)
|
|
if end != -1:
|
|
return md_text[end + 4:].lstrip("\n")
|
|
return md_text
|
|
|
|
|
|
def build_prompt(meta, files, diff_path, agent_md, impact_report_path=None):
|
|
"""Build the prompt string for Claude, embedding the agent definition."""
|
|
files_summary = format_files_summary(files)
|
|
total_additions = sum(f["additions"] for f in files)
|
|
total_deletions = sum(f["deletions"] for f in files)
|
|
|
|
prompt = (
|
|
f"{_strip_frontmatter(agent_md)}\n\n"
|
|
f"---\n\n"
|
|
f"PR #{meta['number']} in repository.\n\n"
|
|
f"PR Title: {meta['title']}\n"
|
|
f"PR Description:\n{meta.get('body', '(no description)')}\n\n"
|
|
f"Changed Files ({len(files)} files, +{total_additions} -{total_deletions}):\n"
|
|
f"{files_summary}\n\n"
|
|
f"The complete diff has been saved to: {diff_path}\n"
|
|
f"Read that file to see every line changed.\n\n"
|
|
)
|
|
|
|
if impact_report_path:
|
|
prompt += (
|
|
f"A global impact report has been saved to: {impact_report_path}\n"
|
|
f"Read that file to see which files import, call, or test the changed code.\n"
|
|
f"You have access to the target repository. Use Grep and Glob tools to "
|
|
f"trace callers, check interface compatibility, and verify that the "
|
|
f"changes do not break existing dependents.\n\n"
|
|
)
|
|
else:
|
|
prompt += (
|
|
f"Note: no global impact report is available (--repo-dir not provided). "
|
|
f"Review the diff in isolation.\n\n"
|
|
)
|
|
|
|
prompt += (
|
|
f"Analyze the diff thoroughly. Focus on: security, correctness, "
|
|
f"error handling, performance, and code quality — in that order.\n\n"
|
|
f"Output your review as a single JSON object inside a ```json code block. "
|
|
f"Do NOT output anything else."
|
|
)
|
|
return prompt
|
|
|
|
|
|
def run_claude(claude_bin, project_root, prompt, extra_dir=None, prompt_file=None):
|
|
"""Run claude -p, return stdout. Prompt is fed via ``< file`` redirect."""
|
|
cmd = [claude_bin, "-p", "--permission-mode", "bypassPermissions"]
|
|
if extra_dir:
|
|
cmd += ["--add-dir", extra_dir]
|
|
|
|
# Write prompt to temp file, then feed via shell redirect to avoid arg-too-long
|
|
if not prompt_file:
|
|
prompt_file = os.path.join(project_root, ".reviews", "_prompt.txt")
|
|
os.makedirs(os.path.dirname(prompt_file), exist_ok=True)
|
|
with open(prompt_file, "w", encoding="utf-8") as f:
|
|
f.write(prompt)
|
|
|
|
shell_cmd = f"{shlex.join(cmd)} < {shlex.quote(prompt_file)}"
|
|
print(f"[review_pr] Command: {shell_cmd}", file=sys.stderr)
|
|
proc = subprocess.Popen(
|
|
shell_cmd,
|
|
cwd=project_root,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
encoding="utf-8",
|
|
shell=True,
|
|
)
|
|
|
|
def _stream(pipe, prefix, dest_lines):
|
|
for line in pipe:
|
|
line = line.rstrip("\n")
|
|
dest_lines.append(line)
|
|
if line.strip():
|
|
print(f"[{prefix}] {line}", file=sys.stderr)
|
|
|
|
stdout_lines = []
|
|
stderr_lines = []
|
|
|
|
heartbeat_stop = threading.Event()
|
|
def _heartbeat(start):
|
|
while not heartbeat_stop.is_set():
|
|
heartbeat_stop.wait(30)
|
|
if not heartbeat_stop.is_set():
|
|
elapsed = int(time.time() - start)
|
|
print(f"[review_pr] ... still running ({elapsed}s elapsed)", file=sys.stderr)
|
|
t_beat = threading.Thread(target=_heartbeat, args=(time.time(),))
|
|
t_beat.start()
|
|
|
|
t_out = threading.Thread(target=_stream, args=(proc.stdout, "claude", stdout_lines))
|
|
t_err = threading.Thread(target=_stream, args=(proc.stderr, "claude:stderr", stderr_lines))
|
|
t_out.start()
|
|
t_err.start()
|
|
|
|
t_out.join(timeout=600)
|
|
heartbeat_stop.set()
|
|
t_err.join(timeout=10)
|
|
t_beat.join(timeout=5)
|
|
proc.wait(timeout=10)
|
|
|
|
if proc.returncode != 0:
|
|
print(f"Claude exited with code {proc.returncode}", file=sys.stderr)
|
|
if stderr_lines:
|
|
print("\n".join(stderr_lines[-50:]), file=sys.stderr)
|
|
sys.exit(1)
|
|
return "\n".join(stdout_lines)
|
|
|
|
|
|
def extract_json_from_output(stdout):
|
|
"""Extract the review JSON from Claude's output."""
|
|
match = re.search(r"```json\s*([\s\S]*)\s*```", stdout)
|
|
if not match:
|
|
print("ERROR: No ```json block found in Claude output.", file=sys.stderr)
|
|
print(stdout[-3000:], file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
try:
|
|
obj = json.loads(match.group(1))
|
|
except json.JSONDecodeError as e:
|
|
print(f"ERROR: Invalid JSON in Claude output: {e}", file=sys.stderr)
|
|
print(match.group(1)[:2000], file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
# Unwrap {review: {...}} envelope if present
|
|
if isinstance(obj, dict) and "review" in obj and isinstance(obj["review"], dict):
|
|
obj = obj["review"]
|
|
|
|
if not isinstance(obj, dict) or "body" not in obj:
|
|
print(f"ERROR: JSON missing required 'body' field. "
|
|
f"Keys: {list(obj.keys()) if isinstance(obj, dict) else type(obj).__name__}",
|
|
file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
return obj
|
|
|
|
|
|
def validate_review(review):
|
|
"""Validate the review JSON structure."""
|
|
if not isinstance(review, dict):
|
|
print(f"ERROR: review is not a JSON object, got {type(review).__name__}", file=sys.stderr)
|
|
print(f"Raw: {json.dumps(review, ensure_ascii=False)[:2000]}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
event = review.get("event", "")
|
|
if event not in ("APPROVED", "REQUEST_CHANGES", "COMMENT", ""):
|
|
print(f"ERROR: invalid event '{event}'", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
body = review.get("body", "")
|
|
if not body or not isinstance(body, str):
|
|
print(f"ERROR: review.body is required and must be a string.", file=sys.stderr)
|
|
print(f"Got type: {type(body).__name__}, value: {json.dumps(body, ensure_ascii=False)[:500]}", file=sys.stderr)
|
|
print(f"Full review keys: {list(review.keys())}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
comments = review.get("comments", [])
|
|
if not isinstance(comments, list):
|
|
print("ERROR: review.comments must be an array", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
for i, c in enumerate(comments):
|
|
if not isinstance(c, dict):
|
|
print(f"ERROR: comment[{i}] is not an object", file=sys.stderr)
|
|
sys.exit(1)
|
|
if "path" not in c or "body" not in c:
|
|
print(f"ERROR: comment[{i}] missing 'path' or 'body'", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
return True
|
|
|
|
|
|
def post_review(gitea_url, api_token, target_repo, pr_num, review):
|
|
"""Post a PR review to Gitea."""
|
|
payload = {
|
|
"body": review["body"],
|
|
"event": review.get("event", "COMMENT"),
|
|
}
|
|
comments = review.get("comments", [])
|
|
if comments:
|
|
# Translate {line} -> {new_line} for Gitea API
|
|
gitea_comments = []
|
|
for c in comments:
|
|
gc = {
|
|
"path": c["path"],
|
|
"body": c["body"],
|
|
}
|
|
if c.get("line"):
|
|
gc["new_line"] = c["line"]
|
|
if c.get("old_line"):
|
|
gc["old_line"] = c["old_line"]
|
|
gitea_comments.append(gc)
|
|
payload["comments"] = gitea_comments
|
|
|
|
body = json.dumps(payload).encode("utf-8")
|
|
url = f"{gitea_url}/api/v1/repos/{target_repo}/pulls/{pr_num}/reviews"
|
|
req = urllib.request.Request(url, data=body, method="POST")
|
|
req.add_header("Authorization", f"token {api_token}")
|
|
req.add_header("Content-Type", "application/json")
|
|
|
|
try:
|
|
with urllib.request.urlopen(req) as resp:
|
|
result = json.loads(resp.read())
|
|
print(f"Review posted: {result.get('html_url', result.get('url', 'unknown'))}")
|
|
return result
|
|
except urllib.error.HTTPError as e:
|
|
err_body = e.read().decode(errors="replace")
|
|
print(f"Failed to post review: {e.code} - {err_body}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
description="Code Review Agent — fetch PR, analyze with Claude, post review"
|
|
)
|
|
parser.add_argument("--target-repo", required=True,
|
|
help="Target repository path (e.g. owner/repo)")
|
|
parser.add_argument("--pr", type=int, required=True,
|
|
help="PR number to review")
|
|
parser.add_argument("--gitea-url", required=True,
|
|
help="Gitea instance URL")
|
|
parser.add_argument("--api-token", required=True,
|
|
help="Gitea API token with read:repository + write:repository")
|
|
parser.add_argument("--repo-dir", default=None,
|
|
help="Path to local clone of the target repo (enables global impact analysis)")
|
|
args = parser.parse_args()
|
|
|
|
claude_bin = _check_claude()
|
|
|
|
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
agent_path = os.path.join(project_root, ".claude", "agents", "code-review-agent.md")
|
|
|
|
if not os.path.exists(agent_path):
|
|
print(f"ERROR: agent definition not found at {agent_path}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
with open(agent_path, "r", encoding="utf-8") as f:
|
|
agent_md = f.read()
|
|
print(f"[review_pr] Loaded agent definition ({len(agent_md)} chars)")
|
|
|
|
# ── 1. Fetch PR metadata ─────────────────────────────────────────────
|
|
print(f"[review_pr] Fetching PR #{args.pr} metadata from {args.target_repo}...")
|
|
meta = fetch_pr_metadata(args.gitea_url, args.api_token, args.target_repo, args.pr)
|
|
print(f" Title: {meta['title']}")
|
|
|
|
# ── 2. Fetch changed files ───────────────────────────────────────────
|
|
print("[review_pr] Fetching changed files...")
|
|
files = fetch_pr_files(args.gitea_url, args.api_token, args.target_repo, args.pr)
|
|
print(f" {len(files)} file(s) changed")
|
|
|
|
if not files:
|
|
print("No files changed — skipping review.")
|
|
return
|
|
|
|
# ── 3. Fetch diff ────────────────────────────────────────────────────
|
|
print("[review_pr] Fetching diff...")
|
|
diff_text = fetch_pr_diff(args.gitea_url, args.api_token, args.target_repo, args.pr)
|
|
|
|
if not diff_text.strip():
|
|
print("Empty diff — skipping review.")
|
|
return
|
|
|
|
if len(diff_text) > MAX_DIFF_CHARS:
|
|
print(
|
|
f"Warning: diff is {len(diff_text)} chars "
|
|
f"(>{MAX_DIFF_CHARS}). Review may be incomplete.",
|
|
file=sys.stderr,
|
|
)
|
|
diff_text = diff_text[:MAX_DIFF_CHARS]
|
|
|
|
print(f" Diff: {len(diff_text)} chars, ~{diff_text.count(chr(10))} lines")
|
|
|
|
# ── 4. Write diff inside project_root so claude CLI can read it ─────
|
|
repo_slug = args.target_repo.replace("/", "_")
|
|
prefix = f"{repo_slug}_pr{args.pr}"
|
|
diff_dir = os.path.join(project_root, ".reviews")
|
|
os.makedirs(diff_dir, exist_ok=True)
|
|
diff_path = os.path.join(diff_dir, f"{prefix}.diff")
|
|
with open(diff_path, "w", encoding="utf-8") as f:
|
|
f.write(diff_text)
|
|
print(f" Diff saved to: {diff_path}")
|
|
|
|
# ── 4b. Build impact report (if repo-dir provided) ──────────────────
|
|
impact_report_path = None
|
|
repo_dir = args.repo_dir
|
|
if repo_dir:
|
|
if not os.path.isdir(repo_dir):
|
|
print(f"ERROR: --repo-dir is not a directory: {repo_dir}", file=sys.stderr)
|
|
sys.exit(1)
|
|
changed_file_names = [f["filename"] for f in files]
|
|
symbols = _extract_symbols(diff_text)
|
|
print(f"[review_pr] Extracted {len(symbols)} changed symbol(s): {', '.join(symbols) if symbols else '(none)'}")
|
|
report = _build_impact_report(repo_dir, changed_file_names, symbols)
|
|
impact_report_path = os.path.join(diff_dir, f"{prefix}_impact.md")
|
|
with open(impact_report_path, "w", encoding="utf-8") as f:
|
|
f.write(report)
|
|
print(f" Impact report saved to: {impact_report_path}")
|
|
else:
|
|
print("[review_pr] No --repo-dir provided — skipping global impact analysis")
|
|
|
|
mon = None
|
|
prompt_file = os.path.join(diff_dir, f"{prefix}_prompt.txt")
|
|
try:
|
|
# ── 5. Build prompt & run Claude ─────────────────────────────────
|
|
prompt = build_prompt(meta, files, diff_path, agent_md, impact_report_path)
|
|
print("[review_pr] Invoking Claude for analysis...")
|
|
|
|
mon = SessionMonitor(project_root)
|
|
mon.start()
|
|
stdout = run_claude(claude_bin, project_root, prompt, extra_dir=repo_dir, prompt_file=prompt_file)
|
|
mon.stop()
|
|
|
|
# ── 6. Parse Claude output ───────────────────────────────────────
|
|
review = extract_json_from_output(stdout)
|
|
validate_review(review)
|
|
print(f" Review event: {review.get('event', 'COMMENT')}")
|
|
print(f" Body length: {len(review['body'])} chars")
|
|
print(f" Inline comments: {len(review.get('comments', []))}")
|
|
|
|
# ── 7. Post review to Gitea ──────────────────────────────────────
|
|
print("[review_pr] Posting review to Gitea...")
|
|
post_review(args.gitea_url, args.api_token, args.target_repo, args.pr, review)
|
|
|
|
print("[review_pr] Done.")
|
|
finally:
|
|
if mon:
|
|
mon.stop()
|
|
# Clean up temp files
|
|
if os.path.exists(diff_path):
|
|
os.unlink(diff_path)
|
|
if os.path.exists(prompt_file):
|
|
os.unlink(prompt_file)
|
|
if impact_report_path and os.path.exists(impact_report_path):
|
|
os.unlink(impact_report_path)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|