github_webhook.py

"""
GitHub Webhook Handler for PR Review Assistant
Processes GitHub PR webhooks and triggers AgentCore reviews
"""

import json
import requests
import os
from typing import Dict, Any

def extract_pr_data(webhook_payload: Dict[Any, Any]) -> Dict[str, Any]:
    """Extract relevant PR data from GitHub webhook payload"""
    
    pr = webhook_payload.get('pull_request', {})
    
    # Get changed files from GitHub API
    files_url = pr.get('url', '') + '/files'
    headers = {'Authorization': f"token {os.getenv('GITHUB_TOKEN')}"}
    
    try:
        files_response = requests.get(files_url, headers=headers)
        files_data = files_response.json() if files_response.status_code == 200 else []
    except:
        files_data = []
    
    # Format files for review
    files = []
    for file_data in files_data:
        files.append({
            'filename': file_data.get('filename', ''),
            'patch': file_data.get('patch', ''),
            'additions': file_data.get('additions', 0),
            'deletions': file_data.get('deletions', 0)
        })
    
    return {
        'title': pr.get('title', ''),
        'description': pr.get('body', ''),
        'author': pr.get('user', {}).get('login', ''),
        'files': files,
        'pr_number': pr.get('number'),
        'repo': webhook_payload.get('repository', {}).get('full_name', ''),
        'pr_url': pr.get('html_url', '')
    }

def post_review_comment(pr_data: Dict[str, Any], review_text: str) -> bool:
    """Post review as a comment on the GitHub PR"""
    
    repo = pr_data.get('repo')
    pr_number = pr_data.get('pr_number')
    
    if not repo or not pr_number:
        return False
    
    url = f"https://api.github.com/repos/{repo}/issues/{pr_number}/comments"
    headers = {
        'Authorization': f"token {os.getenv('GITHUB_TOKEN')}",
        'Accept': 'application/vnd.github.v3+json'
    }
    
    comment_body = f"""## 🤖 Automated Code Review

{review_text}

---
*This review was generated by AgentCore PR Review Assistant*"""
    
    payload = {'body': comment_body}
    
    try:
        response = requests.post(url, headers=headers, json=payload)
        return response.status_code == 201
    except:
        return False

def lambda_handler(event, context):
    """AWS Lambda handler for GitHub webhooks"""
    
    try:
        # Parse webhook payload
        body = json.loads(event.get('body', '{}'))
        
        # Only process opened/synchronize PR events
        action = body.get('action')
        if action not in ['opened', 'synchronize']:
            return {'statusCode': 200, 'body': 'Event ignored'}
        
        # Extract PR data
        pr_data = extract_pr_data(body)
        
        # Call AgentCore for review
        import boto3
        
        agentcore_client = boto3.client('bedrock-agentcore', region_name=os.getenv('AWS_REGION', 'us-west-2'))
        
        response = agentcore_client.invoke_agent_runtime(
            agentRuntimeArn=os.getenv('AGENTCORE_RUNTIME_ARN'),
            payload=json.dumps({'pr_data': pr_data}),
            sessionId=f"pr-{pr_data.get('pr_number', 'unknown')}"
        )
        
        # Parse review result
        review_result = json.loads(response['payload'])
        review_text = review_result.get('review', 'Review failed')
        
        # Post comment to GitHub
        success = post_review_comment(pr_data, review_text)
        
        return {
            'statusCode': 200,
            'body': json.dumps({
                'message': 'Review completed',
                'posted_to_github': success
            })
        }
        
    except Exception as e:
        return {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }

# For local testing
if __name__ == "__main__":
    # Test with sample webhook payload
    sample_event = {
        'body': json.dumps({
            'action': 'opened',
            'pull_request': {
                'title': 'Test PR',
                'body': 'Test description',
                'number': 123,
                'user': {'login': 'testuser'},
                'url': 'https://api.github.com/repos/test/repo/pulls/123',
                'html_url': 'https://github.com/test/repo/pull/123'
            },
            'repository': {
                'full_name': 'test/repo'
            }
        })
    }
    
    result = lambda_handler(sample_event, None)
    print(json.dumps(result, indent=2))