app.py

import os
import shutil
import logging
import time
import requests
import gradio as gr
import atexit
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:
        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
        except Exception as e:
            logger.error(f"Error getting repository info: {str(e)}")
            raise

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

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

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'}
        }

    def initialize_api(self, token: str, ai_provider: AIProvider = None):
        self.github_api = GitHubAPI(token)
        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']}"

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

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

            # Generate AI solution
            context = f"Repository: {owner}/{repo}\nIssue #{issue_number}: {issue['title']}\n\nError Replication:\n{error_replication}\n"
            ai_solution = self.ai_provider.generate_solution(context, issue['body'])

            # Create bash script for resolution
            bash_script = f"""#!/bin/bash
# Bash script to resolve issue #{issue_number} in {owner}/{repo}

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}

echo "Resolution complete."
"""

            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("# GitHub Issue Resolver with AI Assistant")
        
        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():
            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, 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)
                
                # 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,
            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

# 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()