homelab-optimized/scripts/pr-reviewer.py

#!/usr/bin/env python3
"""AI-powered PR reviewer — uses Ollama to review Gitea pull requests."""

import argparse
import json
import logging
import sys
import urllib.request
import urllib.error

log = logging.getLogger("pr-reviewer")

GITEA_URL = "https://git.vish.gg"
GITEA_TOKEN = "REDACTED_TOKEN"  # pragma: allowlist secret
REPO = "vish/homelab"
OLLAMA_URL = "http://192.168.0.145:31434"
OLLAMA_MODEL = "qwen3:32b-fast"
MAX_DIFF_LINES = 500
MAX_FILES = 20


def gitea_api(path: str, method: str = "GET", data: dict | None = None) -> dict | list:
    """Make a Gitea API request."""
    url = f"{GITEA_URL}/api/v1/{path.lstrip('/')}"
    body = json.dumps(data).encode() if data else None
    req = urllib.request.Request(url, data=body, method=method, headers={
        "Authorization": f"token {GITEA_TOKEN}",
        "Content-Type": "application/json",
        "Accept": "application/json",
    })
    with urllib.request.urlopen(req, timeout=30) as resp:
        return json.loads(resp.read())


def get_pr_info(pr_number: int) -> dict:
    """Get PR metadata."""
    return gitea_api(f"repos/{REPO}/pulls/{pr_number}")


def get_pr_files(pr_number: int) -> list[dict]:
    """Get changed files with patches."""
    return gitea_api(f"repos/{REPO}/pulls/{pr_number}/files?limit=50")


def build_diff_summary(files: list[dict]) -> str:
    """Build a truncated diff summary for the LLM."""
    parts = []
    total_lines = 0

    for f in files[:MAX_FILES]:
        filename = f["filename"]
        additions = f.get("additions", 0)
        deletions = f.get("deletions", 0)
        patch = f.get("patch", "") or ""

        # Truncate large patches
        patch_lines = patch.split("\n")
        if total_lines + len(patch_lines) > MAX_DIFF_LINES:
            remaining = MAX_DIFF_LINES - total_lines
            if remaining > 10:
                patch = "\n".join(patch_lines[:remaining]) + "\n... (truncated)"
            else:
                patch = f"(+{additions}/-{deletions}, diff truncated)"

        parts.append(f"--- {filename} (+{additions}/-{deletions}) ---\n{patch}")
        total_lines += len(patch_lines)

        if total_lines >= MAX_DIFF_LINES:
            remaining_files = len(files) - len(parts)
            if remaining_files > 0:
                parts.append(f"\n... and {remaining_files} more files (omitted for brevity)")
            break

    return "\n\n".join(parts)


def review_with_llm(pr_title: str, pr_body: str, diff: str) -> str:
    """Send the PR to Ollama for review."""
    prompt = f"""You are a code reviewer for a homelab infrastructure repository. Review this pull request and provide concise, actionable feedback.

Focus on:
- Security issues (exposed secrets, insecure defaults, missing auth)
- Docker/Compose correctness (syntax, best practices, resource limits)
- YAML validity and formatting
- Potential outages or breaking changes
- Missing documentation for significant changes

Be concise. Skip obvious things. If the PR looks good, say so briefly.

PR Title: {pr_title}
PR Description: {pr_body or "(no description)"}

Changes:
{diff}

Review:"""

    payload = json.dumps({
        "model": OLLAMA_MODEL,
        "prompt": prompt,
        "stream": False,
        "options": {"temperature": 0.3, "num_predict": 1000},
    }).encode()

    req = urllib.request.Request(
        f"{OLLAMA_URL}/api/generate",
        data=payload,
        headers={"Content-Type": "application/json"},
    )

    try:
        with urllib.request.urlopen(req, timeout=120) as resp:
            result = json.loads(resp.read())
    except urllib.error.URLError as e:
        return f"LLM review failed: {e}"

    response = result.get("response", "").strip()
    # Strip thinking tags
    import re
    response = re.sub(r"<think>.*?</think>", "", response, flags=re.DOTALL).strip()
    return response


def post_comment(pr_number: int, body: str):
    """Post a review comment on the PR."""
    gitea_api(f"repos/{REPO}/issues/{pr_number}/comments", method="POST", data={
        "body": body,
    })
    log.info("Posted review comment on PR #%d", pr_number)


def review_pr(pr_number: int, dry_run: bool = False, post: bool = False):
    """Review a PR end-to-end."""
    log.info("Reviewing PR #%d...", pr_number)

    pr = get_pr_info(pr_number)
    title = pr["title"]
    body = pr.get("body", "")
    state = pr["state"]
    log.info("PR: %s (state: %s)", title, state)

    if "[skip-review]" in title.lower() or "[skip-review]" in (body or "").lower():
        log.info("Skipping — [skip-review] tag found")
        return

    files = get_pr_files(pr_number)
    log.info("Files changed: %d", len(files))

    if not files:
        log.info("No files changed, nothing to review")
        return

    total_changes = sum(f.get("additions", 0) + f.get("deletions", 0) for f in files)
    log.info("Total changes: +%d/-%d lines",
             sum(f.get("additions", 0) for f in files),
             sum(f.get("deletions", 0) for f in files))

    diff = build_diff_summary(files)
    log.info("Sending to LLM (%d chars)...", len(diff))

    review = review_with_llm(title, body, diff)

    comment = f"""### AI Review (Qwen3-Coder)

{review}

---
<sub>Automated review by pr-reviewer.py using {OLLAMA_MODEL} on Olares. Add `[skip-review]` to PR title to disable.</sub>"""

    if dry_run:
        print("\n" + "=" * 60)
        print(comment)
        print("=" * 60)
    elif post:
        post_comment(pr_number, comment)
    else:
        print(comment)


def main():
    parser = argparse.ArgumentParser(description="AI-powered PR reviewer")
    parser.add_argument("pr_number", type=int, help="PR number to review")
    parser.add_argument("--post", action="store_true",
                        help="Post review as a Gitea comment (default: print only)")
    parser.add_argument("--dry-run", action="store_true",
                        help="Print review without posting")
    parser.add_argument("-v", "--verbose", action="store_true")

    args = parser.parse_args()
    logging.basicConfig(
        level=logging.DEBUG if args.verbose else logging.INFO,
        format="%(asctime)s %(levelname)-8s %(message)s",
    )

    review_pr(args.pr_number, dry_run=args.dry_run, post=args.post)


if __name__ == "__main__":
    main()