app.py

import gradio as gr 
import langchain
import huggingface_hub 
import 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