Update app.py

app.py

@@ -1,215 +1,173 @@
 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 os
 import requests
+import uuid
+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 tempfile
 import shutil
 
+nltk.download('punkt')
+
 # 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)
+# 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()