""" Fast script to check if top 20 active repositories (from local repos) are checked out and have latest pull. This version checks local repositories first, then verifies they're up to date. """ import sys import logging from pathlib import Path from typing import List, Dict, Tuple, Optional 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_fast.log'), logging.StreamHandler(sys.stdout) ] ) logger = logging.getLogger(__name__) def get_local_repos_sorted_by_activity(repos_dir: Path, limit: int = 20) -> List[Tuple[Path, datetime]]: """ Get local repositories sorted by most recent commit date. Args: repos_dir: Path to repositories directory limit: Maximum number of repos to return Returns: List of tuples (repo_path, last_commit_date) sorted by date (most recent first) """ repos_with_dates: List[Tuple[Path, datetime]] = [] if not repos_dir.exists(): logger.warning(f"Repositories directory does not exist: {repos_dir}") return [] logger.info(f"Scanning local repositories in {repos_dir}...") # Get all directories in repos folder for item in repos_dir.iterdir(): if not item.is_dir(): continue # Check if it's a git repository git_dir = item / ".git" if not git_dir.exists(): continue try: repo = Repo(item) # Get the latest commit date try: # Try to get the latest commit from the active branch or HEAD if repo.head.is_valid(): latest_commit = repo.head.commit commit_date = datetime.fromtimestamp(latest_commit.committed_date) repos_with_dates.append((item, commit_date)) logger.debug(f"Found {item.name}: last commit {commit_date}") else: # Invalid HEAD, use a very old date repos_with_dates.append((item, datetime.min)) except Exception as e: logger.warning(f"Could not get commit date for {item.name}: {e}") repos_with_dates.append((item, datetime.min)) except (InvalidGitRepositoryError, GitCommandError) as e: logger.debug(f"Skipping {item.name}: not a valid git repository") continue except Exception as e: logger.warning(f"Error processing {item.name}: {e}") continue # Sort by commit date descending (most recent first) repos_with_dates.sort(key=lambda x: x[1], reverse=True) # Take top N repos selected = repos_with_dates[:limit] logger.info(f"Found {len(selected)} local repositories (sorted by activity)") return selected def check_repo_status_fast(repo_path: Path, bitbucket: BitbucketClient) -> Dict: """ Check if repository is checked out and up to date. Args: repo_path: Path to repository directory bitbucket: BitbucketClient instance Returns: Dictionary with status information """ repo_slug = repo_path.name status = { "repo_slug": repo_slug, "checked_out": False, "up_to_date": False, "local_commit": None, "remote_commit": None, "branch": None, "error": None } # Check if repository is checked out if not repo_path.exists(): status["error"] = "Repository directory does not exist" return status if not (repo_path / ".git").exists(): status["error"] = "No .git directory" return status status["checked_out"] = True try: # Open repository repo = Repo(repo_path) # Get current branch try: if repo.head.is_valid(): status["branch"] = repo.active_branch.name if not repo.head.is_detached else "detached" status["local_commit"] = repo.head.commit.hexsha else: status["error"] = "Invalid HEAD" return status except Exception as e: status["error"] = f"Could not get local commit: {e}" return status # Fetch latest changes and check remote try: origin = repo.remote(name="origin") origin.fetch() # Get remote commit try: # Try to get remote commit from API first (more reliable) latest_commit = bitbucket.get_latest_commit(repo_slug) if latest_commit: status["remote_commit"] = latest_commit.get("hash") else: # Fallback to git remote if status["branch"] and status["branch"] != "detached": remote_ref = f"origin/{status['branch']}" if remote_ref in repo.refs: status["remote_commit"] = repo.refs[remote_ref].commit.hexsha except Exception as e: logger.debug(f"Could not fetch remote commit from API for {repo_slug}: {e}") # Try git remote as fallback if status["branch"] and status["branch"] != "detached": try: remote_ref = f"origin/{status['branch']}" if remote_ref in repo.refs: status["remote_commit"] = repo.refs[remote_ref].commit.hexsha except Exception: pass # Check if local is up to date if status["local_commit"] and status["remote_commit"]: status["up_to_date"] = (status["local_commit"] == status["remote_commit"]) else: status["error"] = "Could not compare commits (missing local or remote commit)" except ValueError: # Remote doesn't exist status["error"] = "Origin remote not configured" except Exception as e: status["error"] = f"Error checking remote: {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 (Fast Mode)") 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 from local repos top_repos = get_local_repos_sorted_by_activity(repos_dir, limit=20) if not top_repos: logger.warning("No local 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_path, last_commit_date) in enumerate(top_repos, 1): repo_slug = repo_path.name logger.info(f"\n[{i}/{len(top_repos)}] Checking {repo_slug}...") logger.info(f" Last commit: {last_commit_date.strftime('%Y-%m-%d %H:%M:%S')}") status = check_repo_status_fast(repo_path, 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" [OK] Checked out and up to date (branch: {status['branch']})") else: needs_update_count += 1 logger.warning(f" [WARN] Checked out but needs update (branch: {status['branch']})") if status["local_commit"] and status["remote_commit"]: logger.warning(f" Local: {status['local_commit'][:8]}") logger.warning(f" Remote: {status['remote_commit'][:8]}") else: logger.warning(f" {status.get('error', 'Unknown error')}") else: not_checked_out_count += 1 logger.error(f" [ERROR] 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" [OK] Checked out and up to date: {up_to_date_count}") logger.info(f" [WARN] Checked out but needs update: {needs_update_count}") logger.info(f" [ERROR] 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" [ERROR] {status['repo_slug']}: {status['error']}") elif not status["up_to_date"]: logger.info(f" [WARN] {status['repo_slug']}: Needs update (branch: {status.get('branch', 'N/A')})") # Return exit code based on results if not_checked_out_count > 0 or needs_update_count > 0: sys.exit(1) else: logger.info("\n[SUCCESS] 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()