基于 python 的 nexus 私服 jar 包匹配查找程序

发布时间 2023-05-26 11:22:05作者: Nihaorz

程序通过计算 jar 包的 sha1 摘要,到 maven 仓库查询对应路径,适用于 Nexus Repository Manager OSS 2.14.11-01,其他版本未做测试

注意:jar 包路径中不要出现空格
#!/usr/bin/python3

import hashlib
import os
import requests
import sys
import tempfile
from xml.etree import ElementTree
import zipfile

nexus_repo_url = ["http://172.17.0.71:8081/nexus/service/local/repositories/snapshots", "http://172.17.0.71:8081/nexus/service/local/repositories/releases"]

## 欢迎信息
def welcome():
    print("\n程序通过计算jar包的sha1摘要,到maven仓库查询对应路径")
    print("\n程序适用于 Nexus Repository Manager OSS 2.14.11-01,其他版本未做测试")
    print("\n注意:jar包路径中不要出现空格")

## 获取jar包路径
def get_jar_path():
    print("\n拖拽jar包到窗并口回车提交:", end = "")
    try:
        jar_path = input()
        if os.path.exists(jar_path):
            if jar_path.endswith(".jar"):
                return jar_path
            else:
                print("\n请选择以【.jar】结尾的jar包文件")
        else:
            print("\n文件不存在")
        sys.exit()
    except KeyboardInterrupt:
        print("\n\n操作取消,程序退出")
        sys.exit()

## 计算jar包文件的sha1摘要
def get_jar_sha1sum(jar_path):
    with open(jar_path, 'rb') as f:
        return hashlib.new("sha1", f.read()).hexdigest()

## 获取jar包的坐标信息
def get_maven_gav(jar_path):
    temp_dir = tempfile.TemporaryDirectory()
    f = zipfile.ZipFile(jar_path, 'r')
    for file in f.namelist():
        f.extract(file, temp_dir.name)
    maven_dir = temp_dir.name + "/META-INF/maven"
    pom_path = maven_dir
    pom_path = pom_path + "/" + os.listdir(maven_dir)[0]
    pom_path = pom_path + "/" + os.listdir(pom_path)[0] + "/pom.properties"
    f.close()
    groupId = ""
    artifactId = ""
    version = ""
    with open(pom_path, 'r') as f:
        for line in f.readlines():
            if line.startswith("groupId="):
                groupId = line[len("groupId="):].strip()
            elif line.startswith("artifactId="):
                artifactId = line[len("artifactId="):].strip()
            elif line.startswith("version="):
                version = line[len("version="):].strip()
    return groupId, artifactId, version

## 获取jar包及pom文件下载地址
def get_item_url(sha1sum, groupId, artifactId, version):
    for repo_url in nexus_repo_url:
        print(f"\n尝试从【{repo_url}】中检索jar包")
        groupId_path_str = groupId.replace(".", "/")
        url = f"{repo_url}/content/{groupId_path_str}/{artifactId}/{version}"
        response = requests.get(url)
        if response.status_code == 200:
            items = ElementTree.fromstring(response.text).find("data").findall("content-item")
            for item in items:
                item_url = item.find("resourceURI").text
                if item_url.endswith(".jar.sha1"):
                    print("requests.get: " + item_url)
                    url_sha1sum = requests.get(item_url).text
                    if sha1sum == url_sha1sum:
                        # return item_url
                        url_prefix = item_url[:-(len(".jar.sha1"))]
                        return url_prefix + ".jar", url_prefix + ".pom"
        else:
            print(f"\n请求异常,请求地址:{url}")
    print("\n未匹配到正确的jar包")
    return None, None

## 开始处理
def process():
    jar_path = get_jar_path()
    sha1sum = get_jar_sha1sum(jar_path)
    print(f"\nsha1摘要   : {sha1sum}")
    groupId, artifactId, version = get_maven_gav(jar_path)
    print(f"groupId    : {groupId}")
    print(f"artifactId : {artifactId}")
    print(f"version    : {version}")
    jar_url, pom_url = get_item_url(sha1sum, groupId, artifactId, version)
    if jar_url and pom_url:
        print(f"\n成功匹配到jar包")
        print(f"\njar包下载地址: {jar_url}")
        print(f"\npom下载地址: {pom_url}")
    process()

if __name__ == '__main__':
    welcome()
    process()