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