school_compare/scripts/ci/ai_review.py

#!/usr/bin/env python3
"""AI code review for Gitea pull requests, powered by Claude Code.

Reads the PR diff (base branch vs HEAD), asks Claude Code (headless `claude -p`)
to review it, posts the findings as a PR comment via the Gitea API, and exits
non-zero only when the review contains at least one severe finding — so the
job can gate merges without blocking on nitpicks.

Uses only the Python standard library; the review itself runs through the
Claude Code CLI, authenticated with a subscription OAuth token.

Required environment:
    CLAUDE_CODE_OAUTH_TOKEN  token from `claude setup-token` (subscription auth)
    GITEA_TOKEN              Gitea access token for posting PR comments
    GITEA_SERVER_URL         e.g. https://privaterepo.sitaru.org
    GITEA_REPOSITORY         owner/repo
    PR_NUMBER                pull request index
    BASE_REF                 base branch name (e.g. main)
"""

import json
import os
import subprocess
import sys
import urllib.request

MAX_DIFF_CHARS = 150_000

PROMPT = """You are reviewing a pull request for SchoolCompare, a UK school
comparison site (FastAPI backend, Next.js frontend, Airflow/dbt data pipeline,
deployed via Gitea Actions to a staging-then-production Docker setup).

The PR diff is provided on stdin.

Report correctness bugs, security issues, data-loss risks, and broken deploy/CI
configuration. Mark a finding "severe" only if it would break production, leak
data, or corrupt data — severe findings block the merge. Everything else
(style, performance suggestions, minor cleanups) is "minor". Do not invent
findings: an empty findings list is a perfectly good review of a clean diff.

Respond with ONLY a JSON object (no markdown fences, no prose) of this shape:
{
  "summary": "two or three sentences on what the change does and its health",
  "findings": [
    {"severity": "severe" | "minor", "file": "path", "issue": "description"}
  ]
}"""


def get_diff(base_ref: str) -> str:
    subprocess.run(
        ["git", "fetch", "origin", base_ref],
        check=True,
        capture_output=True,
    )
    diff = subprocess.run(
        ["git", "diff", f"origin/{base_ref}...HEAD"],
        check=True,
        capture_output=True,
        text=True,
    ).stdout
    if len(diff) > MAX_DIFF_CHARS:
        diff = diff[:MAX_DIFF_CHARS] + "\n\n[diff truncated for review]"
    return diff


def review(diff: str) -> dict:
    proc = subprocess.run(
        ["claude", "-p", PROMPT, "--output-format", "json"],
        input=diff,
        capture_output=True,
        text=True,
        timeout=900,
    )
    if proc.returncode != 0:
        raise RuntimeError(f"claude CLI failed:\n{proc.stderr}")
    envelope = json.loads(proc.stdout)
    result = envelope["result"].strip()
    # Defensive: strip markdown fences if the model added them anyway
    if result.startswith("```"):
        result = result.split("\n", 1)[1].rsplit("```", 1)[0]
    return json.loads(result)


def format_comment(result: dict) -> str:
    lines = ["## 🤖 AI Code Review (Claude Code)", "", result["summary"], ""]
    severe = [f for f in result["findings"] if f["severity"] == "severe"]
    minor = [f for f in result["findings"] if f["severity"] == "minor"]
    if severe:
        lines.append("### 🔴 Severe (blocks merge)")
        lines += [f"- **{f['file']}**: {f['issue']}" for f in severe]
        lines.append("")
    if minor:
        lines.append("### 🟡 Minor")
        lines += [f"- **{f['file']}**: {f['issue']}" for f in minor]
        lines.append("")
    if not result["findings"]:
        lines.append("✅ No issues found.")
    return "\n".join(lines)


def post_comment(body: str) -> None:
    server = os.environ["GITEA_SERVER_URL"].rstrip("/")
    repo = os.environ["GITEA_REPOSITORY"]
    pr = os.environ["PR_NUMBER"]
    req = urllib.request.Request(
        f"{server}/api/v1/repos/{repo}/issues/{pr}/comments",
        data=json.dumps({"body": body}).encode(),
        headers={
            "Authorization": f"token {os.environ['GITEA_TOKEN']}",
            "Content-Type": "application/json",
        },
        method="POST",
    )
    with urllib.request.urlopen(req, timeout=30) as resp:
        if resp.status >= 300:
            raise RuntimeError(f"Comment post failed: HTTP {resp.status}")


def main() -> int:
    diff = get_diff(os.environ["BASE_REF"])
    if not diff.strip():
        print("Empty diff, nothing to review")
        return 0
    result = review(diff)
    comment = format_comment(result)
    print(comment)
    post_comment(comment)
    severe = [f for f in result["findings"] if f["severity"] == "severe"]
    if severe:
        print(f"\n{len(severe)} severe finding(s) — failing the check", file=sys.stderr)
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())