github_client.py

"""GitHub API client — authentication, PR comments, and metadata."""

from __future__ import annotations

import logging
import time
from pathlib import Path

import httpx
import jwt

from models import AnalysisResult

logger = logging.getLogger(__name__)

GITHUB_API = "https://api.github.com"


# ---------------------------------------------------------------------------
# Authentication
# ---------------------------------------------------------------------------


def _build_jwt(app_id: str, private_key_path: str) -> str:
    """Create a short-lived JWT for GitHub App authentication."""
    private_key = Path(private_key_path).read_text()
    now = int(time.time())
    payload = {
        "iat": now - 60,  # issued at (allow clock drift)
        "exp": now + (10 * 60),  # expires in 10 min
        "iss": app_id,
    }
    return jwt.encode(payload, private_key, algorithm="RS256")


async def _get_installation_token(
    app_id: str,
    private_key_path: str,
    installation_id: int,
) -> str:
    """Exchange the JWT for a short-lived installation access token."""
    token = _build_jwt(app_id, private_key_path)
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            f"{GITHUB_API}/app/installations/{installation_id}/access_tokens",
            headers={
                "Authorization": f"Bearer {token}",
                "Accept": "application/vnd.github+json",
                "X-GitHub-Api-Version": "2022-11-28",
            },
        )
        resp.raise_for_status()
        return resp.json()["token"]


# ---------------------------------------------------------------------------
# GitHub API Client
# ---------------------------------------------------------------------------


class GitHubClient:
    """Authenticated GitHub API client scoped to a single installation."""

    def __init__(self, installation_token: str) -> None:
        self._token = installation_token
        self._headers = {
            "Authorization": f"Bearer {installation_token}",
            "Accept": "application/vnd.github+json",
            "X-GitHub-Api-Version": "2022-11-28",
        }

    @classmethod
    async def for_installation(
        cls,
        app_id: str,
        private_key_path: str,
        installation_id: int,
    ) -> GitHubClient:
        """Factory: authenticate and return a client for the installation."""
        token = await _get_installation_token(app_id, private_key_path, installation_id)
        return cls(token)

    # -- PR metadata -------------------------------------------------------

    async def get_pr_diff(self, owner: str, repo: str, pr_number: int) -> str:
        """Fetch the unified diff for a pull request."""
        async with httpx.AsyncClient() as client:
            resp = await client.get(
                f"{GITHUB_API}/repos/{owner}/{repo}/pulls/{pr_number}",
                headers={**self._headers, "Accept": "application/vnd.github.diff"},
            )
            resp.raise_for_status()
            return resp.text

    async def get_pr_files(self, owner: str, repo: str, pr_number: int) -> list[dict]:
        """List files changed in a PR."""
        async with httpx.AsyncClient() as client:
            resp = await client.get(
                f"{GITHUB_API}/repos/{owner}/{repo}/pulls/{pr_number}/files",
                headers=self._headers,
            )
            resp.raise_for_status()
            return resp.json()

    # -- Comments -----------------------------------------------------------

    async def post_comment(
        self,
        owner: str,
        repo: str,
        pr_number: int,
        body: str,
    ) -> dict:
        """Post a comment on a pull request."""
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                f"{GITHUB_API}/repos/{owner}/{repo}/issues/{pr_number}/comments",
                headers=self._headers,
                json={"body": body},
            )
            resp.raise_for_status()
            logger.info("Posted comment on %s/%s#%d", owner, repo, pr_number)
            return resp.json()

    async def post_analysis(
        self,
        owner: str,
        repo: str,
        pr_number: int,
        result: AnalysisResult,
    ) -> dict:
        """Render an AnalysisResult as Markdown and post it as a PR comment."""
        body = result.to_markdown()
        return await self.post_comment(owner, repo, pr_number, body)

    # -- Clone URL ----------------------------------------------------------

    def clone_url(self, owner: str, repo: str) -> str:
        """Return an authenticated HTTPS clone URL."""
        return f"https://x-access-token:{self._token}@github.com/{owner}/{repo}.git"