""" Script to verify all Bitbucket repositories are present and up-to-date. Compares Bitbucket repos with local repos and updates/clones as needed. """ import sys import logging from pathlib import Path from typing import Dict, List, Set # 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('repo_verification.log'), logging.StreamHandler(sys.stdout) ] ) logger = logging.getLogger(__name__) def get_local_repos(repos_dir: Path) -> Set[str]: """ Get list of all local repository directories. Args: repos_dir: Path to repositories directory Returns: Set of repository names/slugs """ local_repos = set() if not repos_dir.exists(): logger.warning(f"Repositories directory does not exist: {repos_dir}") return local_repos # Get all directories in repos folder for item in repos_dir.iterdir(): if item.is_dir(): # Check if it's a git repository if (item / ".git").exists(): local_repos.add(item.name) else: logger.debug(f"Directory {item.name} exists but is not a git repository") return local_repos def get_bitbucket_repos(bitbucket: BitbucketClient, config: ConfigManager) -> List[Dict]: """ Get all repositories from Bitbucket (including additional repos). Args: bitbucket: Bitbucket client instance config: Configuration manager Returns: List of repository info dictionaries """ # Get all repositories from workspace repos = bitbucket.get_all_repositories() or [] # Get additional manually configured repositories additional_repos_config = config.get("scanning.additional_repositories", []) if additional_repos_config: logger.info(f"Found {len(additional_repos_config)} additional repositories in config") for repo_config in additional_repos_config: if isinstance(repo_config, str): # Parse SSH URL or workspace/repo format if repo_config.startswith("git@bitbucket.org:"): path = repo_config.replace("git@bitbucket.org:", "").replace(".git", "") parts = path.split("/") if len(parts) == 2: workspace, repo_name = parts repos.append({ "name": repo_name, "slug": repo_name, "workspace": {"slug": workspace}, "links": { "clone": [ {"name": "ssh", "href": repo_config} ] } }) elif repo_config.startswith("https://github.com/"): path = repo_config.replace("https://github.com/", "").replace(".git", "") parts = path.split("/") if len(parts) == 2: org, repo_name = parts repos.append({ "name": repo_name, "slug": repo_name, "workspace": {"slug": org}, "links": { "clone": [ {"name": "ssh", "href": repo_config}, {"name": "https", "href": repo_config}, ] } }) elif "/" in repo_config: parts = repo_config.split("/") if len(parts) == 2: workspace, repo_name = parts ssh_url = f"git@bitbucket.org:{workspace}/{repo_name}.git" repos.append({ "name": repo_name, "slug": repo_name, "workspace": {"slug": workspace}, "links": { "clone": [ {"name": "ssh", "href": ssh_url} ] } }) elif isinstance(repo_config, dict): repo_name = repo_config.get("name") or repo_config.get("slug") workspace = repo_config.get("workspace") ssh_url = repo_config.get("ssh_url") or repo_config.get("clone_url") if repo_name and ssh_url: if isinstance(workspace, str): workspace = {"slug": workspace} repos.append({ "name": repo_name, "slug": repo_name, "workspace": workspace or {}, "links": { "clone": [ {"name": "ssh", "href": ssh_url} ] } }) return repos def verify_and_update_repos(): """Main function to verify and update all repositories.""" logger.info("=" * 60) logger.info("Starting Repository Verification and Update") 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") ) git_manager = GitManager( repos_dir=config.get("scanning.repos_dir") ) repos_dir = Path(config.get("scanning.repos_dir")) # Get all Bitbucket repositories logger.info("Fetching repositories from Bitbucket...") bitbucket_repos = get_bitbucket_repos(bitbucket, config) if not bitbucket_repos: logger.warning("No repositories found in Bitbucket") return logger.info(f"Found {len(bitbucket_repos)} repositories in Bitbucket") # Get local repositories logger.info("Scanning local repositories...") local_repos = get_local_repos(repos_dir) logger.info(f"Found {len(local_repos)} local repositories") # Create mapping of repo names to repo info bitbucket_repo_map = {} for repo_info in bitbucket_repos: repo_slug = repo_info.get("name") or repo_info.get("slug") if repo_slug: # Fix repo names that are full URLs - extract just the repo name if repo_slug.startswith("http://") or repo_slug.startswith("https://"): # Extract repo name from URL: https://bitbucket.org/workspace/repo -> repo parts = repo_slug.rstrip("/").split("/") if len(parts) >= 2: repo_slug = parts[-1] # Also try to get from slug field if name was a URL if not repo_slug or repo_slug.startswith("http"): repo_slug = repo_info.get("slug") or repo_slug bitbucket_repo_map[repo_slug] = repo_info # Find missing and existing repos bitbucket_repo_names = set(bitbucket_repo_map.keys()) missing_repos = bitbucket_repo_names - local_repos existing_repos = bitbucket_repo_names & local_repos extra_local_repos = local_repos - bitbucket_repo_names # Report findings logger.info("=" * 60) logger.info("Repository Status Summary:") logger.info(f" Total Bitbucket repos: {len(bitbucket_repo_names)}") logger.info(f" Total local repos: {len(local_repos)}") logger.info(f" Missing repos (need clone): {len(missing_repos)}") logger.info(f" Existing repos (need update): {len(existing_repos)}") if extra_local_repos: logger.info(f" Extra local repos (not in Bitbucket): {len(extra_local_repos)}") logger.info("=" * 60) # List missing repos if missing_repos: logger.info("\nMissing Repositories (will be cloned):") for repo_name in sorted(missing_repos): logger.info(f" - {repo_name}") # List extra local repos if extra_local_repos: logger.info("\nExtra Local Repositories (not in Bitbucket):") for repo_name in sorted(extra_local_repos): logger.info(f" - {repo_name}") # Clone missing repositories cloned_count = 0 failed_clone_count = 0 if missing_repos: logger.info(f"\n{'='*60}") logger.info(f"Cloning {len(missing_repos)} missing repositories...") logger.info(f"{'='*60}") for repo_slug in sorted(missing_repos): repo_info = bitbucket_repo_map[repo_slug] logger.info(f"\nCloning: {repo_slug}") try: clone_url = bitbucket.get_clone_url(repo_info, protocol="ssh") if not clone_url: logger.warning(f"No SSH clone URL found for {repo_slug}, skipping") failed_clone_count += 1 continue clone_result = git_manager.clone_or_update( repo_slug=repo_slug, clone_url=clone_url ) if clone_result and (isinstance(clone_result, tuple) and clone_result[0]): cloned_count += 1 logger.info(f"[OK] Successfully cloned {repo_slug}") else: failed_clone_count += 1 logger.error(f"[FAIL] Failed to clone {repo_slug}") except Exception as e: failed_clone_count += 1 logger.error(f"✗ Error cloning {repo_slug}: {e}", exc_info=True) # Update existing repositories updated_count = 0 failed_update_count = 0 if existing_repos: logger.info(f"\n{'='*60}") logger.info(f"Updating {len(existing_repos)} existing repositories...") logger.info(f"{'='*60}") for repo_slug in sorted(existing_repos): repo_info = bitbucket_repo_map[repo_slug] logger.info(f"\nUpdating: {repo_slug}") try: clone_url = bitbucket.get_clone_url(repo_info, protocol="ssh") if not clone_url: logger.warning(f"No SSH clone URL found for {repo_slug}, skipping") failed_update_count += 1 continue clone_result = git_manager.clone_or_update( repo_slug=repo_slug, clone_url=clone_url ) if clone_result and (isinstance(clone_result, tuple) and clone_result[0]): updated_count += 1 logger.info(f"[OK] Successfully updated {repo_slug}") else: failed_update_count += 1 logger.error(f"[FAIL] Failed to update {repo_slug}") except Exception as e: failed_update_count += 1 logger.error(f"✗ Error updating {repo_slug}: {e}", exc_info=True) # Final summary logger.info("\n" + "=" * 60) logger.info("Verification Summary:") logger.info(f" Total Bitbucket repos: {len(bitbucket_repo_names)}") logger.info(f" Cloned: {cloned_count}") logger.info(f" Updated: {updated_count}") logger.info(f" Failed to clone: {failed_clone_count}") logger.info(f" Failed to update: {failed_update_count}") logger.info("=" * 60) if cloned_count + updated_count == len(bitbucket_repo_names): logger.info("[SUCCESS] All repositories are present and up-to-date!") else: logger.warning("[WARNING] Some repositories could not be cloned/updated. Check logs for details.") except KeyboardInterrupt: logger.info("\nVerification interrupted by user") sys.exit(1) except Exception as e: logger.error(f"Fatal error: {e}", exc_info=True) sys.exit(1) if __name__ == "__main__": verify_and_update_repos()