""" GitHub Integration Service Handles OAuth, webhooks, and PR event processing """ from fastapi import FastAPI, HTTPException, Request, Header from pydantic import BaseModel from typing import Dict, List, Optional, Any import json import datetime import hashlib import base64 import hmac import re import urllib.parse from enum import Enum app = FastAPI(title="GitHub Integration Service", version="1.0.0") class EventType(Enum): PULL_REQUEST = "pull_request" PUSH = "push" ISSUE = "issues" class GitHubEvent(BaseModel): action: str repository: Dict[str, Any] pull_request: Optional[Dict[str, Any]] = None sender: Dict[str, Any] class PRFile(BaseModel): filename: str status: str additions: int deletions: int patch: str blob_url: str class GitHubWebhookService: def __init__(self): self.webhook_secret = "demo-secret" self.access_tokens = {} self.repo_cache = {} def verify_webhook_signature(self, payload: str, signature: str) -> bool: if not signature: return False expected_signature = f"sha256={hmac.new( self.webhook_secret.encode(), payload.encode(), hashlib.sha256 ).hexdigest()}" return hmac.compare_digest(signature, expected_signature) def get_access_token(self, code: str) -> str: token_hash = hashlib.sha256(f"{code}-{datetime.datetime.now().isoformat()}".encode()).hexdigest() self.access_tokens[token_hash] = { "token": f"ghp_{token_hash[:16]}", "expires": datetime.datetime.now() + datetime.timedelta(hours=8), "scopes": ["repo", "read:org"] } return self.access_tokens[token_hash]["token"] def get_pr_files(self, repo_owner: str, repo_name: str, pr_number: int, token: str) -> List[PRFile]: cache_key = f"{repo_owner}/{repo_name}/pr/{pr_number}" if cache_key in self.repo_cache: return self.repo_cache[cache_key] mock_files = [ PRFile( filename="src/main.py", status="modified", additions=15, deletions=5, patch="@@ -1,10 +1,20 @@\n+def new_function():\n+ pass\n", blob_url=f"https://github.com/{repo_owner}/{repo_name}/blob/main/src/main.py" ), PRFile( filename="tests/test_main.py", status="added", additions=25, deletions=0, patch="@@ -0,0 +1,25 @@\n+import unittest\n", blob_url=f"https://github.com/{repo_owner}/{repo_name}/blob/main/tests/test_main.py" ) ] self.repo_cache[cache_key] = mock_files return mock_files github_service = GitHubWebhookService() @app.get("/health") async def health(): return {"status": "healthy", "service": "github-integration"} @app.post("/webhook") async def handle_webhook( request: Request, x_hub_signature: str = Header(None), x_github_event: str = Header(None) ): try: body = await request.body() payload = body.decode('utf-8') if not github_service.verify_webhook_signature(payload, x_hub_signature): raise HTTPException(status_code=401, detail="Invalid signature") event_data = json.loads(payload) event = GitHubEvent(**event_data) if x_github_event == "pull_request" and event.action in ["opened", "synchronize"]: await process_pr_event(event) return {"status": "processed", "event": x_github_event} except Exception as e: raise HTTPException(status_code=500, detail=f"Webhook processing failed: {str(e)}") async def process_pr_event(event: GitHubEvent): pr_data = event.pull_request if not pr_data: return repo_data = event.repository pr_number = pr_data.get("number", 0) repo_owner = repo_data.get("owner", {}).get("login", "unknown") repo_name = repo_data.get("name", "unknown") files = github_service.get_pr_files(repo_owner, repo_name, pr_number, "mock-token") analysis_request = { "repository_url": repo_data.get("html_url", ""), "pull_request_id": pr_number, "files": [ { "path": file.filename, "content": file.patch, "additions": file.additions, "deletions": file.deletions, "status": file.status } for file in files ], "user_token": "mock-token", "pr_data": pr_data } print(f"Processing PR #{pr_number} with {len(files)} files") @app.get("/auth/oauth") async def oauth_callback(code: str, state: Optional[str] = None): try: access_token = github_service.get_access_token(code) return { "access_token": access_token, "token_type": "bearer", "scope": ["repo", "read:org"], "expires_in": 28800 } except Exception as e: raise HTTPException(status_code=500, detail=f"OAuth failed: {str(e)}") @app.get("/repos/{owner}/{repo}/pulls/{pr_number}/files") async def get_pr_files(owner: str, repo: str, pr_number: int, authorization: str = Header(None)): try: if not authorization: raise HTTPException(status_code=401, detail="Authorization required") token = authorization.replace("Bearer ", "") files = github_service.get_pr_files(owner, repo, pr_number, token) return { "files": [ { "filename": file.filename, "status": file.status, "additions": file.additions, "deletions": file.deletions, "patch": file.patch, "blob_url": file.blob_url } for file in files ], "total_count": len(files) } except Exception as e: raise HTTPException(status_code=500, detail=f"Failed to fetch PR files: {str(e)}") @app.post("/repos/{owner}/{repo}/pulls/{pr_number}/reviews") async def create_review( owner: str, repo: str, pr_number: int, review_data: Dict[str, Any], authorization: str = Header(None) ): try: if not authorization: raise HTTPException(status_code=401, detail="Authorization required") review_id = hashlib.sha256(f"{owner}-{repo}-{pr_number}-{datetime.datetime.now().isoformat()}".encode()).hexdigest()[:8] return { "id": int(review_id, 16), "pull_request_url": f"https://api.github.com/repos/{owner}/{repo}/pulls/{pr_number}", "commit_id": "mock-commit-id", "user": {"login": "ai-reviewer"}, "body": review_data.get("body", "AI Code Review"), "state": review_data.get("state", "COMMENT"), "html_url": f"https://github.com/{owner}/{repo}/pull/{pr_number}#pullrequestreview-{review_id}", "submitted_at": datetime.datetime.now().isoformat() } except Exception as e: raise HTTPException(status_code=500, detail=f"Failed to create review: {str(e)}") if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8001)