github_integration.py

#!/usr/bin/env python3
"""
GitHub Integration Helper Script
Provides functions to interact with GitHub API using a token.
"""

import os
import json
import requests
from typing import Optional, Dict, Any


class GitHubIntegration:
    """Helper class for GitHub API interactions."""
    
    def __init__(self, token: Optional[str] = None):
        """
        Initialize GitHub integration.
        
        Args:
            token: GitHub Personal Access Token. If None, will try to load from environment or file.
        """
        self.token = token or self._load_token()
        self.base_url = "https://api.github.com"
        self.headers = {
            "Accept": "application/vnd.github.v3+json",
            "Authorization": f"token {self.token}" if self.token else None
        }
        if not self.headers["Authorization"]:
            self.headers.pop("Authorization")
    
    def _load_token(self) -> Optional[str]:
        """Try to load token from environment variable or file."""
        # Try environment variable first
        token = os.getenv("GITHUB_TOKEN")
        if token:
            return token
        
        # Try to load from file
        token_files = [".github_token", "github_token.txt", ".env"]
        for token_file in token_files:
            if os.path.exists(token_file):
                try:
                    with open(token_file, "r") as f:
                        token = f.read().strip()
                        if token:
                            return token
                except Exception:
                    continue
        
        return None
    
    def verify_token(self) -> bool:
        """Verify if the GitHub token is valid."""
        if not self.token:
            return False
        
        try:
            response = requests.get(
                f"{self.base_url}/user",
                headers=self.headers,
                timeout=10
            )
            return response.status_code == 200
        except Exception:
            return False
    
    def get_user_info(self) -> Optional[Dict[str, Any]]:
        """Get authenticated user information."""
        try:
            response = requests.get(
                f"{self.base_url}/user",
                headers=self.headers,
                timeout=10
            )
            if response.status_code == 200:
                return response.json()
            return None
        except Exception as e:
            print(f"Error fetching user info: {e}")
            return None
    
    def create_repository(
        self,
        name: str,
        description: str = "",
        private: bool = False,
        auto_init: bool = True
    ) -> Optional[Dict[str, Any]]:
        """
        Create a new GitHub repository.
        
        Args:
            name: Repository name
            description: Repository description
            private: Whether repository should be private
            auto_init: Whether to initialize with README
        
        Returns:
            Repository information if successful, None otherwise
        """
        if not self.token:
            print("Error: GitHub token is required to create a repository")
            return None
        
        data = {
            "name": name,
            "description": description,
            "private": private,
            "auto_init": auto_init
        }
        
        try:
            response = requests.post(
                f"{self.base_url}/user/repos",
                headers=self.headers,
                json=data,
                timeout=10
            )
            
            if response.status_code == 201:
                return response.json()
            else:
                print(f"Error creating repository: {response.status_code}")
                print(response.text)
                return None
        except Exception as e:
            print(f"Error creating repository: {e}")
            return None
    
    def repository_exists(self, owner: str, repo: str) -> bool:
        """Check if a repository exists."""
        try:
            response = requests.get(
                f"{self.base_url}/repos/{owner}/{repo}",
                headers=self.headers,
                timeout=10
            )
            return response.status_code == 200
        except Exception:
            return False
    
    def get_repository_info(self, owner: str, repo: str) -> Optional[Dict[str, Any]]:
        """Get repository information."""
        try:
            response = requests.get(
                f"{self.base_url}/repos/{owner}/{repo}",
                headers=self.headers,
                timeout=10
            )
            if response.status_code == 200:
                return response.json()
            return None
        except Exception as e:
            print(f"Error fetching repository info: {e}")
            return None


def main():
    """Example usage of GitHubIntegration."""
    print("GitHub Integration Helper")
    print("=" * 60)
    print()
    
    # Try to get token from user
    token = os.getenv("GITHUB_TOKEN")
    if not token:
        token_file = ".github_token"
        if os.path.exists(token_file):
            with open(token_file, "r") as f:
                token = f.read().strip()
        else:
            token = input("Enter your GitHub Personal Access Token: ").strip()
    
    if not token:
        print("Error: Token is required")
        return
    
    # Initialize GitHub integration
    gh = GitHubIntegration(token)
    
    # Verify token
    print("Verifying token...")
    if not gh.verify_token():
        print("✗ Invalid token")
        return
    
    print("✓ Token verified")
    
    # Get user info
    user_info = gh.get_user_info()
    if user_info:
        print(f"\nAuthenticated as: {user_info.get('login')} ({user_info.get('name', 'N/A')})")
        print(f"Profile: {user_info.get('html_url')}")
    
    # Check if repository exists
    repo_name = "PRO_SOW_GPT"
    username = user_info.get('login') if user_info else "surajdeval"
    
    print(f"\nChecking if repository '{repo_name}' exists...")
    if gh.repository_exists(username, repo_name):
        print(f"✓ Repository '{username}/{repo_name}' exists")
        repo_info = gh.get_repository_info(username, repo_name)
        if repo_info:
            print(f"  URL: {repo_info.get('html_url')}")
            print(f"  Description: {repo_info.get('description', 'N/A')}")
    else:
        print(f"✗ Repository '{username}/{repo_name}' does not exist")
        create = input(f"\nWould you like to create it? (y/n): ").strip().lower()
        if create == 'y':
            description = "PRO SOW GPT Workflow - Automated Statement of Work generation"
            repo_info = gh.create_repository(
                name=repo_name,
                description=description,
                private=False,
                auto_init=True
            )
            if repo_info:
                print(f"✓ Repository created: {repo_info.get('html_url')}")
            else:
                print("✗ Failed to create repository")


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("\n\nOperation cancelled")
    except Exception as e:
        print(f"\nError: {e}")