import os from typing import Dict, Callable from .terraform_infra import ( AtlantisPRWaiter, TerraformInfraPromptBuilder, TerraformInfraPRAnalyzer, ) from .python_general import ( StandardPRWaiter, PythonGeneralPromptBuilder, PythonGeneralPRAnalyzer, ) from .database import ( DatabasePRWaiter, DatabasePromptBuilder, DatabasePRAnalyzer, ) from .generic import ( GenericPromptBuilder, GenericPRAnalyzer, ) from .javascript import ( JavaScriptPromptBuilder, JavaScriptPRAnalyzer, ) from .php import ( PHPPromptBuilder, PHPPRAnalyzer, ) from .github_client import GitHubClient from .llm_client import BedrockClient def terraform_infra_comment_filter(comments, last_commit_date): """Filter for Terraform plan comments that came after the last commit.""" return [ comment['body'] for comment in comments if ( comment['body'].startswith('Ran Plan for ') or comment['body'].startswith( 'Continued plan output from previous comment.' ) ) and comment['created_at'] > last_commit_date ] class RepoConfig: """Configuration for a specific repository.""" def __init__( self, name: str, pr_waiter_class, prompt_builder_class, analyzer_class, comment_filters: Dict[str, Callable] = None, ): self.name = name self.pr_waiter_class = pr_waiter_class self.prompt_builder_class = prompt_builder_class self.analyzer_class = analyzer_class self.comment_filters = comment_filters or {} # Default project type configurations PROJECT_TYPE_CONFIGS = { 'terraform': RepoConfig( name='terraform', pr_waiter_class=StandardPRWaiter, prompt_builder_class=TerraformInfraPromptBuilder, analyzer_class=TerraformInfraPRAnalyzer, comment_filters={}, ), 'python': RepoConfig( name='python', pr_waiter_class=StandardPRWaiter, prompt_builder_class=PythonGeneralPromptBuilder, analyzer_class=PythonGeneralPRAnalyzer, comment_filters={}, ), 'database': RepoConfig( name='database', pr_waiter_class=DatabasePRWaiter, prompt_builder_class=DatabasePromptBuilder, analyzer_class=DatabasePRAnalyzer, comment_filters={}, ), 'generic': RepoConfig( name='generic', pr_waiter_class=StandardPRWaiter, prompt_builder_class=GenericPromptBuilder, analyzer_class=GenericPRAnalyzer, comment_filters={}, ), 'javascript': RepoConfig( name='javascript', pr_waiter_class=StandardPRWaiter, prompt_builder_class=JavaScriptPromptBuilder, analyzer_class=JavaScriptPRAnalyzer, comment_filters={}, ), 'php': RepoConfig( name='php', pr_waiter_class=StandardPRWaiter, prompt_builder_class=PHPPromptBuilder, analyzer_class=PHPPRAnalyzer, comment_filters={}, ), } # Repository-specific configurations (overrides project types) REPO_CONFIGS = { 'terraform-infra': RepoConfig( name='terraform-infra', pr_waiter_class=AtlantisPRWaiter, prompt_builder_class=TerraformInfraPromptBuilder, analyzer_class=TerraformInfraPRAnalyzer, comment_filters={'tf_plans': terraform_infra_comment_filter}, ), 'pull-request-review-tools': RepoConfig( name='pull-request-review-tools', pr_waiter_class=StandardPRWaiter, prompt_builder_class=PythonGeneralPromptBuilder, analyzer_class=PythonGeneralPRAnalyzer, comment_filters={}, # No special comment filtering needed ), 'database': RepoConfig( name='database', pr_waiter_class=DatabasePRWaiter, prompt_builder_class=DatabasePromptBuilder, analyzer_class=DatabasePRAnalyzer, comment_filters={}, # No special comment filtering needed for now ), } class PRAnalyzerFactory: """Factory for creating repo-specific PR analyzers.""" @staticmethod def create_analyzer(repo_name: str, github_token: str, repo_owner: str): """Create a PR analyzer for the specified repository.""" # First check for repo-specific config if repo_name in REPO_CONFIGS: config = REPO_CONFIGS[repo_name] else: # Fall back to project type from environment variable project_type = os.getenv('PROJECT_TYPE') if not project_type: raise ValueError( f'Unknown repository: {repo_name}. ' f'Either add to REPO_CONFIGS or set PROJECT_TYPE environment variable. ' f'Supported repos: {list(REPO_CONFIGS.keys())}. ' f'Supported project types: {list(PROJECT_TYPE_CONFIGS.keys())}' ) if project_type not in PROJECT_TYPE_CONFIGS: raise ValueError( f'Unknown project type: {project_type}. ' f'Supported project types: {list(PROJECT_TYPE_CONFIGS.keys())}' ) config = PROJECT_TYPE_CONFIGS[project_type] # Create clients github_client = GitHubClient(github_token, repo_owner, repo_name) llm_client = BedrockClient() # Create repo-specific components pr_waiter = config.pr_waiter_class() prompt_builder = config.prompt_builder_class() # Override GitHub client's get_pr_details to include repo-specific filters original_get_pr_details = github_client.get_pr_details def get_pr_details_with_filters(pr_number: int): return original_get_pr_details(pr_number, config.comment_filters) github_client.get_pr_details = get_pr_details_with_filters return config.analyzer_class( github_client, llm_client, prompt_builder, pr_waiter )