app.py

import gradio as gr 
import langchain
import huggingface-hub 
import python-dotenv
import PyYaml
from typing import Optional, Union, List, Dict, Any, Tuple
import subprocess
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:
            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:
            return "", str(e), 1

class GitUtilities:
    def __init__(self, repo_path: str):
        self.repo_path = Path(repo_path)

    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
        self.logger = logger
        self.ai_provider = None
        self.file_types = {
            '.py': {'language': 'Python', 'parser': 'python'},
            '.js': {'language': 'JavaScript', 'parser': 'javascript'},
            '.ts': {'language': 'TypeScript', 'parser': 'typescript'},
            '.java': {'language': 'Java', 'parser': 'java'},
            '.cpp': {'language': 'C++', 'parser': 'cpp'},
            '.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.logger)
        self.ai_provider = ai_provider
        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()

            # Determine severity
            severity = self._determine_severity(issue)
            
            # Generate AI solution
            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']}"

        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}
## Implementation Details
- Implementation Date: {datetime.now(timezone.utc).isoformat()}
- Severity Level: {severity.name}
- Resolution Type: {'AI-ASSISTED' if ai_solution else 'MANUAL'}
## 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.
"""

    # The following lines should not be part of the _create_resolution_document method
    script_file = f"{RESOLUTIONS_DIRECTORY}/resolve_issue_{issue_number}.sh"
    with open(script_file, "w") as f:
        f.write(bash_script)

    # Make the script executable
    os.chmod(script_file, 0o755)

    return f"Bash script generated: {script_file}"

        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 create_gradio_interface():
    with gr.Blocks() as demo:
        gr.Markdown("# Enhanced GitHub Issue Resolver")
        gr.Markdown("AI-assisted issue resolution with integrated git operations")
        
        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.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")

            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")

            with gr.Row():
                analyze_button = gr.Button("Analyze Issue")
                resolve_button = gr.Button("Resolve Issue")

            result_output = gr.Textbox(label="Result", interactive=False)

        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]
                
                # Initialize bot with AI provider
                ai_provider = CustomAIProvider(ai_api_base, ai_api_key)
                bot.initialize_api(token, ai_provider)
                
                result = bot.resolve_issue(token, owner, repo, int(issue_number), resolution, forked_repo)
                return result
            except Exception as e:
                return f"Error: {str(e)}"

        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]
        )

    return demo