Add reports, integrations, and utils modules

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()