""" Script to check if top 20 active repositories are checked out and have latest pull. """ import sys import logging from pathlib import Path from typing import List, Dict, Tuple from datetime import datetime from git import Repo, GitCommandError from git.exc import InvalidGitRepositoryError # Import modules try: from config_manager import ConfigManager from bitbucket_client import BitbucketClient from git_manager import GitManager except ImportError: from .config_manager import ConfigManager from .bitbucket_client import BitbucketClient from .git_manager import GitManager # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('top_repos_check.log'), logging.StreamHandler(sys.stdout) ] ) logger = logging.getLogger(__name__) def get_top_active_repos(bitbucket: BitbucketClient, config: ConfigManager, limit: int = 20) -> List[Dict]: """ Get top N repositories sorted by most recent commit date. Args: bitbucket: BitbucketClient instance config: ConfigManager instance limit: Maximum number of repos to return Returns: List of top N repositories sorted by commit date (most recent first) """ logger.info(f"Fetching all repositories from Bitbucket...") all_repos = bitbucket.get_all_repositories() or [] if not all_repos: logger.warning("No repositories found in workspace") return [] logger.info(f"Found {len(all_repos)} repositories, fetching commit dates for top {limit}...") repos_with_dates: List[Tuple[datetime, Dict]] = [] for repo_info in all_repos: repo_slug = repo_info.get("name") or repo_info.get("slug") if not repo_slug: continue # Only process repos from the configured workspace workspace_slug = repo_info.get("workspace", {}).get("slug") if workspace_slug != config.get("bitbucket.workspace"): continue try: # Fetch latest commit for this repository latest_commit = bitbucket.get_latest_commit(repo_slug) if latest_commit and latest_commit.get("date"): try: # Parse ISO 8601 date from Bitbucket commit_date_str = latest_commit.get("date") commit_date = datetime.fromisoformat(commit_date_str.replace("Z", "+00:00")) # Convert to UTC if timezone-aware if commit_date.tzinfo: commit_date = commit_date.replace(tzinfo=None) repos_with_dates.append((commit_date, repo_info)) logger.debug(f"Found commit date for {repo_slug}: {commit_date}") except Exception as e: logger.warning(f"Failed to parse commit date for {repo_slug}: {e}") # Use a very old date as fallback repos_with_dates.append((datetime.min, repo_info)) else: logger.warning(f"Could not fetch commit date for {repo_slug}, using fallback") repos_with_dates.append((datetime.min, repo_info)) except Exception as e: logger.warning(f"Error fetching commit date for {repo_slug}: {e}") repos_with_dates.append((datetime.min, repo_info)) # Sort by commit date descending (most recent first) repos_with_dates.sort(key=lambda x: x[0], reverse=True) # Take top N repos selected_repos = [repo_info for _, repo_info in repos_with_dates[:limit]] logger.info(f"Selected {len(selected_repos)} repositories with most recent commits") return selected_repos def check_repo_status(repo_info: Dict, repos_dir: Path, bitbucket: BitbucketClient) -> Dict: """ Check if repository is checked out and up to date. Args: repo_info: Repository information dictionary repos_dir: Path to repositories directory bitbucket: BitbucketClient instance Returns: Dictionary with status information """ repo_slug = repo_info.get("name") or repo_info.get("slug") repo_path = repos_dir / repo_slug status = { "repo_slug": repo_slug, "checked_out": False, "up_to_date": False, "local_commit": None, "remote_commit": None, "error": None } # Check if repository is checked out if not repo_path.exists(): status["error"] = "Repository not checked out (directory does not exist)" return status if not (repo_path / ".git").exists(): status["error"] = "Repository not checked out (no .git directory)" return status status["checked_out"] = True try: # Open repository repo = Repo(repo_path) # Get remote commit try: latest_commit = bitbucket.get_latest_commit(repo_slug) if latest_commit: status["remote_commit"] = latest_commit.get("hash") except Exception as e: logger.warning(f"Could not fetch remote commit for {repo_slug}: {e}") status["error"] = f"Could not fetch remote commit: {e}" return status # Get local commit try: # Fetch latest changes first origin = repo.remote(name="origin") origin.fetch() # Get current branch try: current_branch = repo.active_branch.name # Get local commit hash status["local_commit"] = repo.head.commit.hexsha # Check if local is up to date with remote remote_ref = f"origin/{current_branch}" if remote_ref in repo.refs: remote_commit = repo.refs[remote_ref].commit.hexsha status["up_to_date"] = (status["local_commit"] == remote_commit) else: # Compare with remote commit from API status["up_to_date"] = (status["local_commit"] == status["remote_commit"]) except Exception as e: logger.warning(f"Could not get local commit for {repo_slug}: {e}") status["error"] = f"Could not get local commit: {e}" except Exception as e: logger.warning(f"Error checking local status for {repo_slug}: {e}") status["error"] = f"Error checking local status: {e}" except InvalidGitRepositoryError: status["error"] = "Invalid git repository" except Exception as e: status["error"] = f"Unexpected error: {e}" return status def main(): """Main function to check top 20 active repositories.""" logger.info("=" * 60) logger.info("Checking Top 20 Active Repositories") logger.info("=" * 60) try: # Load configuration config = ConfigManager() # Validate required configuration if not config.get("bitbucket.workspace"): logger.error("Bitbucket workspace not configured") sys.exit(1) has_api_token = bool(config.get("bitbucket.api_token")) has_app_password = bool(config.get("bitbucket.app_password")) if not config.get("bitbucket.username"): logger.error("Bitbucket username/email not configured") sys.exit(1) if not has_api_token and not has_app_password: logger.error("Bitbucket authentication not configured") sys.exit(1) # Initialize components bitbucket = BitbucketClient( workspace=config.get("bitbucket.workspace"), username=config.get("bitbucket.username"), app_password=config.get("bitbucket.app_password") if not has_api_token else None, api_token=config.get("bitbucket.api_token") ) repos_dir = Path(config.get("scanning.repos_dir")) # Get top 20 active repositories top_repos = get_top_active_repos(bitbucket, config, limit=20) if not top_repos: logger.warning("No repositories found") return logger.info("=" * 60) logger.info(f"Checking status of {len(top_repos)} repositories...") logger.info("=" * 60) # Check each repository results = [] checked_out_count = 0 up_to_date_count = 0 needs_update_count = 0 not_checked_out_count = 0 for i, repo_info in enumerate(top_repos, 1): repo_slug = repo_info.get("name") or repo_info.get("slug") logger.info(f"\n[{i}/{len(top_repos)}] Checking {repo_slug}...") status = check_repo_status(repo_info, repos_dir, bitbucket) results.append(status) if status["checked_out"]: checked_out_count += 1 if status["up_to_date"]: up_to_date_count += 1 logger.info(f" ✓ Checked out and up to date") else: needs_update_count += 1 logger.warning(f" ⚠ Checked out but needs update") logger.warning(f" Local: {status['local_commit'][:8] if status['local_commit'] else 'N/A'}") logger.warning(f" Remote: {status['remote_commit'][:8] if status['remote_commit'] else 'N/A'}") else: not_checked_out_count += 1 logger.error(f" ✗ Not checked out: {status['error']}") # Print summary logger.info("\n" + "=" * 60) logger.info("Summary") logger.info("=" * 60) logger.info(f"Total repositories checked: {len(top_repos)}") logger.info(f" ✓ Checked out and up to date: {up_to_date_count}") logger.info(f" ⚠ Checked out but needs update: {needs_update_count}") logger.info(f" ✗ Not checked out: {not_checked_out_count}") logger.info("=" * 60) # List repositories that need attention if needs_update_count > 0 or not_checked_out_count > 0: logger.info("\nRepositories needing attention:") for status in results: if not status["checked_out"]: logger.info(f" ✗ {status['repo_slug']}: {status['error']}") elif not status["up_to_date"]: logger.info(f" ⚠ {status['repo_slug']}: Needs update") # Return exit code based on results if not_checked_out_count > 0 or needs_update_count > 0: sys.exit(1) else: logger.info("\n✓ All top 20 repositories are checked out and up to date!") sys.exit(0) except Exception as e: logger.error(f"Unexpected error: {e}", exc_info=True) sys.exit(1) if __name__ == "__main__": main()