Update app.py

app.py

@@ -1,281 +1,283 @@
-import sys
-import shutil
-import logging
-import time
 import os
-from datetime import datetime
-from typing import List, Dict, Any
+import github
+from github import Github
 import requests
-import gradio as gr
-import atexit
-import subprocess
-from urllib.parse import urlparse, quote
-import webbrowser
-
-# Constants
-INPUT_DIRECTORY = 'input'
-OUTPUT_DIRECTORY = 'output'
-LOGS_DIRECTORY = 'logs'
-RESOLUTIONS_DIRECTORY = 'resolutions'
-REPOS_DIRECTORY = 'repos'
-
-# Set up logging
-def initialize_logger() -> logging.Logger:
-    log_file = f"{LOGS_DIRECTORY}/github_bot_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log"
-    logging.basicConfig(
-        level=logging.INFO,
-        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
-        handlers=[
-            logging.FileHandler(log_file),
-            logging.StreamHandler()
-        ]
-    )
-    return logging.getLogger(__name__)
-
-# Initialize environment and logger
-def initialize_environment():
-    directories = [LOGS_DIRECTORY, RESOLUTIONS_DIRECTORY, REPOS_DIRECTORY, INPUT_DIRECTORY, OUTPUT_DIRECTORY]
-    for directory in directories:
-        os.makedirs(directory, exist_ok=True)
-
-# GitHub API handler
-class GitHubAPI:
-    def __init__(self, token: str, logger: logging.Logger):
-        self.token = token
-        self.logger = logger
-        self.headers = {
-            'Authorization': f'token {token}',
-            'Accept': 'application/vnd.github.v3+json'
+from typing import List, Dict, Optional
+import logging
+from datetime import datetime
+import pytest
+from abc import ABC, abstractmethod
+import base64
+from concurrent.futures import ThreadPoolExecutor
+import re
+from dataclasses import dataclass
+from enum import Enum
+
+class IssueSeverity(Enum):
+    CRITICAL = 5
+    HIGH = 4
+    MEDIUM = 3
+    LOW = 2
+    TRIVIAL = 1
+
+@dataclass
+class CodeContext:
+    file_path: str
+    content: str
+    language: str
+
+class GitHubGuardianAngel:
+    def __init__(self, github_token: str, ai_provider: AIProvider):
+        self.gh = Github(github_token)
+        self.ai = ai_provider
+        self.logger = self._setup_logging()
+        self.supported_extensions = {
+            '.py': 'Python',
+            '.js': 'JavaScript',
+            '.ts': 'TypeScript',
+            '.java': 'Java',
+            '.cpp': 'C++',
+            '.go': 'Go',
+            '.rs': 'Rust'
         }
-        self.base_url = "https://api.github.com"
 
-    def _check_rate_limit(self) -> bool:
+    def _setup_logging(self):
+        logger = logging.getLogger('guardian_angel')
+        logger.setLevel(logging.INFO)
+        handler = logging.FileHandler('guardian_angel.log')
+        formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+        handler.setFormatter(formatter)
+        logger.addHandler(handler)
+        return logger
+
+    def _get_main_branch_content(self, repo) -> List[CodeContext]:
+        """
+        Retrieves and analyzes the content of the main branch
+        """
         try:
-            response = requests.get(f"{self.base_url}/rate_limit", headers=self.headers)
-            response.raise_for_status()
-            limits = response.json()
-            remaining = limits['resources']['core']['remaining']
-            reset_time = limits['resources']['core']['reset']
-
-            if remaining < 10:
-                wait_time = max(0, reset_time - int(time.time()))
-                if wait_time > 0:
-                    self.logger.warning(f"Rate limit nearly exceeded. Waiting {wait_time} seconds before retrying...")
-                    time.sleep(wait_time)
-                return False
-            return True
-        except requests.exceptions.RequestException as e:
-            self.logger.error(f"Error checking rate limit: {str(e)}. Retrying...")
-            return True
+            # Get default branch
+            default_branch = repo.default_branch
+            branch = repo.get_branch(default_branch)
+            tree = repo.get_git_tree(branch.commit.sha, recursive=True)
+            
+            code_contexts = []
+            
+            def process_file(element):
+                if element.type == 'blob':
+                    _, ext = os.path.splitext(element.path)
+                    if ext in self.supported_extensions:
+                        try:
+                            content = repo.get_contents(element.path).decoded_content.decode('utf-8')
+                            return CodeContext(
+                                file_path=element.path,
+                                content=content,
+                                language=self.supported_extensions[ext]
+                            )
+                        except Exception as e:
+                            self.logger.warning(f"Failed to process file {element.path}: {str(e)}")
+                return None
+
+            # Process files in parallel
+            with ThreadPoolExecutor(max_workers=10) as executor:
+                results = list(executor.map(process_file, tree.tree))
+            
+            code_contexts = [r for r in results if r is not None]
+            
+            return code_contexts
 
-    def get_repository(self, owner: str, repo: str) -> Dict:
-        try:
-            response = requests.get(f"{self.base_url}/repos/{owner}/{repo}", headers=self.headers)
-            response.raise_for_status()
-            return response.json()
-        except requests.HTTPError as e:
-            self.logger.error(f"HTTP error getting repository info for {owner}/{repo}: {str(e)}. Please check the repository details.")
-            raise
         except Exception as e:
-            self.logger.error(f"Error getting repository info: {str(e)}")
+            self.logger.error(f"Error getting main branch content: {str(e)}")
             raise
 
-    def get_issues(self, owner: str, repo: str, state: str = 'open') -> List[Dict]:
-        if not self._check_rate_limit():
-            return []
-
+    def _determine_severity(self, issue, codebase_analysis) -> IssueSeverity:
+        """
+        Determines issue severity based on various factors
+        """
         try:
-            response = requests.get(f"{self.base_url}/repos/{owner}/{repo}/issues", headers=self.headers, params={'state': state})
-            response.raise_for_status()
-            issues = response.json()
-            return [issue for issue in issues if 'pull_request' not in issue]
-        except Exception as e:
-            self.logger.error(f"Error fetching issues for repository {owner}/{repo}: {str(e)}. Please verify the repository and token.")
-            return []
-
-# GitHub Bot
-class GitHubBot:
-    def __init__(self, logger: logging.Logger):
-        self.github_api = None
-        self.logger = logger
+            severity_indicators = {
+                'critical': ['crash', 'security', 'vulnerability', 'urgent', 'production down'],
+                'high': ['bug', 'error', 'failure', 'broken'],
+                'medium': ['enhancement', 'improvement', 'update needed'],
+                'low': ['minor', 'cosmetic', 'style', 'documentation'],
+                'trivial': ['typo', 'formatting']
+            }
+
+            # Check labels
+            label_texts = [label.name.lower() for label in issue.labels]
+            
+            # Check title and body
+            text_to_analyze = f"{issue.title.lower()} {issue.body.lower()}"
+            
+            # Calculate severity score
+            severity_score = 0
+            
+            for severity, indicators in severity_indicators.items():
+                for indicator in indicators:
+                    if indicator in text_to_analyze or any(indicator in label for label in label_texts):
+                        if severity == 'critical':
+                            severity_score = max(severity_score, 5)
+                        elif severity == 'high':
+                            severity_score = max(severity_score, 4)
+                        elif severity == 'medium':
+                            severity_score = max(severity_score, 3)
+                        elif severity == 'low':
+                            severity_score = max(severity_score, 2)
+                        else:
+                            severity_score = max(severity_score, 1)
+
+            # Consider issue age
+            age_days = (datetime.now() - issue.created_at).days
+            if age_days > 30:
+                severity_score += 1
+            if age_days > 90:
+                severity_score += 1
+
+            # Map score to severity enum
+            return IssueSeverity(min(severity_score, 5))
 
-    def initialize_api(self, token: str):
-        self.github_api = GitHubAPI(token, self.logger)
-
-    def fetch_issues(self, token: str, owner: str, repo: str) -> List[Dict]:
-        try:
-            self.initialize_api(token)
-            return self.github_api.get_issues(owner, repo)
         except Exception as e:
-            self.logger.error(f"Error fetching issues for repository {owner}/{repo}: {str(e )}")
-            return []
+            self.logger.error(f"Error determining severity: {str(e)}")
+            return IssueSeverity.MEDIUM
 
-    def resolve_issue(self, token: str, owner: str, repo: str, issue_number: int, resolution: str, forked_repo: str) -> str:
+    def _get_issue_context(self, repo, issue) -> str:
+        """
+        Gathers relevant context for the issue including related code and discussions
+        """
         try:
-            self.initialize_api(token)
-            self.github_api.get_repository(owner, repo)
-
-            # Create resolution file
-            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
-            resolution_file = f"{RESOLUTIONS_DIRECTORY}/resolution_{issue_number}_{timestamp}.md"
-
-            with open(resolution_file, "w") as f:
-                f.write(f"# Resolution for Issue #{issue_number}\n\n{resolution}")
-
-            # Validate forked_repo before cloning
-            if not forked_repo:
-                raise ValueError("Forked repository URL cannot be empty.")
-
-            # Clone the forked repo
-            subprocess.run(['git', 'clone', forked_repo, '/tmp/' + forked_repo.split('/')[-1]], check=True)
-
-            # Change to the cloned directory
-            os.chdir('/tmp/' + forked_repo.split('/')[-1])
-
-            # Assuming manual intervention now
-            input("Apply the fix manually and stage the changes (press ENTER)? ")
-
-            # Commit and push the modifications
-            subprocess.run(['git', 'add', '.'], check=True)
-            subprocess.run(['git', 'commit', '-m', f"Resolved issue #{issue_number} ({quote(resolution)})"], check=True)
-            subprocess.run(['git', 'push', 'origin', 'HEAD'], check=True)
-
-            # Open Pull Request page
-            webbrowser.open(f'https://github.com/{forked_repo.split("/")[-1]}/compare/master...{owner}:{forked_repo.split("/")[-1]}_resolved_issue_{issue_number}')
-
-            return f"Resolution saved: {resolution_file}"
+            context_parts = []
+            
+            # Add issue details
+            context_parts.append(f"Issue #{issue.number}: {issue.title}")
+            context_parts.append(f"Description: {issue.body}")
+            
+            # Add labels
+            context_parts.append(f"Labels: {', '.join([l.name for l in issue.labels])}")
+            
+            # Add related files (if mentioned in the issue)
+            file_patterns = re.findall(r'`(.*?)`|\b\w+\.[a-zA-Z]+\b', issue.body)
+            related_files = []
+            
+            for pattern in file_patterns:
+                try:
+                    content = repo.get_contents(pattern)
+                    if isinstance(content, list):
+                        continue
+                    decoded_content = content.decoded_content.decode('utf-8')
+                    related_files.append(f"File: {pattern}\n```\n{decoded_content}\n```")
+                except:
+                    continue
+            
+            if related_files:
+                context_parts.append("Related Files:")
+                context_parts.extend(related_files)
+            
+            # Add comments
+            comments = issue.get_comments()
+            if comments.totalCount > 0:
+                context_parts.append("Relevant Comments:")
+                for comment in comments[:5]:  # Limit to last 5 comments
+                    context_parts.append(f"Comment by {comment.user.login}:\n{comment.body}")
+            
+            return "\n\n".join(context_parts)
 
         except Exception as e:
-            error_msg = f"Error resolving issue #{issue_number} in repository {owner}/{repo}: {str(e)}"
-            self.logger.error(error_msg)
-            return error_msg
-
-    def suggest_automated_fixes(self, issue_title: str) -> str:
-        if "missing README" in issue_title.lower():
-            return "Consider adding a README.md file to provide project documentation."
-        return "No automated fix available for this issue."
-
-def handle_issue_selection(token, owner, repo, issue_number, resolution, forked_repo):
-    bot = GitHubBot(logger)
-    result = bot.resolve_issue(token, owner, repo, issue_number, resolution, forked_repo)
-    return result
-
-def extract_info_from_url(url: str) -> Dict[str, Any]:
-    info = {}
-    try:
-        response = requests.get(url)
-        response.raise_for_status()
-        info['status_code'] = response.status_code
-        info['headers'] = dict(response.headers)
-        info['content'] = response.text[:500]  # Limit content to first 500 characters for brevity
-
-        parsed_url = urlparse(url)
-        if 'github.com' in parsed_url.netloc:
-            parts = parsed_url.path.split('/')
-            if len(parts) > 2:
-                owner = parts[1]
-                repo = parts[2]
-                issues = bot.fetch_issues(github_token, owner, repo)
-                info['issues'] = issues
-        elif 'huggingface.co' in parsed_url.netloc:
-            # Add Hugging Face specific handling if needed
-            pass
-
-    except requests.HTTPError as e:
-        info['error'] = f"HTTP error: {str(e)}"
-    except Exception as e:
-        info['error'] = f"Error: {str(e)}"
-    return info
-
-# Initialize GitHubBot globally
-bot = GitHubBot(logger)
-
-# Define missing functions with validation
-def fetch_issues(token, repo_url):
-    try:
-        parts = repo_url.split('/')
-        if len(parts) < 2:
-            raise ValueError("Repository URL is not in the correct format. Expected format: 'owner/repo'.")
+            self.logger.error(f"Error getting issue context: {str(e)}")
+            return f"Issue #{issue.number}: {issue.title}\n{issue.body}"
 
-        owner, repo = parts[-2], parts[-1]
-        issues = bot.fetch_issues(token, owner, repo)
-        return issues
-    except Exception as e:
-        return str(e)
-
-def resolve_issue(token, repo_url, issue_number, resolution, forked_repo_url):
-    try:
-        parts = repo_url.split('/')
-        if len(parts) < 2:
-            raise ValueError("Repository URL is not in the correct format. Expected format: 'owner/repo'.")
-
-        owner, repo = parts[-2], parts[-1]
-        result = bot.resolve_issue(token, owner, repo, issue_number, resolution, forked_repo_url)
-        return result
-    except Exception as e:
-        return str(e)
-
-def extract_info(url):
-    try:
-        info = extract_info_from_url(url)
-        return info
-    except Exception as e:
-        return str(e)
-
-def create_gradio_interface():
-    with gr.Blocks() as demo:
-        gr.Markdown("# GitHub Issue Resolver")
-        gr.Markdown("This application allows you to fetch and resolve GitHub issues efficiently.")
-        
-        with gr.Row():
-            token_input = gr.Textbox(label="GitHub Token", placeholder="Enter your GitHub token")
-            repo_url_input = gr.Textbox(label="Repository URL", placeholder="Enter the repository URL (owner/repo)")
-        
-        with gr.Row():
-            issue_number_input = gr.Number(label="Issue Number", info="Enter the issue number")
-            resolution_input = gr.Textbox(label="Resolution", placeholder="Describe the resolution for the issue", lines=4)
-        
-        forked_repo_input = gr.Textbox(label="Forked Repository URL", placeholder="Enter the forked repository URL")
-        
-        submit_button = gr.Button("Resolve Issue")
-        result_output = gr.Textbox(label="Result", interactive=False)
-
-        def on_submit(token, repo_url, issue_number, resolution, forked_repo):
-            issues = fetch_issues(token, repo_url)
-            if issues:
-                automated_fix = bot.suggest_automated_fixes(issues[0]['title'])
-                resolution += f"\n\n**Automated Suggestion:** {automated_fix}"
-                result = resolve_issue(token, repo_url, issue_number, resolution, forked_repo)
-                return result
-            return "No issues found or an error occurred."
-
-        submit_button.click(on_submit, inputs=[token_input, repo_url_input, issue_number_input, resolution_input, forked_repo_input], outputs=result_output)
-
-    return demo
+    def _test_solution(self, solution: str, repo) -> Dict:
+        """
+        Tests the proposed solution using pytest
+        """
+        try:
+            # Create temporary test directory
+            test_dir = "temp_test_dir"
+            os.makedirs(test_dir, exist_ok=True)
+            
+            # Extract code blocks from solution
+            code_blocks = re.findall(r'```(?:python)?\n(.*?)```', solution, re.DOTALL)
+            
+            test_results = {
+                'status': 'untested',
+                'details': [],
+                'error': None
+            }
+            
+            if not code_blocks:
+                test_results['status'] = 'no_code_found'
+                return test_results
+            
+            # Write code blocks to test files
+            for i, code in enumerate(code_blocks):
+                test_file = os.path.join(test_dir, f'test_solution_{i}.py')
+                with open(test_file, 'w') as f:
+                    f.write(code)
+                
+                try:
+                    # Run pytest on the file
+                    test_output = pytest.main(['-v', test_file])
+                    test_results['details'].append({
+                        'file': f'test_solution_{i}.py',
+                        'status': 'passed' if test_output == 0 else 'failed',
+                        'output': str(test_output)
+                    })
+                except Exception as e:
+                    test_results['details'].append({
+                        'file': f'test_solution_{i}.py',
+                        'status': 'error',
+                        'error': str(e)
+                    })
+            
+            # Determine overall status
+            if any(d['status'] == 'error' for d in test_results['details']):
+                test_results['status'] = 'error'
+            elif any(d['status'] == 'failed' for d in test_results['details']):
+                test_results['status'] = 'failed'
+            else:
+                test_results['status'] = 'passed'
+            
+            return test_results
+            
+        except Exception as e:
+            self.logger.error(f"Error testing solution: {str(e)}")
+            return {
+                'status': 'error',
+                'details': [],
+                'error': str(e)
+            }
+        finally:
+            # Cleanup
+            if os.path.exists(test_dir):
+                import shutil
+                shutil.rmtree(test_dir)
+
+    def _comment_solution(self, issue, solution: str, test_results: Dict):
+        """
+        Posts a detailed solution comment on the issue
+        """
+        status_emoji = {
+            'passed': '✅',
+            'failed': '❌',
+            'error': '⚠️',
+            'untested': '⚪',
+            'no_code_found': '❓'
+        }
 
-# Cleanup function
-def cleanup():
-    try:
-        temp_dirs = [REPOS_DIRECTORY]
-        for dir_name in temp_dirs:
-            if os.path.exists(dir_name):
-                shutil.rmtree(dir_name)
-        logging.shutdown()
-    except Exception as e:
-        print(f"Error during cleanup: {str(e)}")
+        comment = f"""
+## 🔮 GitHub Guardian Angel Analysis
+### Proposed Solution:
+{solution}
+Test Results {status_emoji.get(test_results['status'], '⚪')}
+Status: {test_results['status'].upper()}
 
-def main():
-    # Initialize environment and logger
-    initialize_environment()
-    global logger
-    logger = initialize_logger()
+{"#### Test Details:" if test_results['details'] else ""} {"".join([f"- {d['file']}: {status_emoji.get(d['status'], '⚪')} {d['status'].upper()}\n" for d in test_results['details']])}
 
-    # Register cleanup handlers
-    atexit.register(cleanup)
+{f"⚠️ Error: {test_results['error']}" if test_results.get('error') else ""}
 
-    # Create Gradio interface
-    demo = create_gradio_interface()
-    demo.launch()
+Implementation Steps:
+Review the proposed solution and test results
+Apply the changes in the code blocks above
+Run the provided tests to verify the fix
+If tests pass, commit and push the changes
+Close this issue with a reference to the fixing commit
+💡 Please provide feedback on this solution. If you need any clarification or adjustments, let me know! """ try: issue.create_comment(comment) self.logger.info(f"Posted solution comment on issue #{issue.number}") except Exception as e: self.logger.error(f"Error posting comment: {str(e)}") raise
 
-if __name__ == "__main__":
-    main()