github.py

"""GitHub data fetching via gh CLI + local git.

File stats come from the local checkout (git diff --numstat), everything
else (PR metadata, reviews, comments, check runs) from the GitHub API.
Also handles team membership checks for the ownership gate.
"""

import json
import subprocess
from dataclasses import dataclass, field
from pathlib import Path


@dataclass
class PRData:
    """All GitHub data needed to evaluate a PR."""

    number: int
    repo: str
    title: str
    state: str
    draft: bool
    mergeable_state: str
    author: str
    labels: list[str]
    base_sha: str
    head_sha: str
    files: list[dict]
    reviews: list[dict]
    review_comments: list[dict]
    check_runs: list[dict]
    pr_reactions: list[dict] = field(default_factory=list)

    @property
    def file_paths(self) -> list[str]:
        return [f["filename"] for f in self.files]

    @property
    def lines_added(self) -> int:
        return sum(f["additions"] for f in self.files)

    @property
    def lines_deleted(self) -> int:
        return sum(f["deletions"] for f in self.files)

    @property
    def lines_total(self) -> int:
        return self.lines_added + self.lines_deleted

    @property
    def has_new_files(self) -> bool:
        return any(f.get("status") == "A" for f in self.files)


_TRUSTED_ASSOCIATIONS = {"MEMBER", "OWNER", "COLLABORATOR"}
TRUSTED_REACTOR_BOTS = {
    "chatgpt-codex-connector[bot]",
    "copilot-pull-request-reviewer[bot]",
    "greptile-apps[bot]",
    "hex-security-app[bot]",
    "posthog[bot]",
    "veria-ai[bot]",
}


def _normalize_reviews_for_prompt(reviews_raw: list[dict], head_sha: str) -> list[dict]:
    """Normalize top-level reviews for the reviewer prompt.

    Preserve trusted/bot reviews, and annotate whether each review was left on
    the current PR head. This lets the LLM distinguish active feedback from
    older context that may already have been addressed in follow-up commits.
    """
    normalized_reviews = []
    for review in reviews_raw:
        if not (
            review.get("author_association") in _TRUSTED_ASSOCIATIONS
            or review.get("author_association") == "BOT"
            or review.get("user", {}).get("type") == "Bot"
        ):
            continue

        commit_id = review.get("commit_id")
        normalized_reviews.append(
            {
                "user": review["user"]["login"],
                "state": review["state"],
                "body": review.get("body", ""),
                "commit_id": commit_id,
                "is_current_head": commit_id == head_sha,
                "submitted_at": review.get("submitted_at"),
            }
        )

    return normalized_reviews


def _normalize_pr_reactions(reactions_raw: list[dict], author: str) -> list[dict]:
    normalized = []
    for reaction in reactions_raw:
        login = (reaction.get("user") or {}).get("login", "")
        if login == author or login.lower() not in TRUSTED_REACTOR_BOTS:
            continue
        normalized.append(
            {
                "user": login,
                "emoji": {"+1": "👍", "-1": "👎", "eyes": "👀"}.get(reaction.get("content"), reaction.get("content")),
                "created_at": reaction.get("created_at"),
            }
        )
    return normalized


def _gh_api(endpoint: str, *, paginate: bool = False) -> dict | list:
    cmd = ["gh", "api", endpoint]
    if paginate:
        cmd.append("--paginate")
    else:
        sep = "&" if "?" in endpoint else "?"
        cmd[2] = f"{endpoint}{sep}per_page=100"
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
    if result.returncode != 0:
        raise RuntimeError(f"gh api {endpoint} failed: {result.stderr.strip()}")
    return json.loads(result.stdout)


_REVIEW_THREADS_QUERY = """
query($owner: String!, $name: String!, $pr: Int!, $threadCursor: String) {
  repository(owner: $owner, name: $name) {
    pullRequest(number: $pr) {
      reviewThreads(first: 100, after: $threadCursor) {
        pageInfo { hasNextPage endCursor }
        nodes {
          isResolved
          isOutdated
          path
          line
          comments(first: 50) {
            pageInfo { hasNextPage }
            nodes {
              author { login __typename }
              authorAssociation
              body
              databaseId
              replyTo { databaseId }
            }
          }
        }
      }
    }
  }
}
"""


def _gh_graphql(query: str, variables: dict | None = None) -> dict:
    """Run a GraphQL query via gh api graphql with proper variable passing."""
    cmd = ["gh", "api", "graphql", "-f", f"query={query}"]
    for key, value in (variables or {}).items():
        if value is None:
            cmd.extend(["-F", f"{key}=null"])
        elif isinstance(value, int):
            cmd.extend(["-F", f"{key}={value}"])
        else:
            cmd.extend(["-f", f"{key}={value}"])
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
    if result.returncode != 0:
        raise RuntimeError(f"gh api graphql failed: {result.stderr.strip()}")
    data = json.loads(result.stdout)
    if "errors" in data:
        raise RuntimeError(f"GraphQL errors: {data['errors']}")
    return data


def _fetch_review_threads(repo: str, pr_number: int) -> list[dict]:
    """Fetch review threads with resolution status via GraphQL.

    Returns a flat list of comment dicts enriched with is_resolved and
    is_outdated from their parent thread. Paginates through all threads
    and raises if any comment page is truncated.
    """
    owner, name = repo.split("/", 1)
    variables: dict = {"owner": owner, "name": name, "pr": pr_number, "threadCursor": None}

    comments: list[dict] = []
    while True:
        data = _gh_graphql(_REVIEW_THREADS_QUERY, variables)
        review_threads = data["data"]["repository"]["pullRequest"]["reviewThreads"]
        threads = review_threads["nodes"]

        for thread in threads:
            comment_page = thread["comments"]
            if comment_page["pageInfo"]["hasNextPage"]:
                raise RuntimeError(
                    f"Review thread on {thread.get('path')}:{thread.get('line')} "
                    f"has >50 comments — pagination not implemented, escalate to human review"
                )
            for c in comment_page["nodes"]:
                assoc = c.get("authorAssociation", "")
                is_bot = (c.get("author") or {}).get("__typename") == "Bot"
                if assoc not in _TRUSTED_ASSOCIATIONS and assoc != "BOT" and not is_bot:
                    continue
                reply_to = c.get("replyTo")
                comments.append(
                    {
                        "user": (c.get("author") or {}).get("login", "ghost"),
                        "body": c.get("body", ""),
                        "path": thread.get("path", ""),
                        "line": thread.get("line"),
                        "in_reply_to_id": reply_to["databaseId"] if reply_to else None,
                        "is_resolved": thread["isResolved"],
                        "is_outdated": thread["isOutdated"],
                    }
                )

        page_info = review_threads["pageInfo"]
        if not page_info["hasNextPage"]:
            break
        variables["threadCursor"] = page_info["endCursor"]

    return comments


def _git_diff_files(base_sha: str, head_sha: str, repo_root: Path) -> list[dict]:
    """Get changed files with line counts and status from the local checkout."""
    diff_range = f"{base_sha}...{head_sha}"
    run_opts = {"capture_output": True, "text": True, "timeout": 30, "cwd": repo_root}

    numstat = subprocess.run(["git", "diff", "--numstat", diff_range], **run_opts)
    if numstat.returncode != 0:
        raise RuntimeError(f"git diff --numstat failed: {numstat.stderr.strip()}")

    name_status = subprocess.run(["git", "diff", "--name-status", diff_range], **run_opts)
    status_map: dict[str, str] = {}
    for line in name_status.stdout.strip().splitlines():
        parts = line.split("\t", 1)
        if len(parts) == 2:
            status_map[parts[1]] = parts[0]

    files = []
    for line in numstat.stdout.strip().splitlines():
        if not line:
            continue
        parts = line.split("\t", 2)
        if len(parts) != 3:
            continue
        added, deleted, filename = parts
        is_binary = added == "-"
        files.append(
            {
                "filename": filename,
                "additions": int(added) if not is_binary else 0,
                "deletions": int(deleted) if not is_binary else 0,
                "binary": is_binary,
                "status": status_map.get(filename, "M"),
            }
        )
    return files


def ensure_commits(pr_number: int, head_sha: str, repo_root: Path) -> None:
    """Fetch PR commits if not available locally."""
    result = subprocess.run(
        ["git", "cat-file", "-t", head_sha],
        cwd=repo_root,
        capture_output=True,
        timeout=5,
    )
    if result.returncode == 0:
        return
    subprocess.run(
        ["git", "fetch", "origin", f"pull/{pr_number}/head"],
        cwd=repo_root,
        capture_output=True,
        timeout=30,
    )


def fetch_pr(pr_number: int, repo: str, repo_root: Path | None = None) -> PRData:
    """Fetch PR data: metadata from API, file stats from local git."""
    pr = _gh_api(f"repos/{repo}/pulls/{pr_number}")
    reviews_raw = _gh_api(f"repos/{repo}/pulls/{pr_number}/reviews", paginate=True)

    base_sha = pr["base"]["sha"]
    head_sha = pr["head"]["sha"]
    check_runs_resp = _gh_api(f"repos/{repo}/commits/{head_sha}/check-runs")
    pr_reactions: list[dict] = []
    try:
        reactions_raw = _gh_api(f"repos/{repo}/issues/{pr_number}/reactions", paginate=True)
        pr_reactions = _normalize_pr_reactions(reactions_raw, pr["user"]["login"])
    except Exception as exc:
        print(f"warning: reaction fetch failed ({exc}); continuing without reaction context")

    git_root = repo_root or Path.cwd()
    ensure_commits(pr_number, head_sha, git_root)
    files = _git_diff_files(base_sha, head_sha, git_root)

    review_comments = _fetch_review_threads(repo, pr_number)

    return PRData(
        number=pr_number,
        repo=repo,
        title=pr["title"],
        state=pr["state"],
        draft=pr.get("draft", False),
        mergeable_state=pr.get("mergeable_state", "unknown"),
        author=pr["user"]["login"],
        labels=[label["name"] for label in pr.get("labels", [])],
        base_sha=base_sha,
        head_sha=head_sha,
        files=files,
        reviews=_normalize_reviews_for_prompt(reviews_raw, head_sha),
        review_comments=review_comments,
        check_runs=check_runs_resp.get("check_runs", []),
        pr_reactions=pr_reactions,
    )


def check_team_membership(author: str, team_slug: str) -> bool:
    """Check if author is an active member of the given GitHub team."""
    try:
        result = subprocess.run(
            ["gh", "api", f"orgs/PostHog/teams/{team_slug}/memberships/{author}"],
            capture_output=True,
            text=True,
            timeout=10,
        )
        if result.returncode == 0:
            return json.loads(result.stdout).get("state") == "active"
    except (subprocess.TimeoutExpired, json.JSONDecodeError):
        pass
    return False