github_client.py

import os
import tempfile
import shutil
from github import Github, GithubException
from git import Repo
from typing import Optional, List, Dict
import time


class GitHubClient:
    def __init__(self, token: Optional[str] = None):
        """Initialize GitHub client."""
        self.token = token or os.environ.get("GITHUB_TOKEN")
        if not self.token:
            raise ValueError("GitHub token is required. Set GITHUB_TOKEN environment variable.")
        self.github = Github(self.token)
        self.user = self.github.get_user()
        
    def clone_repo(self, repo_url: str, branch: str = "main") -> tuple[str, Repo]:
        """Clone a repository to a temporary directory using environment-based authentication."""
        temp_dir = tempfile.mkdtemp(prefix="codesentinel_")
        try:
            # Set up Git credentials using environment variable (no token in URL)
            if self.token:
                git_credentials_path = os.path.expanduser('~/.git-credentials')
                credentials_line = f"https://{self.token}:@github.com\n"
                
                # Ensure credentials directory exists
                os.makedirs(os.path.dirname(git_credentials_path), exist_ok=True)
                
                # Write credentials if not already present
                existing_content = ''
                if os.path.exists(git_credentials_path):
                    with open(git_credentials_path, 'r') as rf:
                        existing_content = rf.read()
                
                if credentials_line not in existing_content:
                    with open(git_credentials_path, 'a') as f:
                        f.write(credentials_line)
                    os.chmod(git_credentials_path, 0o600)
            
            # Ensure URL is clean (no embedded token)
            clean_url = repo_url
            if repo_url.startswith("https://github.com/"):
                # URL is already clean
                pass
            elif '@github.com' in repo_url:
                # Extract clean URL from token-embedded URL
                clean_url = 'https://github.com' + repo_url.split('github.com')[1]
            
            # Clone using clean URL - Git will use credential helper
            repo = Repo.clone_from(
                clean_url, 
                temp_dir, 
                branch=branch,
                env={'GIT_TERMINAL_PROMPT': '0'}  # Disable prompts
            )
            
            # Configure credential helper for this repo
            repo.git.config('credential.helper', 'store')
            
            return temp_dir, repo
        except Exception as e:
            shutil.rmtree(temp_dir, ignore_errors=True)
            raise Exception(f"Failed to clone repository: {e}")
    
    def fork_repository(self, repo_full_name: str) -> Dict:
        """
        Fork a repository to the authenticated user's account.
        Returns dict with original repo info and fork URL.
        """
        try:
            # Get the original repository
            original_repo = self.github.get_repo(repo_full_name)
            
            # Check if fork already exists
            try:
                existing_fork = self.user.get_repo(original_repo.name)
                if existing_fork.fork and existing_fork.parent.full_name == repo_full_name:
                    return {
                        'original_repo': repo_full_name,
                        'fork_repo': existing_fork.full_name,
                        'fork_url': existing_fork.clone_url,
                        'default_branch': existing_fork.default_branch,
                        'already_existed': True
                    }
            except GithubException:
                # Fork doesn't exist, create it
                pass
            
            # Create the fork using the repository's create_fork method
            fork = original_repo.create_fork()
            
            # Wait for fork to be ready
            max_retries = 10
            for i in range(max_retries):
                try:
                    fork = self.github.get_repo(fork.full_name)
                    if fork.size > 0:  # Fork is ready
                        break
                except GithubException:
                    pass
                time.sleep(2)
            
            return {
                'original_repo': repo_full_name,
                'fork_repo': fork.full_name,
                'fork_url': fork.clone_url,
                'default_branch': fork.default_branch,
                'already_existed': False
            }
            
        except GithubException as e:
            raise Exception(f"Failed to fork repository: {e}")
    
    def get_repo_structure(self, repo_path: str, max_depth: int = 3) -> str:
        """Get repository structure as a string."""
        structure = []
        for root, dirs, files in os.walk(repo_path):
            level = root.replace(repo_path, '').count(os.sep)
            if level > max_depth:
                continue
            indent = ' ' * 2 * level
            structure.append(f'{indent}{os.path.basename(root)}/')
            sub_indent = ' ' * 2 * (level + 1)
            for file in files:
                if not file.startswith('.'):
                    structure.append(f'{sub_indent}{file}')
        return '\n'.join(structure[:100])  # Limit to 100 lines
    
    def read_file(self, repo_path: str, file_path: str) -> str:
        """Read a file from the repository."""
        full_path = os.path.join(repo_path, file_path)
        if not os.path.exists(full_path):
            raise FileNotFoundError(f"File not found: {file_path}")
        
        with open(full_path, 'r', encoding='utf-8', errors='ignore') as f:
            return f.read()
    
    def write_file(self, repo_path: str, file_path: str, content: str):
        """Write content to a file in the repository."""
        full_path = os.path.join(repo_path, file_path)
        os.makedirs(os.path.dirname(full_path), exist_ok=True)
        
        with open(full_path, 'w', encoding='utf-8') as f:
            f.write(content)
    
    def create_branch(self, repo: Repo, branch_name: str):
        """Create a new branch in the repository."""
        try:
            repo.git.checkout('-b', branch_name)
        except Exception as e:
            raise Exception(f"Failed to create branch: {e}")
    
    def commit_changes(self, repo: Repo, message: str):
        """Commit all changes in the repository."""
        try:
            repo.git.add(A=True)
            repo.index.commit(message)
        except Exception as e:
            raise Exception(f"Failed to commit changes: {e}")
    
    def push_branch(self, repo: Repo, branch_name: str):
        """Push branch to remote using environment-based authentication."""
        try:
            # Configure Git credentials using environment variable
            # This avoids embedding tokens in URLs which Replit rejects
            if self.token:
                # Set up credential helper with token from environment
                repo.git.config('credential.helper', 'store')
                
                # Create credentials file with proper format
                git_credentials_path = os.path.expanduser('~/.git-credentials')
                credentials_line = f"https://{self.token}:@github.com\n"
                
                # Write credentials (append mode to preserve existing)
                with open(git_credentials_path, 'a') as f:
                    # Check if this credential already exists
                    if os.path.exists(git_credentials_path):
                        with open(git_credentials_path, 'r') as rf:
                            if credentials_line not in rf.read():
                                f.write(credentials_line)
                    else:
                        f.write(credentials_line)
                
                # Ensure proper permissions on credentials file
                os.chmod(git_credentials_path, 0o600)
            
            # Get remote and ensure URL is clean (no embedded token)
            origin = repo.remote(name='origin')
            origin_url = list(origin.urls)[0]
            
            # Clean the URL if it has an embedded token
            if '@github.com' in origin_url and 'https://' in origin_url:
                # Extract the repo path after github.com
                clean_url = 'https://github.com' + origin_url.split('github.com')[1]
                origin.set_url(clean_url)
            
            # Push using clean URL with credential helper
            # GitPython handles ASCII hyphens correctly
            origin.push(branch_name, set_upstream=True)
            
        except Exception as e:
            raise Exception(f"Failed to push branch: {e}")
    
    def create_pull_request(self, repo_full_name: str, title: str, body: str, 
                          head_branch: str, base_branch: str = "main") -> str:
        """Create a pull request."""
        try:
            repo = self.github.get_repo(repo_full_name)
            pr = repo.create_pull(
                title=title,
                body=body,
                head=head_branch,
                base=base_branch
            )
            return pr.html_url
        except GithubException as e:
            raise Exception(f"Failed to create PR: {e}")
    
    def get_open_pull_requests(self, repo_full_name: str) -> List[Dict]:
        """Get all open pull requests for a repository."""
        try:
            repo = self.github.get_repo(repo_full_name)
            pulls = repo.get_pulls(state='open', sort='created', direction='desc')
            
            pr_list = []
            for pr in pulls:
                pr_data = {
                    'number': pr.number,
                    'title': pr.title,
                    'description': pr.body or '',
                    'author': pr.user.login if pr.user else 'unknown',
                    'created_at': pr.created_at.isoformat() if pr.created_at else '',
                    'updated_at': pr.updated_at.isoformat() if pr.updated_at else '',
                    'files_changed': pr.changed_files,
                    'additions': pr.additions,
                    'deletions': pr.deletions,
                    'url': pr.html_url,
                    'state': pr.state,
                    'mergeable': pr.mergeable,
                    'labels': [label.name for label in pr.labels]
                }
                pr_list.append(pr_data)
            
            return pr_list
        except GithubException as e:
            raise Exception(f"Failed to fetch PRs: {e}")
    
    def get_pr_diff(self, repo_full_name: str, pr_number: int) -> str:
        """Get the diff for a pull request."""
        try:
            repo = self.github.get_repo(repo_full_name)
            pr = repo.get_pull(pr_number)
            
            # Get files changed
            files = pr.get_files()
            diff_content = []
            
            for file in files:
                diff_content.append(f"File: {file.filename}")
                diff_content.append(f"Status: {file.status}")
                diff_content.append(f"Changes: +{file.additions} -{file.deletions}")
                if file.patch:
                    diff_content.append(file.patch)
                diff_content.append("-" * 80)
            
            return "\n".join(diff_content)
        except GithubException as e:
            return f"Error fetching diff: {e}"
    
    def parse_repo_url(self, repo_url: str) -> str:
        """Parse repository URL to get owner/repo format."""
        # Handle https://github.com/owner/repo or github.com/owner/repo
        if "github.com/" in repo_url:
            parts = repo_url.split("github.com/")[1]
            # Remove .git if present
            parts = parts.replace(".git", "")
            # Remove trailing slash
            parts = parts.rstrip("/")
            return parts
        return repo_url
    
    def cleanup_temp_dir(self, temp_dir: str):
        """Clean up temporary directory."""
        try:
            shutil.rmtree(temp_dir, ignore_errors=True)
        except Exception as e:
            print(f"Warning: Failed to cleanup temp directory: {e}")