miscellaneous/nexus/nexus_import.py

import requests
import json

nexus_url_base = 'https://nexus-test.devops.lmru.tech/service/rest/v1/repositories'
nexus_username = 'admin'
nexus_password = 'nexus'

with open('artifactory_repositories.json', 'r') as file:
    artifactory_repositories = json.load(file)

headers = {
    'Content-Type': 'application/json'
}

repo_type_mapping = {
    'LOCAL': 'hosted',
    'REMOTE': 'proxy',
    'VIRTUAL': 'group'
}

format_url_mapping = {
    'maven': 'maven',
    'npm': 'npm',
    'nuget': 'nuget',
    'docker': 'docker',
    'pypi': 'pypi',
    'rubygems': 'rubygems',
    'yum': 'yum',
    'debian': 'apt',
    'generic': 'raw',
    'go': 'go',
    'gradle': 'gradle',
    'helm': 'helm',
    'cocoapods': 'cocoapods'
}

default_attributes = {
    'maven': {
        'versionPolicy': 'RELEASE',
        'layoutPolicy': 'STRICT'
    },
    'nugetProxy': {
        'queryCacheItemMaxAge': 1440,
        'nugetVersion': 'V3'
    },
    'dockerProxy': {
        'indexType': 'REGISTRY'
    },
    'apt': {
        'flat': False,
        'distribution': 'default'
    },
    'proxyDefaults': {
        'contentMaxAge': 1440,
        'metadataMaxAge': 1440
    }
}

for repo in artifactory_repositories:
    repo_name = repo['key']
    repo_type = repo_type_mapping.get(repo['type'])
    package_type = repo.get('packageType').lower()
    
    format_url = format_url_mapping.get(package_type)

    if not format_url:
        print(f"Unknown package type {package_type} for {repo_name}")
        continue

    nexus_url = f"{nexus_url_base}/{format_url}/{repo_type}"

    if repo_type == 'proxy':
        config = {
            "name": repo_name,
            "online": True,
            "storage": {
                "blobStoreName": "nexus-test",
                "strictContentTypeValidation": True
            },
            "proxy": {
                "remoteUrl": repo['url'],
                "contentMaxAge": default_attributes['proxyDefaults']['contentMaxAge'],
                "metadataMaxAge": default_attributes['proxyDefaults']['metadataMaxAge']
            },
            "negativeCache": {
                "enabled": True,
                "timeToLive": 1440
            },
            "httpClient": {
                "blocked": False,
                "autoBlock": True
            },
            "routingRuleName": None,
            "format": format_url
        }

        if format_url == 'docker':
            config['dockerProxy'] = default_attributes['dockerProxy']
            config['docker'] = {
                "v1Enabled": False,
                "forceBasicAuth": True
            }

        if format_url == 'apt':
            config['apt'] = default_attributes['apt']
            config['apt']['distribution'] = repo.get('distribution', 'default')

        if format_url == 'yum':
            config['yum'] = {}

        if format_url == 'maven':
            config["maven"] = default_attributes['maven']

        if format_url == 'nuget':
            config["nugetProxy"] = default_attributes['nugetProxy']

    elif repo_type == 'hosted':
        config = {
            "name": repo_name,
            "online": True,
            "storage": {
                "blobStoreName": "nexus-test",
                "strictContentTypeValidation": True,
                "writePolicy": 'ALLOW'
            },
            "cleanup": None,
            "component": {
                "proprietaryComponents": False
            },
            "format": format_url
        }

        if format_url == 'docker':
            config['docker'] = {
                "v1Enabled": False,
                "forceBasicAuth": True
            }
        if format_url == 'maven':
            config["maven"] = default_attributes['maven']
        if format_url == 'yum':
            config['yum'] = {}

    elif repo_type == 'group':
        config = {
            "name": repo_name,
            "online": True,
            "storage": {
                "blobStoreName": "nexus-test",
                "strictContentTypeValidation": True
            },
            "group": {
                "memberNames": repo['repositories']
            },
            "format": format_url
        }

        if format_url == 'docker':
            config['docker'] = {
                "v1Enabled": False,
                "forceBasicAuth": True
            }
        if format_url == 'maven':
            config["maven"] = default_attributes['maven']

    else:
        print(f"Unknown repository type {repo['type']} for {repo_name}")
        continue
    

    try:
        response = requests.post(nexus_url, headers=headers, auth=(nexus_username, nexus_password), data=json.dumps(config))

        if response.status_code == 201:
            print(f"Repository {repo_name} successfully created")
        else:
            try:
                response_json = response.json()
                if isinstance(response_json, dict):
                    if "duplicated key" in response_json.get("message", ""):
                        print(f"!!! Repository {repo_name} already exists")
                    else:
                        print(f"Error creating repository {repo_name}: {response.status_code} {response.text}")
                else:
                    print(f"Error creating repository {repo_name}: {response.status_code} {response.text}")
            except json.JSONDecodeError:
                print(f"Error creating repository {repo_name}: {response.status_code} {response.text}")

    except requests.exceptions.RequestException as e:
        print(f"Error during request execution: {e}")