Update app.py

app.py

@@ -1,132 +1,71 @@
-import os
-import shutil
-import logging
-import time
-import requests
-import gradio as gr
-import atexit
+# Add these new imports
+from typing import Optional, Union, List, Dict, Any, Tuple
 import subprocess
-from urllib.parse import urlparse, quote
-import webbrowser
-from datetime import datetime
-from typing import List, Dict, Any
-from abc import ABC, abstractmethod
-from enum import Enum
-from dataclasses import dataclass
-from concurrent.futures import ThreadPoolExecutor
-import base64
-
-# 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
-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):
-        self.token = token
-        self.headers = {
-            'Authorization': f'token {token}',
-            'Accept': 'application/vnd.github.v3+json'
-        }
-        self.base_url = "https://api.github.com"
-
-    def get_repository(self, owner: str, repo: str) -> Dict:
+from pathlib import Path
+import json
+import tempfile
+from datetime import datetime, timezone
+import re
+
+# Add these utility classes based on git-bob's architecture
+class TerminalCommand:
+    @staticmethod
+    def execute(command: Union[str, List[str]], cwd: Optional[str] = None) -> Tuple[str, str, int]:
+        """
+        Execute a terminal command and return stdout, stderr, and return code
+        """
+        if isinstance(command, str):
+            command = command.split()
+        
         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:
-            logger.error(f"HTTP error getting repository info for {owner}/{repo}: {str(e)}")
-            raise
+            process = subprocess.Popen(
+                command,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                cwd=cwd,
+                text=True
+            )
+            stdout, stderr = process.communicate()
+            return stdout.strip(), stderr.strip(), process.returncode
         except Exception as e:
-            logger.error(f"Error getting repository info: {str(e)}")
-            raise
+            return "", str(e), 1
 
-    def get_issues(self, owner: str, repo: str, state: str = 'open') -> List[Dict]:
-        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:
-            logger.error(f"Error fetching issues for repository {owner}/{repo}: {str(e)}")
-            return []
-
-# Enum for Issue Severity
-class IssueSeverity(Enum):
-    CRITICAL = 5
-    HIGH = 4
-    MEDIUM = 3
-    LOW = 2
-    TRIVIAL = 1
-
-@dataclass
-class CodeContext:
-    file_path: str
-    content: str
-    language: str
-
-class AIProvider(ABC):
-    @abstractmethod
-    def analyze_code(self, content: str) -> str:
-        pass
-    
-    @abstractmethod
-    def generate_solution(self, context: str, issue: str) -> str:
-        pass
-
-class CustomAIProvider(AIProvider):
-    def __init__(self, api_base: str, api_key: str, model: str = "gpt-3.5-turbo"):
-        self.api_base = api_base
-        self.api_key = api_key
-        self.model = model
-        self.headers = {
-            "Authorization": f"Bearer {api_key}",
-            "Content-Type": "application/json"
-        }
+class GitUtilities:
+    def __init__(self, repo_path: str):
+        self.repo_path = Path(repo_path)
 
-    def analyze_code(self, content: str) -> str:
-        payload = {
-            "model": self.model,
-            "messages": [{"role": "user", "content": f"Analyze this code:\n{content}"}]
-        }
-        response = requests.post(f"{self.api_base}/v1/chat/completions", 
-                                 headers = self.headers, 
-            json=payload)
-        return response.json()["choices"][0]["message"]["content"]
-
-    def generate_solution(self, context: str, issue: str) -> str:
-        payload = {
-            "model": self.model,
-            "messages": [{"role": "user", "content": f"Context:\n{context}\nIssue:\n{issue}\nGenerate solution:"}]
-        }
-        response = requests.post(f"{self.api_base}/v1/chat/completions", 
-                                 headers=self.headers, 
-                                 json=payload)
-        return response.json()["choices"][0]["message"]["content"]
+    def clone(self, url: str, branch: str = "main") -> bool:
+        """Clone a repository"""
+        _, stderr, code = TerminalCommand.execute(
+            f"git clone -b {branch} {url} {self.repo_path}"
+        )
+        return code == 0
+
+    def commit(self, message: str) -> bool:
+        """Create a commit with the given message"""
+        _, _, code = TerminalCommand.execute(
+            ["git", "commit", "-m", message],
+            str(self.repo_path)
+        )
+        return code == 0
 
+    def push(self, remote: str = "origin", branch: str = "main") -> bool:
+        """Push changes to remote"""
+        _, _, code = TerminalCommand.execute(
+            ["git", "push", remote, branch],
+            str(self.repo_path)
+        )
+        return code == 0
+
+    def create_branch(self, branch_name: str) -> bool:
+        """Create and checkout a new branch"""
+        _, _, code = TerminalCommand.execute(
+            ["git", "checkout", "-b", branch_name],
+            str(self.repo_path)
+        )
+        return code == 0
+
+# Enhance the GitHubBot class with new features
 class GitHubBot:
     def __init__(self, logger: logging.Logger):
         self.github_api = None
@@ -141,49 +80,135 @@ class GitHubBot:
             '.go': {'language': 'Go', 'parser': 'go'},
             '.rs': {'language': 'Rust', 'parser': 'rust'}
         }
+        self.git = None
+        self.temp_dir = None
 
     def initialize_api(self, token: str, ai_provider: AIProvider = None):
-        self.github_api = GitHubAPI(token)
+        self.github_api = GitHubAPI(token, self.logger)
         self.ai_provider = ai_provider
-
-    def crawl_issues(self, owner: str, repo: str) -> List[Dict]:
-        issues = self.github_api.get_issues(owner, repo)
-        return issues
-
-    def replicate_error(self, issue: Dict) -> str:
-        # Simulate the error in a controlled environment
-        # This is a placeholder for actual error replication logic
-        return f"Simulated error for issue: {issue['title']}"
+        self.temp_dir = tempfile.mkdtemp()
+        self.git = GitUtilities(self.temp_dir)
+
+    def create_pull_request(self, owner: str, repo: str, title: str, body: str, head: str, base: str = "main") -> Dict:
+        """Create a pull request"""
+        url = f"{self.github_api.base_url}/repos/{owner}/{repo}/pulls"
+        data = {
+            "title": title,
+            "body": body,
+            "head": head,
+            "base": base
+        }
+        response = requests.post(url, headers=self.github_api.headers, json=data)
+        response.raise_for_status()
+        return response.json()
 
     def resolve_issue(self, token: str, owner: str, repo: str, issue_number: int, resolution: str, forked_repo: str) -> str:
         try:
             if not self.ai_provider:
                 raise ValueError("AI provider not initialized. Please initialize with an AI provider.")
 
+            # Create a unique branch name for this fix
+            branch_name = f"fix/issue-{issue_number}-{datetime.now().strftime('%Y%m%d-%H%M%S')}"
+
+            # Clone repository and create new branch
+            if not self.git.clone(forked_repo):
+                raise Exception("Failed to clone repository")
+
+            if not self.git.create_branch(branch_name):
+                raise Exception("Failed to create branch")
+
+            # Get repository content and analyze
+            code_contexts = self._get_main_branch_content(owner, repo)
+            
             # Get issue details
             issue_url = f"{self.github_api.base_url}/repos/{owner}/{repo}/issues/{issue_number}"
             response = requests.get(issue_url, headers=self.github_api.headers)
             response.raise_for_status()
             issue = response.json()
 
-            # Replicate the error
-            error_replication = self.replicate_error(issue)
-
+            # Determine severity
+            severity = self._determine_severity(issue)
+            
             # Generate AI solution
-            context = f"Repository: {owner}/{repo}\nIssue #{issue_number}: {issue['title']}\n\nError Replication:\n{error_replication}\n"
+            context = f"Repository: {owner}/{repo}\nIssue #{issue_number}: {issue['title']}\n\nCode contexts:\n"
+            for ctx in code_contexts[:3]:
+                context += f"\nFile: {ctx.file_path}\n```{ctx.language}\n{ctx.content}\n```\n"
+            
             ai_solution = self.ai_provider.generate_solution(context, issue['body'])
+            
+            # Create comprehensive resolution document
+            full_resolution = self._create_resolution_document(
+                issue_number, severity, resolution, ai_solution
+            )
+            
+            # Save resolution
+            resolution_path = Path(RESOLUTIONS_DIRECTORY) / f"resolution_{issue_number}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.md"
+            resolution_path.write_text(full_resolution)
+
+            # Prompt for manual intervention
+            print("\nAI-generated solution and original resolution have been saved.")
+            print(f"Please review and apply changes in: {self.temp_dir}")
+            input("Press Enter when changes are ready to commit...")
+
+            # Commit and push changes
+            if not self.git.commit(f"Fix #{issue_number}: {issue['title']}\n\n{resolution}"):
+                raise Exception("Failed to commit changes")
+
+            if not self.git.push("origin", branch_name):
+                raise Exception("Failed to push changes")
+
+            # Create pull request
+            pr = self.create_pull_request(
+                owner=owner,
+                repo=repo,
+                title=f"Fix #{issue_number}: {issue['title']}",
+                body=full_resolution,
+                head=branch_name
+            )
+
+            return f"Resolution implemented and PR created: {pr['html_url']}"
 
-            # Create bash script for resolution
-            bash_script = f"""#!/bin/bash
-# Bash script to resolve issue #{issue_number} in {owner}/{repo}
+        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
+
+        finally:
+            # Cleanup
+            if self.temp_dir and os.path.exists(self.temp_dir):
+                shutil.rmtree(self.temp_dir)
+
+    def _create_resolution_document(self, issue_number: int, severity: IssueSeverity, 
+                                 resolution: str, ai_solution: str) -> str:
+        """Create a comprehensive resolution document"""
+        return f"""# Resolution for Issue #{issue_number}
+
+## Severity: {severity.name}
+
+## Original Resolution
+{resolution}
+
+## AI-Generated Solution
+{ai_solution}
 
-echo "Starting resolution for issue #{issue_number}..."
-# Add commands to resolve the issue here
-# Example: git clone {forked_repo}
-# Example: cd {repo}
-# Example: {ai_solution}
+## Implementation Details
+- Implementation Date: {datetime.now(timezone.utc).isoformat()}
+- Severity Level: {severity.name}
+- Resolution Type: {'AI-ASSISTED' if ai_solution else 'MANUAL'}
 
-echo "Resolution complete."
+## Testing Notes
+- [ ] Code changes have been tested locally
+- [ ] All tests pass
+- [ ] No new issues introduced
+
+## Review Checklist
+- [ ] Code follows project style guidelines
+- [ ] Documentation has been updated
+- [ ] Changes are properly tested
+- [ ] No security vulnerabilities introduced
+
+## Additional Notes
+Please review the changes carefully before merging.
 """
 
             script_file = f"{RESOLUTIONS_DIRECTORY}/resolve_issue_{issue_number}.sh"
@@ -202,26 +227,54 @@ echo "Resolution complete."
 
 def create_gradio_interface():
     with gr.Blocks() as demo:
-        gr.Markdown("# GitHub Issue Resolver with AI Assistant")
+        gr.Markdown("# Enhanced GitHub Issue Resolver")
+        gr.Markdown("AI-assisted issue resolution with integrated git operations")
         
-        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.Tab("Configuration"):
+            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():
+                ai_api_base = gr.Textbox(label="AI API Base URL", placeholder="Enter your AI API base URL")
+                ai_api_key = gr.Textbox(label="AI API Key", placeholder="Enter your AI API key")
 
-        with gr.Row():
-            ai_api_base = gr.Textbox(label="AI API Base URL", placeholder="Enter your AI API base URL")
-            ai_api_key = gr.Textbox(label="AI API Key", placeholder="Enter your AI API key")
+        with gr.Tab("Issue Resolution"):
+            with gr.Row():
+                issue_number_input = gr.Number(label="Issue Number", info="Enter the issue number")
+                severity_indicator = gr.Label(label="Issue Severity")
 
-        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")
 
-        forked_repo_input = gr.Textbox(label="Forked Repository URL", placeholder="Enter the forked repository URL")
+            with gr.Row():
+                analyze_button = gr.Button("Analyze Issue")
+                resolve_button = gr.Button("Resolve Issue")
 
-        submit_button = gr.Button("Resolve Issue")
-        result_output = gr.Textbox(label="Result", interactive=False)
+            result_output = gr.Textbox(label="Result", interactive=False)
 
-        def on_submit(token, repo_url, ai_api_base, ai_api_key, issue_number, resolution, forked_repo):
+        def on_analyze(token, repo_url, issue_number):
+            try:
+                parts = repo_url.split('/')
+                owner, repo = parts[-2], parts[-1]
+                
+                # Initialize bot
+                bot.initialize_api(token)
+                
+                # Get issue details
+                issue_url = f"{bot.github_api.base_url}/repos/{owner}/{repo}/issues/{issue_number}"
+                response = requests.get(issue_url, headers=bot.github_api.headers)
+                response.raise_for_status()
+                issue = response.json()
+                
+                # Determine severity
+                severity = bot._determine_severity(issue)
+                
+                return f"Issue Severity: {severity.name}"
+            except Exception as e:
+                return f"Error: {str(e)}"
+
+        def on_resolve(token, repo_url, ai_api_base, ai_api_key, issue_number, resolution, forked_repo):
             try:
                 parts = repo_url.split('/')
                 owner, repo = parts[-2], parts[-1]
@@ -230,30 +283,22 @@ def create_gradio_interface():
                 ai_provider = CustomAIProvider(ai_api_base, ai_api_key)
                 bot.initialize_api(token, ai_provider)
                 
-                # Crawl issues and resolve
-                issues = bot.crawl_issues(owner, repo)
                 result = bot.resolve_issue(token, owner, repo, int(issue_number), resolution, forked_repo)
                 return result
             except Exception as e:
                 return f"Error: {str(e)}"
 
-        submit_button.click(
-            on_submit,
+        analyze_button.click(
+            on_analyze,
+            inputs=[token_input, repo_url_input, issue_number_input],
+            outputs=[severity_indicator]
+        )
+
+        resolve_button.click(
+            on_resolve,
             inputs=[token_input, repo_url_input, ai_api_base, ai_api_key, 
                    issue_number_input, resolution_input, forked_repo_input],
-            outputs=result_output
+            outputs=[result_output]
         )
 
-    return demo
-
-# Initialize the logger and environment
-logger = initialize_logger()
-initialize_environment()
-
-# Create the GitHub bot instance
-bot = GitHubBot(logger)
-
-# Launch the Gradio interface
-if __name__ == "__main__":
-    demo = create_gradio_interface()
-    demo.launch()
+    return demo