school_compare/scripts/ci/ai_review.py

#!/usr/bin/env python3
"""AI code review for Gitea pull requests.

Reads the PR diff (base branch vs HEAD), asks Claude 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.

Required environment:
    ANTHROPIC_API_KEY   Anthropic API key
    GITEA_TOKEN         Gitea token with permission to comment on PRs
    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 requests
from anthropic import Anthropic

MAX_DIFF_CHARS = 150_000

REVIEW_SCHEMA = {
    "type": "object",
    "properties": {
        "summary": {
            "type": "string",
            "description": "Two or three sentences on what the change does and its overall health.",
        },
        "findings": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "severity": {"type": "string", "enum": ["severe", "minor"]},
                    "file": {"type": "string"},
                    "issue": {"type": "string"},
                },
                "required": ["severity", "file", "issue"],
                "additionalProperties": False,
            },
        },
    },
    "required": ["summary", "findings"],
    "additionalProperties": False,
}

SYSTEM_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).

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."""


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:
    client = Anthropic()
    with client.messages.stream(
        model="claude-opus-4-8",
        max_tokens=16000,
        thinking={"type": "adaptive"},
        system=SYSTEM_PROMPT,
        output_config={"format": {"type": "json_schema", "schema": REVIEW_SCHEMA}},
        messages=[
            {
                "role": "user",
                "content": f"Review this pull request diff:\n\n```diff\n{diff}\n```",
            }
        ],
    ) as stream:
        message = stream.get_final_message()
    if message.stop_reason == "refusal":
        raise RuntimeError("Claude declined to review this diff")
    text = next(b.text for b in message.content if b.type == "text")
    return json.loads(text)


def format_comment(result: dict) -> str:
    lines = ["## 🤖 AI Code Review (Claude)", "", 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"]
    resp = requests.post(
        f"{server}/api/v1/repos/{repo}/issues/{pr}/comments",
        headers={"Authorization": f"token {os.environ['GITEA_TOKEN']}"},
        json={"body": body},
        timeout=30,
    )
    resp.raise_for_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())