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 import requests import logging import shutil # Configure logging logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s") logger = logging.getLogger(__name__) 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: logger.error(f"Error executing command {command}: {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""" stdout, stderr, code = TerminalCommand.execute( f"git clone -b {branch} {url} {self.repo_path}" ) if code != 0: logger.error(f"Git clone failed: {stderr}") return code == 0 def commit(self, message: str) -> bool: """Create a commit with the given message""" stdout, stderr, code = TerminalCommand.execute( ["git", "commit", "-m", message], str(self.repo_path) ) if code != 0: logger.error(f"Git commit failed: {stderr}") return code == 0 def push(self, remote: str = "origin", branch: str = "main") -> bool: """Push changes to remote""" stdout, stderr, code = TerminalCommand.execute( ["git", "push", remote, branch], str(self.repo_path) ) if code != 0: logger.error(f"Git push failed: {stderr}") return code == 0 def create_branch(self, branch_name: str) -> bool: """Create and checkout a new branch""" stdout, stderr, code = TerminalCommand.execute( ["git", "checkout", "-b", branch_name], str(self.repo_path) ) if code != 0: logger.error(f"Git branch creation failed: {stderr}") return code == 0 class GitHubBot: def __init__(self, logger: logging.Logger): self.github_api = None self.logger = logger self.ai_provider = None self.git = None self.temp_dir = None self.base_url = "https://api.github.com" def initialize_api(self, token: str): """Initialize the GitHub API with a token.""" if not token: raise ValueError("GitHub token is required.") self.github_api = {"Authorization": f"Bearer {token}"} 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.base_url}/repos/{owner}/{repo}/pulls" data = { "title": title, "body": body, "head": head, "base": base } try: response = requests.post(url, headers=self.github_api, json=data) response.raise_for_status() return response.json() except requests.RequestException as e: logger.error(f"Error creating pull request: {e}") raise def resolve_issue(self, token: str, owner: str, repo: str, issue_number: int, resolution: str, forked_repo: str) -> str: """Resolve a GitHub issue.""" try: self.initialize_api(token) branch_name = f"fix/issue-{issue_number}-{datetime.now().strftime('%Y%m%d-%H%M%S')}" # Clone repository if not self.git.clone(forked_repo): raise Exception("Failed to clone repository") # Create a new branch if not self.git.create_branch(branch_name): raise Exception("Failed to create branch") # Generate resolution content resolution_content = self._create_resolution_document(issue_number, resolution) # Save resolution file resolution_path = Path(self.temp_dir) / f"resolution_{issue_number}.md" with open(resolution_path, "w") as f: f.write(resolution_content) # Commit and push changes if not self.git.commit(f"Fix for issue #{issue_number}"): raise Exception("Failed to commit changes") if not self.git.push("origin", branch_name): raise Exception("Failed to push changes") # Create a pull request pr = self.create_pull_request( owner=owner, repo=repo, title=f"Fix for issue #{issue_number}", body="This PR resolves the reported issue with the following resolution.", head=branch_name ) return f"Pull request created: {pr['html_url']}" except Exception as e: logger.error(f"Error resolving issue #{issue_number}: {e}") return f"Error: {e}" finally: if self.temp_dir and os.path.exists(self.temp_dir): shutil.rmtree(self.temp_dir) def _create_resolution_document(self, issue_number: int, resolution: str) -> str: """Create a resolution document.""" return f"""# Resolution for Issue #{issue_number} ## Resolution Details {resolution} ## Metadata - Date: {datetime.now(timezone.utc).isoformat()} - Resolved By: Automated System """ def create_gradio_interface(): """Create the Gradio interface.""" bot = GitHubBot(logger) def on_resolve(token, repo_url, issue_number, resolution, forked_repo): try: parts = repo_url.strip("/").split("/") owner, repo = parts[-2], parts[-1] result = bot.resolve_issue(token, owner, repo, int(issue_number), resolution, forked_repo) return result except Exception as e: logger.error(f"Error in issue resolution: {e}") return f"Error: {e}" with gr.Blocks() as demo: gr.Markdown("# GitHub Issue Resolver") gr.Markdown("Resolve GitHub issues with AI assistance and Git integration.") with gr.Tab("Issue Resolution"): 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") issue_number_input = gr.Number(label="Issue Number", precision=0) resolution_input = gr.Textbox(label="Resolution", placeholder="Describe the resolution for the issue") forked_repo_input = gr.Textbox(label="Forked Repo URL", placeholder="Enter the forked repository URL") resolve_button = gr.Button("Resolve Issue") result_output = gr.Textbox(label="Result", interactive=False) resolve_button.click( fn=on_resolve, inputs=[token_input, repo_url_input, issue_number_input, resolution_input, forked_repo_input], outputs=[result_output] ) return demo if __name__ == "__main__": demo = create_gradio_interface() demo.launch(server_name="0.0.0.0", server_port=7860)