gitlab-migration.py

#!/usr/bin/env python3
import gitlab
import git
import os
import shutil
import time
import logging
import sys
from datetime import datetime

# Configure logging
log_file = f'gitlab_migration_{datetime.now().strftime("%Y%m%d_%H%M%S")}.log'
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler(log_file),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger("gitlab-migration")

# GitLab connection settings
SOURCE_GITLAB_URL = "https://git.source.com"
SOURCE_GITLAB_TOKEN = os.environ.get("SOURCE_GITLAB_TOKEN", "")

TARGET_GITLAB_URL = "https://gitlab.target.com"
TARGET_GITLAB_TOKEN = os.environ.get("TARGET_GITLAB_TOKEN", "")

# Temporary directory for cloning
TEMP_DIR = "/tmp/gitlab_migration"

# Structure mapping
STRUCTURE_MAPPING = {
    "source-repo-group": "reponew/target-repo-group",


    # Add more mappings as needed
}

# Repository name mapping
REPO_NAME_MAPPING = {
    "source-repo": "target-repo",

    
    # Add more mappings as needed
}


# Whitelist repositories - only these will be migrated when whitelist mode is enabled
WHITELIST_REPOS = [
    "allow-this-repo",
    "allow-this-repo2",

    # Add more repositories to whitelist as needed
]

# Global flags
DRY_RUN = False
SKIP_CONFIRMATIONS = False
WHITELIST_MODE = False

def get_user_confirmation(message):
    """Get user confirmation before proceeding."""
    if SKIP_CONFIRMATIONS:
        return True
    
    response = input(f"\n{message} (y/n): ").strip().lower()
    return response == 'y' or response == 'yes'

def connect_gitlab(url, token):
    """Connect to GitLab instance."""
    logger.info(f"Attempting to connect to GitLab at {url}...")
    
    try:
        gl = gitlab.Gitlab(url=url, private_token=token)
        gl.auth()
        
        # Try different methods to get current user (compatibility with different versions)
        try:
            # Method 1: Newer versions
            user = gl.users.current()
            username = user.username
        except (AttributeError, gitlab.exceptions.GitlabError):
            try:
                # Method 2: Alternative in some versions
                user = gl.users.get('current', lazy=True)
                username = user.username
            except (AttributeError, gitlab.exceptions.GitlabError):
                try:
                    # Method 3: Older versions
                    user = gl.user
                    username = user.username if hasattr(user, 'username') else 'unknown'
                except (AttributeError, gitlab.exceptions.GitlabError):
                    # Method 4: Just verify we're authenticated
                    username = "authenticated_user"
        
        logger.info(f"✅ Successfully connected to {url} as {username}")
        return gl
    except Exception as e:
        logger.error(f"❌ Failed to connect to GitLab at {url}: {e}")
        raise

def get_groups_and_projects(gl, parent_path=None):
    """Recursively get all groups and their projects."""
    result = {}
    
    if parent_path:
        try:
            parent_group = gl.groups.get(parent_path)
            groups = parent_group.subgroups.list(all=True)
            logger.info(f"Found {len(groups)} subgroups in {parent_path}")
        except Exception as e:
            logger.error(f"❌ Failed to get subgroups for {parent_path}: {e}")
            return result
    else:
        try:
            # If no parent path specified, we're looking for oct2-pars directly
            groups = [gl.groups.get("oct2-pars")]
            logger.info("Found oct2-pars group")
        except Exception as e:
            logger.error(f"❌ Failed to get oct2-pars group: {e}")
            return result
    
    for group in groups:
        group_obj = gl.groups.get(group.id)
        group_path = group_obj.full_path
        
        logger.info(f"🔍 Scanning group: {group_path}")
        
        # Get projects in this group
        try:
            projects = group_obj.projects.list(all=True)
            logger.info(f"  Found {len(projects)} repositories in group {group_path}")
            result[group_path] = [project.path for project in projects]
            
            # Get subgroups recursively
            subgroups_result = get_groups_and_projects(gl, group_path)
            result.update(subgroups_result)
        except Exception as e:
            logger.error(f"❌ Error scanning group {group_path}: {e}")
    
    return result

def create_target_group(gl, group_path):
    """Create a group in the target GitLab if it doesn't exist."""
    # Map the source group path to target path if defined
    target_group_path = STRUCTURE_MAPPING.get(group_path, group_path)
    
    # Check if the group exists
    try:
        group = gl.groups.get(target_group_path)
        logger.info(f"✅ Group {target_group_path} already exists in target.")
        return group
    except gitlab.exceptions.GitlabGetError:
        # Group doesn't exist, need to create it
        if DRY_RUN:
            logger.info(f"[DRY RUN] Would create group: {target_group_path}")
            return None
        
        if not get_user_confirmation(f"Group {target_group_path} does not exist. Create it?"):
            logger.warning(f"⚠️ Skipped creating group {target_group_path} as per user request")
            return None
    
    # If this is a subgroup, create parent groups first
    parts = target_group_path.split('/')
    parent_id = None
    
    current_path = ""
    for i, part in enumerate(parts):
        current_path = f"{current_path}/{part}" if current_path else part
        try:
            group = gl.groups.get(current_path)
            parent_id = group.id
            logger.debug(f"Group {current_path} already exists with ID {parent_id}")
        except gitlab.exceptions.GitlabGetError:
            # Create this part of the path
            if DRY_RUN:
                logger.info(f"[DRY RUN] Would create group: {current_path}")
                continue
                
            logger.info(f"📁 Creating group: {current_path}")
            try:
                group = gl.groups.create({
                    'name': part,
                    'path': part,
                    'parent_id': parent_id,
                    'visibility': 'private'  # Adjust as needed
                })
                parent_id = group.id
                logger.info(f"✅ Created group {current_path} with ID {parent_id}")
            except Exception as e:
                logger.error(f"❌ Failed to create group {current_path}: {e}")
                raise
    
    return gl.groups.get(target_group_path) if not DRY_RUN else None

def migrate_repository(source_gl, target_gl, source_group, repo_name):
    """Migrate a single repository from source to target."""
    # Determine target group path and repo name
    target_group_path = STRUCTURE_MAPPING.get(source_group, source_group)
    target_repo_name = REPO_NAME_MAPPING.get(repo_name, repo_name)
    
    logger.info(f"\n{'='*80}")
    logger.info(f"📦 Processing repository: {source_group}/{repo_name}")
    logger.info(f"   Target location: {target_group_path}/{target_repo_name}")
    logger.info(f"{'='*80}")
    
    # Get source project
    try:
        source_project_path = f"{source_group}/{repo_name}"
        source_project = source_gl.projects.get(source_project_path)
        logger.info(f"ℹ️ Source project details:")
        logger.info(f"   - Name: {source_project.name}")
        logger.info(f"   - Description: {source_project.description or 'None'}")
        logger.info(f"   - Visibility: {source_project.visibility}")
        logger.info(f"   - Default branch: {source_project.default_branch}")
    except Exception as e:
        logger.error(f"❌ Failed to get source project {source_project_path}: {e}")
        return False
    
    if DRY_RUN:
        logger.info(f"[DRY RUN] Would migrate {source_project_path} to {target_group_path}/{target_repo_name}")
        return True
    
    # Confirm migration
    if not get_user_confirmation(f"Migrate {source_project_path} to {target_group_path}/{target_repo_name}?"):
        logger.warning(f"⚠️ Skipped migration of {source_project_path} as per user request")
        return False
    
    # Create target group if needed
    try:
        target_group = create_target_group(target_gl, source_group)
        if target_group is None and not DRY_RUN:
            logger.error(f"❌ Failed to create target group, skipping repository {repo_name}")
            return False
    except Exception as e:
        logger.error(f"❌ Failed to create target group: {e}")
        return False
    
    # Check if project already exists in target
    target_project_path = f"{target_group_path}/{target_repo_name}"
    target_project = None
    try:
        target_project = target_gl.projects.get(target_project_path)
        logger.info(f"ℹ️ Project {target_project_path} already exists in target.")
        if not get_user_confirmation(f"Project {target_project_path} already exists. Update it?"):
            logger.warning(f"⚠️ Skipped updating {target_project_path} as per user request")
            return False
    except gitlab.exceptions.GitlabGetError:
        # Project doesn't exist, will create it
        logger.info(f"ℹ️ Project {target_project_path} does not exist yet, will create it.")
    
    # Create temp directory for clone
    repo_temp_dir = f"{TEMP_DIR}/{repo_name}_{int(time.time())}"
    if os.path.exists(repo_temp_dir):
        shutil.rmtree(repo_temp_dir)
    os.makedirs(repo_temp_dir, exist_ok=True)
    
    try:
        # Clone source repository
        source_repo_url = source_project.http_url_to_repo.replace("https://", f"https://oauth2:{SOURCE_GITLAB_TOKEN}@")
        logger.info(f"⬇️ Cloning {source_project_path} to temporary directory...")
        
        repo = git.Repo.clone_from(source_repo_url, repo_temp_dir, mirror=True)
        
        # Check repository size
        size_mb = sum(os.path.getsize(os.path.join(dirpath, filename)) 
                      for dirpath, _, filenames in os.walk(repo_temp_dir) 
                      for filename in filenames) / (1024*1024)
        logger.info(f"ℹ️ Repository size: {size_mb:.2f} MB")
        
        # Get branch info
        branches = repo.git.branch('-a').split('\n')
        logger.info(f"ℹ️ Repository has {len(branches)} branches")
        
        # Create project in target if it doesn't exist
        if not target_project:
            logger.info(f"📝 Creating project {target_repo_name} in target group...")
            target_project = target_gl.projects.create({
                'name': target_repo_name,
                'path': target_repo_name,
                'namespace_id': target_group.id,
                'visibility': source_project.visibility,
                'description': source_project.description or ''
            })
            logger.info(f"✅ Created project {target_project_path} in target")
        
        # Set Git remote and push to target
        target_repo_url = target_project.http_url_to_repo.replace("https://", f"https://oauth2:{TARGET_GITLAB_TOKEN}@")
        logger.info(f"⬆️ Pushing to target {target_project_path}...")
        
        if get_user_confirmation(f"Ready to push changes to {target_project_path}. Proceed?"):
            # Add target as remote and push
            repo.create_remote("target", target_repo_url)
            push_result = repo.remote("target").push(mirror=True)
            
            if push_result:
                logger.info(f"✅ Successfully migrated {source_project_path} to {target_project_path}")
                return True
            else:
                logger.error(f"❌ Push operation completed but returned no result")
                return False
        else:
            logger.warning(f"⚠️ Skipped pushing to {target_project_path} as per user request")
            return False
        
    except Exception as e:
        logger.error(f"❌ Failed to migrate repository {repo_name}: {e}")
        return False
    finally:
        # Clean up
        logger.info(f"🧹 Cleaning up temporary files...")
        if os.path.exists(repo_temp_dir):
            shutil.rmtree(repo_temp_dir)

def print_summary(source_structure):
    """Print a summary of what will be migrated."""
    logger.info("\n\n" + "="*50)
    logger.info("MIGRATION SUMMARY")
    logger.info("="*50)
    
    total_groups = len(source_structure)
    
    # Count repos considering whitelist
    total_repos = 0
    repos_to_migrate = 0
    
    for group, repos in source_structure.items():
        for repo in repos:
            total_repos += 1
            if not WHITELIST_MODE or repo in WHITELIST_REPOS:
                repos_to_migrate += 1
    
    logger.info(f"Total groups to process: {total_groups}")
    logger.info(f"Total repositories found: {total_repos}")
    
    if WHITELIST_MODE:
        logger.info(f"Repositories to migrate (whitelist): {repos_to_migrate}")
        if repos_to_migrate > 0:
            logger.info(f"Whitelisted repos: {', '.join(WHITELIST_REPOS)}")
    else:
        logger.info(f"Repositories to migrate: {repos_to_migrate}")
    
    logger.info("\nGroup structure:")
    
    for group, repos in source_structure.items():
        target_group = STRUCTURE_MAPPING.get(group, group)
        logger.info(f"  {group} → {target_group} ({len(repos)} repositories)")
        for repo in repos:
            target_repo = REPO_NAME_MAPPING.get(repo, repo)
            
            if WHITELIST_MODE and repo not in WHITELIST_REPOS:
                logger.info(f"    - {repo} → {target_repo} (SKIPPED - not in whitelist)")
            else:
                logger.info(f"    - {repo} → {target_repo}")
    
    logger.info("="*50 + "\n")

def run_migration():
    """Main migration function."""
    global DRY_RUN, SKIP_CONFIRMATIONS, WHITELIST_MODE
    
    # Parse command-line arguments
    if "--dry-run" in sys.argv:
        DRY_RUN = True
        logger.info("🔍 DRY RUN MODE: No changes will be made")
    
    if "--no-confirm" in sys.argv:
        SKIP_CONFIRMATIONS = True
        logger.info("⚡ AUTOMATIC MODE: Skipping all confirmations")
        
    if "--whitelist" in sys.argv:
        WHITELIST_MODE = True
        logger.info("📋 WHITELIST MODE: Only specified repositories will be migrated")
    
    start_time = datetime.now()
    logger.info(f"🚀 Starting GitLab migration process at {start_time}")
    logger.info(f"   Source: {SOURCE_GITLAB_URL}")
    logger.info(f"   Target: {TARGET_GITLAB_URL}")
    
    # Connect to GitLab instances
    try:
        logger.info("\n--- PHASE 1: CONNECTING TO GITLAB INSTANCES ---")
        source_gl = connect_gitlab(SOURCE_GITLAB_URL, SOURCE_GITLAB_TOKEN)
        target_gl = connect_gitlab(TARGET_GITLAB_URL, TARGET_GITLAB_TOKEN)
    except Exception as e:
        logger.error(f"❌ Failed to connect to GitLab instances: {e}")
        return
    
    # Get structure from source
    logger.info("\n--- PHASE 2: ANALYZING SOURCE REPOSITORY STRUCTURE ---")
    logger.info("🔍 Retrieving source GitLab structure...")
    source_structure = get_groups_and_projects(source_gl)
    
    # Print summary and confirm
    print_summary(source_structure)
    
    if not DRY_RUN and not get_user_confirmation("Do you want to proceed with the migration?"):
        logger.info("⚠️ Migration aborted by user")
        return
    
    # Create temp directory if it doesn't exist
    if not os.path.exists(TEMP_DIR):
        os.makedirs(TEMP_DIR)
    
    logger.info("\n--- PHASE 3: PERFORMING REPOSITORY MIGRATION ---")
    
    total_repos = 0
    total_to_migrate = 0
    migrated_repos = 0
    failed_repos = 0
    skipped_repos = 0
    
    # Count total repos and repos to migrate
    for group, repos in source_structure.items():
        total_repos += len(repos)
        for repo in repos:
            if not WHITELIST_MODE or repo in WHITELIST_REPOS:
                total_to_migrate += 1
    
    # Migrate each repository
    for group, repos in source_structure.items():
        logger.info(f"\n📂 Processing group {group} with {len(repos)} repositories")
        for repo in repos:
            # Skip if not in whitelist when whitelist mode is enabled
            if WHITELIST_MODE and repo not in WHITELIST_REPOS:
                logger.info(f"⏭️ Skipping {repo} (not in whitelist)")
                skipped_repos += 1
                continue
                
            if migrate_repository(source_gl, target_gl, group, repo):
                migrated_repos += 1
            else:
                failed_repos += 1
            
            logger.info(f"Progress: {migrated_repos + failed_repos + skipped_repos}/{total_repos} repositories processed")
    
    # Clean up temp directory
    if os.path.exists(TEMP_DIR):
        shutil.rmtree(TEMP_DIR)
    
    end_time = datetime.now()
    duration = end_time - start_time
    
    logger.info("\n--- MIGRATION COMPLETED ---")
    logger.info(f"⏱️ Total time: {duration}")
    logger.info(f"📊 Total repositories: {total_repos}")
    if WHITELIST_MODE:
        logger.info(f"📋 Whitelisted repositories: {total_to_migrate}")
        logger.info(f"⏭️ Skipped (not in whitelist): {skipped_repos}")
    logger.info(f"✅ Successfully migrated: {migrated_repos}")
    logger.info(f"❌ Failed to migrate: {failed_repos}")
    logger.info(f"\nDetailed log file: {os.path.abspath(log_file)}")

if __name__ == "__main__":
    run_migration()