-
Notifications
You must be signed in to change notification settings - Fork 0
/
nexus-export.py
84 lines (73 loc) · 3.04 KB
/
nexus-export.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
import requests
import os
import json
import urllib.parse
SOURCE_SERVER = "..."
SOURCE_USER = "..."
SOURCE_PASS = "..."
if SOURCE_SERVER == "..." or SOURCE_USER == "..." or SOURCE_PASS == "...":
print("ERROR: Please provide Nexus credentials")
exit(1)
SOURCE_REPOS = {
"maven-public": "",
"npm-hosted": "",
}
EXTS = ["jar", "pom", "tgz"]
def download_repo(sourceRepo, sourceFolder):
outputFile = f"{sourceRepo}-artifacts.txt"
if os.path.exists(outputFile):
os.remove(outputFile)
url = f"{SOURCE_SERVER}/service/rest/v1/assets?repository={sourceRepo}"
if sourceRepo == "maven-public":
url = f"{SOURCE_SERVER}/service/rest/v1/search/assets?repository={sourceRepo}&maven.extension=jar"
contToken = "initial"
while contToken:
if contToken != "initial":
url = f"{SOURCE_SERVER}/service/rest/v1/assets?continuationToken={contToken}&repository={sourceRepo}"
if sourceRepo == "maven-public":
url = f"{SOURCE_SERVER}/service/rest/v1/search/assets?continuationToken={contToken}&repository={sourceRepo}&maven.extension=jar"
response = requests.get(
url, auth=(SOURCE_USER, SOURCE_PASS), headers={"Accept": "application/json"}
)
if response.status_code != 200:
print(
f"ERROR: Failed to get assets from repository: {sourceRepo} with error code: {response.status_code}"
)
break
response = response.json()
artifacts = [item["downloadUrl"] for item in response["items"]]
with open(outputFile, "a") as out:
for line in artifacts:
if line.endswith(tuple(EXTS)) and (
sourceFolder in line or sourceFolder == ""
):
out.write(f"{line}\n")
if line.endswith(".jar"):
out.write(f'{line.replace(".jar", ".pom")}\n')
contToken = response.get("continuationToken")
print(f"Collected {len(artifacts)} artifacts")
print("Downloading artifacts...")
with open(outputFile, "r") as f:
urls = [line.strip() for line in f]
for i, url in enumerate(urls):
path = url.split("/", 3)[-1]
dir = f"{sourceRepo}/{os.path.dirname(path)}"
os.makedirs(dir, exist_ok=True)
url = urllib.parse.quote(url, safe=":/")
response = requests.get(url, auth=(SOURCE_USER, SOURCE_PASS))
if response.status_code == 200:
with open(os.path.join(dir, os.path.basename(url)), "wb") as f:
f.write(response.content)
print(f"Downloaded artifact: {url}")
else:
print(
f"ERROR: Failed to download artifact: {url} with error code: {response.status_code}"
)
print(response.text)
progress = (i + 1) / len(urls) * 100
print(
f'\rProgress: [{"#" * int(progress // 4)}{" " * (25 - int(progress // 4))}] {int(progress)}%',
end="\r",
)
for repo in SOURCE_REPOS:
download_repo(repo, SOURCE_REPOS[repo])