Update app.py

app.py

@@ -1,283 +1,282 @@
-import os
-import github
-from github import Github
-import requests
-from typing import List, Dict, Optional
+import sys
+import shutil
 import logging
+import time
+import os
 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'
+from typing import List, Dict, Any
+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'
         }
+        self.base_url = "https://api.github.com"
 
-    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
-        """
+    def _check_rate_limit(self) -> bool:
         try:
-            # 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
+            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
 
+    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 main branch content: {str(e)}")
+            self.logger.error(f"Error getting repository info: {str(e)}")
             raise
 
-    def _determine_severity(self, issue, codebase_analysis) -> IssueSeverity:
-        """
-        Determines issue severity based on various factors
-        """
-        try:
-            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 get_issues(self, owner: str, repo: str, state: str = 'open') -> List[Dict]:
+        if not self._check_rate_limit():
+            return []
 
+        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 determining severity: {str(e)}")
-            return IssueSeverity.MEDIUM
+            self.logger.error(f"Error fetching issues for repository {owner}/{repo}: {str(e)}. Please verify the repository and token.")
+            return []
 
-    def _get_issue_context(self, repo, issue) -> str:
-        """
-        Gathers relevant context for the issue including related code and discussions
-        """
-        try:
-            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)
+# GitHub Bot
+class GitHubBot:
+    def __init__(self, logger: logging.Logger):
+        self.github_api = None
+        self.logger = logger
+
+    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 getting issue context: {str(e)}")
-            return f"Issue #{issue.number}: {issue.title}\n{issue.body}"
+            self.logger.error(f"Error fetching issues for repository {owner}/{repo}: {str(e)}")
+            return []
 
-    def _test_solution(self, solution: str, repo) -> Dict:
-        """
-        Tests the proposed solution using pytest
-        """
+    def resolve_issue(self, token: str, owner: str, repo: str, issue_number: int, resolution: str, forked_repo: str) -> str:
         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
-            
+            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}"
+
         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': '❓'
-        }
+            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
+logger = initialize_logger()  # Initialize logger before creating the bot
+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'.")
+
+        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
 
-        comment = f"""
-## 🔮 GitHub Guardian Angel Analysis
-### Proposed Solution:
-{solution}
-Test Results {status_emoji.get(test_results['status'], '⚪')}
-Status: {test_results['status'].upper()}
+# 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)}")
 
-{"#### 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']])}
+def main():
+    # Initialize environment and logger
+    initialize_environment()
+    global logger
+    logger = initialize_logger()
 
-{f"⚠️ Error: {test_results['error']}" if test_results.get('error') else ""}
+    # Register cleanup handlers
+    atexit.register(cleanup)
 
-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
+    # Create Gradio interface
+    demo = create_gradio_interface()
+    demo.launch()
 
+if __name__ == "__main__":
+    main()