Update app.py

app.py

@@ -1,173 +1,215 @@
 import gradio as gr
-import os
-import requests
-import uuid
+import langchain
+import huggingface_hub
+import dotenv
+import yaml
+from typing import Optional, Union, List, Dict, Any, Tuple
+import subprocess
 from pathlib import Path
-from typing import Optional, Union, List, Tuple, Dict, Any
-from pypdf import PdfReader
-from bs4 import BeautifulSoup
-import zipfile
-import nltk
-import logging
+import json
 import tempfile
+from datetime import datetime, timezone
+import re
+import requests
+import logging
 import shutil
 
-nltk.download('punkt')
-
 # Configure logging
 logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
 logger = logging.getLogger(__name__)
 
-# Utility to log messages
-def log(message: str):
-    logger.info(message)
-
-# File and Web Processing Utilities
-def chunk_text(text: str, max_chunk_size: int) -> List[str]:
-    """Breaks large text into manageable chunks."""
-    tokenizer = nltk.data.load('tokenizers/punkt/english.pickle')
-    sentences = tokenizer.tokenize(text)
-    chunks = []
-    current_chunk = ""
-    for sentence in sentences:
-        if len(current_chunk) + len(sentence) + 1 > max_chunk_size:
-            chunks.append(current_chunk.strip())
-            current_chunk = ""
-        current_chunk += sentence + " "
-    if current_chunk:
-        chunks.append(current_chunk.strip())
-    return chunks
-
-def read_pdf(file_path: str) -> str:
-    """Reads text content from a PDF file."""
-    try:
-        reader = PdfReader(file_path)
-        return "\n".join(page.extract_text() for page in reader.pages)
-    except Exception as e:
-        logger.error(f"Error reading PDF: {e}")
-        return ""
-
-def read_txt(file_path: str) -> str:
-    """Reads content from a TXT file."""
-    try:
-        with open(file_path, "r", encoding="utf-8") as f:
-            return f.read()
-    except Exception as e:
-        logger.error(f"Error reading TXT file: {e}")
-        return ""
-
-def read_zip(zip_path: str) -> str:
-    """Extracts and processes text and PDF files within a ZIP archive."""
-    extracted_data = []
-    try:
-        with zipfile.ZipFile(zip_path, 'r') as zip_ref:
-            for file_info in zip_ref.infolist():
-                if file_info.filename.endswith((".txt", ".pdf")):
-                    with zip_ref.open(file_info) as file:
-                        try:
-                            if file_info.filename.endswith(".txt"):
-                                extracted_data.append(file.read().decode("utf-8"))
-                            elif file_info.filename.endswith(".pdf"):
-                                temp_path = f"/tmp/{uuid.uuid4()}"
-                                with open(temp_path, "wb") as temp_file:
-                                    temp_file.write(file.read())
-                                extracted_data.append(read_pdf(temp_path))
-                                os.remove(temp_path)
-                        except Exception as e:
-                            logger.error(f"Error processing file in ZIP: {e}")
-        return "\n".join(extracted_data)
-    except Exception as e:
-        logger.error(f"Error extracting ZIP: {e}")
-        return ""
-
-def fetch_url(url: str, max_depth: int) -> str:
-    """Fetches and scrapes text content from a webpage."""
-    visited = set()
-    to_visit = [(url, 0)]
-    results = []
-    while to_visit:
-        current_url, depth = to_visit.pop(0)
-        if current_url in visited:
-            continue
-        visited.add(current_url)
-        if depth < max_depth:
-            try:
-                response = requests.get(current_url, timeout=10)
-                response.raise_for_status()
-                soup = BeautifulSoup(response.content, 'lxml')
-                results.append(soup.text)
-                for link in soup.find_all("a", href=True):
-                    absolute_url = requests.compat.urljoin(current_url, link.get('href'))
-                    if absolute_url.startswith("http") and absolute_url not in visited:
-                        to_visit.append((absolute_url, depth + 1))
-            except Exception as e:
-                logger.error(f"Error fetching URL {current_url}: {e}")
-    return "\n".join(results)
-
-# Main Workflow Processing
-def process_workflow(command: str, issue_details: str, files: List[Path], url: str, token: str, max_depth: int) -> Dict[str, Any]:
-    """Processes user input and performs selected command."""
-    datasets = []
-    errors = []
-    try:
-        # Add issue details to dataset
-        if issue_details:
-            datasets.append(issue_details)
-
-        # Process uploaded files
-        if files:
-            for file in files:
-                if file.name.endswith(".pdf"):
-                    datasets.append(read_pdf(file.name))
-                elif file.name.endswith(".txt"):
-                    datasets.append(read_txt(file.name))
-                elif file.name.endswith(".zip"):
-                    datasets.append(read_zip(file.name))
-
-        # Fetch URL content
-        if url:
-            datasets.append(fetch_url(url, max_depth=max_depth))
-
-        # Execute commands
-        if command == "Analyze Issue":
-            analysis = chunk_text("\n".join(datasets), 8192)
-            return {"analysis": analysis}
-        elif command == "Propose Resolution":
-            resolution = f"Proposed resolution based on:\n\n{'\n'.join(datasets)}"
-            return {"resolution": resolution}
-        elif command == "Generate PR":
-            combined_data = "\n".join(datasets)
-            return {"pr_content": combined_data, "message": "Pull request content generated."}
-        else:
-            return {"error": "Invalid command"}
-    except Exception as e:
-        errors.append(str(e))
-        return {"error": "\n".join(errors)}
-
-# Gradio Interface
-with gr.Blocks() as demo:
-    gr.Markdown("# GitHub Issue Resolver - Advanced Edition")
-    gr.Markdown("Analyze issues, propose resolutions, and generate PRs.")
-
-    # Input Fields
-    with gr.Row():
-        command = gr.Dropdown(["Analyze Issue", "Propose Resolution", "Generate PR"], label="Command")
-        issue_details = gr.Textbox(label="Issue Details", lines=4, placeholder="Describe the issue or paste details.")
-        files = gr.Files(label="Upload Files", file_types=[".pdf", ".txt", ".zip"])
-        url = gr.Textbox(label="Documentation URL", placeholder="Enter related documentation URL.")
-        token = gr.Textbox(label="GitHub Token", type="password", placeholder="Enter your GitHub token securely.")
-        max_depth = gr.Slider(label="Web Crawl Depth", minimum=1, maximum=10, value=3, step=1)
-
-    # Outputs
-    result_output = gr.JSON(label="Results")
-    process_button = gr.Button("Process")
-
-    # Button Logic
-    process_button.click(
-        process_workflow,
-        inputs=[command, issue_details, files, url, token, max_depth],
-        outputs=[result_output]
-    )
-
-# Launch Application
-demo.launch()
+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)