github_api.py

import requests
from datetime import datetime, timedelta
import pandas as pd
from typing import List, Dict, Optional
import time
from functools import lru_cache
import json

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 _make_request(self, endpoint: str, params: Optional[Dict] = None) -> Dict:
        """Make a request to GitHub API with rate limit handling."""
        url = f"{self.base_url}/{endpoint}"
        response = requests.get(url, headers=self.headers, params=params)
        
        if response.status_code == 403:
            # Check if rate limited
            if 'X-RateLimit-Remaining' in response.headers:
                if int(response.headers['X-RateLimit-Remaining']) == 0:
                    reset_time = int(response.headers['X-RateLimit-Reset'])
                    sleep_time = reset_time - int(time.time()) + 1
                    if sleep_time > 0:
                        time.sleep(min(sleep_time, 60))  # Wait max 60 seconds
                        return self._make_request(endpoint, params)
        
        response.raise_for_status()
        return response.json()
    
    @lru_cache(maxsize=128)
    def get_user_repos(self, username: str) -> List[Dict]:
        """Fetch all repositories for a user or organization."""
        repos = []
        page = 1
        
        while True:
            try:
                # Try user endpoint first
                endpoint = f"users/{username}/repos"
                params = {'page': page, 'per_page': 100, 'type': 'all'}
                page_repos = self._make_request(endpoint, params)
            except requests.HTTPError:
                # If user endpoint fails, try org endpoint
                try:
                    endpoint = f"orgs/{username}/repos"
                    params = {'page': page, 'per_page': 100, 'type': 'all'}
                    page_repos = self._make_request(endpoint, params)
                except requests.HTTPError:
                    break
            
            if not page_repos:
                break
                
            repos.extend(page_repos)
            page += 1
            
            if len(page_repos) < 100:
                break
                
        return repos
    
    def get_repo_details(self, owner: str, repo: str) -> Dict:
        """Get detailed information about a repository."""
        endpoint = f"repos/{owner}/{repo}"
        return self._make_request(endpoint)
    
    def get_repo_contributors(self, owner: str, repo: str, limit: int = 10) -> List[Dict]:
        """Get top contributors for a repository."""
        endpoint = f"repos/{owner}/{repo}/contributors"
        params = {'per_page': limit}
        try:
            return self._make_request(endpoint, params)
        except:
            return []
    
    def get_repo_issues(self, owner: str, repo: str, state: str = 'all') -> int:
        """Get issue count for a repository."""
        endpoint = f"repos/{owner}/{repo}/issues"
        params = {'state': state, 'per_page': 1}
        try:
            response = requests.get(
                f"{self.base_url}/{endpoint}", 
                headers=self.headers, 
                params=params
            )
            # Get total count from Link header
            if 'Link' in response.headers:
                links = response.headers['Link'].split(',')
                for link in links:
                    if 'rel="last"' in link:
                        last_page = link.split('page=')[1].split('>')[0]
                        return int(last_page)
            return len(response.json())
        except:
            return 0
    
    def get_repo_pulls(self, owner: str, repo: str, state: str = 'all') -> int:
        """Get pull request count for a repository."""
        endpoint = f"repos/{owner}/{repo}/pulls"
        params = {'state': state, 'per_page': 1}
        try:
            response = requests.get(
                f"{self.base_url}/{endpoint}", 
                headers=self.headers, 
                params=params
            )
            if 'Link' in response.headers:
                links = response.headers['Link'].split(',')
                for link in links:
                    if 'rel="last"' in link:
                        last_page = link.split('page=')[1].split('>')[0]
                        return int(last_page)
            return len(response.json())
        except:
            return 0
    
    def get_repo_languages(self, owner: str, repo: str) -> Dict[str, int]:
        """Get language breakdown for a repository."""
        endpoint = f"repos/{owner}/{repo}/languages"
        try:
            return self._make_request(endpoint)
        except:
            return {}
    
    def get_stargazers_over_time(self, owner: str, repo: str, limit: int = 100) -> pd.DataFrame:
        """Get star history for a repository."""
        endpoint = f"repos/{owner}/{repo}/stargazers"
        headers = self.headers.copy()
        headers['Accept'] = 'application/vnd.github.v3.star+json'
        
        stargazers = []
        page = 1
        
        while len(stargazers) < limit:
            params = {'page': page, 'per_page': min(100, limit - len(stargazers))}
            try:
                response = requests.get(
                    f"{self.base_url}/{endpoint}",
                    headers=headers,
                    params=params
                )
                response.raise_for_status()
                page_data = response.json()
                
                if not page_data:
                    break
                    
                stargazers.extend(page_data)
                page += 1
                
                if len(page_data) < params['per_page']:
                    break
            except:
                break
        
        if stargazers:
            df = pd.DataFrame([
                {'date': sg['starred_at'], 'stars': i + 1}
                for i, sg in enumerate(stargazers)
            ])
            df['date'] = pd.to_datetime(df['date'])
            return df
        
        return pd.DataFrame(columns=['date', 'stars'])

    def get_user_profile(self, username: str) -> Dict:
        """Fetch profile details for a user or organization."""
        try:
            return self._make_request(f"users/{username}")
        except requests.HTTPError:
            try:
                return self._make_request(f"orgs/{username}")
            except Exception:
                return {}

    def get_rate_limit(self) -> Dict:
        """Fetch current rate limit status."""
        try:
            # We don't want rate_limit call itself to sleep if rate limited, 
            # so make a direct request without rate limit sleep handling
            url = f"{self.base_url}/rate_limit"
            response = requests.get(url, headers=self.headers)
            if response.status_code == 200:
                return response.json()
        except Exception:
            pass
        return {
            "resources": {
                "core": {
                    "limit": 60,
                    "remaining": 60,
                    "reset": int(time.time()) + 3600
                }
            }
        }

    def get_recent_commits(self, owner: str, repo: str, limit: int = 100) -> List[Dict]:
        """Fetch recent commits for a repository."""
        endpoint = f"repos/{owner}/{repo}/commits"
        params = {'per_page': limit}
        try:
            return self._make_request(endpoint, params)
        except Exception:
            return []