diff --git a/vibeguard/integrations/github.py b/vibeguard/integrations/github.py new file mode 100644 index 0000000..287baf0 --- /dev/null +++ b/vibeguard/integrations/github.py @@ -0,0 +1,135 @@ +"""GitHub integration for VibeGuard.""" + +import os +from typing import Any + +import requests + + +class GitHubIntegration: + """GitHub API integration for VibeGuard.""" + + def __init__(self, token: str | None = None) -> None: + """Initialize GitHub integration.""" + self.token = token or os.environ.get("VIBEGUARD_GITHUB_TOKEN") + self.base_url = "https://api.github.com" + + def _get_headers(self) -> dict[str, str]: + """Get headers for API requests.""" + headers = { + "Accept": "application/vnd.github.v3+json", + } + if self.token: + headers["Authorization"] = f"Bearer {self.token}" + return headers + + def create_check_run( + self, + repo: str, + name: str, + conclusion: str, + details_url: str | None = None, + output: dict[str, Any] | None = None, + ) -> dict[str, Any]: + """Create a check run for a GitHub repository.""" + url = f"{self.base_url}/repos/{repo}/check-runs" + + data = { + "name": name, + "head_sha": self._get_head_sha(), + "status": "completed", + "conclusion": conclusion, + } + + if details_url: + data["details_url"] = details_url + + if output: + data["output"] = output + + response = requests.post(url, headers=self._get_headers(), json=data) + response.raise_for_status() + + return response.json() + + def update_check_run( + self, + repo: str, + check_run_id: int, + conclusion: str, + output: dict[str, Any] | None = None, + ) -> dict[str, Any]: + """Update an existing check run.""" + url = f"{self.base_url}/repos/{repo}/check-runs/{check_run_id}" + + data = { + "status": "completed", + "conclusion": conclusion, + } + + if output: + data["output"] = output + + response = requests.patch(url, headers=self._get_headers(), json=data) + response.raise_for_status() + + return response.json() + + def create_pull_request_comment( + self, repo: str, pull_number: int, body: str + ) -> dict[str, Any]: + """Create a comment on a pull request.""" + url = f"{self.base_url}/repos/{repo}/issues/{pull_number}/comments" + + data = {"body": body} + + response = requests.post(url, headers=self._get_headers(), json=data) + response.raise_for_status() + + return response.json() + + def upload_sarif( + self, repo: str, sarif_content: dict[str, Any], commit_oid: str + ) -> dict[str, Any]: + """Upload SARIF results to GitHub Code Scanning.""" + url = f"{self.base_url}/repos/{repo}/code-scanning/sarifs" + + data = {"sarif": sarif_content, "commit_oid": commit_oid} + + response = requests.post(url, headers=self._get_headers(), json=data) + response.raise_for_status() + + return response.json() + + def _get_head_sha(self) -> str: + """Get the head SHA from environment or git.""" + sha = os.environ.get("GITHUB_SHA") + if sha: + return sha + + import subprocess + + result = subprocess.run( + ["git", "rev-parse", "HEAD"], + capture_output=True, + text=True, + ) + return result.stdout.strip() + + def get_pull_request_files(self, repo: str, pull_number: int) -> list[dict[str, Any]]: + """Get files changed in a pull request.""" + url = f"{self.base_url}/repos/{repo}/pulls/{pull_number}/files" + + response = requests.get(url, headers=self._get_headers()) + response.raise_for_status() + + return response.json() + + def get_check_runs(self, repo: str, ref: str) -> dict[str, Any]: + """Get check runs for a ref.""" + url = f"{self.base_url}/repos/{repo}/commits/{ref}/check-runs" + + response = requests.get(url, headers=self._get_headers()) + response.raise_for_status() + + return response.json()