Spaces:
Runtime error
Runtime error
| 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: | |
| 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) | |