app.py

import sys
import shutil
import logging
import time
import os
from datetime import datetime
from typing import List, Dict, Any
import requests
import gradio as gr
import atexit
import subprocess
from urllib.parse import urlparse, quote
import webbrowser

# Constants
INPUT_DIRECTORY = 'input'
OUTPUT_DIRECTORY = 'output'
LOGS_DIRECTORY = 'logs'
RESOLUTIONS_DIRECTORY = 'resolutions'
REPOS_DIRECTORY = 'repos'

# Logger setup
def initialize_logger() -> logging.Logger:
    """Sets up and returns a logger instance."""
    os.makedirs(LOGS_DIRECTORY, exist_ok=True)
    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("GitHubBot")

# Global logger instance
logger = initialize_logger()

# Ensure required directories exist
def initialize_environment():
    """Creates necessary directories for the script."""
    directories = [LOGS_DIRECTORY, RESOLUTIONS_DIRECTORY, REPOS_DIRECTORY, INPUT_DIRECTORY, OUTPUT_DIRECTORY]
    for directory in directories:
        os.makedirs(directory, exist_ok=True)

# GitHub API interaction
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 _check_rate_limit(self) -> bool:
        """Checks the GitHub API rate limit and waits if necessary."""
        try:
            response = requests.get(f"{self.base_url}/rate_limit", headers=self.headers)
            response.raise_for_status()
            limits = response.json()
            remaining = limits['resources']['core']['remaining']
            reset_time = limits['resources']['core']['reset']

            if remaining < 10:
                wait_time = max(0, reset_time - int(time.time()))
                if wait_time > 0:
                    logger.warning(f"Rate limit nearly exceeded. Waiting {wait_time} seconds...")
                    time.sleep(wait_time)
                return False
            return True
        except requests.RequestException as e:
            logger.error(f"Error checking rate limit: {str(e)}")
            return True

    def get_repository(self, owner: str, repo: str) -> Dict:
        """Fetches repository details."""
        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 fetching repo {owner}/{repo}: {str(e)}")
            raise
        except Exception as e:
            logger.error(f"Error fetching repo: {str(e)}")
            raise

    def get_issues(self, owner: str, repo: str, state: str = 'open') -> List[Dict]:
        """Fetches issues for a repository."""
        if not self._check_rate_limit():
            return []

        try:
            response = requests.get(
                f"{self.base_url}/repos/{owner}/{repo}/issues",
                headers=self.headers,
                params={'state': state}
            )
            response.raise_for_status()
            return [issue for issue in response.json() if 'pull_request' not in issue]
        except requests.RequestException as e:
            logger.error(f"Error fetching issues for {owner}/{repo}: {str(e)}")
            return []

# GitHub Bot main functionality
class GitHubBot:
    def __init__(self):
        self.github_api = None

    def initialize_api(self, token: str):
        """Initializes the GitHub API client."""
        self.github_api = GitHubAPI(token)

    def fetch_issues(self, token: str, owner: str, repo: str) -> List[Dict]:
        """Fetches issues from a repository."""
        self.initialize_api(token)
        return self.github_api.get_issues(owner, repo)

    def resolve_issue(self, token: str, owner: str, repo: str, issue_number: int, resolution: str, forked_repo: str) -> str:
        """Resolves a GitHub issue and pushes changes to a forked repository."""
        try:
            self.initialize_api(token)
            self.github_api.get_repository(owner, repo)

            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            resolution_file = f"{RESOLUTIONS_DIRECTORY}/resolution_{issue_number}_{timestamp}.md"

            with open(resolution_file, "w") as f:
                f.write(f"# Resolution for Issue #{issue_number}\n\n{resolution}")

            temp_repo_dir = f"/tmp/{forked_repo.split('/')[-1]}"
            subprocess.run(['git', 'clone', forked_repo, temp_repo_dir], check=True)
            os.chdir(temp_repo_dir)

            input("Apply the fix manually, then press ENTER to continue...")

            subprocess.run(['git', 'add', '.'], check=True)
            subprocess.run(['git', 'commit', '-m', f"Resolved issue #{issue_number}: {resolution}"], check=True)
            subprocess.run(['git', 'push', 'origin', 'HEAD'], check=True)

            pull_request_url = f"https://github.com/{forked_repo.split('/')[-1]}/compare/master...{owner}:{repo}_resolved_issue_{issue_number}"
            webbrowser.open(pull_request_url)

            return f"Resolution completed and saved: {resolution_file}"

        except Exception as e:
            logger.error(f"Error resolving issue #{issue_number} for {owner}/{repo}: {str(e)}")
            return f"Error resolving issue: {str(e)}"

# Utility Functions
def extract_info_from_url(url: str) -> Dict[str, Any]:
    """Extracts information from a URL."""
    info = {}
    try:
        response = requests.get(url)
        response.raise_for_status()
        info['status_code'] = response.status_code
        info['headers'] = dict(response.headers)
        info['content'] = response.text[:500]
        parsed_url = urlparse(url)

        if 'github.com' in parsed_url.netloc:
            parts = parsed_url.path.strip('/').split('/')
            if len(parts) >= 2:
                owner, repo = parts[:2]
                bot = GitHubBot()
                issues = bot.fetch_issues(github_token, owner, repo)
                info['issues'] = issues
    except requests.RequestException as e:
        info['error'] = f"Error extracting info: {str(e)}"
    return info

# Initialize environment
initialize_environment()

# Example usage (placeholder for main logic)
if __name__ == "__main__":
    print("GitHub Bot initialized. Add main logic as needed.")

    # Sample interaction for testing purposes
    token = input("Enter your GitHub Personal Access Token: ").strip()
    owner = input("Enter the repository owner: ").strip()
    repo = input("Enter the repository name: ").strip()

    bot = GitHubBot()
    issues = bot.fetch_issues(token, owner, repo)

    if issues:
        print(f"Found {len(issues)} issue(s):")
        for issue in issues:
            print(f"- #{issue['number']}: {issue['title']}")
    else:
        print("No issues found or unable to fetch issues.")